Spaces:
Running
Running
File size: 9,491 Bytes
13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 a866fc9 13b1a91 9327c55 a866fc9 9327c55 a866fc9 9327c55 a866fc9 9327c55 a866fc9 9327c55 a866fc9 bed5ea1 8275360 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | 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
|