tegwick
bd6fd62b0a
A binding put that only refreshes the sync timestamp, a push that reports the remote is already current, and a C-11 skip on a finished workplan are not repairs. Leave those out of FIXES APPLIED so a clean second run stays clean. Assistant: grok Assistant-Session: 01a16071-5425-78d3-80b4-20fe2ee30f0a
200 lines
13 KiB
Python
200 lines
13 KiB
Python
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_get=lambda *a, **k: None,_api_put=lambda *a:calls.append(a) or dict(updated=1,received=1),_bindings_already_synced=lambda index, bindings: False))
|
|
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
|