import threading
import socket
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import pytest

from tht.adapters.evidence import HttpManifestEvidenceSource
from tht.ports.evidence import EvidenceSourceError


class Handler(BaseHTTPRequestHandler):
    etag_requests = 0
    etag_body_responses = 0
    redirect_target = "/redirected-v1"
    redirect_request_validators = []
    final_request_validators = []

    def do_GET(self):
        if self.path.startswith("/etag"):
            type(self).etag_requests += 1
            if self.headers.get("If-None-Match") == '"abc"':
                self.send_response(304)
                self.end_headers()
                return
            self.send_response(200)
            self.send_header("ETag", '"abc"')
            self.send_header("Content-Type", "text/markdown")
            self.end_headers()
            type(self).etag_body_responses += 1
            self.wfile.write(b"hello")
        elif self.path == "/large":
            self.send_response(200)
            self.send_header("Content-Length", "20")
            self.end_headers()
            self.wfile.write(b"x" * 20)
        elif self.path == "/busy":
            self.send_response(503)
            self.end_headers()
        elif self.path == "/missing":
            self.send_response(404)
            self.end_headers()
        elif self.path == "/redirect-private":
            self.send_response(302)
            self.send_header("Location", f"http://127.0.0.1:{self.server.server_port}/etag")
            self.end_headers()
        elif self.path == "/redirect-userinfo":
            self.send_response(302)
            self.send_header(
                "Location", f"http://user:password@127.0.0.1:{self.server.server_port}/etag"
            )
            self.end_headers()
        elif self.path == "/stable-redirect":
            type(self).redirect_request_validators.append(self.headers.get("If-None-Match"))
            self.send_response(302)
            self.send_header("Location", type(self).redirect_target)
            self.end_headers()
        elif self.path in {"/redirected-v1", "/redirected-v2"}:
            type(self).final_request_validators.append(
                (self.path, self.headers.get("If-None-Match"))
            )
            etag = '"v1"' if self.path.endswith("v1") else '"v2"'
            if self.headers.get("If-None-Match") == etag:
                self.send_response(304)
                self.end_headers()
                return
            self.send_response(200)
            self.send_header("ETag", etag)
            self.end_headers()
            self.wfile.write(self.path.encode())
        else:
            self.send_response(200)
            self.send_header("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT")
            self.end_headers()
            self.wfile.write(b"fallback")

    def log_message(self, format, *args):
        pass


@pytest.fixture
def server_url():
    Handler.etag_requests = 0
    Handler.etag_body_responses = 0
    Handler.redirect_target = "/redirected-v1"
    Handler.redirect_request_validators = []
    Handler.final_request_validators = []
    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield f"http://127.0.0.1:{server.server_port}"
    finally:
        server.shutdown()
        thread.join()


def test_http_uses_etag_and_strips_query_from_provenance(server_url):
    source = HttpManifestEvidenceSource(
        [f"{server_url}/etag?token=secret"], allow_private_hosts=True
    )
    item = next(iter(source.discover()))

    assert item.fingerprint.startswith("etag:")
    assert item.fingerprint != "etag:abc"
    assert item.uri == f"{server_url}/etag"
    assert "secret" not in item.model_dump_json()
    assert source.acquire(item).content == b"hello"


def test_http_uses_last_modified_then_content_hash(server_url):
    modified = next(iter(HttpManifestEvidenceSource(
        [f"{server_url}/modified"], allow_private_hosts=True
    ).discover()))
    assert modified.fingerprint.startswith("last-modified:")

    class NoValidators(Handler):
        def do_GET(self):
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"content")

    server = ThreadingHTTPServer(("127.0.0.1", 0), NoValidators)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        item = next(iter(HttpManifestEvidenceSource(
            [f"http://127.0.0.1:{server.server_port}/doc"], allow_private_hosts=True
        ).discover()))
        assert item.fingerprint.startswith("sha256:")
    finally:
        server.shutdown()
        thread.join()


@pytest.mark.parametrize("path,retryable", [("/busy", True), ("/missing", False)])
def test_http_classifies_status_errors(server_url, path, retryable):
    with pytest.raises(EvidenceSourceError) as caught:
        list(HttpManifestEvidenceSource(
            [server_url + path], allow_private_hosts=True
        ).discover())
    assert caught.value.retryable is retryable
    assert server_url not in str(caught.value)


def test_http_rejects_oversize_and_private_redirect(server_url):
    with pytest.raises(EvidenceSourceError) as large:
        list(HttpManifestEvidenceSource(
            [server_url + "/large"], max_bytes=10, allow_private_hosts=True
        ).discover())
    assert not large.value.retryable

    with pytest.raises(EvidenceSourceError) as redirect:
        list(HttpManifestEvidenceSource([server_url + "/redirect-private"]).discover())
    assert not redirect.value.retryable


def test_http_rejects_unsupported_manifest_scheme():
    with pytest.raises(ValueError, match="http"):
        HttpManifestEvidenceSource(["file:///tmp/secret"])


def test_http_conditional_discovery_reuses_cached_verified_bytes(server_url):
    source = HttpManifestEvidenceSource([server_url + "/etag"], allow_private_hosts=True)
    first = next(iter(source.discover()))
    second = next(iter(source.discover()))

    assert second == first
    assert source.acquire(second).content == b"hello"
    assert Handler.etag_requests == 3
    assert Handler.etag_body_responses == 1


def test_http_rejects_mixed_public_private_dns_answers(monkeypatch):
    monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
        (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80)),
        (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("::1", 80, 0, 0)),
    ])
    with pytest.raises(EvidenceSourceError) as caught:
        list(HttpManifestEvidenceSource(["http://example.test/doc"]).discover())
    assert not caught.value.retryable


def test_http_rejects_userinfo_redirect(server_url):
    with pytest.raises(EvidenceSourceError) as caught:
        list(HttpManifestEvidenceSource(
            [server_url + "/redirect-userinfo"], allow_private_hosts=True
        ).discover())
    assert not caught.value.retryable


class FakeSocket:
    def __init__(self, address):
        self.address = address

    def getpeername(self):
        return (self.address, 443)


class FakeResponse:
    status_code = 200
    headers = {}
    is_redirect = False

    def __init__(self, *, peer="127.0.0.1", stream_error=None, location=None):
        connection = type("Connection", (), {"sock": FakeSocket(peer)})()
        self.raw = type("Raw", (), {"_connection": connection})()
        self.stream_error = stream_error
        self.closed = False
        if location:
            self.is_redirect = True
            self.status_code = 302
            self.headers = {"Location": location}
        else:
            self.is_redirect = False
            self.status_code = 200
            self.headers = {}

    def iter_content(self, chunk_size):
        if self.stream_error:
            raise self.stream_error
        yield b"ok"

    def close(self):
        self.closed = True


class FakeSession:
    def __init__(self, response):
        self.response = response

    def get(self, *args, **kwargs):
        return self.response


def test_http_rejects_public_to_private_rebind(monkeypatch):
    monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
        (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
    ])
    source = HttpManifestEvidenceSource(["https://example.test/doc"])
    response = FakeResponse(peer="127.0.0.1")
    source._session = FakeSession(response)
    with pytest.raises(EvidenceSourceError):
        list(source.discover())
    assert response.closed


def test_http_rejects_public_redirect_to_private_destination(monkeypatch):
    monkeypatch.setattr(socket, "getaddrinfo", lambda host, *args, **kwargs: [
        (socket.AF_INET, socket.SOCK_STREAM, 6, "", (
            "93.184.216.34" if host == "example.test" else "127.0.0.1", 443
        )),
    ])
    source = HttpManifestEvidenceSource(["https://example.test/doc"])
    response = FakeResponse(
        peer="93.184.216.34", location="https://private.test/secret"
    )
    source._session = FakeSession(response)
    with pytest.raises(EvidenceSourceError) as caught:
        list(source.discover())
    assert not caught.value.retryable
    assert response.closed


def test_http_closes_response_when_streaming_fails():
    source = HttpManifestEvidenceSource(
        ["https://example.test/doc"], allow_private_hosts=True
    )
    response = FakeResponse(stream_error=socket.timeout("read timed out"))
    source._session = FakeSession(response)
    with pytest.raises(EvidenceSourceError):
        list(source.discover())
    assert response.closed


def test_http_binds_validators_to_exact_final_redirect_url(server_url):
    source = HttpManifestEvidenceSource(
        [server_url + "/stable-redirect"], allow_private_hosts=True
    )
    first = next(iter(source.discover()))
    second = next(iter(source.discover()))

    assert first == second
    assert Handler.redirect_request_validators == [None, None]
    assert Handler.final_request_validators == [
        ("/redirected-v1", None),
        ("/redirected-v1", '"v1"'),
    ]


def test_http_redirect_path_change_fetches_and_replaces_body(server_url):
    source = HttpManifestEvidenceSource(
        [server_url + "/stable-redirect"], allow_private_hosts=True
    )
    first = next(iter(source.discover()))
    assert source.acquire(first).content == b"/redirected-v1"

    Handler.redirect_target = "/redirected-v2"
    with pytest.raises(EvidenceSourceError):
        source.acquire(first)
    current = next(iter(source.discover()))

    assert source.acquire(current).content == b"/redirected-v2"
    assert ("/redirected-v2", None) in Handler.final_request_validators


def test_http_rejects_unsolicited_304_without_bound_validator(monkeypatch):
    monkeypatch.setattr(socket, "getaddrinfo", lambda *args, **kwargs: [
        (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
    ])
    source = HttpManifestEvidenceSource(["https://example.test/doc"])
    response = FakeResponse(peer="93.184.216.34")
    response.status_code = 304
    source._session = FakeSession(response)

    with pytest.raises(EvidenceSourceError) as caught:
        list(source.discover())
    assert not caught.value.retryable
    assert response.closed


def test_http_rejects_cross_origin_304_for_cached_provenance(server_url):
    source = HttpManifestEvidenceSource([server_url + "/etag"], allow_private_hosts=True)
    item = next(iter(source.discover()))
    response = FakeResponse()
    response.status_code = 304
    source._session = FakeSession(response)

    with pytest.raises(EvidenceSourceError) as caught:
        source._download("https://other.example/doc", item.uri)
    assert not caught.value.retryable
    assert response.closed