| """Personalisation rules: vocabulary restriction, threshold clamp, calibration, fine-tune.""" |
|
|
| from __future__ import annotations |
|
|
| import numpy as np |
| import pytest |
|
|
| import config |
| from backend import personalization as pers |
|
|
|
|
| @pytest.fixture(autouse=True) |
| def isolated_profiles(tmp_path, monkeypatch): |
| """Never write profiles into the repository during tests.""" |
| monkeypatch.setattr(pers, "PROFILE_ROOT", tmp_path / "personalized") |
| return tmp_path / "personalized" |
|
|
|
|
| |
| |
| |
| def test_create_profile_defaults(): |
| p = pers.create_profile("alice", base_model="gru") |
| assert p["base_model"] == "gru" |
| assert p["threshold"] == config.CONFIDENCE_THRESHOLD |
| assert p["vocabulary"] == [] |
| assert p["calibration_samples"] == 0 |
| assert "not proven" in p["notes"] |
|
|
|
|
| @pytest.mark.parametrize("name", ["../etc", "with space", "", "x" * 40, "semi;colon"]) |
| def test_profile_name_validation(name): |
| with pytest.raises(pers.PersonalizationError): |
| pers.create_profile(name) |
|
|
|
|
| def test_unknown_base_model_rejected(): |
| with pytest.raises(pers.PersonalizationError): |
| pers.create_profile("bob", base_model="gpt") |
|
|
|
|
| def test_vocabulary_must_come_from_include50(): |
| with pytest.raises(pers.PersonalizationError): |
| pers.create_profile("bob", vocabulary=["TOTALLY_MADE_UP"]) |
| p = pers.create_profile("bob", vocabulary=["hello", "bank"]) |
| assert set(p["vocabulary"]) == {"HELLO", "BANK"} |
|
|
|
|
| def test_preset_vocabulary(): |
| p = pers.create_profile("carol", preset="greetings") |
| assert p["vocabulary"] == config.PERSONALIZATION_DEFAULT_VOCAB_PRESETS["greetings"] |
| assert p["vocabulary"], "a preset must yield a non-empty subset" |
|
|
|
|
| def test_every_preset_entry_is_an_official_class(): |
| """A preset must never invent a sign that the dataset does not contain.""" |
| from backend.preprocessing.include50 import load_vocabulary |
|
|
| official = set(load_vocabulary()) |
| for name, entries in config.PERSONALIZATION_DEFAULT_VOCAB_PRESETS.items(): |
| assert set(entries) <= official, f"preset '{name}' has unknown entries: {set(entries) - official}" |
|
|
|
|
| def test_threshold_is_clamped_to_a_small_band(): |
| p = pers.create_profile("dave", threshold=0.99) |
| assert p["threshold"] == pytest.approx( |
| config.CONFIDENCE_THRESHOLD + config.PERSONALIZATION_MAX_THRESHOLD_DELTA) |
| assert p["threshold_requested"] == 0.99 |
| q = pers.create_profile("dave2", threshold=0.0) |
| assert q["threshold"] == pytest.approx( |
| config.CONFIDENCE_THRESHOLD - config.PERSONALIZATION_MAX_THRESHOLD_DELTA) |
|
|
|
|
| def test_update_and_delete_profile(): |
| pers.create_profile("erin") |
| updated = pers.update_profile("erin", vocabulary=["BANK"], threshold=0.75) |
| assert updated["vocabulary"] == ["BANK"] and updated["threshold"] == pytest.approx(0.75) |
| assert pers.delete_profile("erin") is True |
| assert pers.delete_profile("erin") is False |
| with pytest.raises(pers.PersonalizationError): |
| pers.load_profile("erin") |
|
|
|
|
| def test_list_profiles_returns_created_ones(): |
| pers.create_profile("frank") |
| pers.create_profile("gina") |
| assert {p["name"] for p in pers.list_profiles()} >= {"frank", "gina"} |
|
|
|
|
| |
| |
| |
| def _samples(n: int, classes=("HELLO", "BANK")) -> tuple: |
| rng = np.random.default_rng(0) |
| X = rng.normal(0, 0.1, size=(n, config.SEQUENCE_LENGTH, config.MODEL_INPUT_DIM)).astype(np.float32) |
| labels = [classes[i % len(classes)] for i in range(n)] |
| return X, labels |
|
|
|
|
| def test_add_calibration_samples_counts_per_class(): |
| pers.create_profile("hina", vocabulary=["HELLO", "BANK", "BANK"]) |
| X, labels = _samples(6) |
| p = pers.add_calibration_samples("hina", X, labels) |
| assert p["calibration_samples"] == 6 |
| assert p["per_class_samples"] == {"HELLO": 3, "BANK": 3} |
| Xs, ys = pers.load_calibration("hina") |
| assert Xs.shape == (6, config.SEQUENCE_LENGTH, config.MODEL_INPUT_DIM) |
| assert ys.shape == (6,) |
|
|
|
|
| def test_calibration_appends_instead_of_overwriting(): |
| pers.create_profile("iris", vocabulary=["HELLO"]) |
| X, labels = _samples(2, ("HELLO",)) |
| pers.add_calibration_samples("iris", X, labels) |
| p = pers.add_calibration_samples("iris", X, labels) |
| assert p["calibration_samples"] == 4 |
|
|
|
|
| def test_calibration_rejects_wrong_shape(): |
| pers.create_profile("jack", vocabulary=["HELLO"]) |
| with pytest.raises(pers.PersonalizationError): |
| pers.add_calibration_samples("jack", np.zeros((3, 30, 100), dtype=np.float32), ["HELLO"] * 3) |
|
|
|
|
| def test_calibration_rejects_unknown_label(): |
| pers.create_profile("kim", vocabulary=["HELLO"]) |
| X, _ = _samples(1, ("HELLO",)) |
| with pytest.raises(pers.PersonalizationError): |
| pers.add_calibration_samples("kim", X, ["NOT_A_SIGN"]) |
|
|
|
|
| def test_calibration_rejects_label_outside_profile_vocabulary(): |
| pers.create_profile("lee", vocabulary=["HELLO"]) |
| X, _ = _samples(1, ("BANK",)) |
| with pytest.raises(pers.PersonalizationError): |
| pers.add_calibration_samples("lee", X, ["BANK"]) |
|
|
|
|
| def test_calibration_label_count_must_match(): |
| pers.create_profile("mo", vocabulary=["HELLO"]) |
| X, _ = _samples(2, ("HELLO",)) |
| with pytest.raises(pers.PersonalizationError): |
| pers.add_calibration_samples("mo", X, ["HELLO"]) |
|
|
|
|
| |
| |
| |
| def test_can_fine_tune_requires_minimum_samples(): |
| pers.create_profile("nina", vocabulary=["HELLO", "BANK"]) |
| ok, why = pers.can_fine_tune("nina") |
| assert ok is False and "at least" in why |
| X, labels = _samples(config.PERSONALIZATION_MIN_SAMPLES_FOR_FINETUNE) |
| pers.add_calibration_samples("nina", X, labels) |
| ok, why = pers.can_fine_tune("nina") |
| assert ok is True, why |
|
|
|
|
| def test_compile_vocabulary_keeps_index_space(): |
| pers.create_profile("omar", vocabulary=["BANK", "HELLO"]) |
| classes = pers.compile_vocabulary(pers.load_profile("omar")) |
| assert set(classes) == {"BANK", "HELLO"} |
| |
| from backend.preprocessing.include50 import load_vocabulary |
|
|
| assert classes == [c for c in load_vocabulary() if c in {"BANK", "HELLO"}] |
|
|
|
|
| def test_apply_profile_to_engine(trained_model): |
| from backend.inference.predictor import PredictionEngine |
|
|
| name, model = trained_model |
| pers.create_profile("pia", vocabulary=["HELLO", "BANK"], threshold=0.8) |
| engine = PredictionEngine(model_name=name, model=model) |
| applied = pers.apply_profile_to_engine(engine, "pia") |
| assert applied["threshold"] == pytest.approx(0.8) |
| assert set(engine.vocab_filter) == {"HELLO", "BANK"} |
| x = np.zeros((config.SEQUENCE_LENGTH, config.MODEL_INPUT_DIM), dtype=np.float32) |
| probs = engine.probabilities(x) |
| outside = [i for i, c in enumerate(engine.classes) if c not in ("HELLO", "BANK")] |
| assert float(probs[outside].sum()) < 1e-9 |
|
|
|
|
| def test_fine_tune_end_to_end_and_reset(trained_model): |
| """Optional path: needs a real trained base model and enough calibration samples.""" |
| name, _ = trained_model |
| base = pers.create_profile("quinn", base_model=name, vocabulary=["HELLO", "BANK"]) |
| assert base["fine_tuned"] is False |
| X, labels = _samples(config.PERSONALIZATION_MIN_SAMPLES_FOR_FINETUNE, ("HELLO", "BANK")) |
| pers.add_calibration_samples("quinn", X, labels) |
| result = pers.fine_tune("quinn", epochs=1) |
| assert result["samples_used"] == config.PERSONALIZATION_MIN_SAMPLES_FOR_FINETUNE |
| assert result["train_samples"] > 0 |
| assert 0.0 <= result["train_accuracy_on_calibration"] <= 1.0 |
| assert "NOT evidence" in result["caveat"] |
| assert (pers.profile_dir("quinn") / "model.keras").exists() |
| assert pers.load_profile("quinn")["fine_tuned"] is True |
|
|
| reset = pers.reset_fine_tune("quinn") |
| assert reset["fine_tuned"] is False |
| assert not (pers.profile_dir("quinn") / "model.keras").exists() |
|
|
|
|
| def test_fine_tune_refuses_without_samples(): |
| pers.create_profile("rita", vocabulary=["HELLO", "BANK"]) |
| with pytest.raises(pers.PersonalizationError): |
| pers.fine_tune("rita", epochs=1) |
|
|
