Coverage for mddb_workflow / utils / register.py: 67%

1from sys import argv 

2from os.path import exists 

3from datetime import datetime 

4 

5from mddb_workflow.utils.auxiliar import load_json, save_json, warn, get_git_version 

6from mddb_workflow.utils.constants import DATE_STYLE 

7from mddb_workflow.utils.type_hints import * 

8 

9 

10class Register: 

11 """The register tracks activity along multiple runs and thus avoids repeating some already succeeded tests 

12 It is also responsible for storing test failure warnings to be written in metadata. 

13 """ 

14 def __init__(self, register_file: 'File'): 

15 # Save the previous register 

16 self.file = register_file 

17 # Save the current workflow call 

18 # Quote those arguments including space, since it means they were quoted when inputed 

19 quoted_argv = [f"'{arg}'" if ' ' in arg else arg for arg in argv] 

20 self.call = ' '.join(quoted_argv) 

21 # Save the current date 

22 self.date = datetime.today().strftime(DATE_STYLE) 

23 # Save the current version 

24 self.version = get_git_version() 

25 # Set the tests tracker 

26 self.tests = {} 

27 # Set the warnings list, which will be filled by failing tests 

28 self.warnings = [] 

29 # Inherit test results and warning from the register last entry 

30 self.entries = [] 

31 if self.file.exists: 

32 # Read the register in disk 

33 self.entries = load_json(self.file.path) 

34 last_entry = self.entries[-1] 

35 # Inherit test results 

36 for test_name, test_result in last_entry['tests'].items(): 

37 self.tests[test_name] = test_result 

38 # Inherit warnings 

39 for warning in last_entry['warnings']: 

40 # DANI: Para quitarnos de encima warnings con el formato antiguo 

41 if not warning.get('tag', None): 

42 continue 

43 self.warnings.append(warning) 

44 # Save the entry for the first time 

45 self.save() 

46 

47 def __repr__(self): 

48 return str(self.to_dict()) 

49 

50 def to_dict(self) -> dict: 

51 # Set a dictionary with the current values 

52 dictionary = { 

53 'call': self.call, 

54 'date': self.date, 

55 'version': self.version, 

56 'tests': self.tests, 

57 'warnings': self.warnings, 

58 } 

59 return dictionary 

60 

61 def update_test(self, key: str, value: Optional[bool]): 

62 """Update a test result and save the register.""" 

63 self.tests[key] = value 

64 self.save() 

65 

66 def get_warnings(self, tag: str) -> list[dict]: 

67 """Get current warnings filtered by tag.""" 

68 return [warning for warning in self.warnings if warning['tag'] == tag] 

69 

70 def add_warning(self, tag: str, message: str): 

71 """Add warnings with the right format and save the register. 

72 A flag is to be passed to handle further removal of warnings. 

73 """ 

74 warn(message) 

75 # If we already had this exact warning then do not repeat it 

76 for warning in self.warnings: 

77 if warning['tag'] == tag and warning['message'] == message: return 

78 # Add a new warning 

79 warning = {'tag': tag, 'message': message} 

80 self.warnings.append(warning) 

81 self.save() 

82 

83 def remove_warnings(self, tag: str): 

84 """Remove warnings filtered by tag and save the register.""" 

85 # WARNING: Do not declare again the warnings list to remove values 

86 # WARNING: i.e. don't do a comprehension list 

87 # WARNING: Otherwise if the warnings list is saved somewhere else they will be disconnected 

88 while True: 

89 target_warning = next((warning for warning in self.warnings if warning['tag'] == tag), None) 

90 if not target_warning: break 

91 self.warnings.remove(target_warning) 

92 self.save() 

93 

94 def save(self): 

95 """Save the register to a json file.""" 

96 # If path does not exist then do nothing 

97 # WARNING: I know this looks a bit silent 

98 # WARNING: Otherwise it is a constant spam when something goes wrong close to beginning 

99 if not exists(self.file.basepath): 

100 return 

101 # Set a new entry for the current run 

102 current_entry = self.to_dict() 

103 # Write entries to disk 

104 save_json(self.entries + [current_entry], self.file.path, indent=4)