awg-dash/tests/deploy/test_api_stub.py

"""Тесты DEV-ONLY заглушки health-эндпоинта deploy/api_stub.py (задача #21, issue #56).

Заглушка — единственный код стенда, который реально отдаёт HTTP в контейнере,
поэтому она проверяется настоящим HTTP-сервером на свободном порту (port=0),
а не моками: проверяются коды ответов и тело для /api/v1/health, алиаса
/healthz, HEAD и «всего остального» (404 not_implemented).
"""

from __future__ import annotations

import contextlib
import importlib.util
import io
import json
import pathlib
import sys
import threading
import unittest
import urllib.error
import urllib.request
from unittest import mock

HERE = pathlib.Path(__file__).resolve().parent
ROOT = HERE.parents[1]
STUB_PATH = ROOT / "deploy" / "api_stub.py"


def _load_stub() -> object:
    """Импорт deploy/api_stub.py по пути (deploy/ — не пакет)."""
    spec = importlib.util.spec_from_file_location("awg_api_stub", STUB_PATH)
    if spec is None or spec.loader is None:  # pragma: no cover - защита от битого дерева
        raise RuntimeError(f"не удалось загрузить {STUB_PATH}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


api_stub = _load_stub()


class ApiStubHttpTest(unittest.TestCase):
    """Живой HTTP-сервер заглушки: коды ответов и контракт тела."""

    @classmethod
    def setUpClass(cls) -> None:
        cls._stdout = contextlib.redirect_stdout(io.StringIO())
        cls._stdout.__enter__()
        cls.server = api_stub.make_server("127.0.0.1", 0)
        cls.port = cls.server.server_address[1]
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()

    @classmethod
    def tearDownClass(cls) -> None:
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=5)
        cls._stdout.__exit__(None, None, None)

    def _url(self, path: str) -> str:
        return f"http://127.0.0.1:{self.port}{path}"

    def _get(self, path: str) -> tuple[int, dict]:
        try:
            with urllib.request.urlopen(self._url(path), timeout=5) as resp:
                return resp.status, json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:  # 4xx/5xx — тоже ответ
            return exc.code, json.loads(exc.read().decode("utf-8"))

    def test_health_returns_200_and_stub_marker(self) -> None:
        status, payload = self._get("/api/v1/health")
        self.assertEqual(200, status)
        self.assertEqual("ok", payload["status"])
        self.assertIs(True, payload["stub"], "health-заглушка обязана помечать ответ stub=true")
        self.assertEqual("#31", payload["issue"], "в ответе должна быть ссылка на задачу реального API")
        self.assertIsInstance(payload["uptime_s"], float)

    def test_healthz_is_alias_of_health(self) -> None:
        status, payload = self._get("/healthz")
        self.assertEqual(200, status)
        self.assertIs(True, payload["stub"])

    def test_query_string_is_ignored(self) -> None:
        status, _ = self._get("/api/v1/health?verbose=1&x=2")
        self.assertEqual(200, status)

    def test_unknown_path_returns_404_not_implemented(self) -> None:
        status, payload = self._get("/api/v1/tunnels")
        self.assertEqual(404, status)
        self.assertEqual("not_implemented", payload["error"])
        self.assertEqual("#31", payload["issue"])
        self.assertEqual("/api/v1/tunnels", payload["path"])

    def test_head_mirrors_get_status(self) -> None:
        for path, expected in (("/api/v1/health", 200), ("/api/v1/tunnels", 404)):
            with self.subTest(path=path):
                req = urllib.request.Request(self._url(path), method="HEAD")
                try:
                    with urllib.request.urlopen(req, timeout=5) as resp:
                        self.assertEqual(expected, resp.status)
                except urllib.error.HTTPError as exc:
                    self.assertEqual(expected, exc.code)

    def test_post_is_rejected(self) -> None:
        req = urllib.request.Request(self._url("/api/v1/health"), data=b"{}", method="POST")
        with self.assertRaises(urllib.error.HTTPError) as ctx:
            urllib.request.urlopen(req, timeout=5)
        self.assertEqual(501, ctx.exception.code, "заглушка не должна поддерживать POST")


class ApiStubMainTest(unittest.TestCase):
    """main(): сборка сервера из аргументов командной строки."""

    class _FakeServer:
        instances: list[tuple[tuple, object]] = []
        served = 0

        def __init__(self, address: tuple, handler: object) -> None:
            type(self).instances.append((address, handler))

        def serve_forever(self) -> None:
            type(self).served += 1

    def setUp(self) -> None:
        self._fake = type("Fake", (self._FakeServer,), {"instances": [], "served": 0})
        self._patch = mock.patch.object(api_stub, "ThreadingHTTPServer", self._fake)
        self._patch.start()
        self.addCleanup(self._patch.stop)

    def _run_main(self, argv: list[str]) -> tuple[int, str]:
        out = io.StringIO()
        with mock.patch.object(sys, "argv", ["api_stub.py", *argv]), contextlib.redirect_stdout(out):
            rc = api_stub.main()
        return rc, out.getvalue()

    def test_main_binds_requested_host_and_port(self) -> None:
        rc, out = self._run_main(["--host", "127.0.0.1", "--port", "0", "--domain", str(HERE)])
        self.assertEqual(0, rc)
        self.assertEqual([(("127.0.0.1", 0), api_stub.Handler)], self._fake.instances)
        self.assertEqual(1, self._fake.served)
        self.assertIn("DEV-ONLY", out)
        self.assertIn("доступен", out)

    def test_main_reports_missing_domain_directory(self) -> None:
        rc, out = self._run_main(["--port", "0", "--domain", str(HERE / "нет-такого-каталога")])
        self.assertEqual(0, rc)
        self.assertIn("НЕ доступен", out)

    def test_defaults_listen_on_all_interfaces(self) -> None:
        rc, _ = self._run_main(["--domain", str(HERE)])
        self.assertEqual(0, rc)
        self.assertEqual((("0.0.0.0", 8080), api_stub.Handler), self._fake.instances[0])


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