tegwick
51bc70d1ff
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
67 lines
3.9 KiB
Python
67 lines
3.9 KiB
Python
import shutil
|
|
import uuid
|
|
from pathlib import Path
|
|
import pytest
|
|
from api.services import forge_projection as fp
|
|
from api.services import historical_projection as hp
|
|
from tests.test_forge_projection import _FakeSession,_Repo,_Row,_TaskRow
|
|
|
|
FIXTURE=Path(__file__).parent/'fixtures/railiance-cluster'
|
|
@pytest.fixture
|
|
def bound(tmp_path):
|
|
root=tmp_path/'railiance-cluster';shutil.copytree(FIXTURE,root)
|
|
derived=fp.derive_from_checkout(root,'railiance-cluster','candidate')
|
|
assert len(derived.workplans)==3
|
|
rows=[];tasks=[]
|
|
for parent in derived.workplans:
|
|
row=_Row(parent.record_id.lower(),'finished',parent.relative_path);row.id=uuid.UUID(parent.uuid);rows.append(row)
|
|
for source in parent.tasks:
|
|
task=_TaskRow(source.record_id,status='done',workplan_id=row.id);task.id=uuid.UUID(source.uuid);tasks.append(task)
|
|
assert len(tasks)==15
|
|
return root,derived,rows,tasks
|
|
|
|
def test_exact_three_sources_derive_only_frozen_15_task_uuids(bound):
|
|
root,derived,rows,tasks=bound
|
|
assert all(parent.historical_preservation for parent in derived.workplans)
|
|
assert all(parent.uuid!=fp.derived_record_uuid(parent.record_id) for parent in derived.workplans)
|
|
expected={item['uuid'] for entry in hp.BINDINGS if entry['repo']=='railiance-cluster' for item in entry['tasks']}
|
|
assert {str(t.id) for t in tasks}==expected
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_rows_are_entirely_unchanged_on_normal_receiving(bound):
|
|
root,derived,rows,tasks=bound;before=[dict(x.__dict__) for x in rows+tasks]
|
|
session=_FakeSession(_Repo(),rows,task_rows=tasks)
|
|
outcome=await fp.reset_repository_projection(session,'railiance-cluster',derived=derived)
|
|
assert outcome.status=='noop' and not outcome.refused
|
|
assert [dict(x.__dict__) for x in rows+tasks]==before
|
|
assert not session.added and not session.deleted
|
|
|
|
@pytest.mark.parametrize('defect',['missing-parent','missing-task','foreign-holder','nonterminal-task','nonterminal-parent','duplicate-current-label'])
|
|
@pytest.mark.asyncio
|
|
async def test_unsupported_actual_binding_refuses_before_any_mutation(bound,defect):
|
|
root,derived,rows,tasks=bound
|
|
if defect=='missing-parent':rows.pop()
|
|
elif defect=='missing-task':tasks.pop()
|
|
elif defect=='foreign-holder':tasks[0].workplan_id=uuid.uuid4()
|
|
elif defect=='nonterminal-task':tasks[0].status='progress'
|
|
elif defect=='duplicate-current-label':tasks.append(_TaskRow(tasks[0].record_id,status='done',workplan_id=tasks[0].workplan_id))
|
|
else:rows[0].status='active'
|
|
before=[dict(x.__dict__) for x in rows+tasks];session=_FakeSession(_Repo(),rows,task_rows=tasks)
|
|
outcome=await fp.reset_repository_projection(session,'railiance-cluster',derived=derived)
|
|
assert outcome.status=='refused';assert [dict(x.__dict__) for x in rows+tasks]==before
|
|
assert not session.added and not session.deleted
|
|
|
|
@pytest.mark.parametrize('change',['hash-drift','missing','symlink','live-source','missing-source-uuid'])
|
|
def test_frozen_source_drift_refuses_derivation(bound,change):
|
|
root,*_=bound;p=next((root/'workplans/archived').glob('*.md'));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-source':p.write_bytes(raw.replace(b'status: finished',b'status: active',1))
|
|
elif change=='missing-source-uuid':p.write_bytes(raw.replace(b'state_hub_task_id:',b'old_task_id:',1))
|
|
else:p.write_bytes(raw+b'\nunreviewed\n')
|
|
with pytest.raises(fp.ForgeDeriveError,match='Historical source binding'):fp.derive_from_checkout(root,'railiance-cluster','candidate')
|
|
|
|
def test_unknown_repository_never_trusts_frozen_source_uuids(bound):
|
|
root,derived,*_=bound;other=fp.derive_from_checkout(root,'different-repo','candidate')
|
|
assert not any(parent.historical_preservation for parent in other.workplans)
|
|
assert not {parent.uuid for parent in derived.workplans}&{parent.uuid for parent in other.workplans}
|