"""Tests for ``muse range-diff`` — compare two versions of a commit series.

Coverage tiers:
- Unit:        range parsing, patch-id computation per commit, pairing logic
               (identical series, single changed, dropped commit, added commit,
               reordered, empty series)
- Integration: JSON schema, text output, nonexistent ref exits nonzero,
               trivially equivalent flag, creation-factor=0 no fuzzy pairing,
               both ranges empty
- Security:    ANSI in range arg rejected
- Stress:      50-commit series fully equivalent (sub-second); 50-commit with
               25 changed, 10 dropped, 10 added
"""

from __future__ import annotations

import datetime
import hashlib
import json
import pathlib

import pytest

from tests.cli_test_helper import CliRunner
from muse.core.object_store import write_object
from muse.core.snapshot import compute_commit_id, compute_snapshot_id
from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
from muse.core._types import Manifest

runner = CliRunner()

_REPO_ID = "range-diff-test"
_counter = 0


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


def _sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _init_repo(path: pathlib.Path) -> pathlib.Path:
    muse = path / ".muse"
    for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
        (muse / d).mkdir(parents=True, exist_ok=True)
    (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
    (muse / "repo.json").write_text(
        json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
    )
    return path


def _env(repo: pathlib.Path) -> dict[str, str]:
    return {"MUSE_REPO_ROOT": str(repo)}


def _write_files(root: pathlib.Path, files: dict[str, bytes]) -> Manifest:
    manifest: Manifest = {}
    for rel, content in files.items():
        oid = _sha(content)
        write_object(root, oid, content)
        manifest[rel] = oid
        p = root / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(content)
    return manifest


def _commit(
    root: pathlib.Path,
    files: dict[str, bytes],
    branch: str = "main",
    parent_id: str | None = None,
    message: str | None = None,
) -> str:
    global _counter
    _counter += 1
    manifest = _write_files(root, files)
    snap_id = compute_snapshot_id(manifest)
    write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
    committed_at = datetime.datetime.now(datetime.timezone.utc)
    msg = message or f"commit {_counter}"
    commit_id = compute_commit_id(
        [parent_id] if parent_id else [], snap_id, msg, committed_at.isoformat()
    )
    write_commit(root, CommitRecord(
        commit_id=commit_id, repo_id=_REPO_ID, branch=branch,
        snapshot_id=snap_id, message=msg, committed_at=committed_at,
        parent_commit_id=parent_id,
    ))
    ref_path = root / ".muse" / "refs" / "heads" / branch
    ref_path.parent.mkdir(parents=True, exist_ok=True)
    ref_path.write_text(commit_id, encoding="utf-8")
    return commit_id


def _invoke(repo: pathlib.Path, *args: str):
    from muse.cli.app import main as cli
    return runner.invoke(cli, ["range-diff", *args], env=_env(repo))


# ---------------------------------------------------------------------------
# Unit — range parsing (imported directly)
# ---------------------------------------------------------------------------


def test_parse_range_with_dotdot() -> None:
    from muse.cli.commands.range_diff import _parse_range
    base, tip = _parse_range("abc..def")
    assert base == "abc"
    assert tip == "def"


def test_parse_range_no_dotdot() -> None:
    from muse.cli.commands.range_diff import _parse_range
    base, tip = _parse_range("main")
    assert base is None
    assert tip == "main"


def test_parse_range_preserves_whitespace_stripped() -> None:
    from muse.cli.commands.range_diff import _parse_range
    base, tip = _parse_range("base .. tip")
    assert base == "base"
    assert tip == "tip"


# ---------------------------------------------------------------------------
# Unit — pairing logic
# ---------------------------------------------------------------------------


def test_identical_series_all_equivalent(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    # old series: 3 commits
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
    c2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="old", parent_id=c1)
    c3 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n", "c.py": b"c=3\n"}, branch="old", parent_id=c2)
    # new series: identical content → same patch-ids
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
    n2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="new", parent_id=n1)
    n3 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n", "c.py": b"c=3\n"}, branch="new", parent_id=n2)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["trivially_equivalent"] is True
    assert all(p["status"] == "equivalent" for p in data["pairs"])


def test_single_commit_changed(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    # old: add a.py with content v1
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
    # new: add a.py with content v2 (different patch-id)
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    data = json.loads(result.stdout)
    assert data["trivially_equivalent"] is False
    # One pair, status should be "changed" (different patch-ids, positionally paired)
    assert len(data["pairs"]) == 1
    assert data["pairs"][0]["status"] == "changed"


def test_commit_dropped(tmp_path: pathlib.Path) -> None:
    """Old series has 2 commits; new series only has 1 (one was dropped/squashed)."""
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base, message="add a")
    c2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="old", parent_id=c1, message="add b")
    # new: squashed into one commit with same final content as old c2
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="new", parent_id=base, message="add a and b")
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    data = json.loads(result.stdout)
    statuses = {p["status"] for p in data["pairs"]}
    # At least one commit should be dropped or the squash results in a "changed" pair
    assert "dropped" in statuses or "changed" in statuses


def test_commit_added(tmp_path: pathlib.Path) -> None:
    """New series has an extra commit not in old series."""
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
    # new: same first commit + an extra one
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
    n2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "extra.py": b"e=9\n"}, branch="new", parent_id=n1)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    data = json.loads(result.stdout)
    statuses = [p["status"] for p in data["pairs"]]
    assert "added" in statuses
    assert "equivalent" in statuses  # c1 ↔ n1


def test_empty_old_series_all_added(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
    # old range is base..base → empty
    result = _invoke(root, f"{base}..{base}", f"{base}..new", "--json")
    data = json.loads(result.stdout)
    assert all(p["status"] == "added" for p in data["pairs"])


def test_empty_new_series_all_dropped(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
    # new range is base..base → empty
    result = _invoke(root, f"{base}..old", f"{base}..{base}", "--json")
    data = json.loads(result.stdout)
    assert all(p["status"] == "dropped" for p in data["pairs"])


def test_both_empty_trivially_equivalent(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    result = _invoke(root, f"{base}..{base}", f"{base}..{base}", "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["trivially_equivalent"] is True
    assert data["pairs"] == []


# ---------------------------------------------------------------------------
# Integration — JSON schema
# ---------------------------------------------------------------------------


def test_json_schema_all_fields(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    for key in ("pairs", "trivially_equivalent", "old_range", "new_range"):
        assert key in data
    pair = data["pairs"][0]
    for key in ("old", "new", "status"):
        assert key in pair
    if pair["old"] is not None:
        for key in ("commit_id", "patch_id", "subject"):
            assert key in pair["old"]


def test_json_pair_commit_info(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base, message="add a")
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base, message="add a")
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    data = json.loads(result.stdout)
    pair = data["pairs"][0]
    assert pair["old"]["subject"] == "add a"
    assert pair["new"]["subject"] == "add a"
    assert pair["old"]["commit_id"] == c1
    assert pair["new"]["commit_id"] == n1
    assert pair["status"] == "equivalent"


# ---------------------------------------------------------------------------
# Integration — text output
# ---------------------------------------------------------------------------


def test_text_output_shows_equivalent(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base, message="add a")
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base, message="add a")
    result = _invoke(root, f"{base}..old", f"{base}..new")
    assert result.exit_code == 0
    assert "equivalent" in result.stdout.lower() or "=" in result.stdout


def test_text_output_shows_changed(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base, message="add a v1")
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base, message="add a v2")
    result = _invoke(root, f"{base}..old", f"{base}..new")
    assert result.exit_code != 0  # has changes
    assert "changed" in result.stdout.lower() or "!" in result.stdout


# ---------------------------------------------------------------------------
# Integration — creation-factor
# ---------------------------------------------------------------------------


def test_creation_factor_zero_no_fuzzy_pairing(tmp_path: pathlib.Path) -> None:
    """With --creation-factor=0.0, only exact patch-id matches are paired."""
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--creation-factor", "0.0", "--json")
    data = json.loads(result.stdout)
    # No exact patch-id match, creation-factor=0 → dropped + added
    statuses = {p["status"] for p in data["pairs"]}
    assert "changed" not in statuses
    assert "dropped" in statuses or "added" in statuses


def test_creation_factor_one_all_positionally_paired(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--creation-factor", "1.0", "--json")
    data = json.loads(result.stdout)
    statuses = {p["status"] for p in data["pairs"]}
    assert "changed" in statuses  # positionally paired → changed


# ---------------------------------------------------------------------------
# Integration — error cases
# ---------------------------------------------------------------------------


def test_nonexistent_old_ref_exits_nonzero(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"a.py": b"x\n"}, branch="main")
    result = _invoke(root, "ghost..no-such", f"{base}..main")
    assert result.exit_code != 0


def test_nonexistent_new_ref_exits_nonzero(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"a.py": b"x\n"}, branch="main")
    result = _invoke(root, f"{base}..main", "ghost..no-such")
    assert result.exit_code != 0


def test_exit_zero_on_trivially_equivalent(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    assert result.exit_code == 0


def test_exit_nonzero_on_differences(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
    c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
    n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    assert result.exit_code != 0


# ---------------------------------------------------------------------------
# Security
# ---------------------------------------------------------------------------


def test_ansi_in_old_range_rejected(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    result = _invoke(root, "\x1b[31mbad\x1b[0m..main", "main..main")
    assert result.exit_code != 0


def test_ansi_in_new_range_rejected(tmp_path: pathlib.Path) -> None:
    root = _init_repo(tmp_path)
    result = _invoke(root, "main..main", "\x1b[31mbad\x1b[0m..main")
    assert result.exit_code != 0


# ---------------------------------------------------------------------------
# Stress — 50 commits
# ---------------------------------------------------------------------------


def test_stress_50_commits_trivially_equivalent(tmp_path: pathlib.Path) -> None:
    """50-commit series on each side, all equivalent — must complete quickly."""
    root = _init_repo(tmp_path)
    base = _commit(root, {"base.py": b"base\n"}, branch="main")
    old_id = base
    new_id = base
    for i in range(50):
        content = f"v = {i}\n".encode()
        old_id = _commit(root, {f"f{i}.py": content}, branch="old", parent_id=old_id, message=f"add f{i}")
        new_id = _commit(root, {f"f{i}.py": content}, branch="new", parent_id=new_id, message=f"add f{i}")
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["trivially_equivalent"] is True
    assert len(data["pairs"]) == 50


def test_stress_50_commits_mixed(tmp_path: pathlib.Path) -> None:
    """50-commit old series, new has 25 equivalent + 15 changed + 10 added."""
    root = _init_repo(tmp_path)
    base = _commit(root, {"base.py": b"base\n"}, branch="main")
    old_id = base
    new_id = base
    # First 25: identical
    for i in range(25):
        content = f"v = {i}\n".encode()
        old_id = _commit(root, {f"f{i}.py": content}, branch="old", parent_id=old_id)
        new_id = _commit(root, {f"f{i}.py": content}, branch="new", parent_id=new_id)
    # Next 25: different content
    for i in range(25, 50):
        old_id = _commit(root, {f"f{i}.py": f"old_{i}\n".encode()}, branch="old", parent_id=old_id)
        new_id = _commit(root, {f"f{i}.py": f"new_{i}\n".encode()}, branch="new", parent_id=new_id)
    # New has 10 extra commits
    for i in range(50, 60):
        new_id = _commit(root, {f"extra{i}.py": b"extra\n"}, branch="new", parent_id=new_id)
    result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
    assert result.exit_code != 0  # has changes
    data = json.loads(result.stdout)
    equivalent = [p for p in data["pairs"] if p["status"] == "equivalent"]
    assert len(equivalent) == 25
    added = [p for p in data["pairs"] if p["status"] == "added"]
    assert len(added) == 10