Coverage for src/objectstate/snapshot_model.py: 98%

1""" 

2Snapshot and Timeline dataclasses for git-like time-travel history. 

3 

4This module provides typed data structures for the ObjectStateRegistry's 

5time-travel system, replacing the tuple-based implementation. 

6 

7Design Philosophy: Correct by Construction 

8- No Optional fields for required data 

9- Immutable snapshots (frozen dataclass) 

10- UUID-based identity for snapshots 

11- Direct attribute access (no getattr fallbacks) 

12""" 

13 

14from dataclasses import asdict, dataclass, field, fields 

15from typing import Dict, Optional, List 

16from objectstate.construction_binding import StateConstructionBinding 

17import uuid 

18import time 

19 

20 

21@dataclass(frozen=True) 

22class StateSnapshot: 

23 """Immutable snapshot of a single ObjectState's data. 

24 

25 Captures the resolved values, parameters, and provenance at a point in time. 

26 Construction declarations are typed values, never live lifecycle resources. 

27 """ 

28 

29 saved_resolved: Dict 

30 live_resolved: Dict 

31 parameters: Dict # Current concrete values 

32 saved_parameters: Dict # Concrete values at last save (for concrete dirty detection) 

33 provenance: Dict 

34 construction: StateConstructionBinding 

35 meta: Dict = field(default_factory=dict) 

36 

37 

38@dataclass(frozen=True) 

39class Snapshot: 

40 """Immutable snapshot of ALL ObjectStates at a point in time. 

41 

42 Analogous to a git commit - captures the entire system state. 

43 """ 

44 

45 id: str # UUID string 

46 timestamp: float 

47 label: str 

48 triggering_scope: Optional[str] # scope_id that triggered the snapshot 

49 parent_id: Optional[str] # UUID of parent snapshot (None for root) 

50 all_states: Dict[str, StateSnapshot] # scope_id → StateSnapshot 

51 

52 @classmethod 

53 def create( 

54 cls, 

55 label: str, 

56 all_states: Dict[str, StateSnapshot], 

57 triggering_scope: Optional[str] = None, 

58 parent_id: Optional[str] = None, 

59 ) -> "Snapshot": 

60 """Create a new snapshot with auto-generated ID and timestamp.""" 

61 return cls( 

62 id=str(uuid.uuid4()), 

63 timestamp=time.time(), 

64 label=label, 

65 triggering_scope=triggering_scope, 

66 parent_id=parent_id, 

67 all_states=all_states, 

68 ) 

69 

70 def to_dict(self) -> Dict: 

71 """Export the typed snapshot document payload.""" 

72 return { 

73 "id": self.id, 

74 "timestamp": self.timestamp, 

75 "label": self.label, 

76 "triggering_scope": self.triggering_scope, 

77 "parent_id": self.parent_id, 

78 "states": { 

79 scope_id: { 

80 "saved_resolved": ss.saved_resolved, 

81 "live_resolved": ss.live_resolved, 

82 "parameters": ss.parameters, 

83 "saved_parameters": ss.saved_parameters, 

84 "provenance": ss.provenance, 

85 "construction": ss.construction, 

86 "meta": ss.meta, 

87 } 

88 for scope_id, ss in self.all_states.items() 

89 }, 

90 } 

91 

92 @classmethod 

93 def from_dict(cls, data: Dict) -> "Snapshot": 

94 """Import a typed snapshot document payload.""" 

95 all_states = { 

96 scope_id: StateSnapshot( 

97 saved_resolved=state_data["saved_resolved"], 

98 live_resolved=state_data["live_resolved"], 

99 parameters=state_data["parameters"], 

100 saved_parameters=state_data["saved_parameters"], 

101 provenance=state_data["provenance"], 

102 construction=state_data["construction"], 

103 meta=state_data["meta"], 

104 ) 

105 for scope_id, state_data in data["states"].items() 

106 } 

107 return cls( 

108 id=data["id"], 

109 timestamp=data["timestamp"], 

110 label=data["label"], 

111 triggering_scope=data["triggering_scope"], 

112 parent_id=data["parent_id"], 

113 all_states=all_states, 

114 ) 

115 

116 

117@dataclass 

118class Timeline: 

119 """Named branch of history - analogous to a git branch. 

120 

121 Points to a head snapshot and tracks its base (branch point). 

122 """ 

123 

124 name: str 

125 head_id: str # UUID of current head snapshot 

126 base_id: str # UUID of snapshot this timeline branched from 

127 created_at: float = field(default_factory=time.time) 

128 description: str = "" 

129 

130 def to_dict(self) -> Dict: 

131 """Export to JSON-serializable dict.""" 

132 return asdict(self) 

133 

134 @classmethod 

135 def from_dict(cls, data: Dict) -> "Timeline": 

136 """Require all declared fields and ignore unknown document keys.""" 

137 return cls(**{item.name: data[item.name] for item in fields(cls)})