Codex Lead Engineer
125 lines
4.5 KiB
Python
125 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from project_bus.db import Database
|
|
from project_bus.operations import OperationsError, backup_database, bootstrap_project, load_manifest
|
|
|
|
|
|
class RecordingClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def call(self, name, arguments):
|
|
self.calls.append((name, arguments))
|
|
return {"name": name, "idempotency_key": arguments["idempotency_key"]}
|
|
|
|
|
|
class OperationsTests(unittest.TestCase):
|
|
def manifest(self):
|
|
return {
|
|
"project": {
|
|
"project_id": "sandbox",
|
|
"name": "Sandbox",
|
|
"description": "test",
|
|
},
|
|
"actors": [
|
|
{
|
|
"actor_id": "po",
|
|
"auth_subject": "po",
|
|
"display_name": "PO",
|
|
"role": "PO",
|
|
},
|
|
{
|
|
"actor_id": "reviewer",
|
|
"auth_subject": "reviewer",
|
|
"display_name": "Reviewer",
|
|
"role": "REVIEWER",
|
|
},
|
|
],
|
|
}
|
|
|
|
def test_bootstrap_uses_stable_idempotency_keys(self):
|
|
first = RecordingClient()
|
|
second = RecordingClient()
|
|
bootstrap_project(first, self.manifest())
|
|
bootstrap_project(second, self.manifest())
|
|
self.assertEqual(first.calls, second.calls)
|
|
self.assertEqual(first.calls[0][0], "create_project")
|
|
self.assertEqual([call[0] for call in first.calls[1:]], ["register_actor", "register_actor"])
|
|
|
|
def test_manifest_rejects_system_and_duplicate_subjects(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "manifest.json"
|
|
invalid = self.manifest()
|
|
invalid["actors"][0]["role"] = "SYSTEM"
|
|
path.write_text(json.dumps(invalid), encoding="utf-8")
|
|
with self.assertRaises(OperationsError):
|
|
load_manifest(path)
|
|
invalid = self.manifest()
|
|
invalid["actors"][1]["auth_subject"] = "po"
|
|
path.write_text(json.dumps(invalid), encoding="utf-8")
|
|
with self.assertRaises(OperationsError):
|
|
load_manifest(path)
|
|
|
|
def test_backup_is_consistent_and_integrity_checked(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = Path(directory) / "source.db"
|
|
destination = Path(directory) / "backup.db"
|
|
Database(source).migrate()
|
|
with sqlite3.connect(source) as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO projects(project_id, name, description, created_at)
|
|
VALUES ('sandbox', 'Sandbox', '', '2026-07-30T00:00:00Z')
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO actors(actor_id, auth_subject, display_name, created_at)
|
|
VALUES ('system', 'system', 'System', '2026-07-30T00:00:00Z')
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO event_log(
|
|
event_id, project_id, event_type, actor_id, actor_role,
|
|
aggregate_type, aggregate_id, payload_json, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
"evt_test",
|
|
"sandbox",
|
|
"test.created",
|
|
"system",
|
|
"SYSTEM",
|
|
"test",
|
|
"test-1",
|
|
"{}",
|
|
"2026-07-30T00:00:00Z",
|
|
),
|
|
)
|
|
receipt = backup_database(source, destination)
|
|
self.assertEqual(receipt["integrity"], "ok")
|
|
self.assertEqual(receipt["event_cursor"], 1)
|
|
with sqlite3.connect(destination) as connection:
|
|
self.assertEqual(
|
|
connection.execute("SELECT COUNT(*) FROM event_log").fetchone()[0],
|
|
1,
|
|
)
|
|
|
|
def test_backup_refuses_in_place_destination(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = Path(directory) / "source.db"
|
|
source.touch()
|
|
with self.assertRaises(OperationsError):
|
|
backup_database(source, source)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|