tests/test_appender.py
4.0 KB · 136 lines · python Raw
1 """Tests for LogAppender."""
2
3 from __future__ import annotations
4
5 import os
6 from collections.abc import Callable
7
8 import pytest
9 from quantumshield.identity.agent import AgentIdentity
10
11 from pqc_audit_log_fs.appender import LogAppender, RotationPolicy
12 from pqc_audit_log_fs.errors import AppendToSealedSegmentError
13 from pqc_audit_log_fs.event import InferenceEvent
14
15
16 def test_append_writes_jsonl_no_seal_yet(
17 signer_identity: AgentIdentity,
18 tmp_log_dir: str,
19 event_factory: Callable[..., InferenceEvent],
20 ) -> None:
21 app = LogAppender(
22 tmp_log_dir, signer_identity,
23 rotation=RotationPolicy(max_events_per_segment=100),
24 )
25 app.append(event_factory())
26 app.append(event_factory())
27 jsonl = os.path.join(tmp_log_dir, "segment-00001.log")
28 sig = os.path.join(tmp_log_dir, "segment-00001.sig.json")
29 assert os.path.exists(jsonl)
30 assert not os.path.exists(sig) # not sealed yet
31 app.close()
32
33
34 def test_seal_current_segment_writes_sig(
35 signer_identity: AgentIdentity,
36 tmp_log_dir: str,
37 event_factory: Callable[..., InferenceEvent],
38 ) -> None:
39 app = LogAppender(tmp_log_dir, signer_identity)
40 app.append(event_factory())
41 header = app.seal_current_segment()
42 assert header is not None
43 assert header.segment_number == 1
44 assert len(header.merkle_root) == 64
45 sig_path = os.path.join(tmp_log_dir, "segment-00001.sig.json")
46 assert os.path.exists(sig_path)
47 app.close()
48
49
50 def test_rotation_on_event_count(
51 signer_identity: AgentIdentity,
52 tmp_log_dir: str,
53 event_factory: Callable[..., InferenceEvent],
54 ) -> None:
55 app = LogAppender(
56 tmp_log_dir, signer_identity,
57 rotation=RotationPolicy(max_events_per_segment=3),
58 )
59 for _ in range(3):
60 app.append(event_factory())
61 # Hitting 3 events triggers a seal
62 assert os.path.exists(os.path.join(tmp_log_dir, "segment-00001.sig.json"))
63 app.close()
64
65
66 def test_rotation_on_byte_threshold(
67 signer_identity: AgentIdentity,
68 tmp_log_dir: str,
69 event_factory: Callable[..., InferenceEvent],
70 ) -> None:
71 app = LogAppender(
72 tmp_log_dir, signer_identity,
73 rotation=RotationPolicy(
74 max_events_per_segment=100_000,
75 max_bytes_per_segment=256, # tiny
76 ),
77 )
78 # A single event's JSONL line is > 256 bytes once metadata is present
79 app.append(event_factory())
80 app.append(event_factory())
81 assert os.path.exists(os.path.join(tmp_log_dir, "segment-00001.sig.json"))
82 app.close()
83
84
85 def test_append_after_close_raises(
86 signer_identity: AgentIdentity,
87 tmp_log_dir: str,
88 event_factory: Callable[..., InferenceEvent],
89 ) -> None:
90 app = LogAppender(tmp_log_dir, signer_identity)
91 app.append(event_factory())
92 app.close()
93 with pytest.raises(AppendToSealedSegmentError):
94 app.append(event_factory())
95
96
97 def test_close_seals_final_segment(
98 signer_identity: AgentIdentity,
99 tmp_log_dir: str,
100 event_factory: Callable[..., InferenceEvent],
101 ) -> None:
102 app = LogAppender(
103 tmp_log_dir, signer_identity,
104 rotation=RotationPolicy(max_events_per_segment=1000),
105 )
106 app.append(event_factory())
107 app.append(event_factory())
108 app.close()
109 assert os.path.exists(os.path.join(tmp_log_dir, "segment-00001.sig.json"))
110
111
112 def test_reopen_continues_with_next_segment(
113 signer_identity: AgentIdentity,
114 tmp_log_dir: str,
115 event_factory: Callable[..., InferenceEvent],
116 ) -> None:
117 # First session
118 app1 = LogAppender(
119 tmp_log_dir, signer_identity,
120 rotation=RotationPolicy(max_events_per_segment=2),
121 )
122 app1.append(event_factory())
123 app1.append(event_factory()) # rotate -> seg 1 sealed
124 app1.close()
125 # Re-open
126 app2 = LogAppender(
127 tmp_log_dir, signer_identity,
128 rotation=RotationPolicy(max_events_per_segment=2),
129 )
130 # Next segment number should be 2
131 assert app2._current_segment_number == 2
132 assert app2._previous_segment_root != ""
133 app2.append(event_factory())
134 app2.close()
135 assert os.path.exists(os.path.join(tmp_log_dir, "segment-00002.sig.json"))
136