"""Adversarial contracts for the Atlas-only draft pull-request client."""

from __future__ import annotations

import copy
import importlib.util
import json
import sys
import urllib.request
from pathlib import Path

import pytest

ROOT = Path(__file__).parents[2]
CLIENT_PATH = ROOT / "services/hermes/scripts/gitea_api.py"
HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"


def _load():
    spec = importlib.util.spec_from_file_location("safe_gitea_api", CLIENT_PATH)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def _draft_payload(**updates):
    payload = {
        "base": "main",
        "body": "Review evidence",
        "head": "hermes/review-fix",
        "title": "WIP: Repair review findings",
    }
    payload.update(updates)
    return payload


def _draft_response(**updates):
    response = {
        "number": 3,
        "state": "open",
        "draft": True,
        "merged": False,
        "html_url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
        "url": "https://scm.bstein.dev/api/v1/repos/atlas/cassandra/pulls/3",
        "title": "WIP: Focused fix",
        "body": "Review evidence",
        "base": {"ref": "main", "repo": {"full_name": "atlas/cassandra"}},
        "head": {
            "ref": "hermes/fix",
            "sha": HEAD_SHA,
            "repo": {"full_name": "atlas/cassandra"},
        },
    }
    response.update(updates)
    return response


class Response:
    def __init__(self, body: object):
        self.body = body if isinstance(body, bytes) else json.dumps(body).encode()

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def read(self, limit=-1):
        return self.body if limit < 0 else self.body[:limit]


def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
    client = _load()
    source = urllib.request.Request(
        "https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
        headers={"Authorization": "token redirect-sentinel"},
    )

    with pytest.raises(client.PolicyError, match="redirects are not allowed") as exc:
        client.RejectRedirectHandler().redirect_request(
            source,
            None,
            302,
            "Found",
            {},
            "https://evil.example/collect",
        )

    assert "redirect-sentinel" not in str(exc.value)
    assert any(
        isinstance(handler, client.RejectRedirectHandler)
        for handler in client._SAFE_OPENER.handlers
    )


@pytest.mark.parametrize(
    ("base_url", "path"),
    [
        ("https://evil.example", "/api/v1/repos/atlas/cassandra"),
        ("http://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
        ("https://scm.bstein.dev:443", "/api/v1/repos/atlas/cassandra"),
        ("https://scm.bstein.dev", "https://evil.example/api/v1/repos/atlas/cassandra"),
        ("https://scm.bstein.dev", "/api/v1/repos/evil/cassandra"),
        ("https://scm.bstein.dev", "/api/v1/repos/%61tlas/cassandra"),
        ("https://scm.bstein.dev", "/api/v1/repos/atlas/../admin"),
    ],
)
def test_host_owner_and_path_escape_attempts_are_rejected(base_url: str, path: str):
    client = _load()

    with pytest.raises(client.PolicyError):
        client.build_request("GET", path, base_url=base_url, token="secret")


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/repos/atlas/cassandra",
        "/api/v1/repos/atlas/cassandra/pulls?state=open&limit=20&page=1",
        "/api/v1/repos/atlas/cassandra/pulls/7",
        "/api/v1/repos/atlas/cassandra/pulls/7/commits?limit=20",
        "/api/v1/repos/atlas/cassandra/pulls/7/files?page=1",
        "/api/v1/repos/atlas/cassandra/branches",
        "/api/v1/repos/atlas/cassandra/branches/main",
        "/api/v1/repos/atlas/cassandra/commits?limit=10",
        f"/api/v1/repos/atlas/cassandra/git/commits/{HEAD_SHA}",
        f"/api/v1/repos/atlas/cassandra/commits/{HEAD_SHA}/status",
        f"/api/v1/repos/atlas/cassandra/commits/{HEAD_SHA}/statuses?limit=10",
        f"/api/v1/repos/atlas/cassandra/statuses/{HEAD_SHA}?page=1",
    ],
)
def test_explicit_read_allowlist_accepts_only_engineering_metadata(path: str):
    client = _load()

    request = client.build_request(
        "GET", path, base_url=client.CANONICAL_BASE_URL, token="runtime"
    )
    assert request.method == "GET"


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/repos/atlas/cassandra/hooks",
        "/api/v1/repos/atlas/cassandra/actions/secrets",
        "/api/v1/repos/atlas/cassandra/actions/variables",
        "/api/v1/repos/atlas/cassandra/collaborators",
        "/api/v1/repos/atlas/cassandra/branch_protections",
        "/api/v1/repos/atlas/cassandra/keys",
        "/api/v1/repos/atlas/cassandra/pulls/7/reviews",
        "/api/v1/repos/atlas/cassandra/pulls/7/merge",
        "/api/v1/repos/atlas/cassandra/pulls/7.diff",
        "/api/v1/repos/atlas/cassandra/releases",
    ],
)
def test_privileged_or_content_routes_are_denied_even_for_get(path: str):
    client = _load()

    with pytest.raises(client.PolicyError, match="outside the metadata read allowlist"):
        client.authorize_request("GET", path, None)


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/repos/atlas/cassandra/pulls?limit=51",
        "/api/v1/repos/atlas/cassandra/pulls?state=merged",
        "/api/v1/repos/atlas/cassandra/pulls?private=true",
        "/api/v1/repos/atlas/cassandra/pulls?limit=1&limit=2",
        "/api/v1/repos/atlas/cassandra?p=1",
    ],
)
def test_read_query_is_bounded(path: str):
    client = _load()

    with pytest.raises(client.PolicyError):
        client.authorize_request("GET", path, None)


@pytest.mark.parametrize(
    ("method", "path", "data"),
    [
        ("DELETE", "/api/v1/repos/atlas/cassandra/pulls/4", None),
        ("POST", "/api/v1/repos/atlas/cassandra/pulls/4/merge", {}),
        (
            "POST",
            "/api/v1/repos/atlas/cassandra/pulls/4/reviews",
            {"event": "APPROVED"},
        ),
        ("PATCH", "/api/v1/repos/atlas/cassandra/pulls/4", {"title": "WIP: x"}),
        ("PUT", "/api/v1/repos/atlas/cassandra/branches/main", {}),
    ],
)
def test_merge_approve_close_delete_update_and_other_mutations_are_rejected(
    method: str, path: str, data: object
):
    client = _load()

    with pytest.raises(client.PolicyError):
        client.build_request(
            method, path, base_url=client.CANONICAL_BASE_URL, token="secret", data=data
        )


@pytest.mark.parametrize(
    "ref",
    [
        "foo/.bar",
        "foo/bar.lock/baz",
        "foo..bar",
        "foo@{bar",
        "foo//bar",
        "-danger",
        "danger.",
        "danger~one",
        "danger^one",
        "danger:one",
        "danger one",
    ],
)
def test_complete_git_ref_validation_rejects_invalid_names(ref: str):
    client = _load()

    with pytest.raises(client.PolicyError):
        client._validate_ref(ref, "head")


def test_git_ref_validation_uses_fixed_trusted_binary():
    client = _load()

    assert client.GIT_BIN == "/usr/bin/git"
    assert client._validate_ref("hermes/valid-fix", "head") == "hermes/valid-fix"


def test_create_forces_draft_title_and_same_repository_branch_names():
    client = _load()
    assert (
        client.authorize_request(
            "POST", "/api/v1/repos/atlas/cassandra/pulls", _draft_payload()
        )
        == "create-draft"
    )
    with pytest.raises(client.PolicyError, match="draft-title prefix"):
        client.authorize_request(
            "POST",
            "/api/v1/repos/atlas/cassandra/pulls",
            _draft_payload(title="Not a draft"),
        )
    with pytest.raises(client.PolicyError):
        client.authorize_request(
            "POST",
            "/api/v1/repos/atlas/cassandra/pulls",
            _draft_payload(head="someone:branch"),
        )


def test_runtime_token_is_only_an_authorization_header():
    client = _load()
    request = client.build_request(
        "POST",
        "/api/v1/repos/atlas/cassandra/pulls",
        base_url=client.CANONICAL_BASE_URL,
        token="do-not-leak",
        data=_draft_payload(),
    )

    assert "do-not-leak" not in request.full_url
    assert b"do-not-leak" not in request.data
    assert request.get_header("Authorization") == "token do-not-leak"


@pytest.mark.parametrize(
    "body",
    [
        "pass" + "word=not-a-real-credential",
        "Authorization: " + "Bearer not-a-real-credential-value",
        "token: " + "ghp_" + "notarealcredentialvalue123456",
        "-----BEGIN OPENSSH " + "PRIVATE KEY-----",
        "eyJnotarealheader." + "notarealpayloadvalue." + "notarealsignature",
    ],
)
def test_body_rejects_common_credential_shapes(body: str):
    client = _load()

    with pytest.raises(client.PolicyError, match="credential material"):
        client._validate_body(body)


def test_create_rejects_exact_runtime_token_before_network():
    client = _load()
    called = False

    def opener(*_args, **_kwargs):
        nonlocal called
        called = True
        return Response(_draft_response())

    with pytest.raises(client.PolicyError, match="runtime credential"):
        client.create_draft(
            "cassandra",
            base="main",
            head="hermes/fix",
            head_sha=HEAD_SHA,
            title="Focused fix",
            body="accidental runtime-sentinel value",
            token="runtime-sentinel",
            opener=opener,
        )
    assert called is False


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),
        ("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/atlas/cassandra",
            token="runtime",
            opener=lambda *_a, **_k: Response(b"x" * (client.MAX_RESPONSE_BYTES + 1)),
        )


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