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_valid_search_row_has_all_columns_and_no_pii():
    row = to_row(ev(model_family="Qwen", parameter_bucket="8-15B", quantization="4-bit",
                    target_context=32768, priority="balanced", hardware_source="confirmed",
                    hardware_memory_class=36), CAT)
    assert set(row) == set(COLUMNS)
    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", "name", "cookie"):
        assert forbidden not in row


def test_server_fills_catalogue_facts():
    row = to_row(ev(event_type="model_select", selected_model="mlx-community/Qwen3-8B-4bit",
                    selected_model_rank=0), CAT)
    assert row["hf_downloads_at_selection"] == 42 and row["quant_bits"] == 4
    assert row["model_family"] == "Qwen" and row["parameter_bucket"] == "8-15B"


@pytest.mark.parametrize("bad", [
    {"event_type": "delete_everything"},
    {"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"},
    {"event_type": "search", "compare_models": ["a/b"] * 6},
])
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": "search"}] * 51)


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_submission", 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_submission", 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_submission", 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_submission", 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"]