Coverage for model_workflow/utils/register.py: 76%

1from sys import argv 

2from os.path import exists 

3from datetime import datetime 

4 

5from model_workflow.utils.auxiliar import load_json, save_json, warn 

6from model_workflow.utils.constants import DATE_STYLE 

7from model_workflow.utils.type_hints import * 

8 

9 

10# The register tracks activity along multiple runs and thus avoids repeating some already succeeded tests 

11# It is also responsible for storing test failure warnings to be written in metadata 

12class Register: 

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 # Set the tests tracker 

23 self.tests = {} 

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

25 self.warnings = [] 

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

27 self.entries = [] 

28 if self.file.exists: 

29 # Read the register in disk 

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

31 last_entry = self.entries[-1] 

32 # Inherit test results 

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

34 self.tests[test_name] = test_result 

35 # Inherit warnings 

36 for warning in last_entry['warnings']: 

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

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

39 continue 

40 self.warnings.append(warning) 

41 # Save the entry for the first time 

42 self.save() 

43 

44 def __repr__ (self): 

45 return str(self.to_dict()) 

46 

47 def to_dict (self) -> dict: 

48 # Set a dictionary with the current values 

49 dictionary = { 

50 'call': self.call, 

51 'date': self.date, 

52 'tests': self.tests, 

53 'warnings': self.warnings, 

54 } 

55 return dictionary 

56 

57 # Update a test result and save the register 

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

59 self.tests[key] = value 

60 self.save() 

61 

62 # Get current warnings filtered by tag 

63 def get_warnings (self, tag : str) -> List[dict]: 

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

65 

66 # Add warnings with the right format and save the register 

67 # A flag is to be passed to handle further removal of warnings 

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

69 warn(message) 

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

71 for warning in self.warnings: 

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

73 # Add a new warning 

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

75 self.warnings.append(warning) 

76 self.save() 

77 

78 # Remove warnings filtered by tag and save the register 

79 def remove_warnings (self, tag : str): 

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

81 self.save() 

82 

83 # Save the register to a json file 

84 def save (self): 

85 # If path does not exist then do nothing 

86 # WARNING: I know this looks a bit silent  

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

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

89 return 

90 # Set a new entry for the current run 

91 current_entry = self.to_dict() 

92 # Write entries to disk 

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