"""
HTTP Range on /media/calls/{id}/audio (server-26#233).

The GCS reads are replaced by an in-memory object: ``audio_object_size`` and
``download_audio_range`` are patched where routers/media.py looks them up, and
every span read is recorded so the tests can assert the route never asks GCS
for more than the requested range (plus at most one STREAM_CHUNK_BYTES span).
"""
import time
from unittest.mock import AsyncMock, patch

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.internal import storage
from app.routers import media

app = FastAPI()
app.include_router(media.router)
client = TestClient(app)

CALL_ID = "call-abc"
GCS_URI = "gs://bucket/calls/call-abc.flac"


class FakeObject:
    """A stored object plus a log of every (start, end) span read from it."""

    def __init__(self, data: bytes):
        self.data = data
        self.reads: list[tuple[int, int]] = []
        self.size_calls = 0

    async def size(self, gcs_uri):
        self.size_calls += 1
        return len(self.data)

    async def read(self, gcs_uri, start, end):
        self.reads.append((start, end))
        return self.data[start:end + 1]


@pytest.fixture
def signed(monkeypatch):
    monkeypatch.setattr(storage.settings, "service_key", "test-key", raising=False)
    exp = int(time.time()) + 600
    return {"exp": exp, "sig": storage.sign_audio_link(CALL_ID, exp)}


def _run(obj: FakeObject, params, headers=None, doc=None):
    doc_get = AsyncMock(return_value=doc if doc is not None else {"call_id": CALL_ID, "audio_gcs_uri": GCS_URI})
    with patch.object(media.fstore, "doc_get", doc_get), \
            patch.object(media, "audio_object_size", obj.size), \
            patch.object(media, "download_audio_range", obj.read):
        resp = client.get(f"/media/calls/{CALL_ID}/audio", params=params, headers=headers or {})
    return resp, doc_get


def test_bounded_range_returns_206_with_exact_bytes(signed):
    obj = FakeObject(bytes(range(256)) * 40)  # 10240 bytes
    resp, _ = _run(obj, signed, {"Range": "bytes=100-199"})
    assert resp.status_code == 206
    assert resp.content == obj.data[100:200]
    assert resp.headers["content-range"] == f"bytes 100-199/{len(obj.data)}"
    assert resp.headers["content-length"] == "100"
    assert resp.headers["accept-ranges"] == "bytes"
    assert resp.headers["content-type"] == "audio/flac"
    assert resp.headers["cache-control"] == "private, max-age=3600"
    # Only the requested span was read from GCS — not the whole object.
    assert obj.reads == [(100, 199)]


def test_open_ended_range(signed):
    obj = FakeObject(b"x" * 5000 + b"tail")
    resp, _ = _run(obj, signed, {"Range": "bytes=5000-"})
    assert resp.status_code == 206
    assert resp.content == b"tail"
    assert resp.headers["content-range"] == "bytes 5000-5003/5004"


def test_suffix_range(signed):
    obj = FakeObject(b"0123456789")
    resp, _ = _run(obj, signed, {"Range": "bytes=-3"})
    assert resp.status_code == 206
    assert resp.content == b"789"
    assert resp.headers["content-range"] == "bytes 7-9/10"


def test_suffix_longer_than_object_returns_whole_object(signed):
    obj = FakeObject(b"0123456789")
    resp, _ = _run(obj, signed, {"Range": "bytes=-500"})
    assert resp.status_code == 206
    assert resp.content == obj.data
    assert resp.headers["content-range"] == "bytes 0-9/10"


def test_end_past_object_is_clamped(signed):
    obj = FakeObject(b"0123456789")
    resp, _ = _run(obj, signed, {"Range": "bytes=8-999"})
    assert resp.status_code == 206
    assert resp.content == b"89"
    assert resp.headers["content-range"] == "bytes 8-9/10"


@pytest.mark.parametrize("rng", ["bytes=10-", "bytes=50-60", "bytes=-0"])
def test_unsatisfiable_range_is_416(signed, rng):
    obj = FakeObject(b"0123456789")
    resp, _ = _run(obj, signed, {"Range": rng})
    assert resp.status_code == 416
    assert resp.headers["content-range"] == "bytes */10"
    assert obj.reads == []


def test_no_range_is_200_full_body_and_advertises_ranges(signed):
    obj = FakeObject(b"abcdef" * 100)
    resp, _ = _run(obj, signed)
    assert resp.status_code == 200
    assert resp.content == obj.data
    assert resp.headers["accept-ranges"] == "bytes"
    assert resp.headers["content-length"] == str(len(obj.data))
    assert "content-range" not in resp.headers


@pytest.mark.parametrize("rng", ["bytes=0-1,4-5", "items=0-5", "bytes=5-2", "garbage"])
def test_ignored_range_headers_fall_back_to_200(signed, rng):
    obj = FakeObject(b"0123456789")
    resp, _ = _run(obj, signed, {"Range": rng})
    assert resp.status_code == 200
    assert resp.content == obj.data


def test_long_object_is_streamed_in_bounded_spans(signed):
    # A ~13 MB "600 s FLAC": a seek near the end reads only from there on, in
    # spans no larger than STREAM_CHUNK_BYTES, and the first span is all the
    # route waits on before the response starts.
    size = 13 * 1024 * 1024
    obj = FakeObject(bytes(size))
    start = size - 1_500_000
    resp, _ = _run(obj, signed, {"Range": f"bytes={start}-"})
    assert resp.status_code == 206
    assert len(resp.content) == 1_500_000
    assert obj.reads[0][0] == start
    assert all(e - s + 1 <= media.STREAM_CHUNK_BYTES for s, e in obj.reads)
    assert sum(e - s + 1 for s, e in obj.reads) == 1_500_000


def test_bad_signature_is_403_before_any_lookup(signed):
    obj = FakeObject(b"0123456789")
    resp, doc_get = _run(obj, {"exp": signed["exp"], "sig": "0" * 64}, {"Range": "bytes=0-3"})
    assert resp.status_code == 403
    doc_get.assert_not_called()
    assert obj.size_calls == 0
    assert obj.reads == []


def test_expired_link_is_403(signed, monkeypatch):
    exp = int(time.time()) - 5
    obj = FakeObject(b"0123456789")
    resp, doc_get = _run(obj, {"exp": exp, "sig": storage.sign_audio_link(CALL_ID, exp)})
    assert resp.status_code == 403
    doc_get.assert_not_called()


def test_missing_object_is_404(signed):
    obj = FakeObject(b"")

    async def no_size(gcs_uri):
        return None

    with patch.object(media.fstore, "doc_get", AsyncMock(return_value={"call_id": CALL_ID, "audio_gcs_uri": GCS_URI})), \
            patch.object(media, "audio_object_size", no_size), \
            patch.object(media, "download_audio_range", obj.read):
        resp = client.get(f"/media/calls/{CALL_ID}/audio", params=signed)
    assert resp.status_code == 404


def test_legacy_mp3_keeps_its_content_type(signed):
    obj = FakeObject(b"ID3" + b"\0" * 50)
    resp, _ = _run(obj, signed, {"Range": "bytes=0-2"}, doc={"call_id": CALL_ID, "audio_gcs_uri": "gs://b/calls/x.mp3"})
    assert resp.status_code == 206
    assert resp.headers["content-type"] == "audio/mpeg"
    assert resp.content == b"ID3"


# ── storage.py: the ranged reads never pull the whole object ────────────────
def _fake_client(monkeypatch, size=1234, payload=b"span"):
    from unittest.mock import MagicMock
    client = MagicMock()
    blob = client.bucket.return_value.blob.return_value
    blob.size = size
    blob.download_as_bytes.return_value = payload
    monkeypatch.setattr(storage, "_gcs_client_cache", client)
    return client, blob


def test_storage_size_reloads_metadata_only(monkeypatch):
    import asyncio
    client, blob = _fake_client(monkeypatch, size=13_000_000)
    assert asyncio.run(storage.audio_object_size(GCS_URI)) == 13_000_000
    client.bucket.assert_called_with("bucket")
    client.bucket.return_value.blob.assert_called_with("calls/call-abc.flac")
    blob.reload.assert_called_once()
    blob.download_as_bytes.assert_not_called()


def test_storage_range_read_passes_inclusive_bounds(monkeypatch):
    import asyncio
    _, blob = _fake_client(monkeypatch)
    assert asyncio.run(storage.download_audio_range(GCS_URI, 100, 199)) == b"span"
    blob.download_as_bytes.assert_called_once_with(start=100, end=199)


def test_storage_range_read_failure_returns_none(monkeypatch):
    import asyncio
    _, blob = _fake_client(monkeypatch)
    blob.download_as_bytes.side_effect = RuntimeError("gcs down")
    assert asyncio.run(storage.download_audio_range(GCS_URI, 0, 10)) is None
    blob.reload.side_effect = RuntimeError("404")
    assert asyncio.run(storage.audio_object_size(GCS_URI)) is None