sssf-opencode/tests/test_opencode_adapter.py

import json
import os
import stat
import tempfile
import time
import unittest
from pathlib import Path
from unittest.mock import patch

from sssf_opencode.adapter import OpenCodeAdapter, OpenCodeRequest, OpenCodeRunError


class OpenCodeAdapterTests(unittest.TestCase):
    def make_fake_opencode(self, directory: Path) -> Path:
        binary = directory / "fake-opencode"
        binary.write_text(
            "#!/usr/bin/env python3\n"
            "import json, os, pathlib, sys\n"
            "args = sys.argv[1:]\n"
            "session = args[args.index('--session') + 1] if '--session' in args else 'ses_new'\n"
            "pathlib.Path(os.environ['FAKE_ARGS_PATH']).write_text(json.dumps(args))\n"
            "print(json.dumps({'type': 'step_start', 'sessionID': session}))\n"
            "report = json.dumps({'status': 'success'})\n"
            "print(json.dumps({'type': 'text', 'sessionID': session, 'part': {'text': report[:10]}}))\n"
            "print(json.dumps({'type': 'text', 'sessionID': session, 'part': {'text': report[10:]}}))\n"
            "print(json.dumps({'type': 'step_finish', 'sessionID': session, 'part': {'tokens': {'input': 11, 'output': 7, 'reasoning': 3, 'cache': {'read': 5, 'write': 2}}, 'cost': 0.125}}))\n"
        )
        binary.chmod(binary.stat().st_mode | stat.S_IXUSR)
        return binary

    def test_streams_text_and_persists_open_code_session_id(self):
        with tempfile.TemporaryDirectory() as temp:
            directory = Path(temp)
            args_path = directory / "args.json"
            adapter = OpenCodeAdapter(binary=str(self.make_fake_opencode(directory)))

            with patch.dict(os.environ, {"FAKE_ARGS_PATH": str(args_path)}):
                result = adapter.run(
                    OpenCodeRequest(
                        prompt="return the report",
                        agent="factory-builder",
                        model="openai/gpt-5.3-codex",
                        cwd=directory,
                    )
                )

            self.assertEqual(result.session_id, "ses_new")
            self.assertEqual(result.text, '{"status": "success"}')
            self.assertEqual(result.input_tokens, 11)
            self.assertEqual(result.output_tokens, 7)
            self.assertEqual(result.cache_read_tokens, 5)
            self.assertEqual(result.cache_write_tokens, 2)
            self.assertEqual(result.reasoning_tokens, 3)
            self.assertEqual(result.cost, 0.125)
            self.assertEqual(
                json.loads(args_path.read_text()),
                [
                    "run",
                    "--format",
                    "json",
                    "--agent",
                    "factory-builder",
                    "--model",
                    "openai/gpt-5.3-codex",
                    "--dir",
                    str(directory),
                    "return the report",
                ],
            )
    def test_continues_a_real_open_code_session(self):
        with tempfile.TemporaryDirectory() as temp:
            directory = Path(temp)
            args_path = directory / "args.json"
            adapter = OpenCodeAdapter(binary=str(self.make_fake_opencode(directory)))

            with patch.dict(os.environ, {"FAKE_ARGS_PATH": str(args_path)}):
                result = adapter.run(
                    OpenCodeRequest(
                        prompt="repair the report",
                        agent="factory-builder",
                        model="openai/gpt-5.3-codex",
                        cwd=directory,
                        session_id="ses_existing",
                    )
                )

            self.assertEqual(result.session_id, "ses_existing")
            self.assertEqual(
                json.loads(args_path.read_text())[-3:],
                ["--session", "ses_existing", "repair the report"],
            )

    def test_reports_json_error_events(self):
        with tempfile.TemporaryDirectory() as temp:
            directory = Path(temp)
            binary = directory / "erroring-opencode"
            binary.write_text(
                "#!/bin/sh\n"
                "printf '%s\\n' '{\"type\":\"error\",\"sessionID\":\"ses_bad\",\"error\":{\"data\":{\"message\":\"denied\",\"ref\":\"err_1\"}}}'\n"
                "exit 1\n"
            )
            binary.chmod(binary.stat().st_mode | stat.S_IXUSR)

            with self.assertRaisesRegex(OpenCodeRunError, r"denied.*err_1"):
                OpenCodeAdapter(binary=str(binary)).run(
                    OpenCodeRequest(
                        prompt="fail cleanly",
                        agent="factory-planner",
                        model="zai/glm-5.2",
                        cwd=directory,
                    )
                )

    def test_stops_on_a_nonretryable_json_error(self):
        with tempfile.TemporaryDirectory() as temp:
            directory = Path(temp)
            binary = directory / "hanging-error-opencode"
            binary.write_text(
                "#!/usr/bin/env python3\n"
                "import json, time\n"
                "print(json.dumps({'type': 'error', 'error': {'data': {'message': 'expired', 'isRetryable': False}}}), flush=True)\n"
                "time.sleep(3)\n"
                "raise SystemExit(1)\n"
            )
            binary.chmod(binary.stat().st_mode | stat.S_IXUSR)
            started = time.monotonic()

            with self.assertRaisesRegex(OpenCodeRunError, "expired"):
                OpenCodeAdapter(binary=str(binary)).run(
                    OpenCodeRequest(
                        prompt="stop after terminal error",
                        agent="factory-planner",
                        model="zai/glm-5.2",
                        cwd=directory,
                    )
                )

            self.assertLess(time.monotonic() - started, 1.5)

    def test_times_out_when_opencode_emits_nothing(self):
        with tempfile.TemporaryDirectory() as temp:
            directory = Path(temp)
            binary = directory / "silent-opencode"
            binary.write_text(
                "#!/usr/bin/env python3\n"
                "import time\n"
                "time.sleep(3)\n"
            )
            binary.chmod(binary.stat().st_mode | stat.S_IXUSR)
            adapter = OpenCodeAdapter(binary=str(binary), timeout_seconds=0.05)
            started = time.monotonic()

            with self.assertRaisesRegex(OpenCodeRunError, r"timed out after 0.05 seconds"):
                adapter.run(
                    OpenCodeRequest(
                        prompt="stop a silent turn",
                        agent="factory-planner",
                        model="zai/glm-5.2",
                        cwd=directory,
                    )
                )

            self.assertLess(time.monotonic() - started, 1.5)


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