tegwick
57f2751f7b
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091
135 lines
6.8 KiB
Python
135 lines
6.8 KiB
Python
"""Closed historical views preserve identity; unknown sources never gain authority."""
|
|
import hashlib
|
|
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, TestExistingWorkplanTasks as ExistingTaskTests
|
|
|
|
|
|
def _fixture(tmp_path, monkeypatch):
|
|
parent = str(uuid.uuid4()); task = str(uuid.uuid4())
|
|
paths = ['workplans/archived/old.md', 'workplans/split.md']
|
|
entries = []
|
|
for index, path in enumerate(paths):
|
|
rid = 'custodian-WP-0003' + ('a' if index else '')
|
|
data = (f'---\nid: {rid}\ntype: workplan\nrepo: activity-core\nstatus: done\n'
|
|
f'state_hub_workstream_id: {parent}\ntasks:\n'
|
|
f'- id: T34\n title: Closed task\n status: done\n state_hub_task_id: {task}\n---\n').encode()
|
|
p = tmp_path/path;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(data)
|
|
entries.append(dict(repo='activity-core',path=path,sha256=hashlib.sha256(data).hexdigest(),
|
|
record_id=rid,uuid=parent,parent_record_id='custodian-WP-0003'))
|
|
monkeypatch.setattr(hp,'BINDINGS',entries)
|
|
return parent,task,paths
|
|
|
|
|
|
def test_exact_shared_views_derive_one_preserved_parent(tmp_path, monkeypatch):
|
|
parent,task,_ = _fixture(tmp_path,monkeypatch)
|
|
derived = fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
assert len(derived.workplans)==1
|
|
w = derived.workplans[0]
|
|
assert w.uuid == parent and w.historical_preservation
|
|
assert len(w.tasks)==1 and w.tasks[0].uuid==task
|
|
assert w.tasks[0].record_id=='T34'
|
|
assert w.uuid != fp.derived_record_uuid(w.record_id)
|
|
|
|
|
|
@pytest.mark.parametrize('change',['bytes','missing','symlink'])
|
|
def test_changed_bound_history_refuses_derive(tmp_path,monkeypatch,change):
|
|
_,_,paths = _fixture(tmp_path,monkeypatch)
|
|
p=tmp_path/paths[1]
|
|
if change=='bytes':p.write_text(p.read_text().replace('status: done','status: active'))
|
|
elif change=='missing':p.unlink()
|
|
else:
|
|
data=p.read_bytes();p.unlink();other=tmp_path/'other.md';other.write_bytes(data);p.symlink_to(other)
|
|
with pytest.raises(fp.ForgeDeriveError,match='Historical source binding'):
|
|
fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
|
|
|
|
def test_unknown_repo_cannot_trust_existing_binding(tmp_path,monkeypatch):
|
|
parent,_,_ = _fixture(tmp_path,monkeypatch)
|
|
derived=fp.derive_from_checkout(tmp_path,'other-repo','abc')
|
|
assert all(not w.historical_preservation and w.uuid != parent for w in derived.workplans)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_historical_receiving_preserves_terminal_tombstone_and_null_task_label(tmp_path,monkeypatch):
|
|
parent,task,_=_fixture(tmp_path,monkeypatch)
|
|
derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
row=_Row(slug='old@retired-20260827',status='archived',path='workplans/archived/old.md')
|
|
row.id=uuid.UUID(parent)
|
|
from datetime import datetime,timezone
|
|
row.projection_retired_at=datetime.now(timezone.utc)
|
|
t=_TaskRow(None,status='done',workplan_id=row.id);t.id=uuid.UUID(task)
|
|
before_row=dict(row.__dict__);before_task=dict(t.__dict__)
|
|
session=_FakeSession(_Repo(),[row],task_rows=[t])
|
|
outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
|
|
assert outcome.status=='noop' and outcome.refused==[]
|
|
assert row.__dict__==before_row and t.__dict__==before_task
|
|
assert not session.added and not session.deleted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_historical_task_refuses_before_mutation(tmp_path,monkeypatch):
|
|
parent,_,_=_fixture(tmp_path,monkeypatch)
|
|
derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
row=_Row(slug='old',status='archived',path='workplans/archived/old.md');row.id=uuid.UUID(parent)
|
|
session=_FakeSession(_Repo(),[row],task_rows=[])
|
|
outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
|
|
assert outcome.status=='refused'
|
|
assert outcome.refused[0]['reason'].startswith('historical binding')
|
|
assert not session.added and not session.deleted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_holder_query_includes_retired_and_stale_parents():
|
|
derived=ExistingTaskTests()._derived(('DEMO-WP-0001-T01','Task','todo'))
|
|
current=_Row(slug='demo-wp-0001',status='active',path='workplans/a.md')
|
|
holder=_TaskRow('DEMO-WP-0001-T01',workplan_id=uuid.uuid4())
|
|
holder.id=uuid.UUID(derived.workplans[0].tasks[0].uuid)
|
|
class Session(_FakeSession):
|
|
statements=[]
|
|
async def execute(self,statement,*args,**kwargs):
|
|
self.statements.append(str(statement))
|
|
return await super().execute(statement,*args,**kwargs)
|
|
session=Session(_Repo(),[current],task_rows=[],task_collision_rows=[holder])
|
|
outcome=await fp.reset_repository_projection(session,'demo',derived=derived)
|
|
assert outcome.status=='refused' and not session.added
|
|
query=session.statements[-1]
|
|
assert 'tasks.id IN' in query
|
|
assert 'projection_retired_at' not in query and 'NOT IN' not in query and 'JOIN' not in query
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_historical_parent_never_recreates(tmp_path,monkeypatch):
|
|
_fixture(tmp_path,monkeypatch)
|
|
derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
session=_FakeSession(_Repo(),[])
|
|
outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
|
|
assert outcome.status=='refused' and not session.added and not session.deleted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('field',['parent','task'])
|
|
async def test_historical_nonterminal_hub_record_requires_owner_review(tmp_path,monkeypatch,field):
|
|
parent,task,_=_fixture(tmp_path,monkeypatch)
|
|
derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
row=_Row(slug='old',status='active' if field=='parent' else 'finished',path='workplans/archived/old.md')
|
|
row.id=uuid.UUID(parent)
|
|
t=_TaskRow('preserved-old-label',status='wait' if field=='task' else 'done',workplan_id=row.id)
|
|
t.id=uuid.UUID(task)
|
|
session=_FakeSession(_Repo(),[row],task_rows=[t])
|
|
outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
|
|
assert outcome.status=='refused' and not session.added and not session.deleted
|
|
|
|
|
|
def test_historical_diff_uses_uuid_without_relabeling(tmp_path,monkeypatch):
|
|
parent,task,_=_fixture(tmp_path,monkeypatch)
|
|
derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
|
|
diff=fp.diff_against_hub(derived,[{'id':parent,'slug':'historic@retired-20260827',
|
|
'status':'archived','backing_relative_path':'old-path.md'}],
|
|
{parent:[{'id':task,'record_id':None,'title':'Old title','status':'done'}]})
|
|
assert not diff.missing and not diff.stale and not diff.differing
|