"""Offline startup diagnostics; fake observations are not live qualification.""" import json from pathlib import Path import queue import tempfile from types import SimpleNamespace import unittest from unittest.mock import patch from witcher_devkit import debug_startup as startup, debugger, diagnostics PROCESS = dict(pid=123, start='456', executable='witcher3.exe') BASELINE = dict(saves={'a.sav':'save'}, mods={'a.ws':'mod'}, settings={'user.settings':'user', 'dx12user.settings':'dx12'}) class Listener: def __init__(self, packets=()): self.queue = queue.Queue() for packet in packets: self.queue.put(packet) self.connected = True self.dropped = False self.error = None self.thread = SimpleNamespace(start=lambda: None) self.closed = False def close(self): self.closed = True class StartupTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.folder = Path(self.temp.name) (self.folder/'startup.json').write_text(json.dumps(dict(phase='prepared'))) self.enterContext(patch.object(startup, 'manifests', return_value=BASELINE)) self.enterContext(patch.object(startup, 'log_state', return_value=dict(bytes=0,prefix=b''))) self.enterContext(patch.object(startup, 'fresh_log', return_value=dict(errors=[],fresh_text=''))) self.enterContext(patch.object(startup.time, 'sleep')) self.enterContext(patch.object(startup, 'processes', return_value=[PROCESS])) self.listener = Listener() self.enterContext(patch.object(startup, 'CompilerListener', return_value=self.listener)) def test_real_window_required_and_startup_baseline_recorded(self): with patch.object(startup, 'windows', return_value=[dict(id='1',title='The Witcher 3')]), \ patch.object(startup.wire, 'Connection') as connection: startup.wait_for_window(self.folder, 10, lambda: None) value = startup.read(self.folder) self.assertEqual(value['phase'], 'attaching') self.assertEqual(value['before'], BASELINE) self.assertEqual(value['process'], PROCESS) self.assertTrue(self.listener.closed) connection.assert_not_called() # Fake listener; no exec/reload escape. def test_compiler_packets_preserve_file_line_and_classification(self): packet = ['ScriptCompiler','error',3,'local/probe.ws','Unknown type'] self.listener.queue.put(packet) with patch.object(startup, 'windows', return_value=[]), self.assertRaises(diagnostics.DiagnosticError) as raised: startup.wait_for_window(self.folder, 10, lambda: None) self.assertEqual(raised.exception.code, 'E5001') value = startup.read(self.folder) self.assertEqual(value['failure']['category'], 'script_compiler_error') result = dict(ok=False,startup=value) self.assertEqual(diagnostics.rejected(result).code,'E5001') self.assertEqual(diagnostics.compiler_locations(result), [dict(file='local/probe.ws',line=3,message='Unknown type')]) self.assertTrue(value['cleanup_required']) def test_compiler_window_without_packets_reports_file_line_gap(self): with patch.object(startup, 'windows', return_value=[ dict(id='2',title='Script Compilation Errors')]), \ self.assertRaises(diagnostics.DiagnosticError): startup.wait_for_window(self.folder,10,lambda:None) value = startup.read(self.folder) self.assertEqual(value['failure']['category'],'script_compiler_error') self.assertIn('unavailable',value['failure']['detail']) self.assertEqual(value['compiler']['errors'],[]) def test_absence_mismatch_and_timeout_are_not_compiler_failures(self): cases = [ ([], [], 'game_absent','E2003'), ([PROCESS], [dict(id='1',title='Unexpected window')], 'window_mismatch','E2003'), ([PROCESS], [], 'startup_timeout','E3003'), ] for processes, windows, category, code in cases: with self.subTest(category=category): (self.folder/'startup.json').write_text(json.dumps(dict(phase='prepared'))) with patch.object(startup,'processes',return_value=processes), \ patch.object(startup,'windows',return_value=windows), \ self.assertRaises(diagnostics.DiagnosticError) as raised: startup.wait_for_window(self.folder,0,lambda:None) self.assertEqual(raised.exception.code,code) value=startup.read(self.folder) self.assertEqual(value['failure']['category'],category) self.assertFalse(value['audits_final']) def test_process_exit_and_identity_change_stop_without_adoption(self): for following in ([],[dict(PROCESS,start='different')]): (self.folder/'startup.json').write_text(json.dumps(dict(phase='prepared'))) with patch.object(startup,'processes',side_effect=[[PROCESS],following]), \ patch.object(startup,'windows',return_value=[]), \ self.assertRaises(diagnostics.DiagnosticError): startup.wait_for_window(self.folder,10,lambda:None) def test_recovery_uses_this_startup_baseline_and_is_idempotent(self): from witcher_devkit import game_settings, lifecycle startup.update(self.folder, before=BASELINE, phase='failed', failure=dict(code='E5001',detail='retained')) with patch.object(startup.support.manage,'ensure_game_closed'), \ patch.object(game_settings,'reset',return_value=dict(ok=True)) as reset, \ patch.object(lifecycle,'verify',return_value=dict(ok=True)): result=startup.recover_closed(self.folder) self.assertTrue(result['ok']) self.assertTrue(result['mods_unchanged']) original=(self.folder/'startup.json').read_bytes() self.assertTrue(startup.recover_closed(self.folder)['already_recovered']) self.assertEqual(reset.call_count,1) self.assertEqual((self.folder/'startup.json').read_bytes(),original) def test_changed_mods_remain_failed_audit(self): from witcher_devkit import game_settings, lifecycle startup.update(self.folder,before=BASELINE,phase='failed') with patch.object(startup.support.manage,'ensure_game_closed'), \ patch.object(game_settings,'reset',return_value=dict(ok=True)), \ patch.object(lifecycle,'verify',return_value=dict(ok=True)), \ patch.object(startup,'manifests',return_value=dict(BASELINE,mods={})): result=startup.recover_closed(self.folder) self.assertFalse(result['ok']) self.assertFalse(result['mods_unchanged']) self.assertTrue(result['saves_unchanged']) def test_legacy_commands_cannot_bypass_unrecovered_startup(self): pending=dict(startup=dict(failure=dict(code='E5001'),cleanup_required=True)) with patch.object(debugger,'descriptor',return_value=None), \ patch.object(debugger,'starting_status',return_value=pending): with self.assertRaisesRegex(RuntimeError,'startup_recovery_required'): debugger.reject_legacy() pending['startup']['cleanup_required']=False debugger.reject_legacy() class StartupLogTests(unittest.TestCase): def test_old_errors_not_reused_and_fresh_file_errors_retained(self): old=b'Error old.ws(1): old failure\n' with patch.object(startup,'log_state',return_value=dict(path='scriptslog.txt',bytes=len(old),prefix=old)): self.assertEqual(startup.fresh_log(dict(bytes=len(old),prefix=old))['errors'],[]) fresh=old+b'Error [modProbe]local/probe.ws(4): syntax error\n' with patch.object(startup,'log_state',return_value=dict(path='scriptslog.txt',bytes=len(fresh),prefix=fresh)): errors=startup.fresh_log(dict(bytes=len(old),prefix=old))['errors'] self.assertEqual(errors,[['ScriptCompiler','error',4,'[modProbe]local/probe.ws','syntax error']])