Spaces:
Running
Running
| 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 | |
| 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 | |