from __future__ import annotations

import hashlib
import urllib.error
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pytest

from osm_lead_source_service.errors import PbfDownloadRejected
from osm_lead_source_service.pbf.geofabrik import (
    AU_ACT_PILOT,
    acquire_region,
    approved_region,
)


class Response:
    def __init__(
        self,
        url: str,
        payload: bytes,
        *,
        headers: dict[str, str] | None = None,
    ) -> None:
        self.status = 200
        self._url = url
        self._payload = payload
        self.headers = headers or {"Content-Length": str(len(payload))}

    def __enter__(self) -> "Response":
        return self

    def __exit__(self, *args: object) -> None:
        return None

    def geturl(self) -> str:
        return self._url

    def read(self, size: int = -1) -> bytes:
        if size < 0:
            result, self._payload = self._payload, b""
            return result
        result, self._payload = self._payload[:size], self._payload[size:]
        return result


class Opener:
    def __init__(
        self,
        responses: list[tuple[str, bytes, dict[str, str] | None]],
    ) -> None:
        self.responses = list(responses)
        self.requests: list[Any] = []

    def open(self, request: Any, timeout: float) -> Response:
        del timeout
        self.requests.append(request)
        expected_url, payload, headers = self.responses.pop(0)
        assert request.full_url == expected_url
        return Response(expected_url, payload, headers=headers)


class ScriptedOpener:
    def __init__(self, steps: list[tuple[str, Response | BaseException]]) -> None:
        self.steps = list(steps)
        self.requests: list[Any] = []

    def open(self, request: Any, timeout: float) -> Response:
        del timeout
        self.requests.append(request)
        expected_url, result = self.steps.pop(0)
        assert request.full_url == expected_url
        if isinstance(result, BaseException):
            raise result
        return result


def _redirect(source: str, target: str, code: int = 302) -> urllib.error.HTTPError:
    return urllib.error.HTTPError(
        source,
        code,
        "Found",
        {"Location": target},
        None,
    )


def test_only_act_is_approved() -> None:
    assert approved_region("AU-ACT") is AU_ACT_PILOT
    with pytest.raises(PbfDownloadRejected, match="not approved"):
        approved_region("australia")


def test_download_uses_companion_md5_and_sha256_cache(tmp_path: Path) -> None:
    payload = b"synthetic-pbf-payload"
    md5 = hashlib.md5(payload, usedforsecurity=False).hexdigest()
    opener = Opener(
        [
            (
                AU_ACT_PILOT.checksum_url,
                f"{md5}  act-latest.osm.pbf\n".encode(),
                None,
            ),
            (
                AU_ACT_PILOT.pbf_url,
                payload,
                {"Content-Length": str(len(payload)), "Content-Encoding": "identity"},
            ),
        ]
    )
    clock_values = iter(
        (
            datetime(2026, 7, 17, 1, tzinfo=timezone.utc),
            datetime(2026, 7, 17, 1, 0, 1, tzinfo=timezone.utc),
            datetime(2026, 7, 17, 1, 0, 2, tzinfo=timezone.utc),
        )
    )
    result = acquire_region(
        AU_ACT_PILOT,
        tmp_path,
        opener=opener,
        clock=lambda: next(clock_values),
    )
    expected_sha256 = hashlib.sha256(payload).hexdigest()
    assert result.content_sha256 == expected_sha256
    assert result.local_path == tmp_path / "au-act" / f"{expected_sha256}.osm.pbf"
    assert result.local_path.read_bytes() == payload
    assert result.cache_hit is False

    cached = acquire_region(
        AU_ACT_PILOT,
        tmp_path,
        opener=Opener(
            [
                (
                    AU_ACT_PILOT.checksum_url,
                    f"{md5}  act-latest.osm.pbf\n".encode(),
                    None,
                )
            ]
        ),
        clock=lambda: datetime(2026, 7, 17, 2, tzinfo=timezone.utc),
    )
    assert cached.cache_hit is True
    assert cached.content_sha256 == expected_sha256


def test_one_same_origin_dated_redirect_is_accepted(tmp_path: Path) -> None:
    payload = b"redirected-synthetic-pbf"
    md5 = hashlib.md5(payload, usedforsecurity=False).hexdigest()
    target = "https://download.geofabrik.de/australia-oceania/australia/act-260716.osm.pbf"
    opener = ScriptedOpener(
        [
            (
                AU_ACT_PILOT.checksum_url,
                Response(
                    AU_ACT_PILOT.checksum_url,
                    f"{md5}  act-latest.osm.pbf\n".encode(),
                ),
            ),
            (AU_ACT_PILOT.pbf_url, _redirect(AU_ACT_PILOT.pbf_url, target)),
            (
                target,
                Response(
                    target,
                    payload,
                    headers={
                        "Content-Length": str(len(payload)),
                        "Content-Encoding": "identity",
                    },
                ),
            ),
        ]
    )
    clock_values = iter(
        (
            datetime(2026, 7, 17, 1, tzinfo=timezone.utc),
            datetime(2026, 7, 17, 1, 0, 1, tzinfo=timezone.utc),
            datetime(2026, 7, 17, 1, 0, 2, tzinfo=timezone.utc),
        )
    )

    result = acquire_region(
        AU_ACT_PILOT,
        tmp_path,
        opener=opener,
        clock=lambda: next(clock_values),
    )

    assert result.content_sha256 == hashlib.sha256(payload).hexdigest()
    assert [request.full_url for request in opener.requests] == [
        AU_ACT_PILOT.checksum_url,
        AU_ACT_PILOT.pbf_url,
        target,
    ]


@pytest.mark.parametrize(
    "target",
    [
        "http://download.geofabrik.de/australia-oceania/australia/act-260716.osm.pbf",
        "https://example.com/australia-oceania/australia/act-260716.osm.pbf",
        "https://download.geofabrik.de/australia-oceania/act-260716.osm.pbf",
        "https://download.geofabrik.de/australia-oceania/australia/act-latest.osm.pbf",
    ],
)
def test_unapproved_redirect_target_is_rejected(tmp_path: Path, target: str) -> None:
    payload = b"synthetic-pbf-payload"
    md5 = hashlib.md5(payload, usedforsecurity=False).hexdigest()
    opener = ScriptedOpener(
        [
            (
                AU_ACT_PILOT.checksum_url,
                Response(
                    AU_ACT_PILOT.checksum_url,
                    f"{md5}  act-latest.osm.pbf\n".encode(),
                ),
            ),
            (AU_ACT_PILOT.pbf_url, _redirect(AU_ACT_PILOT.pbf_url, target)),
        ]
    )

    with pytest.raises(PbfDownloadRejected, match="Geofabrik redirect"):
        acquire_region(AU_ACT_PILOT, tmp_path, opener=opener)
    assert list((tmp_path / "au-act").glob("*.osm.pbf")) == []


def test_second_redirect_is_rejected(tmp_path: Path) -> None:
    payload = b"synthetic-pbf-payload"
    md5 = hashlib.md5(payload, usedforsecurity=False).hexdigest()
    first_target = "https://download.geofabrik.de/australia-oceania/australia/act-260716.osm.pbf"
    second_target = "https://download.geofabrik.de/australia-oceania/australia/act-260717.osm.pbf"
    opener = ScriptedOpener(
        [
            (
                AU_ACT_PILOT.checksum_url,
                Response(
                    AU_ACT_PILOT.checksum_url,
                    f"{md5}  act-latest.osm.pbf\n".encode(),
                ),
            ),
            (
                AU_ACT_PILOT.pbf_url,
                _redirect(AU_ACT_PILOT.pbf_url, first_target),
            ),
            (first_target, _redirect(first_target, second_target)),
        ]
    )

    with pytest.raises(PbfDownloadRejected, match="redirected more than once"):
        acquire_region(AU_ACT_PILOT, tmp_path, opener=opener)
    assert list((tmp_path / "au-act").glob("*.osm.pbf")) == []


def test_bad_companion_checksum_does_not_publish_cache(tmp_path: Path) -> None:
    payload = b"synthetic-pbf-payload"
    opener = Opener(
        [
            (
                AU_ACT_PILOT.checksum_url,
                f"{'0' * 32}  act-latest.osm.pbf\n".encode(),
                None,
            ),
            (AU_ACT_PILOT.pbf_url, payload, {"Content-Length": str(len(payload))}),
        ]
    )
    with pytest.raises(PbfDownloadRejected, match="companion checksum"):
        acquire_region(AU_ACT_PILOT, tmp_path, opener=opener)
    assert list((tmp_path / "au-act").glob("*.osm.pbf")) == []


def test_cache_directory_symlink_is_rejected(tmp_path: Path) -> None:
    real = tmp_path / "real"
    real.mkdir()
    link = tmp_path / "cache"
    link.symlink_to(real, target_is_directory=True)
    with pytest.raises(PbfDownloadRejected, match="regular directory"):
        acquire_region(AU_ACT_PILOT, link, opener=Opener([]))