#!/usr/bin/env python3 from __future__ import annotations import argparse import datetime as dt import json import re import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from tests.harness.agent_process import AgentProcess from tests.harness.didactyl_client import DidactylClient from tests.harness.log_watcher import LogWatcher from tests.harness.reporter import Reporter from tests.harness.test_runner import TestRunner def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Didactyl automated test harness") parser.add_argument("--config", default="tests/configs/test_genesis.jsonc") parser.add_argument("--keys", default="tests/configs/test_keys.jsonc", help="Path to local keys JSONC file") parser.add_argument("--binary", default=None, help="Path to didactyl binary; if omitted, build_static --debug is used") parser.add_argument("--suite", action="append", help="Run only this suite (repeatable)") parser.add_argument("--test", action="append", help="Run only this test name (repeatable)") parser.add_argument("--api-port", type=int, default=8485) parser.add_argument("--api-bind", default="127.0.0.1") parser.add_argument("--timeout", type=float, default=60.0) parser.add_argument("--debug-level", type=int, default=5) parser.add_argument("--output-dir", default=None) parser.add_argument("--verbose", action="store_true") parser.add_argument("--no-restart", action="store_true") parser.add_argument( "--skip-build-static-debug", action="store_true", help="Do not invoke ./build_static.sh --debug before running", ) return parser.parse_args() def build_static_debug_binary() -> None: cmd = ["./build_static.sh", "--debug"] print("Building debug static binary via ./build_static.sh --debug ...") subprocess.run(cmd, check=True) def resolve_default_debug_binary() -> Path: candidates = [ Path("./didactyl_static_x86_64_debug"), Path("./didactyl_static_arm64_debug"), Path("./didactyl_static_armv7_debug"), Path("./didactyl_static_armv6_debug"), ] existing = [p for p in candidates if p.exists() and p.is_file()] if existing: return existing[0] return Path("./didactyl_static_x86_64_debug") def _strip_jsonc(text: str) -> str: text = re.sub(r"//.*?$", "", text, flags=re.MULTILINE) text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL) text = re.sub(r",\s*([}\]])", r"\1", text) return text def _load_keys(keys_path: Path) -> tuple[str | None, str | None]: raw = keys_path.read_text(encoding="utf-8") data = json.loads(_strip_jsonc(raw)) agent_nsec = None admin_pubkey = None if isinstance(data, dict): agent = data.get("agent") admin = data.get("admin") if isinstance(agent, dict): nsec = agent.get("nsec") if isinstance(nsec, str) and nsec.strip(): agent_nsec = nsec.strip() if isinstance(admin, dict): pubkey = admin.get("pubkey") if isinstance(pubkey, str) and pubkey.strip(): admin_pubkey = pubkey.strip() return agent_nsec, admin_pubkey def _apply_keys_to_config_text(config_text: str, agent_nsec: str | None, admin_pubkey: str | None) -> str: out = config_text if agent_nsec: out = re.sub( r'("nsec"\s*:\s*")[^"]*(")', lambda m: f"{m.group(1)}{agent_nsec}{m.group(2)}", out, count=1, ) if admin_pubkey: out = re.sub( r'("pubkey"\s*:\s*")[^"]*(")', lambda m: f"{m.group(1)}{admin_pubkey}{m.group(2)}", out, count=1, ) return out def prepare_runtime_config(config_path: Path, keys_path: Path, output_dir: Path) -> Path: config_text = config_path.read_text(encoding="utf-8") if not keys_path.exists(): return config_path agent_nsec, admin_pubkey = _load_keys(keys_path) rendered = _apply_keys_to_config_text(config_text, agent_nsec, admin_pubkey) runtime_config = output_dir / "runtime_test_genesis.jsonc" runtime_config.write_text(rendered, encoding="utf-8") return runtime_config def main() -> int: args = parse_args() run_ts = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") output_dir = args.output_dir or f"tests/results/{run_ts}" Path(output_dir).mkdir(parents=True, exist_ok=True) if not args.skip_build_static_debug: try: build_static_debug_binary() except subprocess.CalledProcessError as exc: print(f"build_static debug build failed with exit code {exc.returncode}") return 2 binary = Path(args.binary) if args.binary else resolve_default_debug_binary() if not binary.exists(): print(f"didactyl binary not found: {binary}") print("Pass --binary or run ./build_static.sh --debug") return 2 config = Path(args.config) if not config.exists(): print(f"config not found: {config}") print("Create it from template: cp tests/configs/test_genesis.jsonc.example tests/configs/test_genesis.jsonc") return 2 keys = Path(args.keys) runtime_config = prepare_runtime_config(config, keys, Path(output_dir)) if not keys.exists(): print(f"keys file not found (optional): {keys}") print("Create it from template for stable test identity: cp tests/configs/test_keys.jsonc.example tests/configs/test_keys.jsonc") log_file = str(Path(output_dir) / "agent_debug.log") base_url = f"http://{args.api_bind}:{args.api_port}" agent = AgentProcess( binary_path=str(binary), config_path=str(runtime_config), api_port=args.api_port, api_bind=args.api_bind, debug_level=args.debug_level, log_file=log_file, base_url=base_url, ) client = DidactylClient(base_url=base_url, timeout=args.timeout, verify_tls=False) log = LogWatcher(log_file) log.start() started = agent.start(timeout=30) if not started: print("Failed to start Didactyl agent") log.stop() return 1 try: runner = TestRunner(agent=agent, client=client, log=log, args=args) tests = runner.discover_suites("tests.suites") if not tests: print("No tests discovered") return 3 results = runner.run_all(tests) run_meta = { "run_timestamp": dt.datetime.now(dt.timezone.utc).isoformat(), "base_url": base_url, "config": str(runtime_config), "binary": str(binary), "test_count": len(results), } reporter = Reporter(results=results, run_meta=run_meta, output_dir=output_dir) reporter.print_summary() reporter.print_details() json_path = reporter.write_json("results.json") txt_path = reporter.write_text("results.txt") print(f"\nWrote: {json_path}") print(f"Wrote: {txt_path}") bad = [r for r in results if r.status in {"fail", "error", "timeout"}] return 1 if bad else 0 finally: agent.stop(timeout=10) log.stop() if __name__ == "__main__": raise SystemExit(main())