import pytest
from pydantic import ValidationError

from app.catalogue import Catalogue, build_record
from app.events import COLUMNS, ClientEvent, EventBatch, clean_notes, to_row

SID = "0123456789abcdef"


class FakeCatalogue:
    def __init__(self, recs):
        self.by_id = {r.id: r for r in recs}


CAT = FakeCatalogue([
    build_record({"id": "mlx-community/Qwen3-8B-4bit", "downloads": 42, "likes": 3,
                  "safetensors": {"total": 8.19e9}, "tags": ["4-bit"], "pipeline_tag": "text-generation"}),
    build_record({"id": "mlx-community/Llama-3.3-70B-Instruct-4bit", "safetensors": {"total": 70e9},
                  "tags": ["4-bit"], "pipeline_tag": "text-generation"}),
])


def ev(**kw):
    return ClientEvent(**{"event_type": "search", "session_id": SID, **kw})


def test_session_row_has_all_columns_and_no_pii():
    row = to_row(ev(event_type="session", model_family="Qwen", parameter_bucket="8-15B", quantization="4-bit",
                    target_context=32768, priority="balanced", hardware_source="confirmed",
                    hardware_memory_class=36, families_searched=["Qwen", "Gemma", "Qwen"], quants_searched=["4-bit"],
                    distinct_queries=3, top_model="mlx-community/Qwen3-8B-4bit", top_model_score=88.2,
                    top_model_fit="Comfortable", models_viewed=["mlx-community/Qwen3-8B-4bit"]), CAT)
    assert set(row) == set(COLUMNS)
    assert row["schema_version"] == 2 and row["event_type"] == "session"
    assert row["families_searched"] == ["Qwen", "Gemma"]  # de-duplicated
    assert row["hardware_confirmed"] is True and row["suspicious_flags"] == []
    assert row["timestamp"].endswith("Z") and "." not in row["timestamp"]
    for forbidden in ("ip", "user_agent", "email", "cookie"):
        assert forbidden not in row


def test_feedback_gets_catalogue_facts():
    row = to_row(ev(event_type="feedback", selected_model="mlx-community/Qwen3-8B-4bit", tried="yes",
                    quality_rating="good"), CAT)
    assert row["hf_downloads_at_selection"] == 42 and row["quant_bits"] == 4 and row["outcome"] == "worked"
    assert row["model_family"] == "Qwen" and row["parameter_bucket"] == "8-15B"


def test_legacy_clickstream_events_are_dropped_and_old_bench_name_is_accepted():
    assert to_row(ev(event_type="search", model_family="Qwen"), CAT) is None
    assert to_row(ev(event_type="model_view", selected_model="mlx-community/Qwen3-8B-4bit"), CAT) is None
    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                    generation_tps=40, benchmark_version="mlxbench-1"), CAT)
    assert row["event_type"] == "mlx_benchmark" and row["suspicious_flags"] == []


def test_session_lists_are_capped_and_checked():
    with pytest.raises(ValidationError):
        ev(event_type="session", models_viewed=[f"mlx-community/m{i}" for i in range(11)])
    with pytest.raises(ValidationError):
        ev(event_type="session", models_compared=["../etc/passwd"])
    with pytest.raises(ValidationError):
        ev(event_type="session", families_searched=["a", "b", "c", "d", "e", "f"])
    row = to_row(ev(event_type="session", models_clicked=["mlx-community/Nope-1B"]), CAT)
    assert "unknown_model_in_session" in row["suspicious_flags"]


def test_latest_sessions_keeps_newest_row_per_visit():
    from app.events import latest_sessions
    rows = [
        {"event_type": "session", "session_id": "a" * 16, "timestamp": "2026-09-14T01:00:00Z", "distinct_queries": 1},
        {"event_type": "feedback", "session_id": "a" * 16, "timestamp": "2026-09-14T01:00:30Z"},
        {"event_type": "session", "session_id": "a" * 16, "timestamp": "2026-09-14T01:05:00Z", "distinct_queries": 4},
        {"event_type": "session", "session_id": "b" * 16, "timestamp": "2026-09-14T01:05:00Z", "distinct_queries": 1},
        {"event_type": "session", "session_id": "b" * 16, "timestamp": "2026-09-14T01:05:00Z", "distinct_queries": 2},
    ]
    out = latest_sessions(rows)
    sessions = {r["session_id"]: r["distinct_queries"] for r in out if r["event_type"] == "session"}
    assert sessions == {"a" * 16: 4, "b" * 16: 2}  # later arrival wins a same-second tie
    assert sum(1 for r in out if r["event_type"] == "feedback") == 1 and len(out) == 3


@pytest.mark.parametrize("bad", [
    {"event_type": "delete_everything"},
    {"event_type": "session", "compare_models": ["a/b"]},
    {"event_type": "search", "session_id": "NOT-HEX"},
    {"event_type": "search", "target_context": 12345},
    {"event_type": "search", "hardware_memory_class": 17},
    {"event_type": "search", "quantization": "7-bit"},
    {"event_type": "feedback", "reported_tokens_per_second": -1},
    {"event_type": "feedback", "reported_tokens_per_second": 99999},
    {"event_type": "search", "selected_model": "../../etc/passwd"},
    {"event_type": "search", "selected_model": "<script>/x"},
    {"event_type": "mlx_benchmark_submission", "chip": "Intel i9"},
    {"event_type": "search", "ip": "1.2.3.4"},
    {"event_type": "search", "model_family": "<img src=x>"},
    {"event_type": "search", "benchmark_version": "1.0; rm -rf"},
])
def test_rejects_malformed(bad):
    with pytest.raises(ValidationError):
        ClientEvent(**bad)


def test_batch_limits():
    with pytest.raises(ValidationError):
        EventBatch(events=[])
    with pytest.raises(ValidationError):
        EventBatch(events=[{"event_type": "session"}] * 21)


def test_notes_are_scrubbed_and_capped():
    n = clean_notes("mail me a@b.com\x00 see https://evil.example/x or call +1 (555) 123-4567 " + "z" * 500)
    assert "a@b.com" not in n and "https" not in n and "555" not in n and "\x00" not in n
    assert len(n) <= 280
    assert clean_notes("   ") is None
    e = ev(event_type="feedback", notes="<script>alert(1)</script>")
    assert e.notes == "<script>alert(1)</script>"  # stored as inert text; UI renders textContent only


def test_flags_unknown_model_and_implausible_tps():
    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Nope-1B",
                    generation_tps=10, benchmark_version="1", benchmark_type="mlx_lm"), CAT)
    assert "unknown_model" in row["suspicious_flags"]
    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Llama-3.3-70B-Instruct-4bit",
                    generation_tps=1500, benchmark_version="1", benchmark_type="mlx_lm"), CAT)
    assert "implausible_tps" in row["suspicious_flags"]
    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                    generation_tps=45, benchmark_version="1", benchmark_type="mlx_lm", peak_memory_gb=5.1,
                    reported_ram_gb=36), CAT)
    assert row["suspicious_flags"] == []


def test_flags_other_inconsistencies():
    row = to_row(ev(event_type="mlx_benchmark", peak_memory_gb=40, reported_ram_gb=16), CAT)
    assert {"peak_memory_exceeds_ram", "incomplete_benchmark"} <= set(row["suspicious_flags"])
    row = to_row(ev(event_type="feedback", tried="yes", quality_rating="good", failure_reason="too_slow"), CAT)
    assert "conflicting_feedback" in row["suspicious_flags"]
    row = to_row(ev(event_type="browser_benchmark", benchmark_type="mlx_lm"), CAT)
    assert "benchmark_type_mismatch" in row["suspicious_flags"]



def test_quality_submission_validation():
    ok = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                   benchmark_type="mlx_lm", benchmark_version="mlxbench-1", perplexity=9.8, perplexity_stderr=0.2,
                   eval_dataset="wikitext2-test-v1", eval_tokens=16368), CAT)
    assert ok["suspicious_flags"] == [] and ok["perplexity"] == 9.8
    bad = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                    benchmark_version="mlxbench-1", perplexity=5000, eval_dataset="wikitext2-test-v1", eval_tokens=16368), CAT)
    assert "implausible_perplexity" in bad["suspicious_flags"]
    inc = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                    benchmark_version="mlxbench-1", perplexity=9.8), CAT)
    assert "incomplete_quality" in inc["suspicious_flags"]
    with pytest.raises(ValidationError):
        ev(event_type="mlx_benchmark", perplexity=0.5)
    with pytest.raises(ValidationError):
        ev(event_type="mlx_benchmark", perplexity=9.8, eval_dataset="my-own-text")


def test_perplexity_from_old_bench_version_is_ignored():
    from app.stats import community_signals

    row = dict(event_type="mlx_benchmark", selected_model="mlx-community/gemma-4-e2b-it-4bit", perplexity=5268.9,
               eval_dataset="wikitext2-test-v1", suspicious_flags=[])
    sig = community_signals([{**row, "benchmark_version": "mlxbench-1"},
                             {**row, "benchmark_version": "mlxbench-2", "perplexity": 21.4}])
    assert sig[row["selected_model"]].median_perplexity == 21.4
    assert sig[row["selected_model"]].perplexity_count == 1


def test_quality_only_benchmark_is_clean():
    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit", benchmark_type="mlx_lm",
                    benchmark_version="mlxbench-2", perplexity=9.8, perplexity_stderr=0.2, eval_dataset="wikitext2-test-v1",
                    eval_tokens=16368, peak_memory_gb=5.1, reported_ram_gb=36, chip="Apple M3 Max"), CAT)
    assert row["suspicious_flags"] == [] and row["generation_tps"] is None