Coverage for suhteita/robot/SourceServerLibrary/source_server_bridge.py: 0.00%
59 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-04 22:42:46 +00:00
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-04 22:42:46 +00:00
1"""Source server abstraction relaying keywords to API methods of the underlying source server system (Bitbucket)."""
3import ast
4from typing import List, no_type_check
6import jmespath
7import wrapt # type: ignore
8from robot.api import ContinuableFailure, logger
10from suhteita.source_server_actions import Bitbucket as Source
13@no_type_check
14def _string_to_data(string):
15 """Parse the string into the underlying data type if successful else return the string."""
16 try:
17 return ast.literal_eval(str(string).strip())
18 except Exception:
19 return string
22@no_type_check
23@wrapt.decorator
24def _string_variables_to_data(function, instance, args, kwargs):
25 """Transform the string variables to data, relay to function, and return the call result."""
26 args = [_string_to_data(arg) for arg in args]
27 kwargs = dict((arg_name, _string_to_data(arg)) for arg_name, arg in kwargs.items())
28 return function(*args, **kwargs)
31@no_type_check
32class SourceServerBridge(object):
33 """Use the robot framework hybrid API to proxy the calls and support discovery of keywords."""
35 ROBOT_LIBRARY_SCOPE = 'Global'
36 _source_server = Source
37 _session = None
39 def get_keyword_names(self) -> List[str]:
40 """Generate the list of keywords from the underlying provider - required hybrid API method."""
41 get_members = self._source_server.__dict__.items
42 kws = [name for name, function in get_members() if hasattr(function, '__call__')]
43 kws += ['extract_fields', 'extract_paths', 'extract_project_keys', 'source_session']
45 return list(
46 set([kw for kw in kws if not kw.startswith('delete_') and kw not in ('__init__', 'get_pullrequest')])
47 )
49 @no_type_check
50 def source_session(self, url=None, username=None, password=None, **kwargs):
51 """Login and fetch the session object."""
52 self._session = self._source_server(url=url, username=username, password=password, **kwargs)
53 logger.debug('Connected to source server')
54 return self._session
56 @no_type_check
57 @staticmethod
58 def extract_fields(data, fields):
59 """Extract dictionary fields from data per key value (field name) to reduce the clutter in logs."""
60 try:
61 return {field: data[field] for field in fields}
62 except KeyError as err:
63 raise ContinuableFailure(f'Extraction of fields failed for (field=={err})')
65 @no_type_check
66 @staticmethod
67 def extract_paths(data, paths):
68 """Extract dictionary fields from data per paths to values to reduce the clutter in logs."""
69 return {path: jmespath.search(path.lstrip('/').replace('/', '.'), data) for path in paths}
71 @no_type_check
72 @staticmethod
73 def extract_project_keys(projects):
74 """Extract dictionary key field values from list of project dicts received per API."""
75 try:
76 return [project['key'] for project in projects]
77 except KeyError as err:
78 raise ContinuableFailure(f'Extraction of key field failed for projects (field=={err})')
80 @no_type_check
81 def __getattr__(self, name):
82 """Relay the function matching the keyword or the lookup error."""
83 func = None
84 if name in self._source_server.__dict__.keys():
85 func = getattr(self._source_server, name)
87 if func:
88 return _string_variables_to_data(func)
90 raise AttributeError(f'Keyword {name} does not exist or has been overridden by this library.')