atlas-iac/testing/tests/test_hermes_scm_server_coverage.py

"""Behavioral branch coverage for the SCM broker HTTP server primitives."""

from __future__ import annotations

import io

import pytest

from testing.tests.test_hermes_scm_broker_support import _load


class _Connection:
    def __init__(self):
        self.timeouts = []

    def settimeout(self, value):
        self.timeouts.append(value)


class _FlushBuffer(io.BytesIO):
    def __init__(self):
        super().__init__()
        self.flushes = 0

    def flush(self):
        self.flushes += 1


def test_absolute_reader_supports_idle_limit_newline_eof_and_attribute_delegation():
    module = _load("scm_broker_server")
    stream = io.BytesIO(b"first\nsecond")
    reader = module._AbsoluteDeadlineReader(stream, _Connection())
    assert reader.readline(3) == b"fir"
    reader.begin(5)
    assert reader.readline() == b"st\n"
    assert reader.readline() == b"second"
    assert reader.readline() == b""
    assert reader.closed is False
    reader.end()
    assert reader.readline() == b""


@pytest.mark.parametrize(
    ("phase", "detail", "expected"),
    [
        ("pack", "Git command does not match its task grant", "grant-command"),
        ("pack", "Git update does not prove fast-forward ancestry", "ancestry"),
        ("pack", "Git delta base is outside the pack; push full packs", "thin-pack"),
        ("pack", "Git push contains runtime credential material", "credential"),
        ("pack", "Git push contains credential-shaped content", "content"),
        ("pack", "Git receive-pack command framing is invalid", "framing"),
        ("pack", "Git push is limited to namespaced feature branches", "ref"),
        ("control", "arbitrary policy detail", "policy"),
    ],
)
def test_rejection_log_uses_fixed_pack_categories_without_error_text(
    monkeypatch, phase, detail, expected
):
    module = _load("scm_broker_server")
    policy = type("PolicyError", (Exception,), {})
    observed = []
    monkeypatch.setattr(module.logging, "warning", lambda message, *args: observed.append((message, args)))

    module.log_rejection(phase, policy(detail))

    assert observed == [("scm_rejected phase=%s category=%s", (phase, expected))]


def _handler_type(module):
    class StubHandler:
        request_version = "HTTP/1.1"
        command = "GET"

        def setup(self):
            self.connection = _Connection()
            self.rfile = io.BytesIO(b"")
            self.wfile = _FlushBuffer()

        def parse_request(self):
            return self.parse_result

        def send_error(self, code, *_args):
            self.errors.append(code)

        def do_GET(self):
            self.calls.append("GET")

    return type(
        "DeadlineHandler", (module.AbsoluteHeaderDeadlineMixin, StubHandler), {}
    )


def _handler(module, raw: bytes, *, parse=True):
    handler_type = _handler_type(module)
    handler = object.__new__(handler_type)
    handler.setup()
    handler._header_reader._stream = io.BytesIO(raw)
    handler.parse_result = parse
    handler.errors = []
    handler.calls = []
    return handler


def test_header_mixin_setup_and_known_request_dispatch():
    module = _load("scm_broker_server")
    handler = _handler(module, b"GET /healthz HTTP/1.1\r\n")
    handler.handle_one_request()
    assert handler.calls == ["GET"]
    assert handler.wfile.flushes == 1
    assert handler._header_reader._deadline is None


@pytest.mark.parametrize(
    ("raw", "parse", "command", "expected_error", "closed"),
    [
        (b"x" * 65537, True, "GET", 414, False),
        (b"", True, "GET", None, True),
        (b"GET / HTTP/1.1\r\n", False, "GET", None, False),
        (b"TRACE / HTTP/1.1\r\n", True, "TRACE", 501, False),
    ],
)
def test_header_mixin_rejects_long_empty_unparsed_and_unknown_requests(
    raw, parse, command, expected_error, closed
):
    module = _load("scm_broker_server")
    handler = _handler(module, raw, parse=parse)
    handler.command = command
    handler.close_connection = False
    handler.handle_one_request()
    assert handler.errors == ([] if expected_error is None else [expected_error])
    assert handler.close_connection is closed


def test_header_mixin_closes_on_absolute_timeout(monkeypatch):
    module = _load("scm_broker_server")
    handler = _handler(module, b"GET / HTTP/1.1\r\n")
    handler.close_connection = False
    monkeypatch.setattr(
        handler._header_reader,
        "readline",
        lambda *_a, **_k: (_ for _ in ()).throw(TimeoutError()),
    )
    handler.handle_one_request()
    assert handler.close_connection is True


class _Slots:
    def __init__(self, available=True):
        self.available = available
        self.releases = 0

    def acquire(self, *, blocking):
        assert blocking is False
        return self.available

    def release(self):
        self.releases += 1


def _server_shell(module, available=True):
    server = object.__new__(module.BoundedThreadingHTTPServer)
    server._slots = _Slots(available)
    server.shutdowns = []
    server.shutdown_request = server.shutdowns.append
    return server


def test_bounded_server_delegates_accepted_request(monkeypatch):
    module = _load("scm_broker_server")
    server = _server_shell(module)
    calls = []
    monkeypatch.setattr(
        module.ThreadingHTTPServer,
        "process_request",
        lambda self, request, address: calls.append((request, address)),
    )
    server.process_request("socket", ("127.0.0.1", 1))
    assert calls == [("socket", ("127.0.0.1", 1))]
    assert server._slots.releases == 0


def test_bounded_server_releases_on_dispatch_failure(monkeypatch):
    module = _load("scm_broker_server")
    server = _server_shell(module)
    monkeypatch.setattr(
        module.ThreadingHTTPServer,
        "process_request",
        lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("dispatch")),
    )
    with pytest.raises(RuntimeError, match="dispatch"):
        server.process_request("socket", ("127.0.0.1", 1))
    assert server._slots.releases == 1
    assert server.shutdowns == ["socket"]


def test_bounded_server_thread_always_releases(monkeypatch):
    module = _load("scm_broker_server")
    server = _server_shell(module)
    monkeypatch.setattr(
        module.ThreadingHTTPServer,
        "process_request_thread",
        lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("handler")),
    )
    with pytest.raises(RuntimeError, match="handler"):
        server.process_request_thread("socket", ("127.0.0.1", 1))
    assert server._slots.releases == 1