Separate radio defence and harden player and objective lifecycles
This commit is contained in:
@@ -0,0 +1,171 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Synthetic comparator regression only; no measured performance evidence."""
|
||||
import copy
|
||||
import csv
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
spec = importlib.util.spec_from_file_location('compare_performance', ROOT / 'tools/compare_performance.py')
|
||||
comparison = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(comparison)
|
||||
|
||||
|
||||
class PerformanceComparisonTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.directory = tempfile.TemporaryDirectory(prefix='erri-synthetic-performance-')
|
||||
self.addCleanup(self.directory.cleanup)
|
||||
self.root = Path(self.directory.name)
|
||||
template = {
|
||||
'variant': 'baseline', 'repeat_id': 'r1', 'artifact_id': 'synthetic-baseline', 'addon_ids': ['synthetic-base'],
|
||||
'engine_version': 'synthetic-engine',
|
||||
'hardware': {'cpu': 'synthetic-cpu', 'gpu': 'synthetic-gpu', 'ram_gib': 32, 'os': 'synthetic-os', 'driver': 'synthetic-driver'},
|
||||
'workload': {'map': 'synthetic-map', 'route': 'synthetic-route', 'players': 12, 'ai': 36, 'vehicles': 4, 'entities': 100,
|
||||
'settings': {'resolution': [1920, 1080], 'fps_limit': 0, 'vsync': False}, 'warmup_seconds': 30, 'duration_seconds': 60},
|
||||
'capture': {'kind': 'client_frame', 'environment': 'client', 'tool': 'synthetic-generator', 'version': '1'},
|
||||
'sample_file': 'unused.csv',
|
||||
}
|
||||
self.data = {'schema_version': 1, 'sample_origin': 'synthetic', 'runs': []}
|
||||
for repeat in range(1, 4):
|
||||
for variant, frame in [('baseline', 20), ('successor', 10)]:
|
||||
run = copy.deepcopy(template)
|
||||
run.update(variant=variant, repeat_id=f'r{repeat}', artifact_id=f'synthetic-{variant}', sample_file=f'{variant}-{repeat}.csv')
|
||||
run['addon_ids'] = [f'synthetic-{variant}']
|
||||
self.data['runs'].append(run)
|
||||
with (self.root / run['sample_file']).open('w', newline='') as stream:
|
||||
writer = csv.writer(stream)
|
||||
writer.writerow(comparison.COLUMNS)
|
||||
for index in range(1, 90000 // frame + 1):
|
||||
writer.writerow([index, index * frame, frame, 12, 36, 4, 100, 2])
|
||||
|
||||
def execute(self):
|
||||
manifest = self.root / 'synthetic.json'
|
||||
manifest.write_text(json.dumps(self.data))
|
||||
return comparison.compare(manifest)
|
||||
|
||||
def change_csv(self, transform):
|
||||
path = self.root / self.data['runs'][0]['sample_file']
|
||||
rows = path.read_text().splitlines()
|
||||
path.write_text('\n'.join(transform(rows)) + '\n')
|
||||
|
||||
def replace_frames(self, frames):
|
||||
path = self.root / self.data['runs'][0]['sample_file']
|
||||
with path.open('w', newline='') as stream:
|
||||
writer = csv.writer(stream)
|
||||
writer.writerow(comparison.COLUMNS)
|
||||
elapsed = 0
|
||||
for index, frame in enumerate(frames, 1):
|
||||
elapsed += frame
|
||||
writer.writerow([index, elapsed, frame, 12, 36, 4, 100, 2])
|
||||
|
||||
def test_paired_arithmetic_is_explicitly_synthetic_and_not_acceptance(self):
|
||||
report = self.execute()
|
||||
self.assertEqual(report['sample_origin'], 'synthetic')
|
||||
self.assertEqual(report['acceptance'], 'not_evaluated')
|
||||
self.assertEqual(report['aggregate']['baseline']['median_fps'], 50)
|
||||
self.assertEqual(report['aggregate']['successor']['p99_frame_ms'], 10)
|
||||
self.assertEqual(report['median_paired_deltas']['median_frame_reduction_percent'], 50)
|
||||
self.assertEqual(report['runs'][0]['samples_measured'], 3000)
|
||||
self.assertEqual(len(report['manifest_sha256']), 64)
|
||||
|
||||
def test_percentiles_and_warmup_exclusion(self):
|
||||
self.assertEqual(comparison.percentile(list(range(1, 101)), .99), 99)
|
||||
self.change_csv(lambda rows: [rows[0]] + [row.rsplit(',', 1)[0] + ',999' if i < 1500 else row for i, row in enumerate(rows[1:])])
|
||||
self.assertEqual(self.execute()['aggregate']['baseline']['median_ai_simulation_ms'], 2)
|
||||
|
||||
def test_rejects_mismatched_conditions(self):
|
||||
original = copy.deepcopy(self.data)
|
||||
mutations = [('hardware', 'cpu', 'other'), ('workload', 'ai', 35), ('workload', 'route', 'other'),
|
||||
('workload', 'settings', {'fps_limit': 30}), ('workload', 'warmup_seconds', 31), ('capture', 'environment', 'workbench_headless')]
|
||||
for group, key, value in mutations:
|
||||
with self.subTest(group=group, key=key):
|
||||
self.data = copy.deepcopy(original)
|
||||
self.data['runs'][1][group][key] = value
|
||||
with self.assertRaisesRegex(ValueError, 'mismatched'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_missing_pair_and_duplicate_repeat(self):
|
||||
self.data['runs'][-1]['repeat_id'] = 'unpaired'
|
||||
with self.assertRaisesRegex(ValueError, 'Every repeat'):
|
||||
self.execute()
|
||||
self.data['runs'][-1]['repeat_id'] = 'r1'
|
||||
with self.assertRaisesRegex(ValueError, 'Duplicate repeat'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_reused_samples(self):
|
||||
self.data['runs'][1]['sample_file'] = self.data['runs'][0]['sample_file']
|
||||
with self.assertRaisesRegex(ValueError, 'reused'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_incomplete_capture(self):
|
||||
self.change_csv(lambda rows: rows[:-1])
|
||||
with self.assertRaisesRegex(ValueError, 'incomplete recording'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_final_stall_hiding_most_of_measurement(self):
|
||||
# Before the coverage check this 1.2-second sample was accepted as a
|
||||
# 60-second measurement with 100 FPS, dropping its 60-second stall.
|
||||
self.replace_frames([20] * 1500 + [10] * 120 + [60000])
|
||||
with self.assertRaisesRegex(ValueError, 'only 2.000%.*boundary stalls'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_warmup_crossing_stall_hiding_measurement(self):
|
||||
# The long frame starts in warm-up and consumes 58.8 measured seconds.
|
||||
self.replace_frames([20] * 1499 + [58820] + [10] * 120)
|
||||
with self.assertRaisesRegex(ValueError, 'only 2.000%.*boundary stalls'):
|
||||
self.execute()
|
||||
|
||||
def test_short_boundary_crossings_are_visible_with_high_coverage(self):
|
||||
self.replace_frames([17] * 5295)
|
||||
run = self.execute()['runs'][0]
|
||||
self.assertGreaterEqual(run['measurement_coverage_fraction'], .99)
|
||||
self.assertLess(run['measurement_coverage_fraction'], 1)
|
||||
self.assertEqual(len(run['boundary_frames_excluded']), 2)
|
||||
overlap = sum(frame['measurement_overlap_ms'] for frame in run['boundary_frames_excluded'])
|
||||
self.assertAlmostEqual(run['measured_frame_seconds'] * 1000 + overlap, 60000)
|
||||
|
||||
def test_rejects_coverage_just_below_threshold(self):
|
||||
self.replace_frames([20] * 1500 + [10] * 5939 + [620])
|
||||
with self.assertRaisesRegex(ValueError, 'only 98.983%'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_missing_or_duplicate_frames(self):
|
||||
self.change_csv(lambda rows: rows[:100] + rows[101:])
|
||||
with self.assertRaisesRegex(ValueError, 'frame index'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_population_drift(self):
|
||||
self.change_csv(lambda rows: rows[:1] + [rows[1].replace(',12,36,', ',12,35,')] + rows[2:])
|
||||
with self.assertRaisesRegex(ValueError, 'ai workload changed'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_nonfinite_sample(self):
|
||||
self.change_csv(lambda rows: rows[:1] + [rows[1].rsplit(',', 1)[0] + ',nan'] + rows[2:])
|
||||
with self.assertRaisesRegex(ValueError, 'invalid frame/time/load'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_extra_columns_and_timing_gaps(self):
|
||||
original = (self.root / self.data['runs'][0]['sample_file']).read_text()
|
||||
self.change_csv(lambda rows: [rows[0], rows[1] + ',extra'] + rows[2:])
|
||||
with self.assertRaisesRegex(ValueError, 'columns'):
|
||||
self.execute()
|
||||
(self.root / self.data['runs'][0]['sample_file']).write_text(original)
|
||||
self.change_csv(lambda rows: [rows[0], rows[1].replace('1,20,20,', '1,30,20,')] + rows[2:])
|
||||
with self.assertRaisesRegex(ValueError, 'elapsed delta'):
|
||||
self.execute()
|
||||
|
||||
def test_rejects_incomplete_metadata_and_infinite_json(self):
|
||||
del self.data['runs'][0]['capture']['version']
|
||||
with self.assertRaisesRegex(ValueError, 'expected fields'):
|
||||
self.execute()
|
||||
self.data['runs'][0]['capture']['version'] = '1'
|
||||
self.data['runs'][0]['workload']['settings']['bad'] = float('inf')
|
||||
with self.assertRaisesRegex(ValueError, 'Non-finite'):
|
||||
self.execute()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Execute source-local admission bodies and the reservation rejection predicate.
|
||||
|
||||
Only vector arithmetic and opaque UUID equality/nullness are substituted. This is
|
||||
not a character collision, controller, callback, reconnect or native engine test.
|
||||
"""
|
||||
import re
|
||||
from policy_harness import ROOT, body, run
|
||||
base = ROOT / 'addons/core/Scripts/Game/ERR'
|
||||
rules = body(base/'ERR_PlayerRoster.c', 'static bool CanEnter(')
|
||||
rules += '\n' + body(base/'ERR_SpawnFinalization.c', 'static bool Allows(')
|
||||
rules += '\n' + body(base/'ERR_SpawnFinalization.c', 'static bool PositionClose(')
|
||||
rules = rules.replace('ERR_Phase.', 'ERR_Phase::').replace('ERR_PlayerLifeRules.', '')
|
||||
rules = rules.replace('vector.', 'vector::').replace('Math.', 'Math::')
|
||||
# Extract the actual condition, rather than maintaining a second implementation.
|
||||
# Native geometry/terrain branches of Clear are deliberately not emulated.
|
||||
clear = body(base/'ERR_SpawnPlacement.c', 'bool Clear(')
|
||||
match = re.search(r'if \((id != ownReservation[^\n]+)\) return false;', clear)
|
||||
assert match, 'Reservation policy moved: update the source-local extraction'
|
||||
predicate = match[1].replace('vector.', 'vector::').replace('Math.', 'Math::')
|
||||
rules += '\nstatic bool ReservationAllows(int id, int ownReservation, vector position, vector reserved) { return !(' + predicate + '); }\n'
|
||||
run('''#include <cassert>
|
||||
#include <cmath>
|
||||
#include <initializer_list>
|
||||
namespace ERR_Phase {enum {PREPARING,PRISON,RADIO_SEARCH,RADIO_DEFENCE,EXTRACTION,VICTORY,DEFEAT};}
|
||||
struct UUID {
|
||||
int identity;
|
||||
bool IsNull() const { return identity==0; }
|
||||
bool operator==(UUID other) const { return identity==other.identity; }
|
||||
};
|
||||
struct vector {
|
||||
float x,y,z;
|
||||
float operator[](int axis) const { return axis==0?x:(axis==1?y:z); }
|
||||
static float DistanceXZ(vector a, vector b) {return std::hypot(a.x-b.x,a.z-b.z);}
|
||||
};
|
||||
namespace Math {static float AbsFloat(float value) {return std::abs(value);}}
|
||||
''' + rules + '''
|
||||
int main() {
|
||||
// Complete independent authorization matrix, including invalid phases and null IDs.
|
||||
for (int phase=-1;phase<=7;++phase)
|
||||
for (int flags=0;flags<32;++flags)
|
||||
for (int oldId=0;oldId<=2;++oldId)
|
||||
for (int currentId=0;currentId<=2;++currentId) {
|
||||
bool connected=flags&1,usedLife=flags&2,sameController=flags&4;
|
||||
bool sameRequest=flags&8,noControlledEntity=flags&16;
|
||||
bool expected=phase>=1&&phase<=3&&connected&&!usedLife&&oldId>0&&oldId==currentId&&sameController&&sameRequest&&noControlledEntity;
|
||||
assert(Allows(phase,connected,usedLife,UUID{oldId},UUID{currentId},sameController,sameRequest,noControlledEntity)==expected);
|
||||
}
|
||||
const vector origin={0,0,0};
|
||||
// Inclusive vertical tolerance and just-inside/outside horizontal boundary.
|
||||
for (float y : {-1.001f,-1.f,-0.2f,0.f,0.2f,1.f,1.001f}) {
|
||||
assert(PositionClose(vector{0,y,0},origin)==(std::abs(y)<=1));
|
||||
assert(PositionClose(vector{.599f,y,0},origin)==(std::abs(y)<=1));
|
||||
assert(!PositionClose(vector{.601f,y,0},origin));
|
||||
}
|
||||
assert(PositionClose(vector{.3f,.2f,.3f},origin));
|
||||
assert(!PositionClose(vector{.5f,0,.5f},origin)); // radial, not independent axis allowance
|
||||
assert(!PositionClose(vector{0,-20,0},origin)); // fallen below unchanged XZ
|
||||
assert(!PositionClose(vector{0,20,0},origin)); // displaced above unchanged XZ
|
||||
// Connection IDs1 and13 must conflict identically; only the caller's own slot is exempt.
|
||||
for (int id : {1,13,99})
|
||||
for (int owner : {-1,1,13,99})
|
||||
for (float x : {-1.201f,-1.199f,0.f,1.199f,1.201f})
|
||||
for (float y : {-2.001f,-2.f,-1.999f,0.f,1.999f,2.f,2.001f}) {
|
||||
bool expected=id==owner || std::abs(x)>=1.2 || std::abs(y)>=2;
|
||||
assert(ReservationAllows(id,owner,origin,vector{x,y,0})==expected);
|
||||
}
|
||||
assert(!ReservationAllows(1,13,origin,vector{.8f,0,.8f}));
|
||||
assert(ReservationAllows(1,13,origin,vector{.9f,0,.9f}));
|
||||
assert(ReservationAllows(1,13,origin,vector{1.5f,0,0})); // authored grid spacing
|
||||
// Own reservation is excluded while finalization still checks native geometry separately.
|
||||
assert(ReservationAllows(13,13,origin,origin));
|
||||
assert(!ReservationAllows(13,1,origin,origin));
|
||||
}
|
||||
''')
|
||||
print('PASS: player admission phase/life/identity/controller matrix, preload displacement limits, pending reservation overlap and owner exclusion')
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Exercise actual radio tuning, defender and continuity policy bodies.
|
||||
Native fixture covers service events and physical character state separately.
|
||||
"""
|
||||
from policy_harness import ROOT, body, run
|
||||
source=ROOT/'addons/core/Scripts/Game/ERR/ERR_RadioDefenceService.c'
|
||||
functions='\n'.join(body(source, signature) for signature in (
|
||||
'static float DefenceRadius(', 'static float TuneRadius(',
|
||||
'static bool CanTune(', 'static bool Defender(', 'static float SinceAfterSample(',
|
||||
)).replace('ERR_Phase.', 'ERR_Phase::')
|
||||
functions+='\n'+body(ROOT/'addons/core/Scripts/Game/ERR/ERR_MissionRules.c','static bool DefenceComplete(')
|
||||
run('''#include <cassert>
|
||||
#include <initializer_list>
|
||||
namespace ERR_Phase {enum {PREPARING,PRISON,RADIO_SEARCH,RADIO_DEFENCE,EXTRACTION,VICTORY,DEFEAT};}
|
||||
'''+functions+'''
|
||||
int main() {
|
||||
for (int phase=-1;phase<=7;++phase)
|
||||
for (int flags=0;flags<16;++flags)
|
||||
for (int knowledge=0;knowledge<=2;++knowledge)
|
||||
for (float distance : {-1.f,0.f,4.999f,5.f,5.001f,60.f}) {
|
||||
bool registered=flags&1,radio=flags&2,player=flags&4,conscious=flags&8;
|
||||
bool expected=phase==ERR_Phase::RADIO_SEARCH && registered && radio && knowledge>=2 && player && conscious && distance>=0 && distance<=5;
|
||||
assert(CanTune(phase,registered,radio,knowledge,player,conscious,distance)==expected);
|
||||
}
|
||||
for (int flags=0;flags<4;++flags)
|
||||
for (float distance : {-1.f,0.f,59.999f,60.f,60.001f})
|
||||
assert(Defender(flags&1,flags&2,distance)==(bool(flags&1)&&bool(flags&2)&&distance>=0&&distance<60));
|
||||
assert(DefenceRadius()==60 && TuneRadius()==5);
|
||||
float since=SinceAfterSample(-1,-1,1000,2);
|
||||
assert(since==1000);
|
||||
since=SinceAfterSample(since,1000,6000,1);
|
||||
assert(since==1000); // second person can leave without resetting remaining defender
|
||||
since=SinceAfterSample(since,6000,7000,0);
|
||||
assert(since==-1); // absent/dead/unconscious/empty-server samples reset
|
||||
since=SinceAfterSample(since,7000,8000,1);
|
||||
assert(since==8000);
|
||||
assert(!DefenceComplete(37999,since,30000,true));
|
||||
assert(DefenceComplete(38000,since,30000,true));
|
||||
assert(!DefenceComplete(38000,since,45000,true)); // live setting increased
|
||||
assert(SinceAfterSample(since,38000,30000,1)==30000); // backward clock resets interval
|
||||
assert(SinceAfterSample(since,38000,-1,1)==-1);
|
||||
assert(SinceAfterSample(since,38000,50000,-1)==-1);
|
||||
}
|
||||
''')
|
||||
print('PASS: radio authorization/discovery/phase matrix, exact distance limits, defender handover, interruption, empty samples, clock regression and duration boundary')
|
||||
@@ -0,0 +1,409 @@
|
||||
"""Release-gate bypass regressions. Synthetic evidence is confined to temp dirs."""
|
||||
from pathlib import Path
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / 'tools'))
|
||||
import release_check as gate
|
||||
from native_report import analyze
|
||||
from source_manifest import addon_manifest
|
||||
from native_provenance import snapshot, bindings, persist, seal_run, archive_source, qualify_archive, MANIFEST, METADATA, PROVENANCE
|
||||
|
||||
|
||||
class ReleaseAuditTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory(prefix='erri-release-test-')
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
self.write('addons/core/addon.gproj', 'Project {}')
|
||||
self.write('addons/core/Scripts/Game/Test.c', 'class Test {}')
|
||||
self.write('addons/core/Assets/test.png', b'nonempty resource bytes')
|
||||
self.write('docs/evidence/current/manual.txt', 'Synthetic test fixture, not real acceptance.\n')
|
||||
self.log = 'docs/evidence/current/console.log.txt'
|
||||
self.report = 'docs/evidence/current/report.json'
|
||||
self.manifest = 'docs/evidence/current/source-sha256.json'
|
||||
self.write(self.log, 'ERR SMOKE PASS: synthetic\nERR SMOKE COMPLETE: synthetic\n')
|
||||
self.refresh_native()
|
||||
ordinary = {'status': 'passed', 'evidence': ['docs/evidence/current/manual.txt']}
|
||||
native = {'status': 'passed', 'evidence': [self.report, self.manifest],
|
||||
'source_manifest': self.manifest, 'native_report': self.report}
|
||||
self.ledger = {'target': 'Everon, vanilla factions, 12 players',
|
||||
'cases': {name: copy.deepcopy(ordinary) for name in gate.REQUIRED_CASES},
|
||||
'development_checks': {'clean_native_runtime': copy.deepcopy(native)}}
|
||||
self.ledger['cases']['independent_native_build'] = copy.deepcopy(native)
|
||||
|
||||
def write(self, relative, value):
|
||||
path = self.root / relative
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(value if isinstance(value, bytes) else value.encode())
|
||||
|
||||
def json(self, relative, value):
|
||||
self.write(relative, json.dumps(value))
|
||||
|
||||
def refresh_native(self):
|
||||
captured = snapshot(self.root, ['synthetic-runner-not-an-engine'])
|
||||
archive = Path(self.manifest).parent
|
||||
self.write(self.manifest, captured[0])
|
||||
self.write(str(archive / MANIFEST), captured[0])
|
||||
self.write(str(archive / METADATA), captured[1])
|
||||
self.json(str(archive / PROVENANCE), {'schema_version': 1, 'kind': 'launch_snapshot'})
|
||||
self.json(str(archive / 'runner.json.txt'), {**bindings(captured), 'started_unix': json.loads(captured[1])['captured_unix'] + 1, 'result': 'smoke-completed', 'rounds_required': 1, 'stopped_unix': json.loads(captured[1])['captured_unix'] + 2, 'source_unchanged_after_run': True, 'end_source_sha256': hashlib.sha256(captured[0]).hexdigest(), 'console_sha256': hashlib.sha256((self.root / self.log).read_bytes()).hexdigest()})
|
||||
self.json(self.report, analyze((self.root / self.log).read_text(), self.log))
|
||||
|
||||
def run_audit(self):
|
||||
self.json('docs/acceptance.json', self.ledger)
|
||||
return gate.audit(self.root)
|
||||
|
||||
def blocked(self, fragment):
|
||||
errors = self.run_audit()
|
||||
self.assertTrue(errors, 'malformed evidence incorrectly qualified')
|
||||
self.assertTrue(any(fragment in error for error in errors), errors)
|
||||
|
||||
def test_complete_synthetic_fixture_qualifies(self):
|
||||
self.assertEqual(self.run_audit(), [])
|
||||
|
||||
def test_failed_clean_runtime_blocks_even_when_all_cases_pass(self):
|
||||
self.ledger['development_checks']['clean_native_runtime']['status'] = 'failed'
|
||||
self.blocked('clean_native_runtime: status is failed')
|
||||
|
||||
def test_every_required_case_is_mandatory(self):
|
||||
for name in gate.REQUIRED_CASES:
|
||||
with self.subTest(name=name):
|
||||
saved = self.ledger['cases'].pop(name)
|
||||
self.blocked(name + ': required first-release case is missing')
|
||||
self.ledger['cases'][name] = saved
|
||||
|
||||
def test_unknown_case_rejected(self):
|
||||
self.ledger['cases']['fake_replacement'] = {'status': 'passed', 'evidence': [self.log]}
|
||||
self.blocked('unknown first-release case')
|
||||
|
||||
def test_missing_runtime_gate_rejected(self):
|
||||
self.ledger['development_checks'].clear()
|
||||
self.blocked('required clean-runtime gate is missing')
|
||||
|
||||
def test_unknown_and_malformed_statuses_rejected_everywhere(self):
|
||||
entry = self.ledger['cases']['escape_loop']
|
||||
for status in ('PASS', 'waived', '', None, True, {}, []):
|
||||
with self.subTest(status=status):
|
||||
entry['status'] = status
|
||||
self.blocked('unknown or malformed status')
|
||||
entry['status'] = 'passed'
|
||||
self.ledger['development_checks']['historical'] = {'status': 'waived', 'evidence': []}
|
||||
self.blocked('unknown or malformed status')
|
||||
|
||||
def test_malformed_sections_entries_and_evidence(self):
|
||||
original = copy.deepcopy(self.ledger)
|
||||
for section in ('cases', 'development_checks'):
|
||||
self.ledger = copy.deepcopy(original)
|
||||
self.ledger[section] = []
|
||||
self.blocked(section + ' must be an object')
|
||||
for entry in (None, [], 'passed'):
|
||||
self.ledger = copy.deepcopy(original)
|
||||
self.ledger['cases']['escape_loop'] = entry
|
||||
self.blocked('entry must be an object')
|
||||
for evidence in (None, self.log, {}, [None], [42]):
|
||||
self.ledger = copy.deepcopy(original)
|
||||
self.ledger['cases']['escape_loop']['evidence'] = evidence
|
||||
self.blocked('evidence')
|
||||
|
||||
def test_passed_requires_nonempty_evidence(self):
|
||||
self.ledger['cases']['escape_loop']['evidence'] = []
|
||||
self.blocked('requires nonempty evidence')
|
||||
|
||||
def test_empty_files_and_whitespace_cannot_qualify(self):
|
||||
for value in (b'', b' \r\n\t'):
|
||||
self.write('docs/evidence/empty.txt', value)
|
||||
self.ledger['cases']['escape_loop']['evidence'] = ['docs/evidence/empty.txt']
|
||||
self.blocked('evidence is empty')
|
||||
|
||||
def test_missing_and_directory_evidence_rejected(self):
|
||||
for path in ('docs/evidence/missing.txt', 'docs/evidence/current'):
|
||||
self.ledger['cases']['escape_loop']['evidence'] = [path]
|
||||
self.blocked('not an existing regular file')
|
||||
|
||||
def test_unsafe_paths_rejected(self):
|
||||
for path in ('/etc/passwd', '../outside', 'docs/../docs/evidence/current/manual.txt',
|
||||
'./docs/evidence/current/manual.txt', 'docs//evidence/current/manual.txt',
|
||||
'C:\\evidence.txt', '.git/config', 'docs/evidence/\x00invalid'):
|
||||
with self.subTest(path=path):
|
||||
self.ledger['cases']['escape_loop']['evidence'] = [path]
|
||||
self.assertTrue(self.run_audit())
|
||||
|
||||
def test_symlinks_even_inside_repository_rejected(self):
|
||||
for target in (self.root / self.log, Path('/etc/passwd')):
|
||||
link = self.root / 'docs/evidence/link.txt'
|
||||
link.symlink_to(target)
|
||||
self.ledger['cases']['escape_loop']['evidence'] = ['docs/evidence/link.txt']
|
||||
self.assertTrue(self.run_audit())
|
||||
link.unlink()
|
||||
|
||||
def test_symlink_loop_returns_error_without_crash(self):
|
||||
link = self.root / 'docs/evidence/loop'
|
||||
link.symlink_to(link)
|
||||
self.ledger['cases']['escape_loop']['evidence'] = ['docs/evidence/loop']
|
||||
self.assertTrue(self.run_audit())
|
||||
|
||||
def test_duplicate_paths_rejected(self):
|
||||
self.ledger['cases']['escape_loop']['evidence'] = [self.log, self.log]
|
||||
self.blocked('duplicate evidence path')
|
||||
|
||||
def test_missing_explicit_native_fields_rejected(self):
|
||||
for field in ('source_manifest', 'native_report'):
|
||||
value = self.ledger['cases']['independent_native_build'].pop(field)
|
||||
self.blocked('independent_native_build.' + field)
|
||||
self.ledger['cases']['independent_native_build'][field] = value
|
||||
|
||||
def test_explicit_native_fields_must_be_listed_as_evidence(self):
|
||||
self.ledger['cases']['independent_native_build']['evidence'] = [self.log]
|
||||
self.blocked('must also appear')
|
||||
|
||||
def test_current_native_qualification_rejects_changed_added_deleted_files(self):
|
||||
path = self.root / 'addons/core/Scripts/Game/Test.c'
|
||||
original = path.read_bytes()
|
||||
path.write_text('class Changed {}')
|
||||
self.blocked('changed=1')
|
||||
path.write_bytes(original)
|
||||
added = self.root / 'addons/core/Assets/new.custom-resource'
|
||||
added.write_bytes(b'resource')
|
||||
self.blocked('missing=1')
|
||||
added.unlink()
|
||||
path.unlink()
|
||||
self.blocked('extra=1')
|
||||
|
||||
def test_clean_runtime_also_requires_current_manifest(self):
|
||||
self.ledger['cases']['independent_native_build']['status'] = 'pending'
|
||||
self.write('addons/core/Scripts/Game/Test.c', 'class Changed {}')
|
||||
self.blocked('clean_native_runtime.source_manifest: manifest does not match')
|
||||
|
||||
def test_incomplete_and_malformed_manifest_rejected(self):
|
||||
for manifest in ({}, [], {'addons/core/addon.gproj': 'not-sha256'}):
|
||||
self.json(self.manifest, manifest)
|
||||
self.blocked('manifest must be a nonempty')
|
||||
self.json(self.manifest, {'addons/core/addon.gproj': '0' * 64})
|
||||
self.blocked('manifest does not match')
|
||||
|
||||
def test_native_report_cannot_hide_raw_errors(self):
|
||||
self.write(self.log, 'ERR SMOKE PASS: synthetic\nWORLD (E): retained error\nERR SMOKE COMPLETE: synthetic\n')
|
||||
self.blocked('does not match analysis')
|
||||
self.refresh_native()
|
||||
self.blocked('no script, engine or end-screen errors')
|
||||
|
||||
def test_each_native_failure_category_blocks(self):
|
||||
for failure in ('ERR SMOKE FAIL: assertion', 'Virtual Machine Exception',
|
||||
'SCRIPT (E): exception', 'ANIMATION (E): unknown resource', 'Could not find expected endscreen',
|
||||
"entity won't take damage and wont replicate"):
|
||||
self.write(self.log, 'ERR SMOKE PASS: test\n' + failure + '\nERR SMOKE COMPLETE: test\n')
|
||||
self.refresh_native()
|
||||
self.blocked('native raw log must contain')
|
||||
|
||||
def test_truncated_and_assertion_free_logs_rejected(self):
|
||||
for text in ('ERR SMOKE PASS: test\n', 'ERR SMOKE COMPLETE: test\n', 'unrelated log\n'):
|
||||
self.write(self.log, text)
|
||||
self.refresh_native()
|
||||
self.blocked('successful completion/assertions')
|
||||
|
||||
def test_native_report_types_and_missing_fields_rejected(self):
|
||||
report = analyze((self.root / self.log).read_text(), self.log)
|
||||
for bad in ({}, [], dict(report, smoke_completions=True), dict(report, engine_errors='')):
|
||||
self.json(self.report, bad)
|
||||
self.blocked('native_report')
|
||||
|
||||
def test_native_artifacts_cannot_be_mixed_between_archives(self):
|
||||
self.write('docs/evidence/other/source-sha256.json', (self.root / self.manifest).read_bytes())
|
||||
entry = self.ledger['cases']['independent_native_build']
|
||||
entry['source_manifest'] = 'docs/evidence/other/source-sha256.json'
|
||||
entry['evidence'].append(entry['source_manifest'])
|
||||
self.blocked('same evidence archive')
|
||||
|
||||
def test_historical_development_manifests_need_not_match_current(self):
|
||||
self.json('docs/evidence/historical/source-sha256.json', {'old/path.c': '0' * 64})
|
||||
self.ledger['development_checks']['historical'] = {
|
||||
'status': 'passed', 'evidence': ['docs/evidence/historical/source-sha256.json']}
|
||||
self.ledger['development_checks']['expected_fault'] = {'status': 'failed', 'evidence': []}
|
||||
self.assertEqual(self.run_audit(), [])
|
||||
|
||||
def test_generated_state_excluded_but_new_resource_formats_included(self):
|
||||
self.write('addons/core/resourceDatabase.rdb', 'generated')
|
||||
self.write('addons/core/resources.pak', 'generated')
|
||||
self.write('addons/core/profiles/runtime.json', 'generated')
|
||||
self.assertEqual(self.run_audit(), [])
|
||||
self.write('addons/core/Scripts/Game/resourceDatabase.rdb_fake.c', 'class New {}')
|
||||
self.blocked('missing=1')
|
||||
|
||||
def test_source_symlink_cannot_hide_from_manifest(self):
|
||||
(self.root / 'addons/core/Scripts/Game/Alias.c').symlink_to(self.root / 'addons/core/Scripts/Game/Test.c')
|
||||
self.blocked('source/resource symlink is not supported')
|
||||
|
||||
def test_duplicate_json_keys_and_invalid_json_fail_closed(self):
|
||||
for raw in ('{"cases":{},"cases":{}}', '{', '[]', '{"target":NaN}'):
|
||||
self.write('docs/acceptance.json', raw)
|
||||
self.assertTrue(gate.audit(self.root))
|
||||
|
||||
def test_symlinked_ledger_rejected(self):
|
||||
self.json('docs/real-ledger.json', self.ledger)
|
||||
(self.root / 'docs/acceptance.json').symlink_to(self.root / 'docs/real-ledger.json')
|
||||
self.assertTrue(gate.audit(self.root))
|
||||
|
||||
def test_checker_is_read_only(self):
|
||||
self.json('docs/acceptance.json', self.ledger)
|
||||
before = {str(path.relative_to(self.root)): path.read_bytes()
|
||||
for path in self.root.rglob('*') if path.is_file()}
|
||||
self.assertEqual(gate.audit(self.root), [])
|
||||
after = {str(path.relative_to(self.root)): path.read_bytes()
|
||||
for path in self.root.rglob('*') if path.is_file()}
|
||||
self.assertEqual(after, before)
|
||||
|
||||
def test_archive_command_uses_shared_manifest_and_report(self):
|
||||
tools = self.root / 'tools'
|
||||
tools.mkdir()
|
||||
for name in ('archive_native.py', 'native_report.py', 'source_manifest.py', 'native_provenance.py'):
|
||||
shutil.copyfile(ROOT / 'tools' / name, tools / name)
|
||||
session = self.root / 'synthetic-session'
|
||||
session.mkdir()
|
||||
(session / 'console.log').write_text((self.root / self.log).read_text())
|
||||
archive = self.root / Path(self.manifest).parent
|
||||
for name in (MANIFEST, METADATA):
|
||||
shutil.copyfile(archive / name, session / name)
|
||||
shutil.copyfile(archive / 'runner.json.txt', session / 'runner.json')
|
||||
subprocess.run([sys.executable, str(tools / 'archive_native.py'), str(session), 'new-archive'],
|
||||
capture_output=True, text=True, check=True)
|
||||
archive = self.root / 'docs/evidence/new-archive'
|
||||
self.assertEqual(json.loads((archive / 'source-sha256.json').read_text()), addon_manifest(self.root))
|
||||
report = json.loads((archive / 'report.json').read_text())
|
||||
self.assertEqual(report, analyze((self.root / report['log']).read_text(), report['log']))
|
||||
|
||||
def test_native_report_command_compatibility(self):
|
||||
result = subprocess.run([sys.executable, str(ROOT / 'tools/native_report.py'), str(self.root / self.log)],
|
||||
capture_output=True, text=True, check=True)
|
||||
self.assertEqual(json.loads(result.stdout), analyze((self.root / self.log).read_text(), self.root / self.log))
|
||||
|
||||
def test_archive_time_provenance_cannot_qualify_native_gates(self):
|
||||
self.json('docs/evidence/current/' + PROVENANCE, {'schema_version': 1, 'kind': 'archive_time_only'})
|
||||
self.blocked('requires launch_snapshot provenance')
|
||||
|
||||
def test_missing_launch_snapshot_is_not_current_evidence(self):
|
||||
(self.root / 'docs/evidence/current' / METADATA).unlink()
|
||||
self.blocked('launch_provenance')
|
||||
|
||||
def test_modified_launch_bytes_and_runner_binding_rejected(self):
|
||||
path = self.root / 'docs/evidence/current' / MANIFEST
|
||||
path.write_bytes(path.read_bytes() + b'\n')
|
||||
self.blocked('launch manifest byte hash')
|
||||
self.refresh_native()
|
||||
runner = self.root / 'docs/evidence/current/runner.json.txt'
|
||||
data = json.loads(runner.read_text())
|
||||
data['launch_id'] = '00000000-0000-0000-0000-000000000000'
|
||||
runner.write_text(json.dumps(data))
|
||||
self.blocked('runner launch binding')
|
||||
|
||||
def test_snapshot_after_process_start_rejected(self):
|
||||
runner = self.root / 'docs/evidence/current/runner.json.txt'
|
||||
data = json.loads(runner.read_text())
|
||||
data['started_unix'] = 1
|
||||
runner.write_text(json.dumps(data))
|
||||
self.blocked('snapshot must precede process launch')
|
||||
|
||||
def test_runner_failure_and_incomplete_repeated_rounds_cannot_qualify(self):
|
||||
runner = self.root / 'docs/evidence/current/runner.json.txt'
|
||||
data = json.loads(runner.read_text())
|
||||
data['result'] = 'process-exited'
|
||||
runner.write_text(json.dumps(data))
|
||||
self.blocked('completed runner session')
|
||||
data['result'] = 'smoke-completed'
|
||||
data['rounds_required'] = 2
|
||||
runner.write_text(json.dumps(data))
|
||||
self.blocked('successful completion/assertions')
|
||||
|
||||
def test_prelaunch_snapshot_remains_old_when_persisted_after_edit(self):
|
||||
captured = snapshot(self.root, ['synthetic'])
|
||||
self.write('addons/core/Scripts/Game/Test.c', 'class ChangedAfterLaunch {}')
|
||||
session = self.root / 'synthetic-late-session'
|
||||
session.mkdir()
|
||||
persist(session, captured)
|
||||
self.assertEqual((session / MANIFEST).read_bytes(), captured[0])
|
||||
(session / 'console.log').write_text('Synthetic old run')
|
||||
self.json('synthetic-late-session/runner.json', {**bindings(captured), 'started_unix': json.loads(captured[1])['captured_unix'] + 1, 'stopped_unix': json.loads(captured[1])['captured_unix'] + 2, 'source_unchanged_after_run': True, 'end_source_sha256': hashlib.sha256(captured[0]).hexdigest(), 'console_sha256': hashlib.sha256((session / 'console.log').read_bytes()).hexdigest()})
|
||||
with self.assertRaisesRegex(ValueError, 'differ from pre-launch snapshot'):
|
||||
archive_source(session, self.root)
|
||||
|
||||
def test_legacy_archive_requires_opt_in_and_is_marked_historical(self):
|
||||
session = self.root / 'synthetic-legacy'
|
||||
session.mkdir()
|
||||
with self.assertRaisesRegex(ValueError, 'no pre-launch source snapshot'):
|
||||
archive_source(session, self.root)
|
||||
manifest, provenance, launched = archive_source(session, self.root, allow_legacy=True)
|
||||
self.assertFalse(launched)
|
||||
self.assertEqual(provenance['kind'], 'archive_time_only')
|
||||
self.assertEqual(provenance['qualification'], 'historical_only')
|
||||
self.assertEqual(json.loads(manifest), addon_manifest(self.root))
|
||||
|
||||
def test_partial_launch_evidence_cannot_fall_back_to_legacy(self):
|
||||
session = self.root / 'synthetic-partial'
|
||||
session.mkdir()
|
||||
(session / METADATA).write_text('{}')
|
||||
with self.assertRaisesRegex(ValueError, 'missing or symlinked launch evidence'):
|
||||
archive_source(session, self.root, allow_legacy=True)
|
||||
|
||||
def test_source_manifest_copy_must_match_exact_launch_bytes(self):
|
||||
path = self.root / self.manifest
|
||||
path.write_bytes(path.read_bytes() + b'\n')
|
||||
self.blocked('differs from exact launch snapshot bytes')
|
||||
|
||||
def test_recomputed_report_cannot_relabel_a_different_old_raw_log(self):
|
||||
self.write(self.log, 'ERR SMOKE PASS: different synthetic run\nERR SMOKE COMPLETE: other\n')
|
||||
self.json(self.report, analyze((self.root / self.log).read_text(), self.log))
|
||||
self.blocked('raw native log differs from stopped runner log hash')
|
||||
|
||||
def test_report_cannot_use_an_unsealed_neighbor_log(self):
|
||||
other = 'docs/evidence/current/other.log.txt'
|
||||
self.write(other, 'ERR SMOKE PASS: unrelated\nERR SMOKE COMPLETE: unrelated\n')
|
||||
self.json(self.report, analyze((self.root / other).read_text(), other))
|
||||
self.blocked('exact console.log.txt sealed by the stopped runner')
|
||||
|
||||
def test_stopped_source_change_blocks_even_if_current_files_match_launch(self):
|
||||
runner = self.root / 'docs/evidence/current/runner.json.txt'
|
||||
data = json.loads(runner.read_text())
|
||||
data['source_unchanged_after_run'] = False
|
||||
data['end_source_sha256'] = '0' * 64
|
||||
runner.write_text(json.dumps(data))
|
||||
self.blocked('changed between launch and stopped-run snapshot')
|
||||
|
||||
def test_seal_uses_stopped_sources_and_exact_log_bytes(self):
|
||||
session = self.root / 'synthetic-seal'
|
||||
session.mkdir()
|
||||
captured = snapshot(self.root, ['synthetic'])
|
||||
(session / 'console.log').write_bytes(b'exact synthetic log\r\n')
|
||||
sealed = seal_run(self.root, session, captured)
|
||||
self.assertTrue(sealed['source_unchanged_after_run'])
|
||||
self.assertEqual(sealed['console_sha256'], hashlib.sha256(b'exact synthetic log\r\n').hexdigest())
|
||||
self.write('addons/core/Scripts/Game/Test.c', 'class ChangedDuringRun {}')
|
||||
self.assertFalse(seal_run(self.root, session, captured)['source_unchanged_after_run'])
|
||||
|
||||
def test_archive_cli_requires_explicit_legacy_import(self):
|
||||
tools = self.root / 'tools'
|
||||
tools.mkdir()
|
||||
for name in ('archive_native.py', 'native_report.py', 'source_manifest.py', 'native_provenance.py'):
|
||||
shutil.copyfile(ROOT / 'tools' / name, tools / name)
|
||||
session = self.root / 'synthetic-legacy-command'
|
||||
session.mkdir()
|
||||
(session / 'console.log').write_text((self.root / self.log).read_text())
|
||||
command = [sys.executable, str(tools / 'archive_native.py'), str(session), 'legacy-archive']
|
||||
rejected = subprocess.run(command, capture_output=True, text=True)
|
||||
self.assertEqual(rejected.returncode, 2)
|
||||
self.assertFalse((self.root / 'docs/evidence/legacy-archive').exists())
|
||||
subprocess.run(command + ['--allow-legacy-archive-time-source'], capture_output=True, text=True, check=True)
|
||||
provenance = json.loads((self.root / 'docs/evidence/legacy-archive' / PROVENANCE).read_text())
|
||||
self.assertEqual(provenance['kind'], 'archive_time_only')
|
||||
with self.assertRaisesRegex(ValueError, 'requires launch_snapshot provenance'):
|
||||
qualify_archive(self.root / 'docs/evidence/legacy-archive', addon_manifest(self.root))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user