"""Synthetic scope tests. No product backlog, model, signing or services.""" import copy import json from pathlib import Path import sys import tempfile import unittest from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import controller as c import local_mode import test_controller as fixture def scope(): return dict(stage='PHASE-10', checkpoint='P10-00', mode='assessment-only', reportDirectory='docs/automation/stages/PHASE-10', handoffCommit='1'*40, handoffBranch='test/handoff') class ScopeTests(unittest.TestCase): def setUp(self): fixture.EngineTests.setUp(self) first=self.manifest['stages'][0] first['id']='PHASE-10' first['checkpoints'][0]['id']='P10-00' self.manifest['stages'][1]['dependencies']=['PHASE-10'] state=self.engine.store.read() state['stages']['PHASE-10']=state['stages'].pop('ONE') state['manifestSha256']=c.digest(self.manifest) self.engine.store.write(state) tearDown = fixture.EngineTests.tearDown git_at = staticmethod(fixture.EngineTests.git_at) def test_scope_requires_protected_stop(self): self.manifest['executionScope'] = scope() with self.assertRaisesRegex(c.GateError, 'protected stop'): c.validate_manifest(self.manifest) def test_scope_rejects_product_write_and_deletion(self): self.manifest['executionScope'] = scope() (self.repo/'README').write_text('unauthorised') with self.assertRaisesRegex(c.GateError, 'assessment'): self.engine.verify_assessment_writes() (self.repo/'README').unlink() with self.assertRaises(c.GateError): self.engine.verify_assessment_writes() def test_assessment_reports_are_permitted(self): self.manifest['executionScope'] = scope() output=self.repo/scope()['reportDirectory']/'P10-00.md' output.parent.mkdir(parents=True) output.write_text('Synthetic assessment') self.engine.verify_assessment_writes() def test_generic_approval_cannot_launch_next_checkpoint(self): first=self.manifest['stages'][0] first['checkpoints'].append(copy.deepcopy(first['checkpoints'][0])) first['checkpoints'][1]['id']='implementation' self.manifest['executionScope']=scope() state=self.engine.store.read() state['manifestSha256']=c.digest(self.manifest) state['stages']['PHASE-10']['checkpoint']=1 self.engine.store.write(state) self.engine.run() self.assertEqual(0,self.runner.calls) self.assertIn('assessment', self.engine.store.read()['latestFailure']) def test_assessment_reaches_protected_stop_and_cannot_advance(self): output='docs/automation/stages/PHASE-10/P10-00.md' class ReportRunner(fixture.SyntheticRunner): def run_session(inner, engine, stage, checkpoint, directory, prompt): inner.calls += 1 report=engine.repo/output report.parent.mkdir(parents=True,exist_ok=True) report.write_text('Synthetic assessment only') return dict(outcome='complete',summary='synthetic',blockers=[],requestedApprovals=[],checkpoint=checkpoint['id']) first=self.manifest['stages'][0] cp=first['checkpoints'][0] cp['requiredArtifacts']=[output] cp['approvalsAfter']=['candidate-specific-cleanup-scope'] first['checkpoints'].append(copy.deepcopy(cp)) first['checkpoints'][1]['id']='implementation' self.manifest['executionScope']={**scope(),'handoffCommit':self.manifest['baselineCommit'],'handoffBranch':'codex/base'} state=self.engine.store.read() state['manifestSha256']=c.digest(self.manifest) self.engine.store.write(state) self.engine.runner=ReportRunner() self.engine.approval_verifier=lambda request: request['kind']!='candidate-specific-cleanup-scope' self.engine.run() self.assertEqual(1,self.engine.runner.calls) self.assertEqual('candidate-specific-cleanup-scope', self.engine.store.read()['stages']['PHASE-10']['awaiting'][0]['kind']) self.engine.approval_verifier=lambda request: True self.engine.run() self.assertEqual(1,self.engine.runner.calls) self.assertIn('assessment',self.engine.store.read()['latestFailure']) def test_policy_cannot_relabel_p10_01_as_assessment(self): self.manifest['executionScope']={**scope(),'checkpoint':'P10-01'} cp=self.manifest['stages'][0]['checkpoints'][0] cp['id']='P10-01' cp['approvalsAfter']=['candidate-specific-cleanup-scope'] with self.assertRaisesRegex(c.GateError,'protected stop'): c.validate_manifest(self.manifest) def test_evidence_inspection_cannot_be_marked_runtime_qualified(self): self.manifest['executionScope']=scope() cp=self.manifest['stages'][0]['checkpoints'][0] cp['approvalsAfter']=['candidate-specific-cleanup-scope'] cp['checks'][0].update(adapter='assessment-evidence-inspection',configured=True,receiptSha256='a'*64) with self.assertRaisesRegex(c.GateError,'cannot qualify'): c.validate_manifest(self.manifest) def test_inspection_requires_explicit_false_and_never_launches_worker(self): self.manifest['executionScope']=scope() cp=self.manifest['stages'][0]['checkpoints'][0] cp['approvalsAfter']=['candidate-specific-cleanup-scope'] cp['checks'][0].update(adapter='assessment-evidence-inspection',receiptSha256='a'*64) cp['checks'][0].pop('configured',None) with self.assertRaisesRegex(c.GateError,'cannot qualify'): c.validate_manifest(self.manifest) state=self.engine.store.read() with self.assertRaisesRegex(c.GateError,'adapters must be configured'): self.engine.checkpoint(self.manifest['stages'][0],cp,state['stages']['PHASE-10'],state) self.assertEqual(0,self.runner.calls) def test_native_isolation_failure_stops_before_additional_probe(self): import qualify_assessment policy=copy.deepcopy(self.manifest) policy['executionEnabled']=False c.atomic(self.control/'backlog.json',policy) c.atomic(self.control/'proposed-p10-assessment.json',policy) c.atomic(self.control/'config.json',self.config) report=self.root/'failed-native.json' with mock.patch.object(qualify_assessment.local_mode,'qualify',return_value={'status':'FAIL'}), \ mock.patch.object(sys,'argv',['qualify',str(self.control),str(self.repo),str(report)]): with self.assertRaisesRegex(c.GateError,'no further child'): qualify_assessment.main() self.assertEqual('NOT_RUN',c.read_json(report)['assessment']) self.assertFalse((self.repo/'.artifacts').exists()) def test_entry_cannot_prepare_approvals_without_composite_adapter(self): self.manifest['executionScope']={**scope(),'handoffCommit':self.manifest['baselineCommit'],'handoffBranch':'codex/base'} cp=self.manifest['stages'][0]['checkpoints'][0] cp['approvalsAfter']=['candidate-specific-cleanup-scope'] cp['checks']=[] c.atomic(self.control/'backlog.json',self.manifest) c.atomic(self.control/'config.json',self.config) with mock.patch.object(c,'Engine',return_value=self.engine), \ mock.patch.object(local_mode,'require_unattended'): with self.assertRaisesRegex(c.GateError,'composite adapter is not qualified'): c.main(['prepare-entry','--control-root',str(self.control)]) self.assertEqual(0,self.runner.calls) self.assertFalse((self.control/'requests').exists()) class ReadOnlyProfileTests(unittest.TestCase): def test_assessment_profile_keeps_product_read_only(self): with tempfile.TemporaryDirectory() as directory: root=Path(directory) config=dict(repository=str(root/'worker'), workerTemp=str(root/'temp'), workerCodeXHome=str(root/'control/auth'), normalCodeXHome=str(root/'normal'), assessmentReportDirectory='docs/automation/stages/PHASE-10') rules=local_mode.filesystem_rules(root/'control',config) self.assertEqual('read',rules[(root/'worker').resolve().as_posix()]) self.assertEqual('write',rules[(root/'worker/docs/automation/stages/PHASE-10').resolve().as_posix()]) class ReceiptTests(unittest.TestCase): def test_receipt_rejects_changed_product_or_changed_receipt(self): import assessment_adapter as adapter with tempfile.TemporaryDirectory() as directory: repo=Path(directory) (repo/'product.txt').write_text('accepted synthetic source') receipt=dict(schemaVersion=1, scope='P10-00-ASSESSMENT-ONLY', integrity='PASS', files={'product.txt':c.sha_file(repo/'product.txt')}, nativeReports=[dict(id='synthetic',status='PASS',exitCode=0, membershipSha256='a'*64)], retainedDecisions={'production':'NO-GO'}) path=repo/'receipt.json' c.atomic(path,receipt) pin=c.sha_file(path) self.assertEqual('REUSED_SOURCE_BOUND_EVIDENCE',adapter.verify(repo,path,pin)['kind']) (repo/'product.txt').write_text('changed') with self.assertRaises(ValueError): adapter.verify(repo,path,pin) with self.assertRaises(ValueError): adapter.verify(repo,path,'0'*64)