state-hub/tests/test_frozen_binding_guard_isolated.py

from pathlib import Path
import ast, asyncio, importlib.util, types, uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
import pytest

ROOT = Path(__file__).resolve().parents[1]
spec=importlib.util.spec_from_file_location('historical',ROOT/'api/services/historical_projection.py')
h=importlib.util.module_from_spec(spec);spec.loader.exec_module(h)
B=h.FROZEN_CLUSTER_BINDINGS[0]
def binding(_bound=B,**changes):
 B=_bound
 d=dict(workplan_id=uuid.UUID(B['uuid']),repo_slug=B['repo'],filename=Path(B['path']).name,relative_path=B['path'],archived=True,status='finished');d.update(changes);return d
class Obj:
 def __init__(self,**kw):self.__dict__.update(kw);self.writes=[]
 def __setattr__(self,k,v):
  if 'writes' in self.__dict__:self.writes.append(k)
  object.__setattr__(self,k,v)
class Entry(Obj):
 def model_dump(self):return {k:v for k,v in self.__dict__.items() if k!='writes'}
class Col:
 def in_(self,x):return self
 def __eq__(self,x):return self
class Query:
 def __init__(self):self.locked=False;self.refreshed=False;self.ordered=False
 def where(self,*a):return self
 def order_by(self,*a):self.ordered=True;return self
 def with_for_update(self):self.locked=True;return self
 def execution_options(self,**kw):self.refreshed=kw.get("populate_existing") is True;return self
class Model: id=Col();slug=Col();workplan_id=Col()
class Result:
 def __init__(self,values):self.values=values
 def scalars(self):return self
 def all(self):return self.values
 def scalar_one_or_none(self):return self.values[0] if self.values else None
class Session:
 def __init__(self,rows):self.results=iter(rows);self.commits=0;self.queries=[]
 async def execute(self,q):self.queries.append(q);return Result(next(self.results))
 async def commit(self):self.commits+=1
class HTTPException(Exception):
 def __init__(self,status_code,detail):self.status_code=status_code;super().__init__(detail)
def extracted(path,name,extra):
 tree=ast.parse((ROOT/path).read_text());node=next(n for n in tree.body if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef)) and n.name==name)
 node.decorator_list=[];node.returns=None
 for a in node.args.args + node.args.kwonlyargs:a.annotation=None
 node.args.defaults=[]
 ns=dict(Path=Path,datetime=datetime,timezone=timezone,**extra)
 exec(compile(ast.Module(body=[node],type_ignores=[]),str(path),'exec'),ns);return ns[name]
def actual_report():
 tree=ast.parse((ROOT/'scripts/consistency_check.py').read_text())
 nodes=[n for n in tree.body if (isinstance(n,ast.ClassDef) and n.name in {'Issue','ConsistencyReport'}) or (isinstance(n,ast.AnnAssign) and isinstance(n.target,ast.Name) and n.target.id=='_AUTOMATION_ERROR_CHECKS')]
 ns=dict(dataclass=dataclass,field=field,_AUTOMATION_ERROR_CHECKS=frozenset({'C-00','C-44'}))
 exec(compile(ast.Module(body=nodes,type_ignores=[]),'actual_report','exec'),ns)
 return ns['ConsistencyReport']('railiance-cluster','/unused')

def closure(B=B):
 repo=Obj(id=uuid.uuid4());parent=Obj(id=uuid.UUID(B['uuid']),repo_id=repo.id,status='finished')
 tasks=[Obj(id=uuid.UUID(t['uuid']),workplan_id=parent.id,status='done') for t in B['tasks']]
 return repo,parent,tasks
async def run_api(entries,repo,parent,tasks,ordinary=None):
 rows=([parent] if parent else [])+([ordinary] if ordinary else [])
 session=Session([rows,[repo] if repo else [],tasks])
 fn=extracted('api/routers/workstreams.py','sync_workplan_bindings',dict(frozen_cluster_binding=h.frozen_cluster_binding,validate_frozen_cluster_rows=h.validate_frozen_cluster_rows,HTTPException=HTTPException,select=lambda *_:Query(),Workplan=Model,Task=Model,ManagedRepo=Model,_invalidate_workplan_index_cache=lambda:None))
 result=await fn(Obj(bindings=[Entry(**e) for e in entries]),session)
 return result,session
@pytest.mark.parametrize('change',[dict(repo_slug='foreign'),dict(workplan_id=uuid.uuid4()),dict(relative_path='workplans/other.md'),dict(filename='other.md'),dict(archived=False),dict(status='active')])
def test_metadata_refuses(change):
 with pytest.raises(ValueError):h.frozen_cluster_binding(binding(**change))
def test_unrelated_binding_unchanged():assert h.frozen_cluster_binding(binding(workplan_id=uuid.uuid4(),filename='ordinary.md',relative_path='workplans/ordinary.md',repo_slug='other')) is None
@pytest.mark.parametrize('fault',['parent_missing','foreign_repo','parent_active','task_missing','task_active','task_foreign','repo_missing'])
@pytest.mark.parametrize("bound",h.FROZEN_CLUSTER_BINDINGS)
def test_mixed_batch_refusal_precedes_all_writes(fault,bound):
 repo,parent,tasks=closure(bound);ordinary=Obj(id=uuid.uuid4())
 if fault=='repo_missing':repo=None
 if fault=='parent_missing':parent=None
 if fault=='foreign_repo':parent.repo_id=uuid.uuid4()
 if fault=='parent_active':parent.status='active'
 if fault=='task_missing':tasks.pop()
 if fault=='task_active':tasks[0].status='progress'
 if fault=='task_foreign':tasks[0].workplan_id=uuid.uuid4()
 if parent:parent.writes.clear()
 for t in tasks:t.writes.clear()
 normal=binding(workplan_id=ordinary.id,repo_slug='ordinary',filename='ordinary.md',relative_path='workplans/ordinary.md')
 with pytest.raises(HTTPException):asyncio.run(run_api([normal,binding(_bound=bound)],repo,parent,tasks,ordinary))
 assert ordinary.writes==[]
 if parent:assert parent.writes==[]
 assert all(t.writes==[] for t in tasks)
@pytest.mark.parametrize("bound",h.FROZEN_CLUSTER_BINDINGS)
def test_exact_frozen_no_setters_or_commit(bound):
 repo,parent,tasks=closure(bound);result,session=asyncio.run(run_api([binding(_bound=bound)],repo,parent,tasks))
 assert result==dict(updated=0,received=1,preserved=1);assert session.commits==0;assert parent.writes==[]
 assert all(q.locked and q.refreshed and q.ordered for q in (session.queries[0],session.queries[2]))
@pytest.mark.parametrize("bound",h.FROZEN_CLUSTER_BINDINGS)
def test_mixed_valid_preserves_frozen_and_updates_ordinary(bound):
 repo,parent,tasks=closure(bound);ordinary=Obj(id=uuid.uuid4())
 normal=binding(workplan_id=ordinary.id,repo_slug='ordinary',filename='ordinary.md',relative_path='workplans/ordinary.md')
 result,session=asyncio.run(run_api([normal,binding(_bound=bound)],repo,parent,tasks,ordinary))
 assert result==dict(updated=1,received=2,preserved=1);assert session.commits==1;assert parent.writes==[]
 assert ordinary.writes==['backing_filename','backing_relative_path','backing_archived','backing_synced_at']
@pytest.mark.parametrize('tamper',['metadata','bytes','symlink','missing_scan','missing_uuid','valid'])
@pytest.mark.parametrize("bound_index", range(3))
def test_cli_exact_source_omission_or_whole_batch_refusal(tmp_path,tamper,bound_index):
 bound=h.FROZEN_CLUSTER_BINDINGS[bound_index]
 import shutil
 infos=[]
 for entry in h.FROZEN_CLUSTER_BINDINGS:
  target=tmp_path/entry['path'];target.parent.mkdir(parents=True,exist_ok=True)
  shutil.copyfile(Path('/home/worsch/railiance-cluster')/entry['path'],target)
  infos.append((target,dict(state_hub_workstream_id=entry['uuid'],status='finished'),''))
 path=tmp_path/bound['path']
 meta=dict(state_hub_workstream_id=bound['uuid'],status='finished')
 if tamper=='metadata':meta['status']='active'
 if tamper=='missing_uuid':meta.pop('state_hub_workstream_id')
 if tamper=='bytes':path.write_text(path.read_text()+'\n')
 if tamper=='symlink':path.unlink();path.symlink_to(Path('/home/worsch/railiance-cluster')/bound['path'])
 ordinary=tmp_path/'workplans/ordinary.md';ordinary.write_text('ordinary')
 if tamper=='missing_scan':infos.pop(bound_index)
 else:infos[bound_index]=(path,meta,'')
 calls=[];report=actual_report()
 fn=extracted('scripts/consistency_check.py','_sync_workplan_bindings',dict(FROZEN_CLUSTER_BINDINGS=h.FROZEN_CLUSTER_BINDINGS,frozen_cluster_binding=h.frozen_cluster_binding,normalise_workstream_status=lambda s:s,workplan_display_path=lambda root,p:p.relative_to(root).as_posix(),_api_put=lambda *a:calls.append(a) or dict(updated=1,received=1)))
 fn('http://unused','railiance-cluster',[(ordinary,dict(state_hub_workstream_id=str(uuid.uuid4()),status='active'),''),*infos],tmp_path,report)
 if tamper=='valid':assert len(calls)==1 and len(calls[0][2]['bindings'])==1
 else:assert calls==[] and report.failures[0].check_id=='C-44' and report.automation_errors


def test_api_tampered_mixed_batch_refuses_before_queries_or_assignments():
 repo,parent,tasks=closure();ordinary=Obj(id=uuid.uuid4())
 with pytest.raises(HTTPException):
  asyncio.run(run_api([binding(workplan_id=ordinary.id,repo_slug='ordinary',filename='ordinary.md',relative_path='workplans/ordinary.md'),binding(archived=False)],repo,parent,tasks,ordinary))
 assert ordinary.writes==[] and parent.writes==[]


def test_foreign_repo_same_filename_path_is_ordinary():
 assert h.frozen_cluster_binding(binding(workplan_id=uuid.uuid4(),repo_slug='foreign')) is None


@pytest.mark.parametrize('entry',h.FROZEN_CLUSTER_BINDINGS)
def test_all_three_registry_bindings_are_guarded(entry):
 good=dict(workplan_id=entry['uuid'],repo_slug=entry['repo'],filename=Path(entry['path']).name,relative_path=entry['path'],archived=True,status='finished')
 assert h.frozen_cluster_binding(good) is entry
 with pytest.raises(ValueError):h.frozen_cluster_binding(dict(good,status='active'))


def test_real_cli_exit_criterion_is_nonzero_including_remote_all():
 report=actual_report();report.add(severity='FAIL',check_id='C-44',message='refused')
 fn=extracted('scripts/consistency_check.py','consistency_exit_code',{})
 assert fn([report])==1 and fn([report],remote_all=True)==1


@pytest.mark.parametrize('check,key,protected_kind',[
 ('C-04','ws_id','parent'),('C-10','task_id','task'),('C-22','task_id','task'),
 ('C-09','db_id','parent'),('C-23','wp_file','path'),
])
@pytest.mark.parametrize('bound',h.FROZEN_CLUSTER_BINDINGS)
def test_complete_fix_batch_preflight_before_ordinary_writes(check,key,protected_kind,bound):
 report=actual_report();report.repo_path='/tmp/fixture-cluster'
 report.add(severity='WARN',check_id='C-05',message='ordinary first',fixable=True,_fix_context={'ws_id':str(uuid.uuid4()),'field':'title','value':'ordinary'})
 value=bound['uuid'] if protected_kind=='parent' else bound['tasks'][0]['uuid'] if protected_kind=='task' else str(Path(report.repo_path)/bound['path'])
 report.add(severity='WARN',check_id=check,message='protected later',fixable=True,db_id=value if key=='db_id' else '',_fix_context={key:value})
 guard=extracted('scripts/consistency_check.py','_refuse_frozen_fix_batch',dict(FROZEN_CLUSTER_BINDINGS=h.FROZEN_CLUSTER_BINDINGS))
 assert guard(report,'railiance-cluster') and report.automation_errors
 calls=[]
 fn=extracted('scripts/consistency_check.py','fix_repo',dict(check_repo=lambda *a,**kw:report,_refuse_frozen_fix_batch=guard,_api_post=lambda *a:calls.append(a),_api_patch=lambda *a:calls.append(a)))
 assert fn('http://unused','railiance-cluster',None,False,False) is report and calls==[]

def test_unrelated_fix_context_not_refused():
 report=actual_report();report.add(severity='WARN',check_id='C-05',message='ordinary',fixable=True,_fix_context={'ws_id':str(uuid.uuid4())})
 guard=extracted('scripts/consistency_check.py','_refuse_frozen_fix_batch',dict(FROZEN_CLUSTER_BINDINGS=h.FROZEN_CLUSTER_BINDINGS))
 assert not guard(report,'other')


def test_check_repo_guard_precedes_binding_sync_and_does_not_duplicate():
 tree=ast.parse((ROOT/'scripts/consistency_check.py').read_text())
 node=next(n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=='check_repo')
 guard_lines=[n.lineno for n in ast.walk(node) if isinstance(n,ast.Call) and isinstance(n.func,ast.Name) and n.func.id=='_refuse_frozen_fix_batch']
 sync_lines=[n.lineno for n in ast.walk(node) if isinstance(n,ast.Call) and isinstance(n.func,ast.Name) and n.func.id=='_sync_workplan_bindings']
 assert len(guard_lines)==len(sync_lines)==1 and guard_lines[0]<sync_lines[0]
 report=actual_report();report.add(severity='FAIL',check_id='C-44',message='already refused')
 guard=extracted('scripts/consistency_check.py','_refuse_frozen_fix_batch',dict(FROZEN_CLUSTER_BINDINGS=h.FROZEN_CLUSTER_BINDINGS))
 assert guard(report,'railiance-cluster') and len(report.issues)==1


@pytest.mark.parametrize('bound',h.FROZEN_CLUSTER_BINDINGS)
@pytest.mark.parametrize('key',['to_workstream_id','to_workplan_id','to_task_id'])
def test_ordinary_dependency_to_history_remains_allowed(bound,key):
 report=actual_report();ordinary=str(uuid.uuid4())
 target=bound['tasks'][0]['uuid'] if key=='to_task_id' else bound['uuid']
 report.add(severity='WARN',check_id='C-20',message='ordinary inbound edge',fixable=True,db_id=ordinary,_fix_context={'from_workstream_id':ordinary,key:target,'relationship_type':'blocks'})
 guard=extracted('scripts/consistency_check.py','_refuse_frozen_fix_batch',dict(FROZEN_CLUSTER_BINDINGS=h.FROZEN_CLUSTER_BINDINGS))
 assert not guard(report,'railiance-cluster') and not report.automation_errors

@pytest.mark.parametrize('bound',h.FROZEN_CLUSTER_BINDINGS)
@pytest.mark.parametrize('key',['from_workstream_id','from_workplan_id','from_task_id','state_hub_workstream_id','state_hub_task_id','desired_record_id'])
def test_protected_dependency_origin_and_nested_binding_refused(bound,key):
 report=actual_report();origin=bound['tasks'][0]['uuid'] if 'task' in key else bound['uuid']
 context={key:origin,'to_workstream_id':str(uuid.uuid4())}
 if key.startswith('state_hub'):context={'meta':context}
 report.add(severity='WARN',check_id='C-20',message='protected origin',fixable=True,_fix_context=context)
 guard=extracted('scripts/consistency_check.py','_refuse_frozen_fix_batch',dict(FROZEN_CLUSTER_BINDINGS=h.FROZEN_CLUSTER_BINDINGS))
 assert guard(report,'railiance-cluster') and report.automation_errors