Coverage for suhteita/store.py: 96.15%

1"""Provide a simple JSON store for event records.""" 

2 

3import copy 

4import datetime as dti 

5import json 

6import pathlib 

7from typing import Dict, Union, no_type_check 

8 

9from suhteita import ENCODING, NODE_INDICATOR, STORE, TS_FORMAT_PAYLOADS, Clocking 

10 

11TS_FORMAT_STORE = '%Y%m%dT%H%M%S.%fZ' 

12 

13 

14@no_type_check 

15class Store: 

16 @no_type_check 

17 def __init__( 

18 self, context: Dict[str, Union[str, dti.datetime]], setup: object, folder_path: Union[pathlib.Path, str] = STORE 

19 ): 

20 self.store = pathlib.Path(folder_path) 

21 self.identity = context['identity'] 

22 self.start_time = context['start_time'] 

23 self.end_ts = None 

24 self.total_secs = 0.0 

25 self.node_indicator = NODE_INDICATOR 

26 self.store.mkdir(parents=True, exist_ok=True) 

27 self.db_name = f'{self.identity}-{self.start_time.strftime(TS_FORMAT_STORE)}-{self.node_indicator}.json' 

28 self.rank = 0 

29 self.db = { 

30 '_meta': { 

31 'scenario': context.get('scenario', 'unknown'), 

32 'identity': self.identity, 

33 'node_indicator': str(self.node_indicator), 

34 'target': context.get('target', 'unknown'), 

35 'mode': context.get('mode', 'unknown'), 

36 'project': context.get('project', 'unknown'), 

37 'db_name': self.db_name, 

38 'db_path': str(self.store / self.db_name), 

39 'start_ts': self.start_time.strftime(TS_FORMAT_PAYLOADS), 

40 'total_secs': self.total_secs, 

41 'end_ts': self.end_ts, 

42 'has_failures_declared': None, 

43 'has_failures_detected': None, 

44 'setup': copy.deepcopy(setup.__dict__), 

45 }, 

46 'events': [], 

47 } 

48 

49 @no_type_check 

50 def add(self, label: str, ok: bool, clk: Clocking, comment: str = ''): 

51 self.rank += 1 

52 self.db['events'].append( 

53 { 

54 'rank': self.rank, 

55 'label': label, 

56 'ok': ok, 

57 'start_ts': clk[0], 

58 'duration_usecs': clk[1], 

59 'end_ts': clk[2], 

60 'comment': comment, 

61 } 

62 ) 

63 

64 @no_type_check 

65 def dump(self, end_time: dti.datetime, has_failures: bool = False): 

66 self.end_time = end_time 

67 self.db['_meta']['end_ts'] = self.end_time.strftime(TS_FORMAT_PAYLOADS) 

68 self.db['_meta']['total_secs'] = (self.end_time - self.start_time).total_seconds() 

69 self.db['_meta']['has_failures_declared'] = has_failures 

70 detect_failures = False 

71 for event in self.db['events']: 

72 if not event['ok']: 72 ↛ 73line 72 didn't jump to line 73, because the condition on line 72 was never true

73 detect_failures = True 

74 self.db['_meta']['has_failures_detected'] = detect_failures 

75 with open(self.store / self.db_name, 'wt', encoding=ENCODING) as handle: 

76 json.dump(self.db, handle)