"""Consolidation `keyword_metrics_latest` (spec 2026-09-01 §3.1, décisions A1 à A3). La règle a **une seule implémentation**, en SQL : ces tests l'éprouvent là où elle vit. Ce qu'ils protègent, c'est la promesse faite à l'écran réservoir depuis la suppression de la hiérarchie de sources — « le mot-clé porte les observations de **toutes** ses sources, côte à côte ; personne ne prend le dessus sur personne » — et la forme exacte du tableau rendu : une entrée par source qui a une valeur, la plus récente de cette source, la préférée en tête, `[]` quand personne n'a rien dit, jamais `null`. """ import json from datetime import UTC, datetime, timedelta from typing import Any import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.keywords.models import Keyword, KeywordMetric from tests.conftest import Fixture OLD = datetime(2026, 1, 1, tzinfo=UTC) RECENT = datetime(2026, 8, 1, tzinfo=UTC) FRESH = datetime(2026, 8, 30, tzinfo=UTC) # Les six champs consolidés, dans l'ordre du `RETURNS TABLE`. METRICS = ("volume", "kd", "cpc", "intent", "serp_features", "trend") _COLUMNS = "keyword_id, org_id, " + ", ".join(METRICS) + ", last_updated" async def _keyword_with(db: AsyncSession, sandbox: Fixture, *observations: dict[str, Any]) -> Any: keyword = Keyword(org_id=sandbox.org_id, locale="fr-FR", keyword="chaussures") db.add(keyword) await db.flush() for observation in observations: db.add( KeywordMetric( org_id=sandbox.org_id, keyword_id=keyword.id, raw={}, **observation, ) ) await db.commit() return keyword.id async def _latest(db: AsyncSession, keyword_id: Any, preferred: str | None = None) -> Any: """La fonction, appelée comme le fera le service API (§3.1). Les six métriques sont désormais des **tableaux jsonb** : chaque colonne rend une liste de dicts `{value, source, fetched_at}`, `fetched_at` étant une chaîne ISO en UTC produite par `to_char` (jamais un timestamptz nu, dont la sérialisation jsonb dépendrait du fuseau de session Postgres). """ result = await db.execute( text( f"SELECT {_COLUMNS} FROM keyword_metrics_latest(:preferred) " "WHERE keyword_id = :keyword_id" ), {"preferred": preferred, "keyword_id": keyword_id}, ) return result.mappings().one_or_none() def _pairs(entries: list[dict[str, Any]]) -> list[tuple[Any, str]]: """(valeur, source) de chaque entrée **dans l'ordre rendu** — cet ordre est contractuel.""" return [(entry["value"], entry["source"]) for entry in entries] # --- toutes les sources, côte à côte ------------------------------------------------------ async def test_two_sources_at_the_same_date_are_both_rendered( db: AsyncSession, sandbox: Fixture ) -> None: """Le cas mesuré du §1 : deux imports du même jour, deux volumes, **deux** entrées. C'est le test qui définit le chantier. Sous l'ancienne règle, `fetched_at` ne départageait rien et la priorité codée dans les `CASE` élisait Semrush : Mangools disparaissait de l'écran sans que rien ne le dise. Il n'y a plus d'élection. """ keyword_id = await _keyword_with( db, sandbox, {"source": "mangools", "fetched_at": FRESH, "volume": 2200, "kd": 14, "cpc": 3.05}, {"source": "semrush_magic", "fetched_at": FRESH, "volume": 720, "kd": 46, "cpc": 2.26}, ) row = await _latest(db, keyword_id) # À fraîcheur égale et sans source préférée, l'ordre se joue sur le nom de la source. assert _pairs(row["volume"]) == [(2200, "mangools"), (720, "semrush_magic")] assert _pairs(row["kd"]) == [(14, "mangools"), (46, "semrush_magic")] assert _pairs(row["cpc"]) == [(3.05, "mangools"), (2.26, "semrush_magic")] async def test_a_holed_source_never_masks_another(db: AsyncSession, sandbox: Fixture) -> None: """**Par champ** : la source fraîche mais pauvre n'apparaît que là où elle a une valeur.""" keyword_id = await _keyword_with( db, sandbox, # Riche mais ancienne. {"source": "semrush_magic", "fetched_at": OLD, "volume": 100, "kd": 42, "cpc": 1.5}, # Fraîche mais elle n'apporte que le volume. {"source": "generic_csv", "fetched_at": FRESH, "volume": 300}, ) row = await _latest(db, keyword_id) # Les deux sur le volume, la plus fraîche en tête. assert _pairs(row["volume"]) == [(300, "generic_csv"), (100, "semrush_magic")] # Une seule sur les champs que `generic_csv` ne porte pas — elle n'y figure pas du tout. assert _pairs(row["kd"]) == [(42, "semrush_magic")] assert _pairs(row["cpc"]) == [(1.5, "semrush_magic")] async def test_only_the_freshest_observation_of_a_source_is_kept( db: AsyncSession, sandbox: Fixture ) -> None: """La **dernière** observation de chaque source (décision A3), pas tout son historique. L'historique complet reste dans `GET /keywords/{id}/metrics` : le réservoir n'affiche jamais deux lignes de la même source. """ keyword_id = await _keyword_with( db, sandbox, {"source": "mangools", "fetched_at": OLD, "volume": 100}, {"source": "mangools", "fetched_at": RECENT, "volume": 200}, {"source": "mangools", "fetched_at": FRESH, "volume": 300}, ) row = await _latest(db, keyword_id) assert _pairs(row["volume"]) == [(300, "mangools")] async def test_a_field_without_any_observation_is_an_empty_array( db: AsyncSession, sandbox: Fixture ) -> None: """`[]`, **jamais `null`** : le web n'a qu'un cas à traiter (§3.1). Sans le `coalesce` de la fonction, `jsonb_agg` rendrait `null` sur l'ensemble vide — un agrégat sans `GROUP BY` rend toujours une ligne. """ keyword_id = await _keyword_with( db, sandbox, {"source": "mangools", "fetched_at": FRESH, "volume": 300}, ) row = await _latest(db, keyword_id) assert _pairs(row["volume"]) == [(300, "mangools")] for field in ("kd", "cpc", "intent", "serp_features", "trend"): assert row[field] == [], field async def test_keyword_without_metrics_is_absent(db: AsyncSession, sandbox: Fixture) -> None: """Cas **différent** du précédent : aucune observation du tout → aucune ligne. La consolidation part des observations (`GROUP BY m.keyword_id`) et ne fabrique jamais de ligne vide ; c'est au listing du réservoir de faire le `LEFT JOIN` (§6). Corollaire pour l'api : sur un mot-clé attaché à la main, le latéral rend `NULL` sur les six colonnes et le `coalesce` SQL n'y peut rien — c'est au service de retomber sur `[]`. """ keyword_id = await _keyword_with(db, sandbox) assert await _latest(db, keyword_id) is None async def test_last_updated_is_the_max_over_all_sources(db: AsyncSession, sandbox: Fixture) -> None: """Filtre « last updated » du réservoir (flow §1.7) : la fraîcheur du mot-clé, pas d'un champ. Une observation entièrement vide compte quand même : la source a bien été interrogée — mais elle n'apparaît dans le tableau d'aucun champ. """ keyword_id = await _keyword_with( db, sandbox, {"source": "semrush_magic", "fetched_at": OLD, "volume": 100}, {"source": "generic_csv", "fetched_at": FRESH}, ) row = await _latest(db, keyword_id) assert row["last_updated"] == FRESH assert _pairs(row["volume"]) == [(100, "semrush_magic")] async def test_all_fields_carry_their_observations(db: AsyncSession, sandbox: Fixture) -> None: """Les six champs rendent la même forme, valeurs qualitatives comprises.""" keyword_id = await _keyword_with( db, sandbox, { "source": "semrush_magic", "fetched_at": OLD, "intent": "commercial", "serp_features": {"snippet": True}, "trend": [1] * 12, }, {"source": "se_ranking", "fetched_at": FRESH, "volume": 300, "kd": 42, "cpc": 1.5}, ) row = await _latest(db, keyword_id) assert _pairs(row["intent"]) == [("commercial", "semrush_magic")] assert _pairs(row["serp_features"]) == [({"snippet": True}, "semrush_magic")] assert _pairs(row["trend"]) == [([1] * 12, "semrush_magic")] assert _pairs(row["volume"]) == [(300, "se_ranking")] for field in METRICS: for entry in row[field]: assert set(entry) == {"value", "source", "fetched_at"}, field assert entry["source"] and entry["fetched_at"], field # --- source préférée : elle ordonne, elle ne masque rien (§3.4) --------------------------- async def test_preferred_source_leads_without_masking(db: AsyncSession, sandbox: Fixture) -> None: """La préférée passe en **tête**, même plus ancienne — et l'autre reste rendue. C'est tout ce qui reste de la décision n°3.d : un ordre de lecture, plus une élection. """ keyword_id = await _keyword_with( db, sandbox, {"source": "semrush_magic", "fetched_at": OLD, "volume": 100}, {"source": "generic_csv", "fetched_at": FRESH, "volume": 300}, ) row = await _latest(db, keyword_id, preferred="semrush_magic") assert _pairs(row["volume"]) == [(100, "semrush_magic"), (300, "generic_csv")] # `last_updated` reste la fraîcheur réelle du mot-clé, indépendante de la préférence. assert row["last_updated"] == FRESH async def test_preferred_source_orders_field_by_field(db: AsyncSession, sandbox: Fixture) -> None: """La préférée n'est en tête que des champs où elle a une valeur. Ailleurs, l'ordre du repli s'applique — et elle n'y laisse aucun trou. """ keyword_id = await _keyword_with( db, sandbox, {"source": "semrush_magic", "fetched_at": OLD, "volume": 100}, {"source": "generic_csv", "fetched_at": RECENT, "volume": 300, "kd": 42}, {"source": "dataforseo", "fetched_at": FRESH, "kd": 10}, ) row = await _latest(db, keyword_id, preferred="semrush_magic") assert _pairs(row["volume"]) == [(100, "semrush_magic"), (300, "generic_csv")] # La préférée n'a pas de kd : ordre par date, sans elle et sans trou. assert _pairs(row["kd"]) == [(10, "dataforseo"), (42, "generic_csv")] async def test_preferred_source_absent_from_the_keyword(db: AsyncSession, sandbox: Fixture) -> None: """Source préférée réglée mais jamais importée ici : ordre par date, aucune erreur.""" keyword_id = await _keyword_with( db, sandbox, {"source": "semrush_magic", "fetched_at": OLD, "volume": 100}, {"source": "generic_csv", "fetched_at": FRESH, "volume": 300}, ) row = await _latest(db, keyword_id, preferred="dataforseo") assert _pairs(row["volume"]) == [(300, "generic_csv"), (100, "semrush_magic")] async def test_order_is_deterministic_for_unknown_sources( db: AsyncSession, sandbox: Fixture ) -> None: """Deux sources inconnues à fraîcheur égale : départagées par leur nom, jamais au hasard. Plus aucune priorité ne les classe — le nom est le dernier critère, et il suffit. """ keyword_id = await _keyword_with( db, sandbox, {"source": "zzz_source", "fetched_at": FRESH, "volume": 1}, {"source": "aaa_source", "fetched_at": FRESH, "volume": 2}, ) for _ in range(3): row = await _latest(db, keyword_id) assert _pairs(row["volume"]) == [(2, "aaa_source"), (1, "zzz_source")] # --- fraîcheur rendue en UTC, quel que soit le fuseau du serveur -------------------------- async def test_fetched_at_is_utc_whatever_the_session_timezone( db: AsyncSession, sandbox: Fixture ) -> None: """Le verrou du `to_char` : `jsonb_build_object` seul rendrait le fuseau de **session**. Mesuré : `jsonb_build_object('f', )` rend `…T17:00:00+07:00` sous `TimeZone = Asia/Bangkok`. Rien ne fixe `TimeZone` côté api — l'offset rendu au web dépendrait donc de la configuration du serveur Postgres, ce qui a déjà mordu le dépôt. `SET LOCAL` : le réglage meurt avec la transaction, il ne fuit pas vers le test suivant. """ keyword_id = await _keyword_with( db, sandbox, {"source": "mangools", "fetched_at": FRESH, "volume": 300}, ) await db.execute(text("SET LOCAL TIME ZONE 'America/Denver'")) # Le garde-fou du garde-fou : sans ce témoin, le test passerait aussi bien si le `SET` # n'avait aucun effet — et ne prouverait plus rien. naive = await db.scalar( text("SELECT jsonb_build_object('f', CAST(:at AS timestamptz)) ->> 'f'"), {"at": FRESH} ) assert naive.endswith("-06:00"), naive row = await _latest(db, keyword_id) rendered = row["volume"][0]["fetched_at"] assert rendered.endswith("+00:00"), rendered assert datetime.fromisoformat(rendered) == FRESH # --- la vue n'est que la fonction sans préférence ------------------------------------------ @pytest.mark.parametrize("delta", [timedelta(0), timedelta(days=1)]) async def test_view_is_the_function_without_preference( db: AsyncSession, sandbox: Fixture, delta: timedelta ) -> None: """La vue par défaut (Matrice, requêtes ad hoc) n'est qu'un appel avec NULL — §3.1. C'est le **seul** garde-fou réel de la révision Alembic : `alembic check` compare `Base.metadata` au schéma et n'y voit ni fonction ni vue. Le `SELECT *` est délibéré des deux côtés — comparer les formes complètes est exactement ce qu'on teste ici. """ keyword_id = await _keyword_with( db, sandbox, {"source": "semrush_magic", "fetched_at": OLD, "volume": 100}, {"source": "generic_csv", "fetched_at": FRESH + delta, "volume": 300}, ) from_view = ( ( await db.execute( text("SELECT * FROM keyword_metrics_latest WHERE keyword_id = :id"), {"id": keyword_id}, ) ) .mappings() .one() ) from_function = ( ( await db.execute( text("SELECT * FROM keyword_metrics_latest(NULL::text) WHERE keyword_id = :id"), {"id": keyword_id}, ) ) .mappings() .one() ) assert dict(from_view) == dict(from_function) assert list(from_view) == ["keyword_id", "org_id", *METRICS, "last_updated"] # --- la forme reste inlinable --------------------------------------------------------------- async def test_the_function_stays_inlinable(db: AsyncSession, sandbox: Fixture) -> None: """Un `Function Scan` dans le plan signerait la fin de la tenue à l'échelle. Tant que la fonction est inlinée, le planificateur pousse le filtre sur `keyword_id` jusque dans chaque latéral, qui se sert alors du btree `uq_keyword_metrics_keyword_source_fetched` — c'est ce qui rend le réservoir tenable sur une page de 50 mots-clés dans 500 000. Un CTE, un `DISTINCT` au niveau supérieur, un agrégat dans la SELECT list de tête ou un `VOLATILE` la détruiraient **en silence**. """ keyword_id = await _keyword_with( db, sandbox, {"source": "mangools", "fetched_at": FRESH, "volume": 300}, ) plan = "\n".join( row for (row,) in ( await db.execute( text( "EXPLAIN (COSTS OFF) " "SELECT k.id, l.volume, l.kd, l.cpc, l.intent, l.serp_features, l.trend " "FROM keywords k LEFT JOIN LATERAL (" " SELECT c.volume, c.kd, c.cpc, c.intent, c.serp_features, c.trend" " FROM keyword_metrics_latest(CAST(:preferred AS text)) c" " WHERE c.keyword_id = k.id LIMIT 1" ") l ON true WHERE k.id = :id" ), {"preferred": None, "id": keyword_id}, ) ).all() ) assert "Function Scan" not in plan, plan # --- la forme rendue est du JSON, pas une chaîne ------------------------------------------- async def test_columns_are_decoded_json_not_text(db: AsyncSession, sandbox: Fixture) -> None: """asyncpg décode le jsonb côté Python : le service reçoit des listes, pas du texte. Le codec est installé par le dialecte asyncpg de SQLAlchemy ; ce test le vérifie pour un `text()` brut, qui est exactement la façon dont `app/keywords/service.py` interroge la fonction. """ keyword_id = await _keyword_with( db, sandbox, {"source": "mangools", "fetched_at": FRESH, "volume": 300}, ) row = await _latest(db, keyword_id) assert isinstance(row["volume"], list) assert isinstance(row["volume"][0], dict) # Et le contenu reste sérialisable tel quel vers le web. assert json.loads(json.dumps(row["volume"])) == row["volume"]