Coverage for suhteita/store.py: 95.35%

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' 

12Context = Dict[str, Union[str, dti.datetime]] 

13 

14 

15@no_type_check 

16class Store: 

17 @no_type_check 

18 def __init__(self, context: Context, setup: object, folder_path: Union[pathlib.Path, str] = STORE): 

19 self.store = pathlib.Path(folder_path) 

20 self.identity = context['identity'] 

21 self.start_time = context['start_time'] 

22 self.end_ts = None 

23 self.total_secs = 0.0 

24 self.node_indicator = NODE_INDICATOR 

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

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

27 self.rank = 0 

28 self.db = { 

29 '_meta': { 

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

31 'identity': self.identity, 

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

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

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

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

36 'db_name': self.db_name, 

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

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

39 'total_secs': self.total_secs, 

40 'end_ts': self.end_ts, 

41 'has_failures_declared': None, 

42 'has_failures_detected': None, 

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

44 }, 

45 'events': [], 

46 } 

47 

48 @no_type_check 

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

50 self.rank += 1 

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

52 { 

53 'rank': self.rank, 

54 'label': label, 

55 'ok': ok, 

56 'start_ts': clk[0], 

57 'duration_usecs': clk[1], 

58 'end_ts': clk[2], 

59 'comment': comment, 

60 } 

61 ) 

62 

63 @no_type_check 

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

65 self.end_time = end_time 

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

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

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

69 detect_failures = False 

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

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

72 detect_failures = True 

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

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

75 json.dump(self.db, handle)