repo-manager/tests/test_cluster_frozen_history.py
tegwick 70cd754996 Admit exact frozen Cluster histories after verified receiving
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
2026-10-17 02:11:20 +02:00

52 lines
3.2 KiB
Python

import shutil
from pathlib import Path
import pytest
from repo_manager.identifiers import ensure_missing_work_record_identifiers as ensure
from repo_manager.record_identity import scan_record_identities
FIXTURE=Path(__file__).parent/'fixtures/railiance-cluster'
@pytest.fixture
def repo(tmp_path):
root=tmp_path/'railiance-cluster';shutil.copytree(FIXTURE,root)
live=root/'workplans/RCLUSTER-WP-0018.md';live.write_text('---\nid: RCLUSTER-WP-0018\nstatus: ready\n---\n\n```task\nid: RCLUSTER-WP-0018-T01\nstatus: todo\n```\n')
return root,live
def test_exact_three_histories_preserve_15_bindings_and_strict_findings(repo):
root,live=repo;archive={p:p.read_bytes() for p in (root/'workplans/archived').glob('*.md')};before=scan_record_identities(root)
assert len(before['invalid_identifiers'])==15 and len(before['identity_collisions'])==5
preview=ensure(root,execute=False);assert len(preview['assignments'])==2
assert preview['historical_invalid_identifiers']==before['invalid_identifiers']
assert preview['historical_identity_collisions']==before['identity_collisions']
result=ensure(root,execute=True);assert len(result['assignments'])==2
assert all(p.read_bytes()==raw for p,raw in archive.items())
assert not scan_record_identities(root)['ok']
assert ensure(root,execute=False)['assignments']==[]
@pytest.mark.parametrize('change',['bytes','missing','symlink','live','nonterminal','uuid'])
def test_unknown_history_change_refuses_before_any_allocation(repo,change):
root,live=repo;p=next((root/'workplans/archived').glob('*.md'));before=live.read_bytes()
raw=p.read_bytes()
if change=='missing':p.unlink()
elif change=='symlink':p.unlink();target=root/'saved';target.write_bytes(raw);p.symlink_to(target)
elif change=='live':p.write_bytes(raw.replace(b'status: finished',b'status: active',1))
elif change=='nonterminal':p.write_bytes(raw.replace(b'status: done',b'status: todo',1))
elif change=='uuid':p.write_bytes(raw.replace(b'state_hub_task_id:',b'old_task_id:',1))
else:p.write_bytes(raw+b'\nunreviewed change\n')
with pytest.raises(ValueError):ensure(root)
assert live.read_bytes()==before
def test_any_unknown_or_live_source_in_collision_still_refuses(repo):
root,live=repo;before=live.read_bytes();other=root/'workplans/unknown.md'
other.write_text('---\nid: RCLUSTER-WP-0099\nstatus: active\n---\n\n```task\nid: T01\nstatus: todo\nstate_hub_task_id: 11111111-1111-4111-8111-111111111111\n```\n')
with pytest.raises(ValueError,match='identities are not safe'):ensure(root)
assert live.read_bytes()==before
def test_other_repo_does_not_inherit_exact_frozen_exception(repo):
root,live=repo;other=root.with_name('other');root.rename(other)
with pytest.raises(ValueError,match='identities are not safe'):ensure(other)
def test_qualified_collision_stays_global(repo):
root,live=repo;before=live.read_bytes()
(root/'workplans/other.md').write_text('---\nid: RCLUSTER-WP-0019\nstatus: active\n---\n\n```task\nid: RCLUSTER-WP-0018-T01\nstatus: done\nstate_hub_task_id: 11111111-1111-4111-8111-111111111111\n```\n')
with pytest.raises(ValueError,match='collisions='):ensure(root)
assert live.read_bytes()==before