Spaces:
Running
Running
| """ | |
| Tests for the feed's self-description: which tier served it, what interests it | |
| spans, and how far the user is from the next tier. | |
| These encode a product rule as much as a code path. The pipeline is gated — | |
| 1 save reaches the similarity tier, 3 the EWMA tier, 5 the clustered | |
| multi-interest tier — and until now none of that reached the screen, so a user | |
| on three saves silently got the weaker path and read it as a bad feed. | |
| """ | |
| from app.recommend.clustering import MIN_PAPERS_FOR_CLUSTERING, MIN_SAVES_FOR_INTERESTS | |
| from app.routers.recommendations import ( | |
| _MIN_EWMA_INTERACTIONS, | |
| _feed_interests, | |
| _profile_strength, | |
| _serving_tier, | |
| ) | |
| def entry(tags: dict, ranked: list[str] | None = None) -> dict: | |
| return {"tags": tags, "ranked": ranked if ranked is not None else list(tags)} | |
| def tag(source: str, *, label: str = "", tone: int = -1) -> dict: | |
| return {"candidate_source": source, "cluster_label": label, "cluster_tone": tone} | |
| # ── Which tier served this feed ────────────────────────────────────────────── | |
| def test_tier_is_read_from_the_tags_the_pipeline_wrote(): | |
| assert _serving_tier(entry({"a": tag("trending_category_fallback")})) == 0 | |
| assert _serving_tier(entry({"a": tag("qdrant_recommend")})) == 3 | |
| assert _serving_tier(entry({"a": tag("ewma_longterm")})) == 2 | |
| assert _serving_tier(entry({"a": tag("cluster_2", label="Vision", tone=2)})) == 1 | |
| def test_short_term_supplements_still_read_as_tier_1(): | |
| # Only Tier 1 emits short_term_supplement at all. | |
| assert _serving_tier(entry({"a": tag("short_term_supplement")})) == 1 | |
| def test_empty_feed_defaults_to_tier_1(): | |
| assert _serving_tier(entry({})) == 1 | |
| # ── Progress toward the next tier ──────────────────────────────────────────── | |
| def test_tier_3_user_is_pointed_at_the_multi_interest_threshold(): | |
| """From two saves every save is its own interest; that is the next step.""" | |
| s = _profile_strength(entry({"a": tag("qdrant_recommend")}), save_count=1) | |
| assert s is not None | |
| assert s["target"] == MIN_SAVES_FOR_INTERESTS | |
| assert s["remaining"] == MIN_SAVES_FOR_INTERESTS - 1 | |
| assert "multi-interest" in s["unlocks"] | |
| def test_tier_2_is_a_fallback_and_never_nags(): | |
| """Tier 2 now only serves when the multi-interest path failed at 2+ saves.""" | |
| assert _profile_strength(entry({"a": tag("ewma_longterm")}), save_count=3) is None | |
| def test_tier_1_has_nothing_left_to_unlock(): | |
| e = entry({"a": tag("cluster_0", label="Vision", tone=0)}) | |
| assert _profile_strength(e, save_count=12) is None | |
| def test_no_nag_when_the_threshold_is_met_but_the_tier_did_not_engage(): | |
| """ | |
| A user can sit on 6 saves and still be served by Tier 2 — no vectors, an | |
| empty retrieval, a cluster that failed to build. Telling them to "save 2 | |
| more" when they are already past the threshold is a promise the pipeline | |
| has just demonstrated it will not keep. | |
| """ | |
| e = entry({"a": tag("ewma_longterm")}) | |
| assert _profile_strength(e, save_count=MIN_PAPERS_FOR_CLUSTERING + 1) is None | |
| def test_percentage_is_bounded(): | |
| s = _profile_strength(entry({"a": tag("qdrant_recommend")}), save_count=1) | |
| assert 0 <= s["pct"] <= 100 | |
| # ── Interest composition ───────────────────────────────────────────────────── | |
| def test_interests_are_counted_over_the_whole_ranked_pool(): | |
| e = entry( | |
| { | |
| "p1": tag("cluster_0", label="Computer vision", tone=0), | |
| "p2": tag("cluster_0", label="Computer vision", tone=0), | |
| "p3": tag("cluster_1", label="Combinatorics", tone=1), | |
| }, | |
| ranked=["p1", "p2", "p3"], | |
| ) | |
| out = _feed_interests(e) | |
| assert [(i["label"], i["count"]) for i in out] == [ | |
| ("Computer vision", 2), ("Combinatorics", 1), | |
| ] | |
| def test_interests_are_ordered_largest_first(): | |
| """A cluster's share of the pool IS its importance weight after | |
| allocate_quotas(), so size order is quota order.""" | |
| e = entry( | |
| { | |
| "p1": tag("cluster_1", label="Small", tone=1), | |
| "p2": tag("cluster_0", label="Big", tone=0), | |
| "p3": tag("cluster_0", label="Big", tone=0), | |
| }, | |
| ranked=["p1", "p2", "p3"], | |
| ) | |
| assert [i["label"] for i in _feed_interests(e)] == ["Big", "Small"] | |
| def test_unlabelled_papers_are_excluded(): | |
| # Exploration picks and short-term supplements carry no cluster label; the | |
| # strip describes clusters, so they must not appear as a blank chip. | |
| e = entry( | |
| { | |
| "p1": tag("cluster_0", label="Vision", tone=0), | |
| "p2": tag("exploration"), | |
| "p3": tag("short_term_supplement"), | |
| }, | |
| ranked=["p1", "p2", "p3"], | |
| ) | |
| out = _feed_interests(e) | |
| assert len(out) == 1 and out[0]["label"] == "Vision" | |
| def test_non_tier_1_feeds_describe_no_interests(): | |
| e = entry({"p1": tag("ewma_longterm")}, ranked=["p1"]) | |
| assert _feed_interests(e) == [] | |
| def test_papers_outside_the_ranked_pool_are_not_counted(): | |
| # `tags` covers the exploration pool too; only `ranked` is the feed. | |
| e = entry( | |
| { | |
| "p1": tag("cluster_0", label="Vision", tone=0), | |
| "p2": tag("cluster_0", label="Vision", tone=0), | |
| }, | |
| ranked=["p1"], | |
| ) | |
| assert _feed_interests(e)[0]["count"] == 1 | |
| def test_tone_travels_with_the_label(): | |
| e = entry({"p1": tag("cluster_3", label="Robotics", tone=3)}, ranked=["p1"]) | |
| assert _feed_interests(e)[0]["tone"] == 3 | |
| # ── Match meter ────────────────────────────────────────────────────────────── | |
| from app.templates_env import _match_pct, _MATCH_CEIL, _MATCH_FLOOR | |
| def test_match_pct_spans_the_band_real_scores_occupy(): | |
| """A bar drawn on the raw 0-1 range would sit between a third and nine | |
| tenths full for every paper ever retrieved -- visually identical, so | |
| useless for comparing two cards.""" | |
| assert _match_pct(_MATCH_FLOOR) == 0 | |
| assert _match_pct(_MATCH_CEIL) == 100 | |
| mid = _match_pct((_MATCH_FLOOR + _MATCH_CEIL) / 2) | |
| assert 45 <= mid <= 55 | |
| def test_match_pct_clamps_outside_the_band(): | |
| assert _match_pct(0.99) == 100 | |
| assert _match_pct(0.10) == 0 | |
| def test_match_pct_is_zero_for_absent_or_bad_values(): | |
| # 0.0 is what a paper with no score carries; it must render no meter. | |
| assert _match_pct(0.0) == 0 | |
| assert _match_pct(None) == 0 | |
| assert _match_pct("") == 0 | |
| assert _match_pct(-0.5) == 0 | |
| # ── The masthead must not depend on the feed fragment ──────────────────────── | |
| # | |
| # These two need a real app + isolated DB, unlike everything above, which is | |
| # pure functions over dicts. | |
| import asyncio | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| def client(tmp_path, monkeypatch): | |
| import app.config as cfg | |
| import app.db as db_mod | |
| db_path = str(tmp_path / "test.db") | |
| monkeypatch.setattr(cfg, "DB_PATH", db_path) | |
| monkeypatch.setattr(db_mod, "DB_PATH", db_path) | |
| import app.user_state as us | |
| us._cache.clear() | |
| from app.qdrant_svc import _client | |
| _client.cache_clear() | |
| from app.main import app | |
| asyncio.run(db_mod.init_db()) | |
| with TestClient(app, raise_server_exceptions=True) as c: | |
| yield c | |
| def test_homepage_ships_an_h1_before_the_feed_loads(client): | |
| """ | |
| The masthead briefly lived inside partials/recommendations.html, which htmx | |
| fetches AFTER first paint. That left the delivered homepage with no <h1> at | |
| all -- nothing for a screen reader to land on, nothing for a crawler, and no | |
| visible title until the whole tier cascade had run (measured at 11s on a | |
| cold production feed). A page's heading must not depend on a slow request. | |
| """ | |
| # A brand-new user is redirected to onboarding, so skip it first — | |
| # otherwise this asserts against the wizard, not the feed page. | |
| client.get("/") | |
| client.post("/api/onboarding/skip") | |
| body = client.get("/").text | |
| assert 'class="issue-title"' in body, "feed page shipped without its masthead" | |
| assert "<h1" in body | |
| # The heading must be in the delivered document, not arriving later in the | |
| # htmx fragment. | |
| assert body.index("<h1") < body.index('id="rec-section"') | |
| def test_feed_fragment_does_not_repeat_the_masthead(client): | |
| """Two mastheads would render if the fragment also kept one.""" | |
| client.get("/") | |
| client.post("/api/onboarding/skip") | |
| frag = client.get("/api/recommendations").text | |
| assert "issue-title" not in frag | |