atlas-iac/testing/tests/test_hermes_gitea_pr_postconditions.py

"""Response and postcondition contracts for Atlas draft creation."""

from __future__ import annotations

import base64
import copy
import json

import pytest

from testing.tests.test_hermes_gitea_support import (
    HEAD_SHA,
    Response,
    _draft_response,
    _load,
)

@pytest.mark.parametrize("status", [200, 202, 204, 206])
def test_create_accepts_only_http_201(status: int):
    client = _load()

    with pytest.raises(client.PolicyError, match="unexpected HTTP status"):
        client.create_draft(
            "cassandra",
            base="main",
            head="hermes/fix",
            head_sha=HEAD_SHA,
            title="Focused fix",
            body="Review evidence",
            token="runtime",
            opener=lambda *_a, **_k: Response(_draft_response(), status=status),
        )


@pytest.mark.parametrize("status", [201, 202, 204, 206])
def test_read_accepts_only_http_200(status: int):
    client = _load()

    with pytest.raises(client.PolicyError, match="unexpected HTTP status"):
        client.read(
            "/api/v1/repos/titan/cassandra",
            token="runtime",
            opener=lambda *_a, **_k: Response(b"{}", status=status),
        )


def test_successful_create_validates_each_ref_once(monkeypatch):
    client = _load()
    calls = []

    def git_run(command, **_kwargs):
        calls.append(command)
        return type("Result", (), {"returncode": 0})()

    monkeypatch.setattr(client._validate_ref.__globals__["subprocess"], "run", git_run)
    client.create_draft(
        "cassandra",
        base="main",
        head="hermes/fix",
        head_sha=HEAD_SHA,
        title="Focused fix",
        body="Review evidence",
        token="runtime",
        opener=lambda *_a, **_k: Response(_draft_response()),
    )

    assert [command[-1] for command in calls] == [
        "refs/heads/main",
        "refs/heads/hermes/fix",
    ]


def test_create_verifies_every_server_postcondition():
    client = _load()
    calls = []

    def opener(request, timeout):
        calls.append((request, timeout))
        return Response(_draft_response())

    result = client.create_draft(
        "cassandra",
        base="main",
        head="hermes/fix",
        head_sha=HEAD_SHA,
        title="Focused fix",
        body="Review evidence",
        token="runtime",
        opener=opener,
    )

    assert json.loads(result) == _draft_response()
    payload = json.loads(calls[0][0].data)
    assert payload == {
        "base": "main",
        "body": "Review evidence",
        "head": "hermes/fix",
        "title": "WIP: Focused fix",
    }
    assert calls[0][1] == 30


def test_create_postcondition_rejects_every_material_mismatch():
    client = _load()
    mutations = [
        ("number", 0),
        ("number", 2_147_483_648),
        ("state", "closed"),
        ("draft", False),
        ("merged", True),
        ("html_url", "https://evil.example/pulls/3"),
        ("url", "https://evil.example/api/pulls/3"),
        ("title", "Focused fix"),
        ("body", "different"),
    ]
    documents = []
    for key, value in mutations:
        document = _draft_response()
        document[key] = value
        documents.append(document)
    for path, value in [
        (("base", "ref"), "master"),
        (("base", "repo", "full_name"), "evil/cassandra"),
        (("head", "ref"), "other"),
        (("head", "sha"), "0" * 40),
        (("head", "repo", "full_name"), "evil/cassandra"),
    ]:
        document = copy.deepcopy(_draft_response())
        target = document
        for key in path[:-1]:
            target = target[key]
        target[path[-1]] = value
        documents.append(document)

    for document in documents:
        with pytest.raises(client.PolicyError):
            client._require_create_response(
                json.dumps(document).encode(),
                repo="cassandra",
                base="main",
                head="hermes/fix",
                head_sha=HEAD_SHA,
                title="WIP: Focused fix",
                body="Review evidence",
            )


def test_read_response_is_bounded():
    client = _load()

    with pytest.raises(client.PolicyError, match="safe size limit"):
        client.read(
            "/api/v1/repos/titan/cassandra",
            token="runtime",
            opener=lambda *_a, **_k: Response(b"x" * (client.MAX_RESPONSE_BYTES + 1)),
        )


def test_direct_api_rejects_unexpected_success_content_type():
    client = _load()
    response = Response(b"{}")
    response.headers.replace_header("Content-Type", "text/html")

    with pytest.raises(client.PolicyError, match="unexpected response type"):
        client.read(
            "/api/v1/repos/titan/cassandra",
            token="runtime",
            opener=lambda *_a, **_k: response,
        )


def test_output_redaction_covers_exact_token_and_authorization_header():
    client = _load()
    raw = b'{"message":"do-not-leak","debug":"Authorization: token do-not-leak"}'
    redacted = client.redact_bytes(raw, "do-not-leak")

    assert b"do-not-leak" not in redacted
    assert redacted.count(b"[REDACTED]") >= 1


@pytest.mark.parametrize(
    "reflected",
    [
        b"runtime-sentinel",
        base64.b64encode(b"runtime-sentinel"),
        base64.b64encode(b"hermes-automation:runtime-sentinel"),
    ],
)
def test_direct_api_rejects_credential_reflection(reflected: bytes):
    client = _load()

    with pytest.raises(client.PolicyError, match="credential material"):
        client.read(
            "/api/v1/repos/titan/cassandra",
            token="runtime-sentinel",
            opener=lambda *_a, **_k: Response(
                b'{"unexpected":"' + reflected + b'"}'
            ),
        )