#!/usr/bin/env python3 """Offline orchestrator fixtures only; never invokes pio or existing suites.""" import contextlib import importlib.util import io import os from pathlib import Path import subprocess import sys import tempfile import time import unittest from unittest.mock import patch sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parents[2] spec = importlib.util.spec_from_file_location('phase9', ROOT / 'tools/validate_phase9.py') runner = importlib.util.module_from_spec(spec) sys.modules[spec.name] = runner spec.loader.exec_module(runner) class ValidationTests(unittest.TestCase): def options(self, *args): return runner.parser().parse_args(args) def command(self, code, name='fixture', timeout=3, required=()): return runner.Command(name, (sys.executable, '-c', code), timeout, required) def execute(self, commands, **kwargs): output = io.StringIO() with contextlib.redirect_stdout(output): code = runner.execute(commands, ROOT, **kwargs) return code, output.getvalue() def test_default_exact_scope_and_real_paths(self): commands = runner.plan(self.options()) self.assertEqual(len(commands), 23) self.assertEqual({c.name for c in commands}, { 'security_build_policy', 'ssh_auth_policy', 'ssh_auth_transport', 'hidden_input', 'ssh_memory', 'sdk_security_overrides', 'wolfssh_auth_contract', 'ssh_protocol_policy', 'wolf_crypto_policy', 'wolfssh_parser_contract', 'wolfssh_order_contract', 'release_notices', 'ssh_management', 'admin_console_boundary', 'admin_ssh_policy', 'web_admin_transport', 'web_admin_tickets', 'web_httpd_idle', 'web_cookie_auth:base', 'web_cookie_auth:admin', 'web_cookie_auth:accounts', 'web_cookie_auth:ssh', 'web_cookie_auth:lifecycle'}) for command in commands: self.assertTrue((ROOT / command.argv[2]).is_file()) self.assertEqual(command.argv[:2], (sys.executable, '-B')) self.assertEqual(command.timeout, 180) self.assertFalse({'--interop', '--host-only', '--candidate', '--target-contracts', '--pio-adapter', 'pio', '--admission'} & set(command.argv)) by_name = {c.name: c for c in commands} database = str(runner.BUILD / 'compile_commands.json') for name in ('wolfssh_auth_contract', 'ssh_protocol_policy', 'wolf_crypto_policy'): self.assertEqual(by_name[name].argv[3:], ('--compile-commands', database)) self.assertEqual(by_name['sdk_security_overrides'].argv[3:], ('--build-dir', str(runner.BUILD))) self.assertEqual(by_name['security_build_policy'].argv[3:], ('--sdkconfig-header', str(runner.BUILD / 'config/sdkconfig.h'))) def test_opt_ins_independent_and_timeout(self): commands = runner.plan(self.options('--build', '--timeout', '4', '--build-timeout', '5')) self.assertEqual(commands[0], runner.Command('build', ('pio', 'run'), 5)) self.assertTrue(all(c.timeout == 4 for c in commands[1:])) self.assertFalse(any('--interop' in c.argv for c in commands)) commands = runner.plan(self.options('--interop')) self.assertNotIn('build', [c.name for c in commands]) self.assertEqual([c.name for c in commands if '--interop' in c.argv], ['wolfssh_order_contract']) def test_bad_timeout(self): for value in ('0', '-1', 'nan', 'inf', '3601', 'junk'): with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): self.options('--timeout', value) def test_dry_run_never_launches_or_checks_prerequisites(self): with patch.object(runner, 'run_command', side_effect=AssertionError('executed')): code, output = self.execute(runner.plan(self.options('--build', '--interop')), dry_run=True) self.assertEqual(code, 0) self.assertIn('PLAN build', output) self.assertNotIn('PASS ', output) with patch.object(runner, 'run_command', side_effect=AssertionError('executed')): with contextlib.redirect_stdout(io.StringIO()) as output: self.assertEqual(runner.main(['--dry-run']), 0) self.assertIn('SKIP optional build', output.getvalue()) self.assertIn('SKIP optional interop', output.getvalue()) def test_collect_failure_and_failfast_skip(self): commands = [self.command('raise SystemExit(7)', 'bad'), self.command('pass', 'good')] code, output = self.execute(commands) self.assertEqual(code, 1) self.assertIn('FAIL bad: exit 7', output) self.assertIn('PASS good', output) code, output = self.execute(commands, fail_fast=True) self.assertEqual(code, 1) self.assertIn('SKIP good', output) self.assertNotIn('PASS good', output) def test_failed_build_blocks_stale_artifact_checks(self): code, output = self.execute([self.command('raise SystemExit(1)', 'build'), self.command('pass', 'host')]) self.assertEqual(code, 1) self.assertIn('SKIP host', output) def test_prerequisites(self): with tempfile.TemporaryDirectory() as tmp: missing = Path(tmp) / 'missing' commands = [runner.Command('missing-executable', (str(missing),), 1), self.command('raise AssertionError()', 'missing-input', required=(missing,))] code, output = self.execute(commands) self.assertEqual(code, 1) self.assertIn('PREREQ missing-executable', output) self.assertIn('PREREQ missing-input', output) def test_environment_and_literal_argv(self): with tempfile.TemporaryDirectory(prefix='phase9 fixture ') as tmp: script = Path(tmp) / 'fixture ; literal.py' script.write_text('import os, sys\nassert os.environ["CCACHE_DISABLE"] == "1"\n' 'assert os.environ["PHASE9_FIXTURE"] == "preserved"\n' 'assert sys.argv[1] == "a ; $(not-a-command)"\n') command = runner.Command('literal', (sys.executable, str(script), 'a ; $(not-a-command)'), 3) with patch.dict(os.environ, {'PHASE9_FIXTURE': 'preserved', 'CCACHE_DISABLE': '0'}): self.assertEqual(self.execute([command])[0], 0) self.assertEqual(os.environ['CCACHE_DISABLE'], '0') def test_timeout_kills_descendant_and_collects(self): with tempfile.TemporaryDirectory() as tmp: marker = Path(tmp) / 'should-not-exist' child = f'import time; from pathlib import Path; time.sleep(1); Path({str(marker)!r}).touch()' parent = f'import subprocess, sys, time; subprocess.Popen([sys.executable, "-c", {child!r}]); time.sleep(10)' started = time.monotonic() code, output = self.execute([self.command(parent, timeout=0.25), self.command('pass', 'next')]) self.assertLess(time.monotonic() - started, 4) self.assertEqual(code, 1) self.assertIn('TIMEOUT fixture', output) self.assertIn('PASS next', output) time.sleep(1.1) self.assertFalse(marker.exists()) def test_streams_not_captured_and_stdin_closed(self): with patch.object(runner.subprocess, 'Popen') as popen, patch.object(runner.os, 'killpg'): popen.return_value.wait.return_value = 0 code, _ = self.execute([self.command('pass')]) self.assertEqual(code, 0) kwargs = popen.call_args.kwargs self.assertNotIn('stdout', kwargs) self.assertNotIn('stderr', kwargs) self.assertNotIn('shell', kwargs) self.assertEqual(kwargs['stdin'], subprocess.DEVNULL) self.assertTrue(kwargs['start_new_session']) if __name__ == '__main__': unittest.main()