cc-agents-iterm-plugin/tests/test_service.py

import asyncio
import json
import os
import socket
import sys
import types
import unittest
import urllib.request
from unittest import mock

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "files"))
import claude_agents as ca  # noqa: E402


def free_port():
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def http_get(port, path):
    with urllib.request.urlopen(f"http://127.0.0.1:{port}{path}", timeout=3) as r:
        return r.status, r.read().decode("utf-8")


class PortAlive(unittest.TestCase):
    def test_port_alive(self):
        s = socket.socket()
        s.bind(("127.0.0.1", 0))
        s.listen()
        port = s.getsockname()[1]
        self.assertTrue(ca.port_alive(port))
        s.close()
        self.assertFalse(ca.port_alive(port))


class Serve(unittest.TestCase):
    def setUp(self):
        self.port = free_port()
        patches = [
            mock.patch.object(ca, "PORT", self.port),
            mock.patch.object(ca, "CONFIG_PATH", "/nonexistent/claude-agents/config.json"),
            mock.patch.object(ca, "load_sessions", lambda: []),
            mock.patch.object(ca, "fetch_usage", lambda: {}),
            mock.patch.object(ca, "fetch_balance", lambda: ({}, "test")),
        ]
        for p in patches:
            p.start()
            self.addCleanup(p.stop)

    async def _with_server(self, check):
        task = asyncio.create_task(ca.serve())
        loop = asyncio.get_running_loop()
        try:
            for _ in range(60):
                if ca.port_alive(self.port):
                    break
                await asyncio.sleep(0.05)
            return await check(loop)
        finally:
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

    def test_serve_answers(self):
        async def check(loop):
            status, body = await loop.run_in_executor(None, http_get, self.port, "/")
            self.assertEqual(status, 200)
            self.assertIn('<div id="root">', body)
            status, raw = await loop.run_in_executor(None, http_get, self.port, "/raw")
            self.assertIn("tree", json.loads(raw))
        asyncio.run(self._with_server(check))
        self.assertFalse(ca.port_alive(self.port))

    def test_serve_survives_usage_failure(self):
        def no_token():
            raise FileNotFoundError("no credentials")

        async def check(loop):
            body = ""
            for _ in range(60):
                _, body = await loop.run_in_executor(None, http_get, self.port, "/limits")
                if "Не найден токен Claude Code" in body:
                    break
                await asyncio.sleep(0.05)
            self.assertIn("Не найден токен Claude Code", body)
            status, _ = await loop.run_in_executor(None, http_get, self.port, "/")
            self.assertEqual(status, 200)
        with mock.patch.object(ca, "fetch_usage", no_token):
            asyncio.run(self._with_server(check))

    def test_serve_busy_port_raises(self):
        s = socket.socket()
        s.bind(("127.0.0.1", self.port))
        s.listen()
        self.addCleanup(s.close)
        with self.assertRaises(OSError):
            asyncio.run(ca.serve())


class RegisterToolbelt(unittest.TestCase):
    def setUp(self):
        self.registered = []
        self.kicks = []

        async def register(connection, **kwargs):
            self.registered.append(kwargs)

        fake = types.SimpleNamespace(tool=types.SimpleNamespace(async_register_web_view_tool=register))
        patches = [
            mock.patch.object(ca, "iterm2", fake),
            mock.patch.object(ca, "PORT", 8766),
            mock.patch.object(ca, "CONFIG_PATH", "/nonexistent/claude-agents/config.json"),
            mock.patch.object(ca, "kickstart_service", lambda: self.kicks.append(1)),
        ]
        for p in patches:
            p.start()
            self.addCleanup(p.stop)

    def assert_registered(self):
        self.assertEqual(self.registered, [{
            "display_name": "Claude", "identifier": "com.andrey.claude-agents",
            "reveal_if_already_registered": False, "url": "http://127.0.0.1:8766/"}])

    def test_register_toolbelt_kickstarts_when_port_dead(self):
        with mock.patch.object(ca, "port_alive", lambda *a, **k: False):
            asyncio.run(ca.register_toolbelt(object()))
        self.assertEqual(self.kicks, [1])
        self.assert_registered()

    def test_register_toolbelt_skips_kickstart_when_alive(self):
        with mock.patch.object(ca, "port_alive", lambda *a, **k: True):
            asyncio.run(ca.register_toolbelt(object()))
        self.assertEqual(self.kicks, [])
        self.assert_registered()


class Kickstart(unittest.TestCase):
    def test_kickstart_command(self):
        with mock.patch.object(ca.subprocess, "run") as run:
            ca.kickstart_service()
        self.assertEqual(run.call_args.args[0],
                         ["launchctl", "kickstart", f"gui/{os.getuid()}/com.andrey.claude-agents"])
        with mock.patch.object(ca.subprocess, "run", side_effect=OSError("no launchctl")):
            ca.kickstart_service()  # не бросает


class Run(unittest.TestCase):
    def test_run_headless_calls_serve(self):
        called = []

        async def fake_serve():
            called.append(1)
        with mock.patch.object(ca, "serve", fake_serve):
            ca.run(["--headless"])
        self.assertEqual(called, [1])

    def test_run_without_iterm_exits(self):
        with mock.patch.object(ca, "iterm2", None):
            with self.assertRaises(SystemExit):
                ca.run([])


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