Coverage for suhteita/robot/SourceServerLibrary/source_server_bridge.py: 0.00%

1"""Source server abstraction relaying keywords to API methods of the underlying source server system (Bitbucket).""" 

2 

3import ast 

4from typing import List, no_type_check 

5 

6import jmespath 

7import wrapt # type: ignore 

8from robot.api import ContinuableFailure, logger 

9 

10from suhteita.source_server_actions import Bitbucket as Source 

11 

12 

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 

20 

21 

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) 

29 

30 

31@no_type_check 

32class SourceServerBridge(object): 

33 """Use the robot framework hybrid API to proxy the calls and support discovery of keywords.""" 

34 

35 ROBOT_LIBRARY_SCOPE = 'Global' 

36 _source_server = Source 

37 _session = None 

38 

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'] 

44 

45 return list( 

46 set([kw for kw in kws if not kw.startswith('delete_') and kw not in ('__init__', 'get_pullrequest')]) 

47 ) 

48 

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 

55 

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})') 

64 

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} 

70 

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})') 

79 

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) 

86 

87 if func: 

88 return _string_variables_to_data(func) 

89 

90 raise AttributeError(f'Keyword {name} does not exist or has been overridden by this library.')