"""TDD — push_stream route must stream the request body (not buffer it).

Rules (from musewire-performance.md):

  S1  push_stream route source must NOT contain `await request.body()`.
  S2  push_stream route source must call `request.stream()`.
  S3  A valid push still yields RESULT ok=True after the streaming change.
  S4  Auth is checked before any stream bytes are consumed.

Why this matters: `await request.body()` causes Cloudflare to buffer the
entire request body before forwarding it to the origin, which triggers
TLS bad_record_mac errors on large payloads.  request.stream() lets
Cloudflare forward chunks as they arrive — matching GitHub's push behavior.
"""
from __future__ import annotations

import hashlib
import inspect
from datetime import datetime, timezone
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import msgpack
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.models.wire import (
    SFRAME_COMMIT_PACK,
    SFRAME_END,
    SFRAME_HEADER,
    SFRAME_RESULT,
)
from muse.core.mpack import GRPC_CONTENT_TYPE, GrpcFrameReader, MuseWireFrameWriter, grpc_frame
from tests.factories import create_repo

_fw = MuseWireFrameWriter()


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _pack(obj: object) -> bytes:
    return msgpack.packb(obj, use_bin_type=True)


def _wrap(ft: str, data: object) -> bytes:
    return grpc_frame(_fw.wrap(frame_type=ft, payload=_pack(data)))


def _sha256_oid(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()


def _utc() -> str:
    return datetime.now(tz=timezone.utc).isoformat()


def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes:
    return _wrap(SFRAME_HEADER, {
        "t": SFRAME_HEADER,
        "branch": "main",
        "force": False,
        "have": [],
        "head": _sha256_oid(b"head"),
        "n_objects": n_objects,
        "n_commits": n_commits,
    })


def _commit_pack_frame(commits: list[dict], snapshots: list[dict]) -> bytes:
    return _wrap(SFRAME_COMMIT_PACK, {"t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots})


def _end_frame() -> bytes:
    return _wrap(SFRAME_END, {"t": SFRAME_END})


def _make_commit(snapshot_id: str | None = None, branch: str = "main") -> dict:
    snap = snapshot_id or _sha256_oid(b"default-snap")
    return {
        "commit_id": _sha256_oid(f"commit-{_utc()}".encode()),
        "parent_ids": [],
        "snapshot_id": snap,
        "branch": branch,
        "message": "test commit",
        "author": "gabriel",
        "committed_at": _utc(),
        "signature": "",
        "signer_key_id": "",
        "agent_id": "",
        "model_id": "",
        "metadata": {},
    }


def _make_snapshot(snap_id: str) -> dict:
    return {"snapshot_id": snap_id, "manifest": {}}


def _stub_r2_backend(monkeypatch: pytest.MonkeyPatch) -> None:
    backend = MagicMock()
    backend.store_object = AsyncMock(return_value=None)
    backend.object_exists = AsyncMock(return_value=False)
    backend.get_object = AsyncMock(return_value=None)
    monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend)


# ---------------------------------------------------------------------------
# S1 — route source must NOT contain `await request.body()`
# ---------------------------------------------------------------------------

def test_s1_push_stream_route_does_not_call_request_body() -> None:
    """push_stream must use request.stream(), not await request.body().

    `await request.body()` forces Cloudflare to buffer the entire upload
    before forwarding it to the origin server — causing bad_record_mac
    on large payloads.  This is a static check on the route source code.
    """
    from musehub.api.routes import wire

    source = inspect.getsource(wire.push_stream)
    assert "await request.body()" not in source, (
        "push_stream route calls `await request.body()` which buffers the full body "
        "before processing. Replace with `request.stream()` so Cloudflare streams "
        "chunks as they arrive. See docs/protocol/musewire-performance.md."
    )


# ---------------------------------------------------------------------------
# S2 — route source must call `request.stream()`
# ---------------------------------------------------------------------------

def test_s2_push_stream_route_uses_request_stream() -> None:
    """push_stream must pass request.stream() to wire_push_stream."""
    from musehub.api.routes import wire

    source = inspect.getsource(wire.push_stream)
    assert "request.stream()" in source, (
        "push_stream route must call `request.stream()` to pass an AsyncIterator "
        "to wire_push_stream. Currently it buffers via `await request.body()`. "
        "See docs/protocol/musewire-performance.md."
    )


# ---------------------------------------------------------------------------
# S3 — valid push still succeeds after streaming fix
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_s3_valid_push_still_succeeds_with_streaming(
    client: AsyncClient,
    db_session: AsyncSession,
    auth_headers: dict,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """After switching to request.stream(), a valid push must still yield RESULT ok=True."""
    _stub_r2_backend(monkeypatch)
    repo = await create_repo(db_session, owner="testuser", name="s3-success")

    snap_id = _sha256_oid(b"s3-snap")
    commit = _make_commit(snapshot_id=snap_id)
    snap = _make_snapshot(snap_id)
    body = _header_frame() + _commit_pack_frame([commit], [snap]) + _end_frame()

    resp = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=body,
        headers={**auth_headers, "Content-Type": GRPC_CONTENT_TYPE},
    )

    assert resp.status_code == 200
    reader = GrpcFrameReader()
    reader.feed(resp.content)
    frames = list(reader.frames())
    result_frames = [f for f in frames if f.get("t") == SFRAME_RESULT]
    assert result_frames, f"No RESULT frame in response; frames: {[f.get('t') for f in frames]}"
    assert result_frames[0].get("ok") is True


# ---------------------------------------------------------------------------
# S4 — auth is checked before stream consumption
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_s4_auth_checked_before_stream_consumed(
    client: AsyncClient,
    db_session: AsyncSession,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Unauthenticated push must be rejected without consuming the request stream."""
    _stub_r2_backend(monkeypatch)
    repo = await create_repo(db_session, owner="testuser", name="s4-auth")

    wire_was_called = []

    async def _spy(session, repo_id, body_iter, pusher_id):
        wire_was_called.append(True)
        yield _pack({"t": SFRAME_RESULT, "ok": True, "msg": "ok", "heads": {}, "head": ""})

    monkeypatch.setattr("musehub.api.routes.wire.wire_push_stream", _spy)

    body = _header_frame() + _end_frame()
    resp = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=body,
        headers={"Content-Type": GRPC_CONTENT_TYPE},
    )

    if resp.status_code in (401, 403):
        assert not wire_was_called, "wire_push_stream called despite auth failure"
    else:
        unpacker = msgpack.Unpacker(raw=False)
        unpacker.feed(resp.content)
        frames = list(unpacker)
        error_frames = [f for f in frames if f.get("t") == "X"]
        assert error_frames or not wire_was_called