BQUAT/tests/test_runtime_storage.py at development · fabiendostie/BQUAT · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import sys
import unittest
from pathlib import Path
from uuid import uuid4

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from runtime import storage  # noqa: E402


def _sandbox_root() -> Path:
    root = ROOT / "runs" / "tmp-tests" / uuid4().hex
    root.mkdir(parents=True, exist_ok=True)
    return root


class RuntimeStorageTests(unittest.TestCase):
    def test_manifest_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-1")
        manifest = {"run_id": "run-1", "status": "pending"}
        storage.write_manifest(run_dir, manifest)
        loaded = storage.read_manifest(run_dir)
        self.assertEqual(loaded["run_id"], "run-1")

    def test_approvals_default(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-2")
        approvals = storage.read_approvals(run_dir)
        self.assertEqual(approvals, {"approvals": []})

    def test_schema_defaults(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-3")
        artifacts = storage.read_artifact_index(run_dir)
        evidence = storage.read_evidence_links(run_dir)
        drrs = storage.read_drrs(run_dir)
        gates = storage.read_human_gates(run_dir)
        events = storage.read_events(run_dir)
        timeline = storage.read_timeline(run_dir)
        self.assertEqual(artifacts["artifacts"], [])
        self.assertEqual(evidence, {"evidence": []})
        self.assertEqual(drrs, {"drrs": []})
        self.assertEqual(gates, {"gates": []})
        self.assertEqual(events, {"events": []})
        self.assertEqual(timeline["entries"], [])

    def test_artifact_index_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-4")
        payload = {"run_id": "run-4", "artifacts": [{"path": "a"}], "updated_at": "now"}
        storage.write_artifact_index(run_dir, payload)
        loaded = storage.read_artifact_index(run_dir)
        self.assertEqual(loaded, payload)

    def test_evidence_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-5")
        payload = {"evidence": [{"id": "e1"}]}
        storage.write_evidence_links(run_dir, payload)
        loaded = storage.read_evidence_links(run_dir)
        self.assertEqual(loaded, payload)

    def test_drr_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-5b")
        payload = {"drrs": [{"id": "d1"}]}
        storage.write_drrs(run_dir, payload)
        loaded = storage.read_drrs(run_dir)
        self.assertEqual(loaded, payload)

    def test_gates_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-6")
        payload = {"gates": [{"gate_id": "g1"}]}
        storage.write_human_gates(run_dir, payload)
        loaded = storage.read_human_gates(run_dir)
        self.assertEqual(loaded, payload)

    def test_events_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-7")
        payload = {"events": [{"event_type": "WorkflowStarted"}]}
        storage.write_events(run_dir, payload)
        loaded = storage.read_events(run_dir)
        self.assertEqual(loaded, payload)

    def test_tool_results_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-8")
        payload = {"results": [{"name": "getCurrentTime", "status": "completed"}]}
        storage.write_tool_results(run_dir, payload)
        loaded = storage.read_tool_results(run_dir)
        self.assertEqual(loaded, payload)

    def test_timeline_round_trip(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-9")
        payload = {
            "run_id": "run-9",
            "generated_at": "2025-12-30T00:00:00-05:00",
            "entries": [{"type": "event", "timestamp": "2025-12-30T00:01:00-05:00"}],
        }
        storage.write_timeline(run_dir, payload)
        loaded = storage.read_timeline(run_dir)
        self.assertEqual(loaded, payload)

    def test_list_runs(self) -> None:
        root = _sandbox_root()
        run_dir = storage.init_run_dir(root, "run-10")
        storage.write_manifest(run_dir, {"run_id": "run-10"})
        storage.init_run_dir(root, "run-11")
        runs = storage.list_runs(root)
        self.assertEqual([path.name for path in runs], ["run-10"])


if __name__ == "__main__":
    unittest.main()