import asyncio
import time
from pathlib import Path
from types import SimpleNamespace

import pytest

from utils import download_manager
from utils.download_manager import DownloadConfig, DownloadMetrics, ResilientDownloader


@pytest.mark.asyncio
async def test_download_deduplicates_same_target(monkeypatch, tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    downloader._inflight_downloads.clear()
    call_count = {"value": 0}

    async def fake_submit(runner, **_kwargs):
        return await runner()

    def fake_queue():
        return SimpleNamespace(submit=fake_submit)

    def fake_download_sync(
        url, filename, _headers, _skip_if_exists, _progress, _max_size_bytes
    ):
        call_count["value"] += 1
        time.sleep(0.05)
        path = tmp_path / filename
        path.write_bytes(b"data")
        return DownloadMetrics(
            url=url,
            path=str(path),
            size=path.stat().st_size,
            elapsed=0.01,
            used_multipart=False,
            resumed=False,
        )

    monkeypatch.setattr(download_manager, "get_download_queue", fake_queue)
    monkeypatch.setattr(downloader, "_download_sync", fake_download_sync)

    first, second = await asyncio.gather(
        downloader.download("https://cdn.example.com/video.mp4", "video.mp4"),
        downloader.download("https://cdn.example.com/video.mp4", "video.mp4"),
    )

    assert call_count["value"] == 1
    assert first.path == second.path


@pytest.mark.asyncio
async def test_download_does_not_deduplicate_different_urls_with_same_filename(
    monkeypatch, tmp_path
):
    downloader = ResilientDownloader(str(tmp_path))
    downloader._inflight_downloads.clear()
    call_count = {"value": 0}

    async def fake_submit(runner, **_kwargs):
        return await runner()

    def fake_queue():
        return SimpleNamespace(submit=fake_submit)

    def fake_download_sync(
        url, filename, _headers, _skip_if_exists, _progress, _max_size_bytes
    ):
        call_count["value"] += 1
        path = tmp_path / filename
        path.write_text(url, encoding="utf-8")
        return DownloadMetrics(
            url=url,
            path=str(path),
            size=path.stat().st_size,
            elapsed=0.01,
            used_multipart=False,
            resumed=False,
        )

    monkeypatch.setattr(download_manager, "get_download_queue", fake_queue)
    monkeypatch.setattr(downloader, "_download_sync", fake_download_sync)

    await asyncio.gather(
        downloader.download("https://cdn.example.com/video-a.mp4", "video.mp4"),
        downloader.download("https://cdn.example.com/video-b.mp4", "video.mp4"),
    )

    assert call_count["value"] == 2


def test_probe_uses_probe_retry_limit(monkeypatch, tmp_path):
    downloader = ResilientDownloader(
        str(tmp_path), config=DownloadConfig(probe_max_retries=1, retry_backoff=0.0)
    )
    attempts = {"count": 0}

    class FailingSession:
        def head(self, *args, **kwargs):
            attempts["count"] += 1
            raise RuntimeError("boom")

    monkeypatch.setattr(downloader, "_get_session", lambda: FailingSession())

    total_size, supports_range = downloader._probe(
        "https://cdn.example.com/video.mp4", {}
    )

    assert (total_size, supports_range) == (0, False)
    assert attempts["count"] == 2


def test_download_single_rejects_resume_when_origin_ignores_range(
    monkeypatch, tmp_path
):
    downloader = ResilientDownloader(str(tmp_path))

    class DummyResponse:
        status_code = 200
        headers = {"content-length": "5"}

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def raise_for_status(self):
            return None

        def iter_bytes(self, chunk_size):
            yield b"fresh"

    class DummySession:
        def stream(self, method, url, **kwargs):
            return DummyResponse()

    monkeypatch.setattr(downloader, "_get_session", lambda: DummySession())

    with pytest.raises(download_manager._ResumeNotSupportedError):
        downloader._download_single(
            "https://cdn.example.com/video.mp4",
            str(tmp_path / "video.part"),
            {"Range": "bytes=10-"},
        )


def test_download_sync_preserves_too_large_error_and_keeps_existing_target(
    monkeypatch, tmp_path
):
    downloader = ResilientDownloader(str(tmp_path))
    target = tmp_path / "video.mp4"
    target.write_bytes(b"existing")

    monkeypatch.setattr(downloader, "_probe", lambda url, headers: (10, False))

    with pytest.raises(download_manager.DownloadTooLargeError):
        downloader._download_sync(
            "https://cdn.example.com/video.mp4",
            "video.mp4",
            {},
            False,
            None,
            5,
        )

    assert target.read_bytes() == b"existing"


def test_download_sync_falls_back_when_multipart_range_is_not_supported(
    monkeypatch, tmp_path
):
    downloader = ResilientDownloader(
        str(tmp_path),
        config=DownloadConfig(multipart_threshold=1, max_workers=2),
    )
    target = tmp_path / "video.mp4"

    monkeypatch.setattr(downloader, "_probe", lambda url, headers: (4, True))

    def fail_multipart(*_args, **_kwargs):
        raise download_manager._ResumeNotSupportedError("range ignored")

    def download_sequential(_url, target_path, _headers, *, progress_state=None, max_size_bytes=None):
        Path(target_path).write_bytes(b"data")

    monkeypatch.setattr(downloader, "_download_multipart", fail_multipart)
    monkeypatch.setattr(downloader, "_download_single", download_sequential)

    metrics = downloader._download_sync(
        "https://cdn.example.com/video.mp4",
        "video.mp4",
        {},
        False,
        None,
        None,
    )

    assert target.read_bytes() == b"data"
    assert metrics.used_multipart is False


def test_download_skip_if_exists_enforces_max_size(tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    (tmp_path / "video.mp4").write_bytes(b"123456")

    with pytest.raises(download_manager.DownloadTooLargeError):
        downloader._download_sync(
            "https://cdn.example.com/video.mp4",
            "video.mp4",
            {},
            True,
            None,
            5,
        )


@pytest.mark.asyncio
async def test_download_rejects_parent_path_traversal(tmp_path):
    downloader = ResilientDownloader(str(tmp_path))

    with pytest.raises(
        download_manager.DownloadError, match="escapes output directory"
    ):
        await downloader.download("https://cdn.example.com/video.mp4", "../evil.txt")


@pytest.mark.asyncio
async def test_download_rejects_absolute_target_path(tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    outside_path = tmp_path.parent / "evil.txt"

    with pytest.raises(
        download_manager.DownloadError, match="escapes output directory"
    ):
        await downloader.download(
            "https://cdn.example.com/video.mp4", str(outside_path)
        )


def test_download_sync_allows_nested_relative_paths(monkeypatch, tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    target = tmp_path / "tweet123" / "video.mp4"

    monkeypatch.setattr(downloader, "_probe", lambda url, headers: (4, False))

    def fake_download_single(
        url, target_path, headers, *, progress_state=None, max_size_bytes=None
    ):
        Path(target_path).write_bytes(b"data")

    monkeypatch.setattr(downloader, "_download_single", fake_download_single)

    metrics = downloader._download_sync(
        "https://cdn.example.com/video.mp4",
        "tweet123/video.mp4",
        {},
        False,
        None,
        None,
    )

    assert target.exists()
    assert metrics.path == str(target)


@pytest.mark.asyncio
async def test_download_subprocess_times_out_and_kills_worker(monkeypatch, tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    downloader._subprocess_timeout_seconds = 0.01

    class HangingProcess:
        def __init__(self):
            self.returncode = None
            self.killed = False
            self.communicate_calls = 0

        async def communicate(self, _stdin=None):
            self.communicate_calls += 1
            if self.communicate_calls == 1:
                await asyncio.sleep(60)
            return b"", b""

        def kill(self):
            self.killed = True
            self.returncode = -9

    process = HangingProcess()

    async def fake_create_subprocess_exec(*_args, **_kwargs):
        return process

    monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess_exec)

    with pytest.raises(download_manager.DownloadError, match="timed out"):
        await downloader._download_subprocess(
            url="https://cdn.example.com/video.mp4",
            filename="video.mp4",
            headers={},
            skip_if_exists=False,
        )

    assert process.killed is True
    assert process.communicate_calls == 2


def test_download_single_retry_recomputes_range_offset(monkeypatch, tmp_path):
    downloader = ResilientDownloader(
        str(tmp_path), config=DownloadConfig(max_retries=1, retry_backoff=0.0)
    )
    part_path = tmp_path / "video.part"
    part_path.write_bytes(b"0123")
    full_payload = b"0123456789"
    seen_ranges = []

    class FlakyResponse:
        def __init__(self, start, fail):
            self.start = start
            self.fail = fail
            self.status_code = 206
            self.headers = {
                "content-range": f"bytes {start}-9/10",
                "content-length": str(10 - start),
            }

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def raise_for_status(self):
            return None

        def iter_bytes(self, chunk_size):
            if self.fail:
                yield full_payload[self.start : self.start + 3]
                raise RuntimeError("connection dropped")
            yield full_payload[self.start :]

    class FlakySession:
        def __init__(self):
            self.calls = 0

        def stream(self, method, url, headers=None, **kwargs):
            self.calls += 1
            seen_ranges.append(headers["Range"])
            start = int(headers["Range"].removeprefix("bytes=").rstrip("-"))
            return FlakyResponse(start, fail=self.calls == 1)

    session = FlakySession()
    monkeypatch.setattr(downloader, "_get_session", lambda: session)

    downloader._download_single(
        "https://cdn.example.com/video.mp4",
        str(part_path),
        {"Range": "bytes=4-"},
    )

    assert seen_ranges == ["bytes=4-", "bytes=7-"]
    assert part_path.read_bytes() == full_payload


@pytest.mark.asyncio
async def test_download_leader_cancellation_releases_joiners(monkeypatch, tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    downloader._inflight_downloads.clear()
    started = asyncio.Event()

    async def hanging_submit(runner, **_kwargs):
        started.set()
        await asyncio.sleep(60)

    def fake_queue():
        return SimpleNamespace(submit=hanging_submit)

    monkeypatch.setattr(download_manager, "get_download_queue", fake_queue)

    leader = asyncio.create_task(
        downloader.download("https://cdn.example.com/video.mp4", "video.mp4")
    )
    await started.wait()
    joiner = asyncio.create_task(
        downloader.download("https://cdn.example.com/video.mp4", "video.mp4")
    )
    await asyncio.sleep(0)
    await asyncio.sleep(0)
    leader.cancel()

    with pytest.raises(asyncio.CancelledError):
        await leader
    with pytest.raises(asyncio.CancelledError):
        await asyncio.wait_for(joiner, timeout=1)


def test_download_sync_discards_full_size_partial_instead_of_resuming(
    monkeypatch, tmp_path
):
    downloader = ResilientDownloader(str(tmp_path))
    temp_path = tmp_path / f"video.mp4{downloader.config.temp_suffix}"
    temp_path.write_bytes(b"\x00" * 10)
    seen_headers = []

    monkeypatch.setattr(downloader, "_probe", lambda url, headers: (10, True))

    def fake_download_single(
        url, target_path, headers, *, progress_state=None, max_size_bytes=None
    ):
        seen_headers.append(dict(headers))
        Path(target_path).write_bytes(b"0123456789")

    monkeypatch.setattr(downloader, "_download_single", fake_download_single)

    metrics = downloader._download_sync(
        "https://cdn.example.com/video.mp4",
        "video.mp4",
        {},
        False,
        None,
        None,
    )

    assert metrics.resumed is False
    assert "Range" not in seen_headers[0]
    assert not temp_path.exists()
    assert (tmp_path / "video.mp4").read_bytes() == b"0123456789"


def test_download_sync_resumes_partial_smaller_than_total(monkeypatch, tmp_path):
    downloader = ResilientDownloader(str(tmp_path))
    temp_path = tmp_path / f"video.mp4{downloader.config.temp_suffix}"
    temp_path.write_bytes(b"0123")
    seen_headers = []

    monkeypatch.setattr(downloader, "_probe", lambda url, headers: (10, True))

    def fake_download_single(
        url, target_path, headers, *, progress_state=None, max_size_bytes=None
    ):
        seen_headers.append(dict(headers))
        with open(target_path, "ab") as outfile:
            outfile.write(b"456789")

    monkeypatch.setattr(downloader, "_download_single", fake_download_single)

    metrics = downloader._download_sync(
        "https://cdn.example.com/video.mp4",
        "video.mp4",
        {},
        False,
        None,
        None,
    )

    assert metrics.resumed is True
    assert seen_headers[0]["Range"] == "bytes=4-"
    assert (tmp_path / "video.mp4").read_bytes() == b"0123456789"


def test_close_download_http_clients_closes_registered_clients(tmp_path):
    download_manager.close_download_http_clients()
    downloader = ResilientDownloader(str(tmp_path))

    client = downloader._get_session()
    assert client.is_closed is False

    download_manager.close_download_http_clients()

    assert client.is_closed is True
    assert downloader._get_session() is not client