"""Tests for checklist 2.4 — Commit integrity on push/stream.

Covers:
  - Forged parent_id rejection at receive time
  - Root commit with no parent accepted
  - Parent from different repo rejected
"""
from __future__ import annotations

import secrets
import struct
from datetime import datetime, timezone

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

from tests.factories import create_repo as factory_create_repo
from muse.core.types import blob_id, now_utc_iso
from musehub.types.json_types import JSONObject, StrDict


# ── helpers ────────────────────────────────────────────────────────────────────




def _last_frame(raw: bytes) -> JSONObject:
    """Return the last msgpack frame from a push-stream response body."""
    unpacker = msgpack.Unpacker(raw=False)
    unpacker.feed(raw)
    last: JSONObject = {}
    for frame in unpacker:
        last = frame
    return last


def _is_rejection(resp_content: bytes) -> bool:
    """True if the push-stream response ended with an error frame."""
    frame = _last_frame(resp_content)
    return frame.get("t") == "X" or not frame.get("ok", True)


def _sha256_object_id(content: bytes) -> str:
    return blob_id(content)


def _mp(data: JSONObject) -> bytes:
    return msgpack.packb(data, use_bin_type=True)


def _sha256_id(seed: str) -> str:
    return blob_id(seed.encode())


def _rand8() -> str:
    return secrets.token_hex(4)


def _make_commit(
    commit_id: str | None = None,
    parent: str | None = None,
    snap_id: str | None = None,
) -> JSONObject:
    uid = secrets.token_hex(16)
    return {
        "commit_id": commit_id or _sha256_id(f"commit-{uid}"),
        "branch": "main",
        "snapshot_id": snap_id or _sha256_id(f"snap-{uid}"),
        "message": "test commit",
        "committed_at": now_utc_iso(),
        "parent_commit_id": parent,
        "author": "Test User <test@example.com>",
    }


def _make_valid_object(content: bytes) -> JSONObject:
    """Object whose object_id is the correct sha256 of content."""
    return {
        "object_id": _sha256_object_id(content),
        "content": content,
        "path": "file.bin",
        "encoding": "raw",
    }


def _make_tampered_object(content: bytes) -> JSONObject:
    """Object whose object_id claims sha256 of DIFFERENT content."""
    wrong_content = content + b"\x00"
    return {
        "object_id": _sha256_object_id(wrong_content),  # hash of wrong_content
        "content": content,                              # but we send content
        "path": "file.bin",
        "encoding": "raw",
    }


def _make_non_sha256_object(content: bytes) -> JSONObject:
    """Object_id without sha256: prefix — rejected because all IDs are canonically sha256."""
    return {
        "object_id": f"blob:{secrets.token_hex(16)}",
        "content": content,
        "path": "file.bin",
        "encoding": "raw",
    }


# ── MWP frame builders ─────────────────────────────────────────────────────────

def _mwp_frame(ft: str, data: JSONObject) -> bytes:
    payload = msgpack.packb(data, use_bin_type=True)
    envelope = msgpack.packb(
        {"ft": ft, "sz": len(payload), "id": blob_id(payload)},
        use_bin_type=True,
    )
    return (
        b"muse"
        + b"\x01"
        + struct.pack(">I", len(envelope))
        + envelope
        + struct.pack(">Q", len(payload))
        + payload
    )


def _mwp_stream(
    commits: list[JSONObject],
    *,
    objects: list[JSONObject] | None = None,
    branch: str = "main",
) -> bytes:
    objects = objects or []
    frames: list[bytes] = [
        _mwp_frame("H", {
            "t": "H",
            "branch": branch,
            "force": False,
            "head": None,
            "have": [],
            "n_objects": len(objects),
            "n_commits": len(commits),
        })
    ]
    for obj in objects:
        frames.append(_mwp_frame("O", {
            "t": "O",
            "id": obj["object_id"],
            "path": obj.get("path", "file.bin"),
            "content": obj["content"],
            "enc": "raw",
        }))
    frames.append(_mwp_frame("C", {
        "t": "C",
        "commits": commits,
        "snapshots": [],
    }))
    frames.append(_mwp_frame("E", {
        "t": "E",
        "n_objects": len(objects),
        "n_commits": len(commits),
    }))
    return b"".join(frames)


# ── Parent commit integrity: /push/stream ─────────────────────────────────────

async def test_push_commit_with_parent_in_bundle_succeeds(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """A commit whose parent_commit_id is in the same push bundle must be accepted."""
    repo = await factory_create_repo(
        db_session, slug=f"integrity-parent-in-bundle-{_rand8()}", owner="test-user-wire"
    )
    parent_id = _sha256_id(f"parent-{secrets.token_hex(16)}")
    child_id = _sha256_id(f"child-{secrets.token_hex(16)}")
    parent = _make_commit(commit_id=parent_id)
    child = _make_commit(commit_id=child_id, parent=parent_id)

    resp = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=_mwp_stream([parent, child]),
        headers={**wire_headers, "Content-Type": "application/x-muse-wire"},
    )
    assert resp.status_code == 200


async def test_push_commit_with_parent_in_db_succeeds(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """A commit whose parent is already in the DB for this repo must be accepted."""
    repo = await factory_create_repo(
        db_session, slug=f"integrity-parent-in-db-{_rand8()}", owner="test-user-wire"
    )
    parent_id = _sha256_id(f"parent-{secrets.token_hex(16)}")
    child_id = _sha256_id(f"child-{secrets.token_hex(16)}")
    parent = _make_commit(commit_id=parent_id)
    stream_headers = {**wire_headers, "Content-Type": "application/x-muse-wire"}

    r1 = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=_mwp_stream([parent]),
        headers=stream_headers,
    )
    assert r1.status_code == 200

    child = _make_commit(commit_id=child_id, parent=parent_id)
    r2 = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=_mwp_stream([child]),
        headers=stream_headers,
    )
    assert r2.status_code == 200


async def test_push_commit_with_forged_parent_id_is_rejected(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """A commit referencing a parent that exists in neither the bundle nor the repo DB
    must be rejected."""
    repo = await factory_create_repo(
        db_session, slug=f"integrity-forged-parent-{_rand8()}", owner="test-user-wire"
    )
    forged_parent_id = _sha256_id(f"forged-{secrets.token_hex(16)}")
    child_id = _sha256_id(f"child-{secrets.token_hex(16)}")
    child = _make_commit(commit_id=child_id, parent=forged_parent_id)

    resp = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=_mwp_stream([child]),
        headers={**wire_headers, "Content-Type": "application/x-muse-wire"},
    )
    assert resp.status_code == 200
    assert _is_rejection(resp.content), f"Expected rejection, got: {_last_frame(resp.content)}"
    frame = _last_frame(resp.content)
    msg = frame.get("msg", "").lower()
    assert "parent" in msg or "rejected" in msg, f"Expected parent/rejected in error msg, got: {msg}"


async def test_push_commit_with_parent_from_different_repo_is_rejected(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """A commit whose parent_id exists in a DIFFERENT repo must be rejected."""
    repo_a = await factory_create_repo(
        db_session, slug=f"integrity-repo-a-{_rand8()}", owner="test-user-wire"
    )
    repo_b = await factory_create_repo(
        db_session, slug=f"integrity-repo-b-{_rand8()}", owner="test-user-wire"
    )
    stream_headers = {**wire_headers, "Content-Type": "application/x-muse-wire"}

    commit_in_a_id = _sha256_id(f"commit-a-{secrets.token_hex(16)}")
    commit_in_a = _make_commit(commit_id=commit_in_a_id)
    r1 = await client.post(
        f"/{repo_a.owner}/{repo_a.slug}/push/stream",
        content=_mwp_stream([commit_in_a]),
        headers=stream_headers,
    )
    assert r1.status_code == 200

    child_id = _sha256_id(f"child-b-{secrets.token_hex(16)}")
    child = _make_commit(commit_id=child_id, parent=commit_in_a_id)
    r2 = await client.post(
        f"/{repo_b.owner}/{repo_b.slug}/push/stream",
        content=_mwp_stream([child]),
        headers=stream_headers,
    )
    assert r2.status_code == 200
    assert _is_rejection(r2.content), f"Expected rejection, got: {_last_frame(r2.content)}"
    frame = _last_frame(r2.content)
    msg = frame.get("msg", "").lower()
    assert "parent" in msg or "rejected" in msg, f"Expected parent/rejected in error msg, got: {msg}"


async def test_push_root_commit_with_no_parent_succeeds(
    client: AsyncClient,
    db_session: AsyncSession,
    wire_headers: StrDict,
) -> None:
    """A root commit (no parent_commit_id) must be accepted — this is the initial push."""
    repo = await factory_create_repo(
        db_session, slug=f"integrity-root-commit-{_rand8()}", owner="test-user-wire"
    )
    root = _make_commit(parent=None)

    resp = await client.post(
        f"/{repo.owner}/{repo.slug}/push/stream",
        content=_mwp_stream([root]),
        headers={**wire_headers, "Content-Type": "application/x-muse-wire"},
    )
    assert resp.status_code == 200