enduro-trails/tests/unit/test_healthcheck_oneliner.py

"""Unit-тесты исполняемого поведения healthcheck-one-liner'а (ET-015).

Тестируем именно тот код, который зашит в `docker-compose.yml`, чтобы
гарантировать поведение exit-кода в трёх сценариях (UT-01..UT-03):

    UT-01 (AC-06):       exit 0 при HTTP 200.
    UT-02 (AC-05/AC-06): exit ≠ 0 при недоступном порту.
    UT-03 (AC-06):       exit ≠ 0 при HTTP 500.

URL в one-liner подменяется на адрес мок-сервера, остальной код
выполняется ровно тот же, что и внутри контейнера.
"""
from __future__ import annotations

import socket
import subprocess
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

import pytest
import yaml

REPO_ROOT = Path(__file__).resolve().parents[2]
COMPOSE_PATH = REPO_ROOT / "docker-compose.yml"
PROD_HEALTH_URL = "http://localhost:5556/api/health"


def _load_oneliner() -> str:
    """Возвращает 4-й элемент массива test (сам python-код), как в compose."""
    data = yaml.safe_load(COMPOSE_PATH.read_text(encoding="utf-8"))
    test = data["services"]["app"]["healthcheck"]["test"]
    assert isinstance(test, list) and len(test) >= 4, f"unexpected healthcheck.test: {test!r}"
    code = test[3]
    assert isinstance(code, str)
    return code


def _pick_unused_port() -> int:
    """Свободный TCP-порт на 127.0.0.1 (грязно, но достаточно для теста)."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _retarget(code: str, url: str) -> str:
    """Заменяет prod-URL внутри one-liner на тестовый.

    Используем именно подмену строки (а не отдельный код), чтобы под тест
    шла та же логика урлопен + проверки статуса, что и в проде.
    """
    assert PROD_HEALTH_URL in code, (
        f"one-liner должен содержать {PROD_HEALTH_URL!r}, иначе ретаргет небезопасен. "
        f"Код: {code!r}"
    )
    return code.replace(PROD_HEALTH_URL, url)


def _run(code: str, timeout: float = 10.0) -> subprocess.CompletedProcess:
    return subprocess.run(
        [sys.executable, "-c", code],
        capture_output=True,
        text=True,
        timeout=timeout,
    )


class _FixedStatusHandler(BaseHTTPRequestHandler):
    status_code = 200
    body = b'{"status":"ok"}'

    def do_GET(self):  # noqa: N802 — имя задано BaseHTTPRequestHandler
        self.send_response(self.status_code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(self.body)))
        self.end_headers()
        self.wfile.write(self.body)

    def log_message(self, *_args, **_kwargs):  # тишина в pytest-логах
        pass


@pytest.fixture()
def mock_server():
    """Поднимает локальный HTTPServer с настраиваемым статусом."""
    started = []

    def _start(status_code: int):
        handler = type(
            "_H",
            (_FixedStatusHandler,),
            {"status_code": status_code},
        )
        server = HTTPServer(("127.0.0.1", 0), handler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        started.append((server, thread))
        port = server.server_address[1]
        return f"http://127.0.0.1:{port}/"

    yield _start

    for server, thread in started:
        server.shutdown()
        server.server_close()
        thread.join(timeout=2)


# ───────────────────────── UT-01 (AC-06) ─────────────────────────

def test_ut01_returns_zero_on_http_200(mock_server):
    url = mock_server(200)
    code = _retarget(_load_oneliner(), url)
    result = _run(code)
    assert result.returncode == 0, (
        f"ожидался exit code 0 при HTTP 200, получили {result.returncode}. "
        f"stderr={result.stderr!r}"
    )


# ───────────────────────── UT-02 (AC-05/AC-06) ─────────────────────────

def test_ut02_returns_nonzero_when_port_unused():
    port = _pick_unused_port()
    code = _retarget(_load_oneliner(), f"http://127.0.0.1:{port}/")
    result = _run(code)
    assert result.returncode != 0, (
        f"ожидался ненулевой exit code, когда никто не слушает порт, "
        f"но получили 0. stderr={result.stderr!r}"
    )


# ───────────────────────── UT-03 (AC-06) ─────────────────────────

@pytest.mark.parametrize("status_code", [301, 404, 500, 503])
def test_ut03_returns_nonzero_on_non_2xx(mock_server, status_code):
    """Любой не-200 ответ должен трактоваться как unhealthy.

    one-liner из ADR-020 проверяет `status == 200`, всё остальное → exit 1
    (либо HTTPError → ненулевой exit). Параметризация — защита от
    регресса, если кто-то сменит условие на `< 400` и т.п.
    """
    url = mock_server(status_code)
    code = _retarget(_load_oneliner(), url)
    result = _run(code)
    assert result.returncode != 0, (
        f"ожидался ненулевой exit code при HTTP {status_code}, "
        f"получили 0. stderr={result.stderr!r}"
    )