diff --git a/tests/deploy/.coveragerc b/tests/deploy/.coveragerc new file mode 100644 index 0000000..9b41183 --- /dev/null +++ b/tests/deploy/.coveragerc @@ -0,0 +1,10 @@ +# Конфиг покрытия для тестов dev/тест-стенда (#21, issue #56). +# +# Из покрытия исключаются сами тестовые модули: они исполняются по определению, +# и без omit «покрытие 100 %» означало бы только то, что тесты запустились. +# Измеряется код стенда: deploy/*.py и tests/deploy/validate_compose.py +# (валидатор манифестов, который вызывает `make test-manifests`). +[report] +omit = + tests/deploy/test_*.py + tests/deploy/conftest.py diff --git a/tests/deploy/pytest.ini b/tests/deploy/pytest.ini new file mode 100644 index 0000000..ab6bac0 --- /dev/null +++ b/tests/deploy/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = . +addopts = -q --cov=deploy --cov=tests/deploy --cov-config=tests/deploy/.coveragerc --cov-report=term-missing --cov-fail-under=70 diff --git a/tests/deploy/test_api_stub.py b/tests/deploy/test_api_stub.py new file mode 100644 index 0000000..1d5cec3 --- /dev/null +++ b/tests/deploy/test_api_stub.py @@ -0,0 +1,157 @@ +"""Тесты 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)