atlas-iac/testing/tests/test_hermes_gitea_branch_coverage.py

"""Behavioral branch coverage for the Gitea protection policy checker."""

from __future__ import annotations

import json
import os
import stat
import sys
from pathlib import Path

import pytest

from testing.tests.test_hermes_scm_broker_support import ROOT, _load_path


def _module(name: str = "branch_protection_coverage"):
    return _load_path(
        name,
        ROOT / "services/gitea/scripts/gitea_branch_protection_check.py",
    )


def _rule(module, name: str = "main", **updates):
    value = {
        "rule_name": name,
        "priority": 1,
        "created_at": "2026-01-01T00:00:00Z",
        **module._required("bstein"),
    }
    value.update(updates)
    return value


def test_read_bounded_accepts_regular_file_and_closes_descriptor(
    tmp_path: Path, monkeypatch
):
    module = _module("branch_read_regular")
    source = tmp_path / "rules.json"
    source.write_bytes(b"[]")
    closed: list[int] = []
    real_close = os.close

    def close(descriptor: int) -> None:
        closed.append(descriptor)
        real_close(descriptor)

    monkeypatch.setattr(module.os, "close", close)
    assert module._read_bounded(source) == b"[]"
    assert len(closed) == 1


@pytest.mark.parametrize("kind", ["directory", "oversized", "short"])
def test_read_bounded_rejects_unsafe_or_changed_input(
    tmp_path: Path, monkeypatch, kind
):
    module = _module(f"branch_read_{kind}")
    source = tmp_path / "rules"
    source.write_bytes(b"[]")
    real_fstat = module.os.fstat
    real_read = module.os.read

    if kind == "directory":
        monkeypatch.setattr(
            module.os,
            "fstat",
            lambda descriptor: os.stat_result(
                (stat.S_IFDIR | 0o700, 0, 0, 0, 0, 0, 2, 0, 0, 0)
            ),
        )
    elif kind == "oversized":
        monkeypatch.setattr(
            module.os,
            "fstat",
            lambda descriptor: os.stat_result(
                (stat.S_IFREG | 0o600, 0, 0, 0, 0, 0, module.MAX_INPUT + 1, 0, 0, 0)
            ),
        )
    else:
        monkeypatch.setattr(module.os, "fstat", real_fstat)
        monkeypatch.setattr(module.os, "read", lambda descriptor, maximum: b"[")
    with pytest.raises(module.PolicyError):
        module._read_bounded(source)
    monkeypatch.setattr(module.os, "read", real_read)


@pytest.mark.parametrize(
    "pattern",
    [
        "{main}",
        "{main,master}",
        "m?in",
        "m[ai]in",
        "m[!z]in",
        "m[a-z]in",
        "m\\ain",
        "m\\",
        "m[",
        "m[]",
        "m[!]",
        "m[[a]",
        "m[\\a]",
        "m[-a]",
        "m[a-]",
        "m[a-b-c]",
        "m[z-a]",
        "m[aa-b]",
        "m{ain}",
        "m{ain,}",
        "m{ain,aster",
    ],
)
def test_every_special_or_malformed_glob_fails_closed(pattern):
    module = _module("branch_glob_invalid")
    with pytest.raises(module.PolicyError, match="glob branch protection"):
        module._literal_rule(pattern)


@pytest.mark.parametrize("pattern", ["", "x" * 256, "máin"])
def test_rule_names_are_ascii_and_bounded(pattern):
    module = _module("branch_name_bounds")
    with pytest.raises(module.PolicyError, match="rule name"):
        module._literal_rule(pattern)


def test_plain_rule_matching_is_exact_and_case_sensitive():
    module = _module("branch_literal_case")
    assert module._literal_rule("main") == "main"
    assert module._literal_rule("MAIN") == "MAIN"
    assert module.evaluate(
        json.dumps([_rule(module, name="MAIN")]).encode(), "main", "bstein"
    ) == "ABSENT"


@pytest.mark.parametrize("value", [None, "x" * 65, "invalid", "2026-01-01T00:00:00"])
def test_created_at_requires_short_timezone_aware_iso(value):
    module = _module("branch_created_bounds")
    with pytest.raises(module.PolicyError, match="creation time"):
        module._created(value)


@pytest.mark.parametrize(
    "value",
    [b"x" * (1024 * 1024 + 1), b"{}", json.dumps([None]).encode()],
)
def test_evaluate_rejects_invalid_top_level_inputs(value):
    module = _module("branch_evaluate_top")
    with pytest.raises((module.PolicyError, json.JSONDecodeError)):
        module.evaluate(value, "main", "bstein")


def test_evaluate_rejects_unknown_branch_and_excess_rules():
    module = _module("branch_evaluate_bounds")
    with pytest.raises(module.PolicyError, match="input"):
        module.evaluate(b"[]", "release", "bstein")
    with pytest.raises(module.PolicyError, match="response"):
        module.evaluate(json.dumps([{}] * 101).encode(), "main", "bstein")


@pytest.mark.parametrize("priority", [True, -1, 1_000_001, "1"])
def test_evaluate_rejects_ambiguous_priority_types(priority):
    module = _module("branch_priority_bounds")
    rule = _rule(module, priority=priority)
    with pytest.raises(module.PolicyError, match="priority"):
        module.evaluate(json.dumps([rule]).encode(), "main", "bstein")


def test_legacy_zero_priority_is_accepted_but_duplicate_ties_fail_closed():
    module = _module("branch_priority_zero")
    assert module.evaluate(
        json.dumps([_rule(module, priority=0)]).encode(), "main", "bstein"
    ) == "PRESENT"

    first = _rule(module, priority=0, created_at="2026-01-01T00:00:00Z")
    second = _rule(
        module,
        priority=0,
        created_at="2026-01-01T00:00:00Z",
        enable_force_push=True,
    )
    with pytest.raises(module.PolicyError, match="duplicate primary-branch"):
        module.evaluate(json.dumps([first, second]).encode(), "main", "bstein")


def test_any_special_rule_blocks_absent_create_and_present_readback():
    module = _module("branch_special_readback")
    for rules in (
        [_rule(module, name="feature/*")],
        [_rule(module), _rule(module, name="{main}", priority=0)],
    ):
        with pytest.raises(module.PolicyError, match="human review"):
            module.evaluate(json.dumps(rules).encode(), "main", "bstein")


def test_main_reports_success_and_failures(tmp_path: Path, monkeypatch, capsys):
    module = _module("branch_main_paths")
    source = tmp_path / "rules.json"
    source.write_text(json.dumps([_rule(module)]), encoding="utf-8")
    monkeypatch.setattr(sys, "argv", ["checker", str(source), "main", "bstein"])
    assert module.main() == 0
    assert capsys.readouterr().out.strip() == "PRESENT"

    source.write_text("not-json", encoding="utf-8")
    assert module.main() == 1
    assert "branch protection check failed" in capsys.readouterr().err

    monkeypatch.setattr(
        module, "_read_bounded", lambda _path: (_ for _ in ()).throw(OSError())
    )
    assert module.main() == 1