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 def __init__ (self, register_file : 'File'): 

14 # Save the previous register 

15 self.file = register_file 

16 # Save the current workflow call 

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

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

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

20 # Save the current date 

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

22 # Save the current version 

23 self.version = get_git_version() 

24 # Set the tests tracker 

25 self.tests = {} 

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

27 self.warnings = [] 

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

29 self.entries = [] 

30 if self.file.exists: 

31 # Read the register in disk 

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

33 last_entry = self.entries[-1] 

34 # Inherit test results 

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

36 self.tests[test_name] = test_result 

37 # Inherit warnings 

38 for warning in last_entry['warnings']: 

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

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

41 continue 

42 self.warnings.append(warning) 

43 # Save the entry for the first time 

44 self.save() 

45 

46 def __repr__ (self): 

47 return str(self.to_dict()) 

48 

49 def to_dict (self) -> dict: 

50 # Set a dictionary with the current values 

51 dictionary = { 

52 'call': self.call, 

53 'date': self.date, 

54 'version': self.version, 

55 'tests': self.tests, 

56 'warnings': self.warnings, 

57 } 

58 return dictionary 

59 

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

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

62 self.tests[key] = value 

63 self.save() 

64 

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

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

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

68 

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

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

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

72 warn(message) 

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

74 for warning in self.warnings: 

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

76 # Add a new warning 

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

78 self.warnings.append(warning) 

79 self.save() 

80 

81 def remove_warnings (self, tag : str): 

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

83 self.warnings = [ warning for warning in self.warnings if warning['tag'] != tag ] 

84 self.save() 

85 

86 def save (self): 

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

88 # If path does not exist then do nothing 

89 # WARNING: I know this looks a bit silent  

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

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

92 return 

93 # Set a new entry for the current run 

94 current_entry = self.to_dict() 

95 # Write entries to disk 

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