.pouch/skills/agent-runtime/tests/test_runner.py

#!/usr/bin/env python3

from __future__ import annotations

import importlib.util
import base64
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock


SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "agent.py"
SPEC = importlib.util.spec_from_file_location("agent_runtime_runner", SCRIPT)
assert SPEC and SPEC.loader
agent = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(agent)


class RunnerTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        pouch = self.root / ".pouch"
        pouch.mkdir()
        (pouch / "agent.yaml").write_text(
            """schemaVersion: 1
locations:
  local-app:
    environment: local
    host: local
    project: local-project
    path: /tmp/writes
  remote-mac:
    environment: macmini
    host: runtime:0123456789abcdef0123
    project: remote-project
    path: /Users/tester/writes
profiles:
  test-agent:
    cli: codex
    model: null
    reasoningEffort: medium
    permissionMode: workspace-write
bindings:
  tester.android:
    adapter: tester
    location: local-app
    profile: test-agent
  tester.ios:
    adapter: tester
    location: remote-mac
    profile: test-agent
""",
            encoding="utf-8",
        )

    def tearDown(self) -> None:
        self.temp.cleanup()

    def test_local_binding_routes_to_fixed_probe(self):
        def command(argv, **_kwargs):
            if argv == ["adb", "version"]:
                return 0, "Android Debug Bridge version 1"
            if argv == ["adb", "devices"]:
                return 0, "List of devices attached\nemulator-5554\tdevice\n"
            self.assertEqual(argv, ["adb", "-s", "emulator-5554", "shell", "getprop", "sys.boot_completed"])
            return 0, "1"

        with mock.patch.object(agent, "run_fixed_command", side_effect=command) as runner:
            result = agent.execute_runner_request(self.root, {
                "binding": "tester.android", "operation": "platform_probe", "platform": "android", "node": "local",
            })
        self.assertEqual(result["status"], "ready")
        self.assertEqual(set(result["checks"]), {"adb", "device", "boot"})
        self.assertEqual(runner.call_count, 3)

    def test_remote_binding_uses_orca_terminal_transport(self):
        marker = "POUCH_RUNNER_RESULT_" + ("0" * 31) + "1"
        payload = {"status": "ready", "observed": "Darwin/xcrun/simctl probe completed", "checks": {"macos": {"status": "ready"}}}

        def fake_orca(argv, _environment=None):
            if argv[:2] == ["terminal", "create"]:
                return {"terminal": {"handle": "term-1"}}
            if argv[:2] == ["terminal", "read"]:
                return {"terminal": {"tail": [marker + base64.b64encode(json.dumps(payload).encode()).decode()]}}
            return {}

        with mock.patch.object(agent, "worktree_for", return_value={"id": "wt-1", "path": "/Users/tester/writes"}) as resolve, \
                mock.patch.object(agent, "orca_json", side_effect=fake_orca), \
                mock.patch.object(agent.uuid, "uuid4", return_value=mock.Mock(hex="0" * 31 + "1")):
            result = agent.execute_runner_request(self.root, {
                "binding": "tester.ios", "operation": "platform_probe", "platform": "ios", "node": "macmini",
            })
        self.assertEqual(result["status"], "ready")
        self.assertEqual(result["checks"]["binding"]["status"], "ready")
        resolve.assert_called_once()

    def test_remote_transport_submits_startup_command_when_first_read_has_no_marker(self):
        marker = "POUCH_RUNNER_RESULT_" + ("0" * 31) + "1"
        payload = {"status": "ready", "observed": "Darwin/xcrun/simctl probe completed", "checks": {"macos": {"status": "ready"}}}
        reads = [
            {"terminal": {"tail": ["Welcome to fish"]}},
            {"terminal": {"tail": [marker + base64.b64encode(json.dumps(payload).encode()).decode()]}},
        ]

        def fake_orca(argv, _environment=None):
            if argv[:2] == ["terminal", "create"]:
                return {"terminal": {"handle": "term-1"}}
            if argv[:2] == ["terminal", "read"]:
                return reads.pop(0)
            return {}

        with mock.patch.object(agent, "worktree_for", return_value={"id": "wt-1", "path": "/Users/tester/writes"}), \
                mock.patch.object(agent, "orca_json", side_effect=fake_orca) as orca, \
                mock.patch.object(agent.uuid, "uuid4", return_value=mock.Mock(hex="0" * 31 + "1")):
            result = agent.execute_runner_request(self.root, {
                "binding": "tester.ios", "operation": "platform_probe", "platform": "ios", "node": "macmini",
            })
        self.assertEqual(result["status"], "ready")
        self.assertTrue(any(call.args[0][:2] == ["terminal", "send"] for call in orca.call_args_list))

    def test_unknown_operation_and_command_failure_are_structured(self):
        unknown = agent.execute_runner_request(self.root, {
            "binding": "tester.android", "operation": "exec", "platform": "android",
        })
        self.assertEqual(unknown["status"], "blocked")
        with mock.patch.object(agent, "run_fixed_command", return_value=(127, "adb missing")):
            failed = agent.execute_runner_request(self.root, {
                "binding": "tester.android", "operation": "platform_probe", "platform": "android",
            })
        self.assertEqual(failed["status"], "environment-fail")

    def test_json_stdout_is_single_protocol_object(self):
        process = subprocess.run(
            [sys.executable, str(SCRIPT), "run", "exec", "--root", str(self.root)],
            input=json.dumps({"binding": "tester.android", "operation": "unknown", "platform": "android"}),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
        self.assertEqual(process.returncode, 0)
        self.assertEqual(json.loads(process.stdout)["status"], "blocked")
        self.assertEqual(process.stderr, "")


if __name__ == "__main__":
    unittest.main()