atlas-iac/testing/tests/test_hermes_scm_broker_internal_coverage.py

"""Behavioral branch coverage for SCM broker protocol helpers."""

from __future__ import annotations

import io
from email.message import Message

import pytest

from testing.tests.test_hermes_scm_broker_support import (
    Response,
    _load,
    _receive_command,
)


def test_broker_redirect_status_and_bounded_read_helpers(monkeypatch):
    broker = _load("scm_broker")
    with pytest.raises(broker.PolicyError, match="redirects"):
        broker.RejectRedirect().redirect_request(
            object(), None, 302, "redirect", {}, "https://elsewhere.invalid"
        )

    class CodeOnly:
        def getcode(self):
            return 204

    assert broker._status(type("Status", (), {"status": 200})()) == 200
    assert broker._status(CodeOnly()) == 204
    assert broker._status(object()) is None

    with pytest.raises(broker.PolicyError, match="safe size"):
        broker._read_bounded(io.BytesIO(b"x"), 4, -1)
    with pytest.raises(broker.PolicyError, match="safe size"):
        broker._read_bounded(io.BytesIO(b"xxxxx"), 4)
    with pytest.raises(broker.PolicyError, match="safe size"):
        broker._read_bounded(io.BytesIO(b"x"), 4, 2)
    with pytest.raises(broker.PolicyError, match="deadline"):
        broker._read_bounded(io.BytesIO(b"x"), 4, 1, deadline=0)

    timeouts = []
    assert (
        broker._read_bounded(
            io.BytesIO(b"xy"),
            4,
            2,
            deadline=broker.time.monotonic() + 5,
            set_timeout=timeouts.append,
        )
        == b"xy"
    )
    assert timeouts


def test_spool_bounded_rejects_bounds_early_eof_and_closes_on_read_error():
    broker = _load("scm_broker")
    for length in (-1, 5):
        with pytest.raises(broker.PolicyError, match="safe size"):
            broker._spool_bounded(
                io.BytesIO(b"x"),
                4,
                length,
                token="sentinel",
                context="request",
                deadline=999999999,
            )
    with pytest.raises(broker.PolicyError, match="ended early"):
        broker._spool_bounded(
            io.BytesIO(b"x"),
            4,
            2,
            token="sentinel",
            context="request",
            deadline=999999999,
        )

    class Broken:
        def read(self, _size):
            raise OSError("broken stream")

    with pytest.raises(OSError, match="broken"):
        broker._spool_bounded(
            Broken(), 4, 1, token="sentinel", context="request", deadline=999999999
        )


class _HandlerInput:
    def __init__(self, body: bytes, *, content_type="application/json", length=None):
        self.headers = Message()
        self.headers["Content-Type"] = content_type
        if length is None:
            length = len(body)
        self.headers["Content-Length"] = str(length)
        self.rfile = io.BytesIO(body)
        self.connection = type(
            "Connection", (), {"settimeout": lambda self, value: None}
        )()


def test_control_json_loader_requires_fixed_json_object():
    broker = _load("scm_broker")
    handler = _HandlerInput(b"{}")
    handler.headers["Transfer-Encoding"] = "chunked"
    with pytest.raises(broker.PolicyError, match="chunked"):
        broker._load_json(handler)
    with pytest.raises(broker.PolicyError, match="must be JSON"):
        broker._load_json(_HandlerInput(b"{}", content_type="text/plain"))
    with pytest.raises(broker.PolicyError, match="must be an object"):
        broker._load_json(_HandlerInput(b"[]"))
    assert broker._load_json(_HandlerInput(b'{"path":"safe"}')) == {"path": "safe"}


@pytest.mark.parametrize(
    ("target", "expected"),
    [
        (
            "/git/atlas/cassandra.git/info/refs?service=git-upload-pack",
            ("cassandra", "info/refs", "git-upload-pack"),
        ),
        (
            "/git/atlas/cassandra.git/info/refs?service=git-receive-pack",
            ("cassandra", "info/refs", "git-receive-pack"),
        ),
        (
            "/git/atlas/cassandra.git/git-upload-pack",
            ("cassandra", "git-upload-pack", "git-upload-pack"),
        ),
        (
            "/git/atlas/cassandra.git/git-receive-pack",
            ("cassandra", "git-receive-pack", "git-receive-pack"),
        ),
    ],
)
def test_git_target_accepts_each_exact_smart_http_route(target, expected):
    broker = _load("scm_broker")
    assert broker._git_target(target) == expected


def test_content_length_and_credential_helpers_cover_safe_and_rejected_values():
    broker = _load("scm_broker")
    headers = Message()
    headers["Content-Length"] = "0"
    assert broker._content_length(headers, 1) == 0
    forms = broker._credential_forms("sentinel")
    assert len(forms) == 3 and forms[0] == b"sentinel"
    zero = b"0" * 40
    commit = b"1" * 40
    line = zero + b" " + commit + b" refs/heads/hermes/x " + forms[2] + b"\n"
    framed = f"{len(line) + 4:04x}".encode() + line + b"0000"
    with pytest.raises(broker.PolicyError, match="credential material"):
        broker._validate_receive_pack(framed, "sentinel")


def test_receive_pack_validation_rewinds_stream_and_accepts_bytes():
    broker = _load("scm_broker")
    body = _receive_command(b"0" * 40, b"1" * 40, b"refs/heads/hermes/rewind")
    stream = io.BytesIO(body)
    stream.seek(9)
    broker._validate_receive_pack(stream, "sentinel")
    assert stream.tell() == 0
    broker._validate_receive_pack(body, "sentinel")


@pytest.mark.parametrize(
    ("body", "match"),
    [
        (b"bad!", "framing"),
        (b"0000", "no ref command"),
        (b"0003", "framing"),
        (b"0008abc", "framing"),
        (b"0008a b\n0000", "ref command"),
    ],
)
def test_receive_pack_rejects_bad_framing_and_commands(body, match):
    broker = _load("scm_broker")
    with pytest.raises(broker.PolicyError, match=match):
        broker._validate_receive_pack(body, "sentinel")


def _packet(old: bytes, new: bytes, ref: bytes, *, terminator=True) -> bytes:
    command = old + b" " + new + b" " + ref + b"\n"
    framed = f"{len(command) + 4:04x}".encode() + command
    return framed + (b"0000" if terminator else b"")


def test_receive_pack_rejects_nonascii_many_commands_and_missing_terminator(
    monkeypatch,
):
    broker = _load("scm_broker")
    zero = b"0" * 40
    commit = b"1" * 40
    with pytest.raises(broker.PolicyError, match="canonical ASCII"):
        broker._validate_receive_pack(
            _packet(zero, commit, b"refs/heads/hermes/\xff"), "sentinel"
        )
    command = _packet(zero, commit, b"refs/heads/hermes/fix", terminator=False)
    with pytest.raises(broker.PolicyError, match="terminator"):
        broker._validate_receive_pack(command, "sentinel")
    limit = _load("receive_pack_scan").MAX_PUSH_COMMANDS
    many = command * (limit + 1) + b"0000"
    with pytest.raises(broker.PolicyError, match="too many"):
        broker._validate_receive_pack(many, "sentinel")


def test_upstream_request_infers_byte_length_and_requires_stream_length():
    broker = _load("scm_broker")
    seen = []

    def opener(request, timeout):
        seen.append((request, timeout))
        return Response(b"result", content_type="application/x-git-upload-pack-result")

    assert (
        broker._upstream_git_request(
            "/atlas/cassandra.git/git-upload-pack",
            method="POST",
            body=b"request",
            content_type=None,
            expected_type="application/x-git-upload-pack-result",
            token="sentinel",
            opener=opener,
        )
        == b"result"
    )
    assert seen[0][0].get_header("Content-length") == "7"
    with pytest.raises(broker.PolicyError, match="length is missing"):
        broker._upstream_git_request(
            "/atlas/cassandra.git/git-upload-pack",
            method="POST",
            body=io.BytesIO(b"request"),
            content_type=None,
            expected_type="application/x-git-upload-pack-result",
            token="sentinel",
            opener=opener,
        )


def test_upstream_request_rejects_wrong_type_and_supports_getcode_status():
    broker = _load("scm_broker")

    class CodeOnly(Response):
        def __init__(self, body, content_type):
            super().__init__(body, content_type=content_type)
            del self.status

        def getcode(self):
            return 200

    assert (
        broker._upstream_git_request(
            "/atlas/cassandra.git/info/refs?service=git-upload-pack",
            method="GET",
            body=None,
            content_type=None,
            expected_type="application/x-git-upload-pack-advertisement",
            token="sentinel",
            opener=lambda *_a, **_k: CodeOnly(
                b"advertisement", "application/x-git-upload-pack-advertisement"
            ),
        )
        == b"advertisement"
    )
    with pytest.raises(broker.PolicyError, match="response type"):
        broker._upstream_git_request(
            "/atlas/cassandra.git/info/refs?service=git-upload-pack",
            method="GET",
            body=None,
            content_type=None,
            expected_type="application/x-git-upload-pack-advertisement",
            token="sentinel",
            opener=lambda *_a, **_k: Response(b"bad", content_type="text/plain"),
        )