Coverage for suhteita/robot/TicketSystemLibrary/ticket_system_bridge.py: 100.00%

1"""Ticket system abstraction relaying keywords to API methods of the underlying ticket system (JIRA).""" 

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.ticket_system_actions import Jira as Ticket 

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 TicketSystemBridge(object): 

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

34 

35 ROBOT_LIBRARY_SCOPE = 'Global' 

36 _ticket_system = Ticket 

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._ticket_system.__dict__.items 

42 kws = [name for name, function in get_members() if hasattr(function, '__call__')] 

43 kws += ['extract_fields', 'extract_paths', 'extract_project_keys', 'ticket_session'] 

44 

45 return [kw for kw in kws if not kw.startswith('delete_') and kw not in ('__init__', 'get_issue_remotelinks')] 

46 

47 @no_type_check 

48 def ticket_session(self, url=None, username=None, password=None, **kwargs): 

49 """Login and fetch the session object.""" 

50 self._session = self._ticket_system(url=url, username=username, password=password, **kwargs) 

51 logger.debug('Connected to ticket system') 

52 return self._session 

53 

54 @no_type_check 

55 @staticmethod 

56 def extract_fields(data, fields): 

57 """Extract dictionary fields from data per key value (field name) to reduce the clutter in logs.""" 

58 try: 

59 return {field: data[field] for field in fields} 

60 except KeyError as err: 

61 raise ContinuableFailure(f'Extraction of fields failed for (field=={err})') 

62 

63 @no_type_check 

64 @staticmethod 

65 def extract_paths(data, paths): 

66 """Extract dictionary fields from data per paths to values to reduce the clutter in logs.""" 

67 return {path: jmespath.search(path.lstrip('/').replace('/', '.'), data) for path in paths} 

68 

69 @no_type_check 

70 @staticmethod 

71 def extract_project_keys(projects): 

72 """Extract dictionary key field values from list of project dicts received per API.""" 

73 try: 

74 return [project['key'] for project in projects] 

75 except KeyError as err: 

76 raise ContinuableFailure(f'Extraction of key field failed for projects (field=={err})') 

77 

78 @no_type_check 

79 def __getattr__(self, name): 

80 """Relay the function matching the keyword or the lookup error.""" 

81 func = None 

82 if name in self._ticket_system.__dict__.keys(): 

83 func = getattr(self._ticket_system, name) 

84 

85 if func: 

86 return _string_variables_to_data(func) 

87 

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