30dc3182ab
Record the annotated rebuild, bounded graphics matrices, private release-shape smoke, and remaining integration gates. Preserve STEP20C NEXT and production multi-output NOT YET SUPPORTED.
127 lines
7.4 KiB
Python
127 lines
7.4 KiB
Python
"""Offline fail-closed tests for the G smoke fixture; never launch compositor."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
HERE=Path(__file__).resolve().parent
|
|
sys.path[:0]=[str(HERE),str(HERE.parent)]
|
|
import adapter,gate
|
|
from horizontal_adapter import replace_one
|
|
|
|
class GOffline(unittest.TestCase):
|
|
def test_c2_transform_unique_and_compiles(self):
|
|
text=adapter.generate_g();compile(text,'G-smoke/runner.py','exec')
|
|
self.assertEqual(text.count(adapter.G_IMPORT),1)
|
|
self.assertIn("open_chain('E1')",text);self.assertIn('fresh_chain(',text)
|
|
self.assertIn('dismissal(log',text);self.assertIn('final_restoration(baseline,final)',text)
|
|
self.assertIn('BINARY.resolve() and sha(BINARY)==BINARY_SHA',text)
|
|
def test_transform_rejects_missing_and_duplicate(self):
|
|
for source in ('missing','x x'):
|
|
with self.assertRaises(ValueError):replace_one(source,'x','y')
|
|
def test_old_A_and_non_G_ELF_rejected(self):
|
|
old_a=gate.LAB/'stage-static/usr/bin/swayfx-mosaic'
|
|
with self.assertRaises(ValueError):gate.validate_binary_identity(old_a,'979dcd95ab43fa1e465d6e9d262485b73d186cedacd0f957059dd24f482ce6f4')
|
|
with self.assertRaises(ValueError):gate.validate_binary_identity('/home/fy59/swayfx-mosaic-overnight-mcolx0gq/matrix/D4-mosaic/stage/usr/bin/swayfx-mosaic','0'*64)
|
|
with self.assertRaises(ValueError):gate.validate_binary_identity('/usr/bin/sway','0'*64)
|
|
def test_old_D_or_other_source_set_rejected(self):
|
|
rows=dict(gate.TRACE_SOURCES);rows[next(iter(rows))]='0'*64
|
|
with self.assertRaises(ValueError):gate.source_bindings({'trace_sources':rows})
|
|
def test_wrong_wlroots_and_mutated_freeze_rejected(self):
|
|
with self.assertRaises(ValueError):gate.validate_wlroots(gate.WLROOTS,'0'*64)
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p=Path(d)/'input';p.write_bytes(b'changed')
|
|
with self.assertRaises(ValueError):gate.validate_file_map({str(p):hashlib.sha256(b'original').hexdigest()})
|
|
def test_quota_and_no_retry_rejected(self):
|
|
state={'limits':{'scene_build':2,'mosaic_build':2,'stage':2,'run':18,'capture':220},
|
|
'campaign_limits':{'C':3,'D1':2,'D2':2,'D3':2,'D4':2,'E1':1,'E2':1,'E3':2,'F':2,'G':1},
|
|
'used':dict.fromkeys(('scene_build','mosaic_build','stage','run','capture'),0),
|
|
'campaign_used':dict.fromkeys(('C','D1','D2','D3','D4','E1','E2','E3','F','G'),0),
|
|
'runs':{},'events':[]}
|
|
state['used']['capture']=204
|
|
with patch('ledger._validate'):
|
|
with self.assertRaises(ValueError):gate.validate_budget_room(state)
|
|
with tempfile.TemporaryDirectory() as d:
|
|
work=Path(d);(work/'consumed.json').write_text('{}')
|
|
with self.assertRaises(ValueError):gate.validate_one_shot(work)
|
|
def test_incomplete_frame_and_window_evidence_rejected(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
w=Path(d);(w/'results').mkdir();(w/'run').mkdir()
|
|
with self.assertRaises(ValueError):gate.validate_capture_batch(w,[])
|
|
with self.assertRaises(ValueError):gate.validate_window_results(w)
|
|
def test_real_c2_events_satisfy_finish_predicate_and_mutations_fail(self):
|
|
c2=gate.LAB/'horizontal/C2'
|
|
evidence=(c2/'run/events.jsonl',c2/'actions.json',c2/'results/captures.json',
|
|
c2/'results/clicks.json',c2/'results/runtime.json')
|
|
for path in evidence:
|
|
self.assertTrue(path.is_file(),str(path))
|
|
self.assertRegex(gate.sha(path),r'^[0-9a-f]{64}$')
|
|
if (gate.WORK/'freeze.json').is_file():
|
|
frozen=gate.expected_files()
|
|
for path in evidence:self.assertEqual(frozen[str(path)],gate.sha(path),str(path))
|
|
events=[json.loads(line) for line in evidence[0].read_text().splitlines()]
|
|
state=json.loads(evidence[1].read_text())
|
|
captures=json.loads(evidence[2].read_text())
|
|
clicks=json.loads(evidence[3].read_text())
|
|
runtime=json.loads(evidence[4].read_text())
|
|
gate.validate_capture_batch(c2,captures)
|
|
gate.validate_window_results(c2)
|
|
valid=gate.finish_predicate(state,events,captures,clicks,runtime,True,True)
|
|
self.assertTrue(valid['complete'],valid)
|
|
self.assertEqual((valid['injections'],valid['click_injections'],
|
|
valid['standalone_move_injections'],valid['grabs']),(10,8,2,4))
|
|
injections=[i for i,e in enumerate(events) if e.get('kind')=='injection']
|
|
self.assertEqual(len(injections),10)
|
|
missing=events[:injections[-1]]+events[injections[-1]+1:]
|
|
extra=events+[dict(events[injections[0]])]
|
|
wrong_action=[dict(e) for e in events]
|
|
wrong_action[injections[1]]['action']='move'
|
|
unknown_action=[dict(e) for e in events]
|
|
unknown_action[injections[1]]['action']='unknown'
|
|
for mutation in (missing,extra,wrong_action,unknown_action):
|
|
with self.subTest(mutation=len(mutation)):
|
|
self.assertFalse(gate.finish_predicate(
|
|
state,mutation,captures,clicks,runtime,True,True)['complete'])
|
|
def test_preflight_refuses_absent_or_rejected_review_without_side_effects(self):
|
|
if not (gate.WORK/'freeze.json').is_file():
|
|
self.skipTest('review-gated preflight is exercised after the fresh freeze exists')
|
|
real_root=gate.GROOT
|
|
with tempfile.TemporaryDirectory() as d:
|
|
review_dir=Path(d)/'reviews';review_dir.mkdir()
|
|
absent=review_dir/'release-smoke-review.json'
|
|
rejected=review_dir/'rejected.json'
|
|
rejected.write_text(json.dumps({'status':'CORRECTION_REQUIRED'}))
|
|
|
|
class RootWithIsolatedReview:
|
|
def __truediv__(self, other):
|
|
if str(other)=='reviews/release-smoke-review.json':
|
|
return absent
|
|
return real_root/other
|
|
|
|
budget=(gate.LAB/'budget.json').read_bytes()
|
|
for path,expected in ((absent,'independent G smoke byte review missing'),
|
|
(rejected,'G smoke review scope/status')):
|
|
if path==absent:
|
|
self.assertFalse(absent.exists())
|
|
else:
|
|
rejected.rename(absent)
|
|
with patch.object(gate,'GROOT',RootWithIsolatedReview()):
|
|
with self.assertRaisesRegex(ValueError,expected):
|
|
gate.preflight(require_review=True)
|
|
self.assertEqual((gate.LAB/'budget.json').read_bytes(),budget)
|
|
self.assertFalse((gate.WORK/'consumed.json').exists())
|
|
self.assertFalse((gate.WORK/'run').exists())
|
|
|
|
if __name__=='__main__':
|
|
p=argparse.ArgumentParser();p.add_argument('--output',type=Path,required=True);args=p.parse_args()
|
|
result=unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(GOffline))
|
|
data={'status':'PASS' if result.wasSuccessful() else 'FAIL','executed':result.testsRun,
|
|
'skipped':[(str(t),reason) for t,reason in result.skipped],
|
|
'failures':[(str(t),s) for t,s in result.failures], 'errors':[(str(t),s) for t,s in result.errors],
|
|
'runtime':'NOT RUN','build':'NOT RUN','stage':'NOT RUN','review_acceptance_created':False}
|
|
args.output.write_text(json.dumps(data,indent=2)+'\n')
|
|
raise SystemExit(not result.wasSuccessful())
|