107 lines
5.8 KiB
Python
107 lines
5.8 KiB
Python
"""Execute actual scope policy; structural cleanup checks are not native proof."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from policy_harness import body, run
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SOURCE = ROOT / 'addons/core/Scripts/Game/ERR/ERR_ItemStationTeardown.c'
|
|
PROBE = ROOT / 'addons/core/Scripts/Game/ERR/ERR_NativeItemStationTeardownProbe.c'
|
|
|
|
class ItemStationTeardown(unittest.TestCase):
|
|
def test_actual_recovery_gate_all_inputs(self):
|
|
method = body(SOURCE, 'static bool Recover(')
|
|
run('#include <cassert>\n' + method + '''
|
|
int main() {
|
|
for (int mask=0; mask<32; ++mask) {
|
|
bool initialized=mask&1, scoped=mask&2, offWorld=mask&4;
|
|
bool catalogPresent=mask&8, editMode=mask&16;
|
|
assert(Recover(initialized,scoped,offWorld,catalogPresent,editMode)==(mask==7));
|
|
}
|
|
}
|
|
''')
|
|
|
|
def test_real_validation_precedes_owned_subscription_capture(self):
|
|
source = SOURCE.read_text()
|
|
self.assertNotIn('override bool InitValidSetup', source)
|
|
init = body(SOURCE, 'protected override void DelayedInit(')
|
|
self.assertIn('ERR_GameMode.Get() && ERR_KnownOffWorldKit(owner)', init)
|
|
self.assertIn('super.DelayedInit(owner)', init)
|
|
subscription = body(SOURCE, 'protected override void AddRemoveOnDamageStateChanged(')
|
|
self.assertLess(subscription.index('super.AddRemove'), subscription.index('m_ERRTeardownInitialized = true'))
|
|
self.assertIn('if (!subscribe || !m_ERRTeardownInitializing) return;', subscription)
|
|
scope = body(SOURCE, 'protected bool ERR_KnownOffWorldKit(')
|
|
self.assertIn('owner.GetWorld() == GetGame().GetWorld()', scope)
|
|
self.assertIn('{AE578EEA4244D41F}', scope)
|
|
self.assertIn('{21EF98BFC1EB3793}', scope)
|
|
self.assertEqual(scope.count('prefab =='), 2)
|
|
self.assertIn('modded class SCR_ResupplyMedicalGadgetSupportStationComponent', source)
|
|
|
|
def test_cleanup_has_real_detach_and_unchanged_normal_path(self):
|
|
deletion = body(SOURCE, 'override void OnDelete(')
|
|
self.assertIn('super.OnDelete(owner);', deletion)
|
|
self.assertIn('m_EntityCatalogManager != null', deletion)
|
|
self.assertLess(deletion.index('if (m_ERRTeardownDone) return;'), deletion.index('Remove(DelayedInit)'))
|
|
self.assertIn('AddRemoveOnDamageStateChanged(owner, false);', deletion)
|
|
self.assertIn('TEMP_GetOnInteractorReplicated().Remove(TEMP_OnInteractorReplicated)', deletion)
|
|
self.assertIn('m_ERRTeardownResource = null;', deletion)
|
|
self.assertNotIn('InitValidSetup(', deletion)
|
|
self.assertNotIn('LogLevel', deletion)
|
|
|
|
def test_fault_injection_is_owned_bounded_and_measures_callbacks(self):
|
|
source = PROBE.read_text()
|
|
self.assertTrue(source.startswith('#ifdef WORKBENCH'))
|
|
self.assertIn('BaseWorld.CreateWorld("Preview", "ERRStationTeardownProbe")', source)
|
|
self.assertIn('m_World != GetGame().GetWorld()', source)
|
|
self.assertIn('m_EntityCatalogManager = null;', source)
|
|
self.assertNotIn('s_Instance =', source)
|
|
self.assertNotIn('InitValidSetup(', source)
|
|
self.assertIn('CallLater(DelayedInit, 100, false, GetOwner())', source)
|
|
self.assertEqual(source.count('OnDelete(GetOwner());'), 2)
|
|
self.assertIn('now + 500', source)
|
|
self.assertIn('now + 10000', source)
|
|
self.assertIn('m_Station.ERR_TeardownProbeInitCalls() == 0', source)
|
|
self.assertGreaterEqual(source.count('m_Station.ERR_TeardownProbeResourceCalls() == 1'), 3)
|
|
self.assertIn('m_Index < 2', source)
|
|
self.assertIn('if (m_Kit) delete m_Kit;', source)
|
|
|
|
def test_integrated_smoke_steps_finish_only_after_probe_success(self):
|
|
source = (ROOT / 'addons/core/Scripts/Game/ERR/ERR_NativeBaseSmoke.c').read_text()
|
|
self.assertIn('protected ref ERR_NativeItemStationTeardownProbe m_ERRItemStationTeardownProbe;', source)
|
|
self.assertIn('if (m_ERRItemStationTeardownProbe) m_ERRItemStationTeardownProbe.Stop();', source)
|
|
self.assertIn('if (m_ERRBaseStep == 2000)', source)
|
|
self.assertIn('int result = m_ERRItemStationTeardownProbe.Tick(now);', source)
|
|
self.assertIn('if (result == 1) { m_ERRBaseStep = 2002; Transition(ERR_Phase.DEFEAT); }', source)
|
|
self.assertIn('current == ERR_Phase.DEFEAT && m_ERRBaseStep == 2002', source)
|
|
for preserved in ('m_ERRPrisonBreachProbe.Tick(now)', 'm_ERRPoweredFlightProbe.Tick(now)', 'ERR_NativeDiscoveryMapProbe.Run()'):
|
|
self.assertIn(preserved, source)
|
|
|
|
def invoke_runner(self, profile, *extra):
|
|
return subprocess.run([sys.executable, str(ROOT/'tools/headless_workbench.py'),
|
|
'--compat', str(profile/'missing-prefix'), '--profile', str(profile),
|
|
'--smoke', 'station-teardown', '--project', 'core', '--rounds', '1',
|
|
*extra], text=True, capture_output=True)
|
|
|
|
def test_runner_rejects_wrong_project_or_round_count_before_launch(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
profile = Path(directory)
|
|
for extra in [('--project', 'chat'), ('--rounds', '2')]:
|
|
with self.subTest(extra=extra):
|
|
result = self.invoke_runner(profile, *extra)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('station-teardown requires --project core --rounds 1', result.stderr)
|
|
self.assertEqual(list(profile.iterdir()), [])
|
|
|
|
def test_runner_valid_case_stops_at_missing_prefix_without_side_effects(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
profile = Path(directory)
|
|
result = self.invoke_runner(profile)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('must be a disposable cloned prefix', result.stderr)
|
|
self.assertEqual(list(profile.iterdir()), [])
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|