from __future__ import annotations

import hashlib
import http.client
import ssl
from contextlib import contextmanager
from pathlib import Path
from urllib.parse import urlsplit

import pytest

from adw_modules import result_mailbox

ATTEMPT = "a" * 24


@contextmanager
def mailbox(output: Path, attempt_id: str = ATTEMPT):
    server = result_mailbox.ResultMailbox("127.0.0.1", attempt_id, output)
    target = server.start()
    try:
        yield server, target
    finally:
        server.close()


def direct_put(target: result_mailbox.UploadTarget, path: str, body: bytes, *, token: str | None = None, digest: str | None = None, length: int | None = None) -> int:
    parsed = urlsplit(target.url)
    context = ssl.create_default_context(cadata=target.ca_pem)
    connection = http.client.HTTPSConnection(parsed.hostname, parsed.port, context=context, timeout=5)
    try:
        connection.request(
            "PUT",
            path,
            body=body,
            headers={
                "Authorization": f"Bearer {token or target.token}",
                "Content-Type": "application/octet-stream",
                "Content-Length": str(len(body) if length is None else length),
                "X-Content-SHA256": digest or hashlib.sha256(body).hexdigest(),
            },
        )
        response = connection.getresponse()
        response.read()
        return response.status
    finally:
        connection.close()


def test_real_tls_upload_uses_pinned_certificate_and_exact_bytes(tmp_path: Path) -> None:
    bundle = tmp_path / "worker.bundle"
    bundle.write_bytes(b"git bundle bytes")
    output = tmp_path / "output"
    output.mkdir()
    with mailbox(output) as (server, target):
        result_mailbox.upload_result(target, bundle)
        assert server.wait(1).read_bytes() == bundle.read_bytes()


def test_wrong_token_and_attempt_path_are_rejected_without_secret_logs(tmp_path: Path, capsys) -> None:
    output = tmp_path / "output"
    output.mkdir()
    with mailbox(output) as (server, target):
        path = urlsplit(target.url).path
        assert direct_put(target, path, b"bundle", token="z" * 43) == 401
        assert direct_put(target, f"/v1/results/{'b' * 24}", b"bundle") == 404
        assert not server.bundle_path.exists()
        captured = capsys.readouterr()
        assert target.token not in captured.out + captured.err


def test_wrong_ca_is_rejected_before_upload(tmp_path: Path) -> None:
    bundle = tmp_path / "worker.bundle"
    bundle.write_bytes(b"bundle")
    first_output = tmp_path / "first"
    second_output = tmp_path / "second"
    first_output.mkdir()
    second_output.mkdir()
    with mailbox(first_output) as (_, target), mailbox(second_output, "b" * 24) as (_, other):
        wrong = result_mailbox.UploadTarget(target.url, target.token, other.ca_pem)
        with pytest.raises(ssl.SSLCertVerificationError):
            result_mailbox.upload_result(wrong, bundle)


def test_digest_mismatch_empty_and_oversized_bodies_are_rejected(tmp_path: Path) -> None:
    for name, body, digest, length in (
        ("digest", b"bundle", "0" * 64, None),
        ("empty", b"", hashlib.sha256(b"").hexdigest(), None),
        ("oversized", b"", hashlib.sha256(b"").hexdigest(), result_mailbox.MAX_BUNDLE_BYTES + 1),
    ):
        output = tmp_path / name
        output.mkdir()
        with mailbox(output) as (server, target):
            assert direct_put(target, urlsplit(target.url).path, body, digest=digest, length=length) == 400
            assert not server.bundle_path.exists()
            assert not server.temporary_path.exists()


def test_replay_is_rejected_after_first_authenticated_claim(tmp_path: Path) -> None:
    bundle = tmp_path / "worker.bundle"
    bundle.write_bytes(b"bundle")
    output = tmp_path / "output"
    output.mkdir()
    with mailbox(output) as (_, target):
        result_mailbox.upload_result(target, bundle)
        with pytest.raises(RuntimeError, match="HTTP 409"):
            result_mailbox.upload_result(target, bundle)


def test_partial_upload_never_becomes_final_bundle(tmp_path: Path) -> None:
    output = tmp_path / "output"
    output.mkdir()
    with mailbox(output) as (server, target):
        parsed = urlsplit(target.url)
        context = ssl.create_default_context(cadata=target.ca_pem)
        connection = http.client.HTTPSConnection(parsed.hostname, parsed.port, context=context, timeout=5)
        connection.putrequest("PUT", parsed.path)
        connection.putheader("Authorization", f"Bearer {target.token}")
        connection.putheader("Content-Type", "application/octet-stream")
        connection.putheader("Content-Length", "10")
        connection.putheader("X-Content-SHA256", hashlib.sha256(b"x" * 10).hexdigest())
        connection.endheaders()
        connection.send(b"x")
        connection.close()
        assert server.finished.wait(1)
        with pytest.raises(RuntimeError, match="failed validation"):
            server.wait(0)
        assert not server.bundle_path.exists()
        assert not server.temporary_path.exists()


@pytest.mark.parametrize(
    "url",
    [
        f"http://127.0.0.1:4443/v1/results/{ATTEMPT}",
        f"https://localhost:4443/v1/results/{ATTEMPT}",
        f"https://user:password@127.0.0.1:4443/v1/results/{ATTEMPT}",
        f"https://127.0.0.1/v1/results/{ATTEMPT}",
        f"https://127.0.0.1:4443/v1/results/{ATTEMPT}?redirect=1",
        f"https://127.0.0.1:4443/v1/results/{ATTEMPT}#fragment",
        "https://127.0.0.1:4443/v1/results/not-an-attempt",
        f"https://127.0.0.1:invalid/v1/results/{ATTEMPT}",
    ],
)
def test_invalid_upload_urls_fail_closed(url: str) -> None:
    target = result_mailbox.UploadTarget(
        url,
        "t" * 43,
        "-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n",
    )
    with pytest.raises(ValueError, match="URL is invalid"):
        result_mailbox.validate_target(target)


def test_attempt_token_and_certificate_validation_fail_closed() -> None:
    target = result_mailbox.UploadTarget(
        f"https://127.0.0.1:4443/v1/results/{ATTEMPT}",
        "short",
        "not a certificate",
    )
    with pytest.raises(ValueError, match="authorization is invalid"):
        result_mailbox.validate_target(target)
    valid = result_mailbox.UploadTarget(
        target.url,
        "t" * 43,
        "-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n",
    )
    with pytest.raises(ValueError, match="does not match execution"):
        result_mailbox.validate_target(valid, "b" * 24)


@pytest.mark.parametrize("host", ["", "localhost", "::1", "999.1.1.1"])
def test_mailbox_requires_explicit_valid_ipv4_host(tmp_path: Path, host: str) -> None:
    with pytest.raises(ValueError, match="IPv4 address|identity is invalid"):
        result_mailbox.ResultMailbox(host, ATTEMPT, tmp_path)


def test_server_caps_concurrent_request_handlers(tmp_path: Path) -> None:
    output = tmp_path / "output"
    output.mkdir()
    with mailbox(output) as (server, _):
        assert server._server is not None
        acquired = [server._server.connections.acquire(blocking=False) for _ in range(result_mailbox.MAX_CONNECTIONS)]
        try:
            assert all(acquired)
            assert server._server.connections.acquire(blocking=False) is False
        finally:
            for held in acquired:
                if held:
                    server._server.connections.release()


def test_close_removes_upload_artifacts_and_certificate_state(tmp_path: Path) -> None:
    output = tmp_path / "output"
    output.mkdir()
    server = result_mailbox.ResultMailbox("127.0.0.1", ATTEMPT, output)
    temporary_directory = Path(server._temporary.name)
    server.start()
    server.temporary_path.write_bytes(b"partial")
    server.bundle_path.write_bytes(b"complete")
    assert temporary_directory.exists()
    server.close()
    assert not server.temporary_path.exists()
    assert not server.bundle_path.exists()
    assert not temporary_directory.exists()