tests: покрытие health-заглушки и отдельный профиль pytest (#56)
- tests/deploy/pytest.ini + .coveragerc: гейт ≥70 % для кода стенда (deploy/ и validate_compose.py), тестовые модули из покрытия исключены. Refs #21, Refs #56
This commit is contained in:
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user