490 lines
19 KiB
Python
490 lines
19 KiB
Python
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "sync_state.py"
|
|
SPEC = importlib.util.spec_from_file_location("sync_state", SCRIPT)
|
|
sync_state = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
SPEC.loader.exec_module(sync_state)
|
|
|
|
|
|
def write_json(path: Path, value: dict) -> str:
|
|
payload = (json.dumps(value, indent=2) + "\n").encode()
|
|
path.write_bytes(payload)
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
class SyncStateTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
self.original_cwd = Path.cwd()
|
|
os.chdir(self.root)
|
|
self.state_path = self.root / "agent-ui" / ".sync-state.json"
|
|
self.state_path.parent.mkdir(parents=True)
|
|
self.base_state = {
|
|
"schema_version": 1,
|
|
"surface_type": "ops-dev",
|
|
"baseline_mode": "migration",
|
|
"last_sync_mode": "baseline",
|
|
"last_synced_head": "a" * 40,
|
|
"last_synced_at": "2026-01-01T00:00:00Z",
|
|
"agent_ui_paths": [],
|
|
"code_paths": [],
|
|
"notes": [],
|
|
}
|
|
|
|
def tearDown(self):
|
|
os.chdir(self.original_cwd)
|
|
self.temporary.cleanup()
|
|
|
|
def prepare_args(self, digest: str, **overrides):
|
|
values = {
|
|
"state": self.state_path,
|
|
"expected_sha256": digest,
|
|
"task_path": "agent-task/ui_work",
|
|
"previous_task_path": None,
|
|
"scope": "view:dashboard",
|
|
"status_path": ["agent-ui/definition/views/dashboard/index.md"],
|
|
"frame_path": ["agent-ui/frame/views/dashboard/index.md"],
|
|
"code_path": ["src/dashboard.py"],
|
|
"verification_requirement": ["python -m unittest: PASS"],
|
|
"prepared_at": "2026-07-29T00:00:00Z",
|
|
"replace": False,
|
|
}
|
|
values.update(overrides)
|
|
return type("Args", (), values)
|
|
|
|
def reconcile_args(self, digest: str, completion_log: Path, **overrides):
|
|
values = {
|
|
"state": self.state_path,
|
|
"expected_sha256": digest,
|
|
"task_path": "agent-task/ui_work",
|
|
"completion_log": completion_log,
|
|
"validation_evidence": "validate-agent-ui scope=view:dashboard PASS",
|
|
"reconciled_at": "2026-07-29T01:00:00Z",
|
|
"sync_result_head": None,
|
|
}
|
|
values.update(overrides)
|
|
return type("Args", (), values)
|
|
|
|
def prepare_milestone_args(self, digest: str, **overrides):
|
|
values = {
|
|
"state": self.state_path,
|
|
"expected_sha256": digest,
|
|
"milestone_path": (
|
|
"agent-roadmap/phase/product/milestones/dashboard-refresh.md"
|
|
),
|
|
"scope": "view:dashboard",
|
|
"status_path": ["agent-ui/definition/views/dashboard/index.md"],
|
|
"frame_path": ["agent-ui/frame/views/dashboard/index.md"],
|
|
"code_path": ["src/dashboard.py"],
|
|
"verification_requirement": ["python -m unittest: PASS"],
|
|
"prepared_at": "2026-07-29T00:00:00Z",
|
|
"replace": False,
|
|
}
|
|
values.update(overrides)
|
|
return type("Args", (), values)
|
|
|
|
def reconcile_milestone_args(self, digest: str, **overrides):
|
|
values = {
|
|
"state": self.state_path,
|
|
"expected_sha256": digest,
|
|
"milestone_path": (
|
|
"agent-roadmap/phase/product/milestones/dashboard-refresh.md"
|
|
),
|
|
"closure_evidence": "complete-milestone check-only: closure-ready",
|
|
"validation_evidence": "validate-agent-ui scope=view:dashboard PASS",
|
|
"reconciled_at": "2026-07-29T01:00:00Z",
|
|
"sync_result_head": None,
|
|
}
|
|
values.update(overrides)
|
|
return type("Args", (), values)
|
|
|
|
def record_sync_args(self, digest: str, **overrides):
|
|
values = {
|
|
"state": self.state_path,
|
|
"expected_sha256": digest,
|
|
"sync_mode": "incremental",
|
|
"sync_result_head": "b" * 40,
|
|
"synced_at": "2026-07-29T02:00:00Z",
|
|
"agent_ui_path": ["agent-ui/definition/views/dashboard/index.md"],
|
|
"code_path": ["src/dashboard.py"],
|
|
"note": ["direct sync"],
|
|
}
|
|
values.update(overrides)
|
|
return type("Args", (), values)
|
|
|
|
def test_prepare_migrates_v1_and_records_typed_paths(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
result = sync_state._prepare(self.prepare_args(digest))
|
|
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("prepared", result["status"])
|
|
self.assertEqual(4, state["schema_version"])
|
|
self.assertEqual(
|
|
["agent-ui/definition/views/dashboard/index.md"],
|
|
state["pending_code_work"][0]["status_paths"],
|
|
)
|
|
self.assertEqual(
|
|
["agent-ui/frame/views/dashboard/index.md"],
|
|
state["pending_code_work"][0]["frame_paths"],
|
|
)
|
|
self.assertEqual([], state["reconciled_code_work"])
|
|
self.assertEqual([], state["pending_milestone_work"])
|
|
self.assertEqual([], state["reconciled_milestone_work"])
|
|
|
|
def test_prepare_is_idempotent_and_requires_replace_for_scope_change(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
first = sync_state._prepare(self.prepare_args(digest))
|
|
second = sync_state._prepare(
|
|
self.prepare_args(
|
|
first["after_sha256"],
|
|
prepared_at="2026-07-29T00:05:00Z",
|
|
)
|
|
)
|
|
self.assertEqual("already_prepared", second["status"])
|
|
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._prepare(
|
|
self.prepare_args(
|
|
second["after_sha256"],
|
|
code_path=["src/other.py"],
|
|
)
|
|
)
|
|
|
|
replaced = sync_state._prepare(
|
|
self.prepare_args(
|
|
second["after_sha256"],
|
|
code_path=["src/other.py"],
|
|
replace=True,
|
|
)
|
|
)
|
|
self.assertEqual("replaced", replaced["status"])
|
|
|
|
def test_prepare_rejects_stale_digest(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._prepare(self.prepare_args("0" * 64))
|
|
self.assertEqual(digest, hashlib.sha256(self.state_path.read_bytes()).hexdigest())
|
|
|
|
def test_prepare_rejects_malformed_digest(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._prepare(self.prepare_args("not-a-sha256"))
|
|
self.assertEqual(digest, hashlib.sha256(self.state_path.read_bytes()).hexdigest())
|
|
|
|
def test_prepare_rebinds_refined_task_path(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
rebound = sync_state._prepare(
|
|
self.prepare_args(
|
|
prepared["after_sha256"],
|
|
previous_task_path="agent-task/ui_work",
|
|
task_path="agent-task/ui_work/01_closure",
|
|
)
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("rebound", rebound["status"])
|
|
self.assertEqual(
|
|
"agent-task/ui_work/01_closure",
|
|
state["pending_code_work"][0]["task_path"],
|
|
)
|
|
|
|
def test_prepare_rejects_status_path_owned_by_another_task(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._prepare(
|
|
self.prepare_args(
|
|
prepared["after_sha256"],
|
|
task_path="agent-task/other_ui_work",
|
|
frame_path=[],
|
|
code_path=["src/other.py"],
|
|
)
|
|
)
|
|
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual(1, len(state["pending_code_work"]))
|
|
self.assertEqual(
|
|
"agent-task/ui_work",
|
|
state["pending_code_work"][0]["task_path"],
|
|
)
|
|
|
|
def test_prepare_rebind_requires_replace_for_changed_scope(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._prepare(
|
|
self.prepare_args(
|
|
prepared["after_sha256"],
|
|
previous_task_path="agent-task/ui_work",
|
|
task_path="agent-task/ui_work/01_closure",
|
|
code_path=["src/closure.py"],
|
|
)
|
|
)
|
|
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual(
|
|
"agent-task/ui_work",
|
|
state["pending_code_work"][0]["task_path"],
|
|
)
|
|
|
|
def test_prepare_milestone_records_mapping_and_blocks_shared_ui_owner(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
milestone = self.root / (
|
|
"agent-roadmap/phase/product/milestones/dashboard-refresh.md"
|
|
)
|
|
milestone.parent.mkdir(parents=True)
|
|
milestone.write_text("# Milestone: Dashboard refresh\n")
|
|
prepared = sync_state._prepare_milestone(
|
|
self.prepare_milestone_args(digest)
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("prepared", prepared["status"])
|
|
self.assertEqual(1, len(state["pending_milestone_work"]))
|
|
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._prepare(
|
|
self.prepare_args(
|
|
prepared["after_sha256"],
|
|
frame_path=[],
|
|
)
|
|
)
|
|
|
|
def test_reconcile_milestone_moves_mapping_and_is_idempotent(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
milestone = self.root / (
|
|
"agent-roadmap/phase/product/milestones/dashboard-refresh.md"
|
|
)
|
|
milestone.parent.mkdir(parents=True)
|
|
milestone.write_text("# Milestone: Dashboard refresh\n")
|
|
prepared = sync_state._prepare_milestone(
|
|
self.prepare_milestone_args(digest)
|
|
)
|
|
|
|
reconciled = sync_state._reconcile_milestone(
|
|
self.reconcile_milestone_args(prepared["after_sha256"])
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("reconciled", reconciled["status"])
|
|
self.assertEqual([], state["pending_milestone_work"])
|
|
self.assertEqual(1, len(state["reconciled_milestone_work"]))
|
|
self.assertEqual("a" * 40, state["last_synced_head"])
|
|
|
|
milestone.unlink()
|
|
repeated = sync_state._reconcile_milestone(
|
|
self.reconcile_milestone_args(reconciled["after_sha256"])
|
|
)
|
|
self.assertEqual("already_reconciled", repeated["status"])
|
|
|
|
def test_v2_legacy_pending_paths_are_typed_during_migration(self):
|
|
state = dict(self.base_state)
|
|
state["schema_version"] = 2
|
|
state["pending_code_work"] = [
|
|
{
|
|
"task_path": "agent-task/ui_work",
|
|
"scope": "all",
|
|
"agent_ui_paths": [
|
|
"agent-ui/definition/views/dashboard/index.md",
|
|
"agent-ui/frame/views/dashboard/index.md",
|
|
],
|
|
"code_paths": ["src/dashboard.py"],
|
|
"verification_requirements": [],
|
|
"prepared_at": "2026-07-29T00:00:00Z",
|
|
"state": "pending",
|
|
}
|
|
]
|
|
normalized = sync_state._normalize_state(state)
|
|
entry = normalized["pending_code_work"][0]
|
|
self.assertEqual(
|
|
["agent-ui/definition/views/dashboard/index.md"],
|
|
entry["status_paths"],
|
|
)
|
|
self.assertEqual(
|
|
["agent-ui/frame/views/dashboard/index.md"],
|
|
entry["frame_paths"],
|
|
)
|
|
|
|
def test_normalize_rejects_non_list_and_duplicate_pending_entries(self):
|
|
invalid = dict(self.base_state)
|
|
invalid["pending_code_work"] = {}
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._normalize_state(invalid)
|
|
|
|
entry = {
|
|
"task_path": "agent-task/ui_work",
|
|
"scope": "all",
|
|
"status_paths": ["agent-ui/definition/views/dashboard/index.md"],
|
|
"frame_paths": [],
|
|
"code_paths": ["src/dashboard.py"],
|
|
"verification_requirements": [],
|
|
"prepared_at": "2026-07-29T00:00:00Z",
|
|
"state": "pending",
|
|
}
|
|
duplicate = dict(self.base_state)
|
|
duplicate["schema_version"] = 3
|
|
duplicate["pending_code_work"] = [entry, dict(entry)]
|
|
duplicate["reconciled_code_work"] = []
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._normalize_state(duplicate)
|
|
|
|
second_owner = dict(entry)
|
|
second_owner["task_path"] = "agent-task/other"
|
|
duplicate_owner = dict(self.base_state)
|
|
duplicate_owner["schema_version"] = 3
|
|
duplicate_owner["pending_code_work"] = [entry, second_owner]
|
|
duplicate_owner["reconciled_code_work"] = []
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._normalize_state(duplicate_owner)
|
|
|
|
def test_reconcile_moves_pending_entry_and_is_idempotent(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
completion_log = Path("agent-task/archive/2026/07/ui_work/complete.log")
|
|
(self.root / completion_log).parent.mkdir(parents=True)
|
|
(self.root / completion_log).write_text(
|
|
"# Complete - ui_work\n\n"
|
|
"## 루프 이력\n\n"
|
|
"| Plan | Review | Verdict | 메모 |\n"
|
|
"|---|---|---|---|\n"
|
|
"| `plan.log` | `review.log` | PASS | done |\n"
|
|
)
|
|
|
|
reconciled = sync_state._reconcile(
|
|
self.reconcile_args(prepared["after_sha256"], self.root / completion_log)
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("reconciled", reconciled["status"])
|
|
self.assertEqual([], state["pending_code_work"])
|
|
self.assertEqual(1, len(state["reconciled_code_work"]))
|
|
self.assertEqual("a" * 40, state["last_synced_head"])
|
|
self.assertIn(
|
|
"baseline=preserved:no validated sync_result_head",
|
|
state["notes"][-1],
|
|
)
|
|
|
|
repeated = sync_state._reconcile(
|
|
self.reconcile_args(reconciled["after_sha256"], self.root / completion_log)
|
|
)
|
|
self.assertEqual("already_reconciled", repeated["status"])
|
|
|
|
def test_reconcile_rejects_completion_log_for_another_task(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
completion_log = Path("agent-task/archive/2026/07/other/complete.log")
|
|
(self.root / completion_log).parent.mkdir(parents=True)
|
|
(self.root / completion_log).write_text(
|
|
"# Complete - other\n\n"
|
|
"## 루프 이력\n\n"
|
|
"| Plan | Review | Verdict | 메모 |\n"
|
|
"|---|---|---|---|\n"
|
|
"| `plan.log` | `review.log` | PASS | done |\n"
|
|
)
|
|
with self.assertRaises(sync_state.StateError):
|
|
sync_state._reconcile(
|
|
self.reconcile_args(prepared["after_sha256"], completion_log)
|
|
)
|
|
|
|
def test_record_sync_preserves_pending_and_reconciled_state(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
recorded = sync_state._record_sync(
|
|
self.record_sync_args(
|
|
prepared["after_sha256"],
|
|
agent_ui_path=[
|
|
"agent-ui/definition/views/other-dashboard/index.md"
|
|
],
|
|
code_path=["src/other_dashboard.py"],
|
|
)
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("sync_recorded", recorded["status"])
|
|
self.assertEqual("incremental", state["last_sync_mode"])
|
|
self.assertEqual("b" * 40, state["last_synced_head"])
|
|
self.assertEqual(1, len(state["pending_code_work"]))
|
|
self.assertEqual([], state["reconciled_code_work"])
|
|
|
|
def test_record_sync_preserves_pending_milestone_state(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
milestone = self.root / (
|
|
"agent-roadmap/phase/product/milestones/dashboard-refresh.md"
|
|
)
|
|
milestone.parent.mkdir(parents=True)
|
|
milestone.write_text("# Milestone: Dashboard refresh\n")
|
|
prepared = sync_state._prepare_milestone(
|
|
self.prepare_milestone_args(digest)
|
|
)
|
|
state_before = self.state_path.read_bytes()
|
|
|
|
with self.assertRaisesRegex(
|
|
sync_state.StateError,
|
|
"overlaps pending work paths",
|
|
):
|
|
sync_state._record_sync(
|
|
self.record_sync_args(prepared["after_sha256"])
|
|
)
|
|
self.assertEqual(state_before, self.state_path.read_bytes())
|
|
|
|
recorded = sync_state._record_sync(
|
|
self.record_sync_args(
|
|
prepared["after_sha256"],
|
|
agent_ui_path=[
|
|
"agent-ui/definition/views/other-dashboard/index.md"
|
|
],
|
|
code_path=["src/other_dashboard.py"],
|
|
)
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("sync_recorded", recorded["status"])
|
|
self.assertEqual(1, len(state["pending_milestone_work"]))
|
|
self.assertEqual([], state["reconciled_milestone_work"])
|
|
|
|
def test_record_sync_rejects_paths_owned_by_pending_work(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
prepared = sync_state._prepare(self.prepare_args(digest))
|
|
state_before = self.state_path.read_bytes()
|
|
|
|
for agent_ui_path in (
|
|
"agent-ui/definition/views/dashboard/index.md",
|
|
"agent-ui/definition/views/dashboard",
|
|
):
|
|
with self.subTest(agent_ui_path=agent_ui_path):
|
|
with self.assertRaisesRegex(
|
|
sync_state.StateError,
|
|
"overlaps pending work paths",
|
|
):
|
|
sync_state._record_sync(
|
|
self.record_sync_args(
|
|
prepared["after_sha256"],
|
|
agent_ui_path=[agent_ui_path],
|
|
)
|
|
)
|
|
|
|
self.assertEqual(state_before, self.state_path.read_bytes())
|
|
|
|
def test_record_migration_allows_no_code_paths(self):
|
|
digest = write_json(self.state_path, self.base_state)
|
|
recorded = sync_state._record_sync(
|
|
self.record_sync_args(
|
|
digest,
|
|
sync_mode="migration",
|
|
code_path=[],
|
|
)
|
|
)
|
|
state = json.loads(self.state_path.read_text())
|
|
self.assertEqual("sync_recorded", recorded["status"])
|
|
self.assertEqual("migration", state["last_sync_mode"])
|
|
self.assertEqual([], state["code_paths"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|