"""Agent-facing contracts, tested without launching or mutating the game.""" import contextlib import io import json import os from pathlib import Path import shlex import shutil import subprocess import sys import tempfile import unittest from unittest.mock import patch from witcher_devkit import arguments, catalog, cli_app, diagnostics, native, ui_driver class CliTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.enterContext(patch.object(native.legacy.manage,'state_directory',return_value=self.root)) for name, value in (('ROOT', self.root), ('STATE_ROOT', self.root), ('STATE', self.root / 'native-session.json')): p = patch.object(native, name, value) p.start() self.addCleanup(p.stop) def run_cli(self, *args): out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): try: status = cli_app.main(list(args)) except SystemExit as error: status = error.code return status, out.getvalue(), err.getvalue() def test_all_commands_have_contracts_and_all_examples_parse(self): self.assertEqual(native.COMMANDS, set(arguments.SPECS)) self.assertEqual(native.COMMANDS, {n[5:] for n in catalog.CATALOG if n.startswith('call ')}) parser = cli_app.parser() for name, spec in catalog.CATALOG.items(): with self.subTest(command=name): for key in ('summary', 'arguments', 'prerequisites', 'effects', 'completion', 'examples'): self.assertTrue(spec[key], (name, key)) for example in spec['examples']: parsed = parser.parse_args(shlex.split(example)) if parsed.operation != 'experiment':cli_app.preflight(parsed) status, out, err = self.run_cli(*name.split(), '--help') self.assertEqual(status, 0) self.assertIn(spec['summary'], out) self.assertIn('Completion / output:', out) self.assertFalse(err) def test_discovery_is_offline_and_uses_runtime_constraints(self): with patch.object(native.wire, 'Connection', side_effect=AssertionError('No network')), \ patch.object(cli_app, 'dispatch', side_effect=AssertionError('No operations')): status, out, err = self.run_cli('commands', '--json') self.assertEqual(status, 0) value = json.loads(out) self.assertEqual(value['contract_version'], 1) spawn = next(c for c in value['commands'] if c['name'] == 'call spawn') self.assertEqual(spawn['parameters'], arguments.SPECS['spawn']) for topic in catalog.GUIDES: self.assertEqual(self.run_cli('guide', topic, '--json')[0], 0) for code in diagnostics.ERRORS: status, out, err = self.run_cli('explain', code, '--json') self.assertEqual(status, 0) self.assertEqual(json.loads(out)['code'], code) self.assertFalse((self.root / '.lock').exists()) def test_unknown_command_suggests_known_command(self): status, out, err = self.run_cli('verfy') self.assertEqual(status, 2) self.assertEqual(json.loads(out)['diagnostic']['code'], 'E1001') self.assertIn('Did you mean verify?', err) def test_unknown_documentation_target_is_structured(self): for args in [('help', 'call', 'spwan', '--json'), ('explain', 'E1234')]: status, out, err = self.run_cli(*args) self.assertEqual(status, 2) self.assertEqual(json.loads(out)['diagnostic']['code'], 'E1001') self.assertFalse((self.root / '.lock').exists()) def test_invalid_fields_are_rejected_before_lock_or_observation(self): examples = [ (['call', 'spawn', 'nekker|9|6|4'], "'count'"), (['input', '0|0|0|0|Bogus|60'], "'button'"), (['input', '2|0|0|0|None|60'], "'move_x'"), (['advance', '0'], "'ticks'"), (['call', 'pad', '0|0|0|0|1024|0|0|300'], "'buttons'"), (['call', 'ui_rows', '1|0|33'], "'count'"), (['ui-find', 'label', 'Settings', '--timeout', 'nan'], '--timeout'), (['ui-tree', '--root', 'anything'], "'root'"), ] with patch.object(native, 'call', side_effect=AssertionError('No game call')): for args, field in examples: with self.subTest(args=args): status, out, err = self.run_cli(*args) self.assertEqual(status, 2) self.assertEqual(json.loads(out)['diagnostic']['code'], 'E1001') self.assertIn(field, err) self.assertFalse((self.root / '.lock').exists()) def test_direct_bounded_action_validates_before_snapshot(self): with patch.object(native, 'call') as call: with self.assertRaisesRegex(ValueError, 'ticks'): native.bounded_action('input', '0|0|0|0|None|999') call.assert_not_called() def test_missing_session_points_to_attach(self): status, out, err = self.run_cli('call', 'snapshot') self.assertEqual(status, 1) self.assertEqual(json.loads(out)['diagnostic']['code'], 'E2004') self.assertIn('attach', err) def test_success_stdout_is_one_json_object(self): with patch.object(cli_app, 'dispatch', return_value={'ok': True, 'data': {'ready': True}}): status, out, err = self.run_cli('call', 'snapshot') self.assertEqual(status, 0) self.assertTrue(json.loads(out)['ok']) self.assertEqual(len(out.splitlines()), 1) self.assertFalse(err) def test_nested_ui_rejection_is_a_failure_with_original_evidence(self): original = {'ok': True, 'request': 2, 'data': {'error': 'ui_node_detached'}} with patch.object(cli_app, 'dispatch', return_value=original): status, out, err = self.run_cli('call', 'ui_node', '1|3') self.assertEqual(status, 1) result = json.loads(out) self.assertEqual(result['diagnostic']['code'], 'E4001') self.assertEqual(result['result'], original) def test_ui_failed_postcondition_retains_delivered_input(self): responses = [{'ok': True, 'data': {'node': {'bounds': [0, 0, 20, 20]}}}] * 2 + [ {'ok': True, 'data': {'dispatched': True}}] with patch.object(ui_driver, 'find', side_effect=[{'revision': 1, 'node': {'id': 3}}, TimeoutError('No Sound label')]), \ patch.object(native, 'call', side_effect=responses) as call, patch.object(ui_driver.time, 'sleep'): status, out, err = self.run_cli('ui-dispatch', 'label', 'Settings', '--expect-label', 'Sound') result = json.loads(out) self.assertEqual(status, 1) self.assertEqual(result['diagnostic']['code'], 'E4002') self.assertTrue(result['result']['delivery']['dispatched']) self.assertFalse(result['result']['effect_verified']) self.assertEqual(sum(c.args[0] == 'ui_dispatch' for c in call.call_args_list), 1) def test_compile_failure_exposes_locations_and_rollback(self): original = dict(ok=False, compile=dict(ok=False, errors=[ ['ScriptCompiler', 'error', 42, 'local/pathdevkit.ws', 'unexpected token']]), rollback={'ok': False}, artifacts='/tmp/compiler-evidence', executed=False) with patch.object(cli_app, 'dispatch', return_value=original): status, out, err = self.run_cli('evaluate', '/tmp/experiment.ws') result = json.loads(out) self.assertEqual(status, 1) self.assertEqual(result['diagnostic']['code'], 'E5001') self.assertEqual(result['result'], original) self.assertIn('local/pathdevkit.ws:42', err) self.assertFalse(result['result']['executed']) def test_timeout_keeps_uncertainty_context_and_does_not_retry(self): self.root.joinpath('native-session.json').write_text(json.dumps( dict(session='test', request=5, uncertain={'command': 'spawn', 'request': 5}))) with patch.object(cli_app, 'dispatch', side_effect=TimeoutError('lost acknowledgement')) as dispatch: status, out, err = self.run_cli('call', 'spawn', 'nekker|1|6|4') d = json.loads(out)['diagnostic'] self.assertEqual(status, 1) self.assertEqual(d['code'], 'E3003') self.assertEqual(d['context']['uncertain']['command'], 'spawn') self.assertIn('do not replay blindly', err) dispatch.assert_called_once() def test_failed_save_audit_does_not_exit_successfully(self): with patch.object(cli_app, 'dispatch', return_value=dict(ok=True, state='stopped', save_files_unchanged=False)): status, out, err = self.run_cli('stop') self.assertEqual(status, 1) self.assertEqual(json.loads(out)['diagnostic']['code'], 'E6001') def test_lock_collision_does_not_dispatch(self): with (self.root / '.lock').open('w') as lock: cli_app.fcntl.flock(lock, cli_app.fcntl.LOCK_EX | cli_app.fcntl.LOCK_NB) with patch.object(cli_app, 'dispatch') as dispatch: status, out, err = self.run_cli('call', 'snapshot') dispatch.assert_not_called() self.assertEqual(status, 1) self.assertEqual(json.loads(out)['diagnostic']['code'], 'E2005') def test_source_only_checkout_help_needs_no_local_state(self): source = Path(__file__).resolve().parents[1] checkout = self.root / 'checkout' devkit = checkout / 'devkit' devkit.mkdir(parents=True) for name in ('cli.py', 'native.py', 'w3dev'): shutil.copy2(source / name, devkit / name) for name in ('witcher_devkit', 'scripts', 'tests', 'fixtures'): shutil.copytree(source / name, devkit / name, ignore=shutil.ignore_patterns('__pycache__', '*.pyc')) environment = dict(os.environ, PYTHONDONTWRITEBYTECODE='1') environment.pop('PYTHONPATH', None) for entry, args in [('cli.py', ['commands', '--json']), ('native.py', ['help', 'call', 'spawn', '--json']), ('cli.py', ['guide', '--json']), ('cli.py', ['call', 'pad', '--help'])]: result = subprocess.run([sys.executable, str(devkit / entry), *args], cwd='/', env=environment, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) if '--json' in args: self.assertTrue(json.loads(result.stdout)['ok']) result = subprocess.run( [sys.executable, '-m', 'unittest', 'discover', '-s', '.', '-p', 'test_asset_lz4.py', '-q'], cwd=devkit, env=environment, capture_output=True, text=True, timeout=30) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((devkit / '.lock').exists()) self.assertFalse((devkit / 'native-session.json').exists()) def test_doctor_without_receipts_is_actionable_and_does_not_lock(self): from witcher_devkit import lifecycle with patch.object(lifecycle, 'ROOT', self.root), patch.object(native.wire, 'Connection', side_effect=AssertionError('No network')): status, out, err = self.run_cli('doctor') result = json.loads(out) self.assertEqual(status, 1) self.assertEqual(result['diagnostic']['code'], 'E2001') self.assertIn('deployment.json', result['error']) self.assertFalse((self.root / '.lock').exists()) if __name__ == '__main__': unittest.main()