#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Host-only tests for the Phase-1.0AC dormant fake-syscall adapter.""" from __future__ import annotations import argparse from datetime import datetime, timezone import importlib.util from pathlib import Path import sys CURRENT_COMMANDS = "authid browse cat cd chgrp chmod chown chroot cmp cp df echo env exec exit export file find grep hbdbg hbldr hexdump http2_get id kill launch ln ls mkdir mknod mount mv notify pkg_install procstat ps pwd reptyr rm rmdir sfocreate sfoinfo sleep stat sum suspend sync sysctl touch umount".split() def load(path: Path): spec = importlib.util.spec_from_file_location("phase10ac_dormant", path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) args = parser.parse_args() root = args.root.resolve() sys.path.insert(0, str(root / "tools")) module = load(root / "tools/phase10ac_dormant_adapter.py") from phase10w_shsrv_client_policy import SessionPlan from phase10z_passive_batch_contract import build_passive_batch def plan(window="T2_GREETING_AND_HELP", path=None): commands = ("help",) if window == "T2_GREETING_AND_HELP" else ("stat", "sum") return SessionPlan("synthetic_ac", "not-retained.invalid", 2323, window, path, commands, 10, datetime(2030, 1, 1, tzinfo=timezone.utc)) def greeting(extra=""): return ("Welcome to shsrv.elf running on pid 1, compiled Jul 22 2026 at 12:34:56\n" "Model: synthetic\nS/N: SECRET-SERIAL\nS/W: 9.60\n" "SoC temp: 40 C\nCPU temp: 41 C\nCPU freq: 3500 MHz\n" + extra).encode("ascii") def help_text(): return "Builtin commands:\n" + "".join( f" {command} - synthetic\n" for command in CURRENT_COMMANDS) + "\n" def step(op, result, value=0, data=b"", advance=0.0): return module.FakeSyscallStep(op, result, value, data, advance) def valid_steps(data=None, pending=True, writes=(2, 3), extra=()): payload = data or greeting(help_text()) values = [step(module.CREATE_STREAM, module.OK), step(module.SET_NONBLOCKING, module.OK)] if pending: values += [step(module.START_CONNECT, module.PENDING), step(module.WAIT_WRITE, module.READY, advance=.1), step(module.GET_SO_ERROR, module.ZERO)] else: values += [step(module.START_CONNECT, module.IMMEDIATE, advance=.1)] for count in writes: values += [step(module.WAIT_WRITE, module.READY, advance=.1), step(module.WRITE_BYTES, module.PROGRESS, count)] values += list(extra) values += [step(module.WAIT_READ, module.READY, advance=.1), step(module.READ_BYTES, module.PROGRESS, len(payload), payload), step(module.WAIT_READ, module.TIMEOUT, advance=9.6)] return tuple(values) def execute(steps=None, batch=None, close_result=module.OK, receipt=True): clock = module.OfflineFakeClock() actual_batch = batch or build_passive_batch(plan()) facade = module.OfflineFakeSyscallFacade( clock, steps or valid_steps(writes=(2, 3)), close_result) return module.run_dormant_adapter( actual_batch, facade, clock, receipt), facade def expect_failure(function): try: function() except (module.DormantAdapterError, module.FakeSyscallError): return raise RuntimeError("invalid dormant adapter scenario was accepted") cases = [] def case(name): def register(function): cases.append((name, function)); return function return register @case("01 pending connect lifecycle succeeds") def _(): outcome, facade = execute() assert outcome.classification == "OFFLINE_DORMANT_FAKE_SYSCALL_ADAPTER_COMPLETE" and facade.close_count == 1 @case("02 immediate connect lifecycle succeeds") def _(): outcome, _ = execute(valid_steps(pending=False)) assert outcome.batch_bytes_sent == 5 @case("03 partial writes complete exact batch") def _(): outcome, _ = execute(valid_steps(writes=(1, 1, 3))) assert outcome.write_calls == 3 and outcome.batch_bytes_sent == 5 @case("04 interrupted connect wait is bounded") def _(): prefix = (step(module.CREATE_STREAM, module.OK), step(module.SET_NONBLOCKING, module.OK), step(module.START_CONNECT, module.PENDING), step(module.WAIT_WRITE, module.INTERRUPTED, advance=.1), step(module.WAIT_WRITE, module.READY, advance=.1), step(module.GET_SO_ERROR, module.ZERO)) suffix = valid_steps(pending=False)[3:] outcome, _ = execute(prefix + suffix) assert outcome.interrupted_waits == 1 @case("05 interrupted read wait is bounded") def _(): extra = (step(module.WAIT_READ, module.INTERRUPTED, advance=.1),) outcome, _ = execute(valid_steps(extra=extra)) assert outcome.interrupted_waits == 1 @case("06 timeout only seals at deadline") def _(): outcome, _ = execute() assert outcome.timed_out_waits == 1 and outcome.result_classification == "SOURCE_FAMILY_FINGERPRINT_ONLY" @case("07 receipt is required before create") def _(): expect_failure(lambda: execute(receipt=False)) @case("08 exact fake facade type is required") def _(): batch = build_passive_batch(plan()); clock = module.OfflineFakeClock() expect_failure(lambda: module.run_dormant_adapter(batch, object(), clock, True)) @case("09 exact fake clock type is required") def _(): class Derived(module.OfflineFakeClock): pass expect_failure(lambda: module.OfflineFakeSyscallFacade(Derived(), valid_steps())) @case("10 create failure does not close an unowned descriptor") def _(): steps = (step(module.CREATE_STREAM, module.ERROR),) clock = module.OfflineFakeClock(); facade = module.OfflineFakeSyscallFacade(clock, steps) expect_failure(lambda: module.run_dormant_adapter(build_passive_batch(plan()), facade, clock, True)) assert facade.close_count == 0 @case("11 nonblocking failure closes once") def _(): steps = (step(module.CREATE_STREAM, module.OK), step(module.SET_NONBLOCKING, module.ERROR)) _, facade = None, module.OfflineFakeSyscallFacade(module.OfflineFakeClock(), steps) expect_failure(lambda: module.run_dormant_adapter(build_passive_batch(plan()), facade, facade.clock, True)) assert facade.close_count == 1 @case("12 pending connect requires write readiness") def _(): values = list(valid_steps()); values[3] = step(module.GET_SO_ERROR, module.ZERO) expect_failure(lambda: execute(tuple(values))) @case("13 pending connect requires zero SO_ERROR") def _(): values = list(valid_steps()); values[4] = step(module.GET_SO_ERROR, module.NONZERO) expect_failure(lambda: execute(tuple(values))) @case("14 connect error fails") def _(): values = list(valid_steps()); values[2] = step(module.START_CONNECT, module.ERROR) expect_failure(lambda: execute(tuple(values))) @case("15 write requires readiness") def _(): values = list(valid_steps()); del values[5] expect_failure(lambda: execute(tuple(values))) @case("16 zero write fails") def _(): values = list(valid_steps()); values[6] = step(module.WRITE_BYTES, module.ZERO) expect_failure(lambda: execute(tuple(values))) @case("17 excess write count fails") def _(): values = list(valid_steps(writes=(6,))); expect_failure(lambda: execute(tuple(values))) @case("18 incomplete write script fails") def _(): values = valid_steps(writes=(2, 2)); expect_failure(lambda: execute(values)) @case("19 read requires readiness") def _(): values = list(valid_steps()); index = next(i for i, item in enumerate(values) if item.operation == module.WAIT_READ); del values[index] expect_failure(lambda: execute(tuple(values))) @case("20 EOF is never completion") def _(): values = list(valid_steps()); index = next(i for i, item in enumerate(values) if item.operation == module.READ_BYTES); values[index] = step(module.READ_BYTES, module.EOF) expect_failure(lambda: execute(tuple(values))) @case("21 receive bound is enforced") def _(): data = b"x" * 65537; expect_failure(lambda: execute(valid_steps(data=data))) @case("22 data at deadline fails") def _(): values = list(valid_steps()); index = next(i for i, item in enumerate(values) if item.operation == module.READ_BYTES); item = values[index]; values[index] = step(item.operation, item.result, item.value, item.data, 9.6) expect_failure(lambda: execute(tuple(values))) @case("23 deadline wins readiness race") def _(): values = list(valid_steps()); values[-1] = step(module.WAIT_READ, module.READY, advance=9.6) expect_failure(lambda: execute(tuple(values))) @case("24 early script exhaustion fails") def _(): expect_failure(lambda: execute(valid_steps()[:-1])) @case("25 close failure invalidates success") def _(): expect_failure(lambda: execute(close_result=module.ERROR)) @case("26 unused post-deadline steps are discarded") def _(): values = valid_steps() + (step(module.READ_BYTES, module.ERROR),) outcome, facade = execute(values) assert outcome.discarded_steps_after_close == 1 and facade.discarded_steps == 1 @case("27 malformed result fails at deadline") def _(): expect_failure(lambda: execute(valid_steps(data=greeting("partial\n")))) @case("28 path result is sanitized") def _(): path = "/data/a.elf" data = greeting(f"filename: {path}\nsize: 123\n12345 {path}\n") batch = build_passive_batch(plan("T3_ONE_EXACT_PATH", path)) outcome, _ = execute(valid_steps(data=data, writes=(10, 10, len(batch.payload) - 20)), batch) assert outcome.result_classification == "WEAK_FILE_CORRELATION_ONLY" and not outcome.exact_identity @case("29 outcome retains no target and proves no device") def _(): outcome, _ = execute() assert not outcome.target_retained and not outcome.live_transport_present and not outcome.device_behavior_proven @case("30 no live API exists") def _(): names = set(dir(module)) assert not ({"connect", "send", "recv", "main"} & names) @case("31 fake step validation rejects hidden data") def _(): expect_failure(lambda: step(module.CREATE_STREAM, module.OK, data=b"x")) @case("32 fake script bound is enforced") def _(): clock = module.OfflineFakeClock(); one = step(module.CREATE_STREAM, module.OK) expect_failure(lambda: module.OfflineFakeSyscallFacade(clock, tuple(one for _ in range(1025)))) failures = [] for name, function in cases: try: function(); print(f"PASS {name}") except Exception as error: failures.append(f"{name}: {error}"); print(f"FAIL {name}: {error}") if failures: return 1 print(f"Phase-1.0AC dormant-adapter tests passed: {len(cases)}") return 0 if __name__ == "__main__": raise SystemExit(main())