"""Le DDL du réservoir (spec Imports §3) : contraintes, index partiels, cascades. Ces tests parlent directement à la base. Ce qu'ils gardent honnête n'est pas du code Python mais des **garanties de la base** — celles sur lesquelles le pipeline T2 s'appuiera pour faire ses `ON CONFLICT` sans relire, et celles que la suppression d'une cible doit tenir sans qu'aucun code applicatif n'ait à y penser (décision n°4 : rétention org). """ import uuid from datetime import UTC, datetime, timedelta from typing import Any import pytest from sqlalchemy import func, null, select, text from sqlalchemy.exc import DBAPIError, IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.jobs.models import Job from app.keywords.models import ( Import, Keyword, KeywordList, KeywordListItem, KeywordMetric, KeywordPageMap, KeywordTag, OrgSettings, Tag, ) from app.sites.models import Site from tests.conftest import Fixture NOW = datetime(2026, 8, 31, 12, 0, tzinfo=UTC) async def _keyword(db: AsyncSession, sandbox: Fixture, word: str = "chaussures") -> Keyword: keyword = Keyword(org_id=sandbox.org_id, locale="fr-FR", keyword=word) db.add(keyword) await db.commit() return keyword async def _metric(db: AsyncSession, sandbox: Fixture, keyword_id: uuid.UUID, **kwargs: Any) -> None: db.add( KeywordMetric( org_id=sandbox.org_id, keyword_id=keyword_id, source=kwargs.pop("source", "semrush_magic"), fetched_at=kwargs.pop("fetched_at", NOW), raw=kwargs.pop("raw", {}), **kwargs, ) ) await db.commit() async def _list(db: AsyncSession, sandbox: Fixture, name: str = "Semrush") -> KeywordList: keyword_list = KeywordList(org_id=sandbox.org_id, site_target_id=sandbox.target_id, name=name) db.add(keyword_list) await db.commit() return keyword_list async def _import(db: AsyncSession, sandbox: Fixture, **kwargs: Any) -> Import: job = Job(org_id=sandbox.org_id, type="import_csv") db.add(job) await db.flush() row = Import( org_id=sandbox.org_id, site_target_id=sandbox.target_id, source="semrush_magic", kind=kwargs.pop("kind", "csv"), fetched_at=NOW, job_id=job.id, status=kwargs.pop("status", "queued"), **kwargs, ) db.add(row) await db.commit() return row async def _map( db: AsyncSession, sandbox: Fixture, keyword_id: uuid.UUID, url: str, source: str ) -> KeywordPageMap: row = KeywordPageMap( org_id=sandbox.org_id, site_target_id=sandbox.target_id, keyword_id=keyword_id, url_normalized=url, source=source, seen_at=NOW, ) db.add(row) await db.commit() return row # --- keywords ------------------------------------------------------------------------- async def test_uuidv7_and_defaults(db: AsyncSession, sandbox: Fixture) -> None: keyword = await _keyword(db, sandbox) assert keyword.id.version == 7 assert keyword.created_at.tzinfo is not None @pytest.mark.parametrize("locale", ["fr-FR", "en-US", "pt-BR"]) async def test_locale_accepted(db: AsyncSession, sandbox: Fixture, locale: str) -> None: db.add(Keyword(org_id=sandbox.org_id, locale=locale, keyword=f"kw {locale}")) await db.commit() @pytest.mark.parametrize("locale", ["fr", "FR", "fr-fr", "FR-FR", "fr_FR", "fra-FRA", ""]) async def test_locale_rejected(db: AsyncSession, sandbox: Fixture, locale: str) -> None: """Le marché est langue-pays (décision n°2) : `fr` seul ne dit pas `fr-FR` vs `fr-BE`.""" db.add(Keyword(org_id=sandbox.org_id, locale=locale, keyword="chaussures")) with pytest.raises(IntegrityError): await db.commit() await db.rollback() async def test_keyword_unique_per_org_and_locale(db: AsyncSession, sandbox: Fixture) -> None: await _keyword(db, sandbox) # Même marché, même org : rejeté (c'est ce qui rend l'upsert du pipeline sûr). db.add(Keyword(org_id=sandbox.org_id, locale="fr-FR", keyword="chaussures")) with pytest.raises(IntegrityError): await db.commit() await db.rollback() # Autre marché : accepté — `fr-FR` et `fr-BE` sont deux mots-clés distincts. db.add(Keyword(org_id=sandbox.org_id, locale="fr-BE", keyword="chaussures")) # Autre org : accepté — le réservoir n'est jamais partagé entre organisations. db.add(Keyword(org_id=f"{sandbox.org_id}_bis", locale="fr-FR", keyword="chaussures")) await db.commit() async def test_keyword_tsv_is_generated(db: AsyncSession, sandbox: Fixture) -> None: """La colonne générée n'est jamais écrite par l'application — la base la tient à jour.""" keyword = await _keyword(db, sandbox, "chaussures de sport") tsv = await db.scalar(select(Keyword.keyword_tsv).where(Keyword.id == keyword.id)) assert tsv == "'chaussures':1 'de':2 'sport':3" # Recherche plein texte : c'est le chemin de `GET /keywords?q=` (§6). found = await db.scalar( select(func.count()) .select_from(Keyword) .where(Keyword.org_id == sandbox.org_id) .where(Keyword.keyword_tsv.bool_op("@@")(func.plainto_tsquery("simple", "sport"))) ) assert found == 1 async def test_trigram_search_available(db: AsyncSession, sandbox: Fixture) -> None: """Repli flou de la recherche : l'extension pg_trgm est bien posée par la migration.""" await _keyword(db, sandbox, "chaussures de sport") similarity = await db.scalar(text("SELECT similarity('chaussures', 'chausures')")) assert similarity > 0.5 async def test_embedding_is_nullable_halfvec(db: AsyncSession, sandbox: Fixture) -> None: """Colonne prête pour les embeddings v2.x, vide et sans index en v1.""" keyword = await _keyword(db, sandbox) dimension = await db.scalar( text( "SELECT atttypmod FROM pg_attribute " "WHERE attrelid = 'keywords'::regclass AND attname = 'embedding'" ) ) assert dimension == 1024 assert (await db.scalar(select(Keyword.embedding).where(Keyword.id == keyword.id))) is None # --- keyword_metrics ------------------------------------------------------------------ async def test_metric_unique_is_one_observation(db: AsyncSession, sandbox: Fixture) -> None: """`(keyword_id, source, fetched_at)` = **une** observation, corrigeable (review C1).""" keyword = await _keyword(db, sandbox) # Capturé avant le commit qui échoue : `rollback()` expire l'instance ORM. keyword_id = keyword.id await _metric(db, sandbox, keyword_id, volume=100) db.add( KeywordMetric( org_id=sandbox.org_id, keyword_id=keyword_id, source="semrush_magic", fetched_at=NOW, volume=120, raw={}, ) ) with pytest.raises(IntegrityError): await db.commit() await db.rollback() # Une autre source ou une autre date empile, elle n'écrase jamais. await _metric(db, sandbox, keyword_id, source="generic_csv", volume=120) await _metric(db, sandbox, keyword_id, fetched_at=NOW - timedelta(days=7), volume=90) assert ( await db.scalar( select(func.count()) .select_from(KeywordMetric) .where(KeywordMetric.keyword_id == keyword_id) ) ) == 3 async def test_metric_values_are_all_nullable(db: AsyncSession, sandbox: Fixture) -> None: """Une source qui ne fournit rien reste une observation valide (canon data §7).""" keyword = await _keyword(db, sandbox) await _metric(db, sandbox, keyword.id, source="generic_csv") stored = await db.scalar(select(KeywordMetric).where(KeywordMetric.keyword_id == keyword.id)) assert (stored.volume, stored.kd, stored.cpc, stored.intent) == (None, None, None, None) assert (stored.serp_features, stored.trend) == (None, None) async def test_metric_trend_is_smallint_array(db: AsyncSession, sandbox: Fixture) -> None: keyword = await _keyword(db, sandbox) await _metric(db, sandbox, keyword.id, trend=list(range(12)), serp_features={"snippet": True}) stored = await db.scalar(select(KeywordMetric).where(KeywordMetric.keyword_id == keyword.id)) assert stored.trend == list(range(12)) assert stored.serp_features == {"snippet": True} async def test_metric_raw_is_required(db: AsyncSession, sandbox: Fixture) -> None: """La ligne source complète est toujours conservée : rien n'est irrécupérable.""" keyword = await _keyword(db, sandbox) db.add( # `null()` et non `None` : le type JSONB écrit `None` comme un JSON `null`, qui # satisferait le NOT NULL — seul un vrai NULL SQL éprouve la contrainte. KeywordMetric( org_id=sandbox.org_id, keyword_id=keyword.id, source="x", fetched_at=NOW, raw=null() ) ) with pytest.raises(IntegrityError): await db.commit() await db.rollback() # --- listes, items, tags --------------------------------------------------------------- async def test_list_name_unique_per_target(db: AsyncSession, sandbox: Fixture) -> None: await _list(db, sandbox) db.add(KeywordList(org_id=sandbox.org_id, site_target_id=sandbox.target_id, name="Semrush")) with pytest.raises(IntegrityError): await db.commit() await db.rollback() async def test_list_item_pair_is_unique(db: AsyncSession, sandbox: Fixture) -> None: """PK composite : l'ajout en masse `ON CONFLICT DO NOTHING` du T2 s'appuie dessus.""" keyword = await _keyword(db, sandbox) keyword_list = await _list(db, sandbox) db.add(KeywordListItem(org_id=sandbox.org_id, list_id=keyword_list.id, keyword_id=keyword.id)) await db.commit() db.add(KeywordListItem(org_id=sandbox.org_id, list_id=keyword_list.id, keyword_id=keyword.id)) with pytest.raises(IntegrityError): await db.commit() await db.rollback() async def test_tag_name_unique_per_org(db: AsyncSession, sandbox: Fixture) -> None: db.add(Tag(org_id=sandbox.org_id, name="marque")) await db.commit() db.add(Tag(org_id=sandbox.org_id, name="marque")) with pytest.raises(IntegrityError): await db.commit() await db.rollback() # --- keyword_page_map : la sortie du module -------------------------------------------- async def test_map_page_id_stays_null_without_crawl(db: AsyncSession, sandbox: Fixture) -> None: """Une URL déclarée à la main vit sans page : c'est la planification (flow §1.8).""" keyword = await _keyword(db, sandbox) row = await _map(db, sandbox, keyword.id, "https://exemple.fr/a", "manual") assert row.page_id is None # Aucune FK vers `pages` : la table n'existe pas encore (module Crawl). row.page_id = uuid.uuid4() await db.commit() @pytest.mark.parametrize("source", ["import_url", "manual", "gsc"]) async def test_map_sources_accepted(db: AsyncSession, sandbox: Fixture, source: str) -> None: keyword = await _keyword(db, sandbox) await _map(db, sandbox, keyword.id, "https://exemple.fr/a", source) async def test_map_unknown_source_rejected(db: AsyncSession, sandbox: Fixture) -> None: keyword = await _keyword(db, sandbox) with pytest.raises(IntegrityError): await _map(db, sandbox, keyword.id, "https://exemple.fr/a", "crawl") await db.rollback() async def test_map_observations_can_be_multiple(db: AsyncSession, sandbox: Fixture) -> None: """Cannibalisation : plusieurs URL observées pour un mot-clé, nativement (§3.7).""" keyword = await _keyword(db, sandbox) await _map(db, sandbox, keyword.id, "https://exemple.fr/a", "import_url") await _map(db, sandbox, keyword.id, "https://exemple.fr/b", "import_url") # `gsc` observera la même URL sans entrer en conflit avec l'import (zéro migration T3). await _map(db, sandbox, keyword.id, "https://exemple.fr/a", "gsc") assert ( await db.scalar( select(func.count()) .select_from(KeywordPageMap) .where(KeywordPageMap.keyword_id == keyword.id) ) ) == 3 # La même observation deux fois, en revanche, est refusée : l'upsert du T2 la met à jour. with pytest.raises(IntegrityError): await _map(db, sandbox, keyword.id, "https://exemple.fr/a", "import_url") await db.rollback() async def test_manual_map_is_unique_per_target_and_keyword( db: AsyncSession, sandbox: Fixture ) -> None: """Décision n°6 : la planification désigne **une** page cible, pas deux.""" keyword = await _keyword(db, sandbox) keyword_id = keyword.id await _map(db, sandbox, keyword_id, "https://exemple.fr/a", "manual") with pytest.raises(IntegrityError): await _map(db, sandbox, keyword_id, "https://exemple.fr/b", "manual") await db.rollback() # L'index est partiel : les observations, elles, restent multiples sur la même cible… await _map(db, sandbox, keyword_id, "https://exemple.fr/b", "import_url") await _map(db, sandbox, keyword_id, "https://exemple.fr/c", "import_url") # …et l'attache manuelle déjà posée n'a pas bougé (l'automatique ne l'écrase jamais). manual = await db.scalar( select(KeywordPageMap).where( KeywordPageMap.keyword_id == keyword_id, KeywordPageMap.source == "manual" ) ) assert manual.url_normalized == "https://exemple.fr/a" # --- imports --------------------------------------------------------------------------- @pytest.mark.parametrize("kind", ["csv", "api"]) async def test_import_kinds(db: AsyncSession, sandbox: Fixture, kind: str) -> None: await _import(db, sandbox, kind=kind) async def test_import_unknown_kind_rejected(db: AsyncSession, sandbox: Fixture) -> None: with pytest.raises(IntegrityError): await _import(db, sandbox, kind="xlsx") await db.rollback() @pytest.mark.parametrize("status", ["queued", "running", "done", "failed"]) async def test_import_statuses(db: AsyncSession, sandbox: Fixture, status: str) -> None: """Même vocabulaire que `jobs.status` : la recopie (§3.11) ne peut pas dériver.""" await _import(db, sandbox, status=status) async def test_import_unknown_status_rejected(db: AsyncSession, sandbox: Fixture) -> None: with pytest.raises(IntegrityError): await _import(db, sandbox, status="pending") await db.rollback() async def test_import_sha256_is_not_unique(db: AsyncSession, sandbox: Fixture) -> None: """Pas d'UNIQUE dur : le re-import après correction d'adaptateur reste possible (§3.11).""" await _import(db, sandbox, file_sha256="abc", filename="export.csv") await _import(db, sandbox, file_sha256="abc", filename="export.csv") async def test_import_requires_a_job(db: AsyncSession, sandbox: Fixture) -> None: """Un import sans job serait un rapport sans exécution — la FK le refuse.""" db.add( Import( org_id=sandbox.org_id, site_target_id=sandbox.target_id, source="semrush_magic", kind="csv", fetched_at=NOW, job_id=uuid.uuid4(), status="queued", ) ) with pytest.raises(IntegrityError): await db.commit() await db.rollback() async def test_list_and_import_reference_each_other(db: AsyncSession, sandbox: Fixture) -> None: """Le cycle `keyword_lists` ↔ `imports` (§3.4 / §3.11) est bien posé des deux côtés.""" keyword_list = await _list(db, sandbox) row = await _import(db, sandbox, keyword_list_id=keyword_list.id) keyword_list.created_from_import_id = row.id await db.commit() # SET NULL : supprimer l'import ne supprime jamais la liste qu'il a créée. await db.delete(row) await db.commit() await db.refresh(keyword_list) assert keyword_list.created_from_import_id is None # --- org_settings ---------------------------------------------------------------------- async def test_org_settings_is_one_row_per_org(db: AsyncSession, sandbox: Fixture) -> None: db.add(OrgSettings(org_id=sandbox.org_id, preferred_source="semrush_magic")) await db.commit() db.add(OrgSettings(org_id=sandbox.org_id)) with pytest.raises(IntegrityError): await db.commit() await db.rollback() # Aucune contrainte sur la valeur : le registre d'adaptateurs est du code, la validation # se fait au PATCH (422), pas en base — la liste des sources bouge à chaque tranche. settings = await db.get(OrgSettings, sandbox.org_id) settings.preferred_source = None await db.commit() # --- suppressions : rétention org (décision n°4) ---------------------------------------- async def _counts(db: AsyncSession, sandbox: Fixture) -> dict[str, int]: async def count(model: Any, column: Any, value: Any) -> int: return await db.scalar(select(func.count()).select_from(model).where(column == value)) return { "lists": await count(KeywordList, KeywordList.org_id, sandbox.org_id), "items": await count(KeywordListItem, KeywordListItem.org_id, sandbox.org_id), "maps": await count(KeywordPageMap, KeywordPageMap.org_id, sandbox.org_id), "imports": await count(Import, Import.org_id, sandbox.org_id), "keywords": await count(Keyword, Keyword.org_id, sandbox.org_id), "metrics": await count(KeywordMetric, KeywordMetric.org_id, sandbox.org_id), "tags": await count(Tag, Tag.org_id, sandbox.org_id), } async def _full_target(db: AsyncSession, sandbox: Fixture) -> None: keyword = await _keyword(db, sandbox) await _metric(db, sandbox, keyword.id, volume=100) keyword_list = await _list(db, sandbox) db.add(KeywordListItem(org_id=sandbox.org_id, list_id=keyword_list.id, keyword_id=keyword.id)) tag = Tag(org_id=sandbox.org_id, name="marque") db.add(tag) await db.flush() db.add(KeywordTag(org_id=sandbox.org_id, keyword_id=keyword.id, tag_id=tag.id)) await db.commit() await _map(db, sandbox, keyword.id, "https://exemple.fr/a", "manual") await _import(db, sandbox, keyword_list_id=keyword_list.id) async def test_deleting_target_keeps_the_org_reservoir(db: AsyncSession, sandbox: Fixture) -> None: """Confirmation pédagogique du flow §1.10, garantie par la base et non par du code.""" await _full_target(db, sandbox) assert await _counts(db, sandbox) == { "lists": 1, "items": 1, "maps": 1, "imports": 1, "keywords": 1, "metrics": 1, "tags": 1, } await db.execute(text("DELETE FROM site_targets WHERE id = :id"), {"id": sandbox.target_id}) await db.commit() assert await _counts(db, sandbox) == { "lists": 0, "items": 0, # via la CASCADE de la liste, pas de la cible. "maps": 0, "imports": 0, "keywords": 1, # le réservoir de l'org survit… "metrics": 1, # …avec son historique… "tags": 1, # …et ses tags. } async def test_deleting_site_cascades_through_its_targets( db: AsyncSession, sandbox: Fixture ) -> None: await _full_target(db, sandbox) site = await db.get(Site, sandbox.site_id) await db.delete(site) await db.commit() counts = await _counts(db, sandbox) assert (counts["lists"], counts["maps"], counts["imports"]) == (0, 0, 0) assert (counts["keywords"], counts["metrics"], counts["tags"]) == (1, 1, 1) async def test_deleting_a_keyword_takes_everything_that_points_to_it( db: AsyncSession, sandbox: Fixture ) -> None: """L'autre sens : purger un mot-clé du réservoir (job différé) ne laisse aucun orphelin.""" await _full_target(db, sandbox) keyword = await db.scalar(select(Keyword).where(Keyword.org_id == sandbox.org_id)) await db.delete(keyword) await db.commit() counts = await _counts(db, sandbox) assert (counts["keywords"], counts["metrics"], counts["items"], counts["maps"]) == ( 0, 0, 0, 0, ) # La liste, le tag et l'import restent : ils appartiennent à la cible, pas au mot-clé. assert (counts["lists"], counts["tags"], counts["imports"]) == (1, 1, 1) assert ( await db.scalar( select(func.count()).select_from(KeywordTag).where(KeywordTag.org_id == sandbox.org_id) ) ) == 0 async def test_org_id_is_never_optional(db: AsyncSession, sandbox: Fixture) -> None: """Architecture §8 : pas de ligne hors organisation, donc pas de fuite possible.""" db.add(Keyword(locale="fr-FR", keyword="chaussures")) with pytest.raises((IntegrityError, DBAPIError)): await db.commit() await db.rollback()