"""L'adaptateur API Mangools et le job `import_api` (règle « fixture réelle d'abord », T8).

Le format de réponse n'est documenté nulle part : il a été **relevé** avec une vraie clé le
2026-09-02 (`capture_mangools`). Les captures brutes ne sont pas commitées — `api/.gitignore`
exclut `tests/fixtures/real/mangools_api/`, parce qu'elles portent l'état du compte de
l'utilisateur et tous les mots-clés du domaine interrogé. Ce qui est versionné, et ce sur quoi
tous les tests de parse s'appuient, ce sont les extraits de `tests/fixtures/mangools_api/` :
**deux lignes par forme**, volume réduit, forme strictement intacte.

Ce qui a été anonymisé, et **seulement** ça : `kw` et `_id`. Tout ce que le parse regarde
vraiment — `sv`, `seo`, `cpc`, `msv`, les `null`, l'enveloppe, `target_type` — est verbatim.
Neutraliser un nombre aurait vidé la fixture de son intérêt ; neutraliser un libellé de
mot-clé ne coûte rien puisque le parse n'en fait qu'une chaîne normalisée.

L'ancien `test_the_adapter_is_still_locked` a disparu avec le verrou : sa raison d'être était
d'échouer le jour où le parse serait écrit. Il est remplacé par les tests de parse ci-dessous
— et par `test_the_lock_is_open_and_needs_no_capture_on_disk`, qui garde le piège inverse.
"""

import json
import uuid
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any

import httpx
import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import crypto
from app.core.errors import AppError
from app.integrations import pipeline as api_pipeline
from app.integrations import service
from app.integrations.models import Integration
from app.jobs import worker
from app.jobs.models import Job
from app.keywords import pipeline
from app.keywords import service as keywords_service
from app.keywords.models import Import, Keyword, KeywordList, KeywordListItem
from app.keywords.sources import mangools_api
from app.keywords.sources.base import CanonicalKeywordRow, RowError
from tests.conftest import Fixture

KEY = "mangools-cle-de-test-77aa"

FIXTURES = Path(__file__).resolve().parent / "fixtures" / "mangools_api"


def fixture(name: str) -> Any:
    return json.loads((FIXTURES / f"{name}.json").read_text())


def rows_of(name: str) -> list[Any]:
    return fixture(name)["keywords"]


async def _integration(
    db: AsyncSession, org_id: str, config: dict[str, Any] | None = None
) -> Integration:
    record = Integration(
        org_id=org_id,
        provider="mangools",
        credentials=crypto.encrypt(KEY),
        credentials_hint=crypto.hint(KEY),
        config=config or {},
    )
    db.add(record)
    await db.commit()
    return record


async def _run(db: AsyncSession, job: Job) -> Job | None:
    """Exécute **ce** job, sans passer par `worker.claim`.

    `claim` prend le plus vieux job `queued` de *toute* la base : l'appeler ici volerait le
    job d'une autre suite qui tourne en parallèle (ou d'un autre test du même fichier). On
    reproduit donc exactement ce que le claim écrit — `running`, `started_at`, `attempts` — sur
    la ligne qu'on veut, ce qui satisfait le jeton de propriété de `finish`.
    """
    job.status = "running"
    job.started_at = datetime.now(UTC)
    job.attempts = 1
    await db.commit()
    return await worker.execute(db, job)


def _api(handler: Any, **kwargs: Any) -> mangools_api.MangoolsApi:
    """Un client branché sur un bouchon, **sans espacement** : les tests n'attendent pas.

    `min_request_interval=0.0` est passé explicitement plutôt que déduit d'un flag « test » :
    c'est un paramètre du client, il se lit dans la signature, et le débit réel garde donc sa
    valeur de production partout ailleurs.
    """
    return mangools_api.MangoolsApi(
        KEY,
        client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
        min_request_interval=0.0,
        **kwargs,
    )


def _router(**bodies: Any) -> Any:
    """Un bouchon qui répond par chemin, et compte les appels reçus.

    Compter n'est pas cosmétique : « une unité de quota par requête » est *la* propriété qui
    gouverne l'adaptateur, et le seul moyen de la tester est de compter les requêtes.
    """
    calls: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        body = bodies.get(request.url.path.rsplit("/", 1)[-1].replace("-", "_"))
        if body is None:
            return httpx.Response(404, json={"error": "chemin non bouchonné"})
        if callable(body):
            return body(request)
        return httpx.Response(200, json=body)

    handler.calls = calls  # type: ignore[attr-defined]
    return handler


# --- le verrou, et ce qui l'a remplacé ----------------------------------------------------


def test_the_lock_is_open_and_needs_no_capture_on_disk(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Le parse est écrit — et `parsing_ready()` ne dépend plus des captures brutes.

    Le piège que ce test garde : `parsing_ready()` a longtemps exigé la présence des JSON de
    `tests/fixtures/real/`, que `api/.gitignore` exclut. Le laisser tel quel après l'écriture
    du parse aurait interdit tout import API partout sauf sur le poste qui a pris les captures
    — en production, silencieusement, avec un message parlant de fichiers introuvables.

    `CAPTURE_DIR` est donc pointé vers un répertoire **vide**, seule façon de reproduire ici
    ce qu'est n'importe quelle autre machine. Sur le poste qui a lancé la capture, les fichiers
    existent et le test passerait sans rien prouver.
    """
    monkeypatch.setattr(mangools_api, "CAPTURE_DIR", tmp_path)
    assert mangools_api.missing_captures() == list(mangools_api.REQUIRED_CAPTURES)
    assert mangools_api.PARSER_WRITTEN is True
    assert mangools_api.parsing_ready() is True


def test_the_source_name_is_shared_with_the_csv_adapter() -> None:
    """Décision 2026-09-01 : un seul nom de provenance pour le CSV et l'API Mangools."""
    from app.keywords import sources

    assert mangools_api.SOURCE_NAME == "mangools"
    csv_adapter = sources.get("mangools")
    assert csv_adapter is not None
    assert csv_adapter.name == mangools_api.SOURCE_NAME


# --- les trois modes de récupération ------------------------------------------------------


def test_a_spec_carries_exactly_one_mode() -> None:
    """Ni deux, ni aucun : l'ambiguïté est refusée à la construction."""
    assert mangools_api.FetchSpec(locale="fr-FR", domain="example.com").is_discovery is True
    assert mangools_api.FetchSpec(locale="fr-FR", keywords=("chaise",)).is_discovery is False
    assert mangools_api.FetchSpec(locale="fr-FR", list_id="abc").is_discovery is False
    with pytest.raises(ValueError, match="exactement un mode"):
        mangools_api.FetchSpec(locale="fr-FR")
    with pytest.raises(ValueError, match="exactement un mode"):
        mangools_api.FetchSpec(locale="fr-FR", domain="example.com", keywords=("chaise",))
    with pytest.raises(ValueError, match="exactement un mode"):
        mangools_api.FetchSpec(locale="fr-FR", list_id="abc", domain="example.com")


# --- le parse, sur les extraits versionnés ------------------------------------------------


def test_a_discovery_row_becomes_a_canonical_row() -> None:
    rows = mangools_api.parse_rows(rows_of("competitor_keywords"))
    rich, poor = rows
    assert isinstance(rich, CanonicalKeywordRow)
    assert isinstance(poor, CanonicalKeywordRow)

    assert rich.keyword == "mot cle riche"
    assert rich.volume == 90
    assert rich.kd == Decimal(44)
    #  `2.940000057220459` dans la réponse : du float32 élargi, pas de la précision.
    assert rich.cpc == Decimal("2.94")
    #  L'API ne rend ni intention ni features SERP.
    assert rich.intent is None
    assert rich.serp_features is None
    #  Ni URL ni position portées : le réservoir ne range une position qu'attachée à une URL.
    assert rich.url is None
    assert rich.position is None
    #  Rien n'est perdu : la ligne source entière est dans `raw`, position comprise.
    assert rich.raw["position"] == 100
    assert rich.raw["et"] == pytest.approx(0.357, rel=1e-3)
    assert rich.raw["svn"] == 90


def test_an_absent_metric_stays_none_and_never_becomes_zero() -> None:
    """`seo` est **souvent nul** sur `competitor-keywords` : c'est une absence, pas un zéro.

    Un KD à 0 signifierait « ce mot-clé est trivial à positionner » — l'exact contraire de
    « cette source ne connaît pas son KD ». Le même piège vaut pour le CPC.
    """
    _, poor = mangools_api.parse_rows(rows_of("competitor_keywords"))
    assert isinstance(poor, CanonicalKeywordRow)
    assert poor.kd is None
    assert poor.cpc is None
    assert poor.volume == 40


def test_an_enrichment_row_carries_the_difficulty() -> None:
    """C'est toute la raison d'être du second mode : `keyword-imports` renseigne le KD."""
    rows = mangools_api.parse_rows(rows_of("keyword_imports"))
    full, without_cpc = rows
    assert isinstance(full, CanonicalKeywordRow)
    assert isinstance(without_cpc, CanonicalKeywordRow)
    assert full.kd == Decimal(17)
    assert full.volume == 2200
    assert full.cpc == Decimal("2.97")
    #  Même endpoint, même ligne, un champ en moins : l'absence reste une absence.
    assert without_cpc.kd == Decimal(17)
    assert without_cpc.cpc is None


def test_the_trend_keeps_the_last_twelve_known_months() -> None:
    """`msv` va de 12 à 131 points ; la tendance canonique en fait douze, en 0–100."""
    rich, _ = mangools_api.parse_rows(rows_of("competitor_keywords"))
    assert isinstance(rich, CanonicalKeywordRow)
    assert rich.trend is not None
    assert len(rich.trend) == 12
    assert max(rich.trend) == 100
    assert all(0 <= value <= 100 for value in rich.trend)
    #  Les douze **derniers** : l'extrait porte quatorze mois, les deux premiers sont hors
    #  fenêtre. Le pic de la fenêtre est 390 (2025-09) et le dernier mois vaut 70 → 18.
    assert rich.trend[-1] == 18
    assert rich.trend.index(100) == 1


def test_the_trend_is_read_in_chronological_order_not_response_order() -> None:
    """La série n'est pas garantie ordonnée : trier est ce qui empêche une sparkline fausse."""
    ordered = [[2025, month, 10 * (month + 1)] for month in range(12)]
    shuffled = [ordered[i] for i in (5, 0, 11, 3, 8, 1, 9, 2, 10, 4, 7, 6)]
    row = mangools_api.parse_row(1, {"kw": "k", "msv": shuffled})
    assert isinstance(row, CanonicalKeywordRow)
    assert row.trend == [8, 17, 25, 33, 42, 50, 58, 67, 75, 83, 92, 100]


def test_a_month_without_volume_is_skipped_never_filled_with_zero() -> None:
    """Un zéro inventé se lirait comme un effondrement du volume — donc on saute le trou."""
    series = [[2024, 0, 100], [2024, 1, None]]
    series += [[2024, month, 100] for month in range(2, 12)]
    series += [[2025, 0, 100], [2025, 1, 100]]
    row = mangools_api.parse_row(1, {"kw": "k", "msv": series})
    assert isinstance(row, CanonicalKeywordRow)
    #  Treize mois écrits, un vide : douze connus, tous à 100.
    assert row.trend == [100] * 12


def test_a_short_or_stale_series_yields_no_trend() -> None:
    """Une tendance incomplète ou trop vieille est `None`, jamais une ligne en erreur."""
    short = mangools_api.parse_row(1, {"kw": "k", "msv": [[2025, m, 10] for m in range(11)]})
    assert isinstance(short, CanonicalKeywordRow)
    assert short.trend is None

    #  Douze mois pleins, mais qui s'arrêtent bien avant le mois le plus récent de la réponse :
    #  une sparkline de 2019 est indiscernable à l'écran d'une tendance actuelle.
    stale = [{"kw": "vieux", "msv": [[2019, m, 10] for m in range(12)]}]
    fresh = [{"kw": "frais", "msv": [[2026, m, 10] for m in range(12)]}]
    parsed = mangools_api.parse_rows(stale + fresh)
    assert isinstance(parsed[0], CanonicalKeywordRow)
    assert parsed[0].trend is None


def test_a_row_without_a_keyword_is_an_error_not_a_crash() -> None:
    """Contrat des adaptateurs (§4.2) : les lignes fautives sont **rendues**, jamais levées."""
    parsed = mangools_api.parse_rows([{"kw": "  "}, {"sv": 10}, "pas un objet", {"kw": "bon"}])
    assert [type(row) for row in parsed] == [RowError, RowError, RowError, CanonicalKeywordRow]
    assert parsed[0].line == 1
    assert parsed[3].line == 4


def test_a_boolean_is_never_read_as_a_number() -> None:
    """`bool` est un `int` en Python : sans garde, `True` deviendrait un volume de 1."""
    row = mangools_api.parse_row(1, {"kw": "k", "sv": True, "seo": False})
    assert isinstance(row, CanonicalKeywordRow)
    assert row.volume is None
    assert row.kd is None


# --- localisations : le cloisonnement pays du chemin API -----------------------------------


def test_only_countries_are_kept_from_the_locations_index() -> None:
    """Une subdivision cible un marché plus étroit que la cible : la retenir fausserait tout."""
    found = mangools_api.country_locations(fixture("locations"))
    assert found == {"US": 2840, "GB": 2826}
    #  La Californie est dans l'extrait, avec `country_code: "US"` — c'est `target_type` qui
    #  la disqualifie, pas son code pays.
    assert any(entry["target_type"] == "State" for entry in fixture("locations"))


def test_the_bridge_is_the_iso_code_not_the_mangools_code() -> None:
    """Le Royaume-Uni est `code: "gb"`, `tld: "uk"` et `country_code: "GB"`. Seul l'ISO colle."""
    entry = next(x for x in fixture("locations") if x["country_code"] == "GB")
    assert (entry["code"], entry["tld"]) == ("gb", "uk")
    assert mangools_api.country_locations([entry]) == {"GB": 2826}


def test_the_search_term_is_the_common_name_not_the_iso_name() -> None:
    """Le nom ISO n'est pas cherchable : « Korea, Republic of » ne trouvera jamais rien.

    Ce que `/mangools/locations` indexe, ce sont des noms courants anglais. `common_name` de
    pycountry est exactement ça, et les trois pays qui n'en ont pas sont dans `_SEARCH_TERMS`.
    """
    assert mangools_api._search_term("KR") == "South Korea"
    assert mangools_api._search_term("VN") == "Vietnam"
    assert mangools_api._search_term("IR") == "Iran"
    assert mangools_api._search_term("LA") == "Laos"
    #  Sans `common_name` : la table à la main.
    assert mangools_api._search_term("RU") == "Russia"
    assert mangools_api._search_term("TR") == "Turkey"
    #  Et les pays dont le nom ISO est déjà le nom courant passent sans détour.
    assert mangools_api._search_term("FR") == "France"
    assert mangools_api._search_term("GB") == "United Kingdom"


def test_an_unknown_country_code_falls_back_to_the_code_itself() -> None:
    """Un code hors ISO ne doit pas faire tomber le fetch avant même l'appel."""
    assert mangools_api._search_term("ZZ") == "ZZ"


async def test_an_unknown_market_fails_rather_than_falling_back() -> None:
    """Importer le marché américain dans une cible française serait indétectable à l'écran."""
    handler = _router(locations=[])
    client = _api(handler)
    with pytest.raises(mangools_api.MangoolsApiError) as caught:
        await client.resolve_location("fr-FR")
    assert "FR" in str(caught.value)
    #  Le message ne doit **pas** renvoyer vers `config.location_id` : l'écran Intégrations
    #  n'envoie jamais de `config`, ce serait un conseil que l'utilisateur ne peut pas suivre.
    assert "config" not in str(caught.value)


async def test_a_locale_without_a_country_is_refused() -> None:
    client = _api(_router())
    with pytest.raises(mangools_api.MangoolsApiError, match="pas de pays exploitable") as caught:
        await client.resolve_location("fr")
    #  Le message renvoie vers un écran qui existe, pas vers un réglage que l'UI n'écrit pas.
    assert "Sites" in str(caught.value)
    assert "config" not in str(caught.value)


# --- plan, débit, pagination ---------------------------------------------------------------


def test_the_plan_is_read_from_limits_never_hard_coded() -> None:
    rows, interval = mangools_api.plan_limits(fixture("limits"))
    assert rows == 2000
    #  `reqs-per-short-period: 3` → une requête toutes les ~3,3 s.
    assert interval == pytest.approx(mangools_api.SHORT_PERIOD_SECONDS / 3)


def test_an_unreadable_limits_response_falls_back_instead_of_failing() -> None:
    """L'import doit pouvoir tourner : un plan illisible n'est pas une raison de refuser."""
    rows, interval = mangools_api.plan_limits({"competitor-keywords-rows": True})
    assert rows == mangools_api.FALLBACK_COMPETITOR_ROWS
    assert interval == mangools_api.MIN_REQUEST_INTERVAL


async def test_two_requests_are_never_closer_than_the_plan_allows() -> None:
    """Le débit se **prouve** en mesurant : sans espacement, l'API rendrait des 429."""
    handler = _router(limits={"reqs-per-short-period": 3})
    client = mangools_api.MangoolsApi(
        KEY,
        client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
        #  Un intervalle court, mais non nul : on teste le mécanisme, pas la patience.
        min_request_interval=0.05,
    )
    started = datetime.now(UTC)
    await client.check()
    await client.check()
    await client.check()
    elapsed = (datetime.now(UTC) - started).total_seconds()
    #  Deux attentes pour trois requêtes : la première ne patiente jamais.
    assert elapsed >= 0.1
    assert len(handler.calls) == 3


async def test_discovery_paginates_until_the_provider_says_stop() -> None:
    """`__hasNextPage` gouverne la pagination — et chaque page coûte une unité de quota."""
    pages = {
        1: {"keywords": [{"kw": "un"}], "__hasNextPage": True, "isAllowedPagination": True},
        2: {"keywords": [{"kw": "deux"}], "__hasNextPage": False, "isAllowedPagination": True},
    }

    def competitor(request: httpx.Request) -> httpx.Response:
        page = int(request.url.params["page"])
        return httpx.Response(200, json=pages[page])

    handler = _router(
        limits=fixture("limits"),
        locations=fixture("locations"),
        competitor_keywords=competitor,
    )
    client = _api(handler)
    rows = await client.fetch(mangools_api.FetchSpec(locale="en-US", domain="example.com"))
    assert [row.keyword for row in rows] == ["un", "deux"]
    #  limits + locations (gratuits) + deux pages (payantes) = quatre requêtes.
    assert len(handler.calls) == 4


async def test_pagination_stops_at_the_requested_limit_without_paying_a_second_page() -> None:
    """Une page de plus, c'est une unité de quota du client : on ne la demande pas pour rien."""
    body = {"keywords": [{"kw": f"k{i}", "_id": f"k{i}"} for i in range(5)], "__hasNextPage": True}
    handler = _router(
        limits=fixture("limits"),
        competitor_keywords=body,
    )
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", domain="example.com", location_id=2840, limit=5)
    )
    assert len(rows) == 5
    #  `location_id` forcé : pas d'appel à `/locations`. limits + une page = deux requêtes.
    assert len(handler.calls) == 2
    #  Le `limit` envoyé est le plafond du **plan**, pas les 5 lignes voulues : la taille de
    #  page ne dépend jamais de ce qu'il reste à collecter (voir
    #  `test_every_page_asks_for_the_same_size`). C'est la sortie qui est tronquée à 5.
    assert handler.calls[-1].url.params["limit"] == "2000"


async def test_pagination_never_exceeds_the_plan_row_cap() -> None:
    """`competitor-keywords-rows` est le plafond du **plan** : le demander plus haut est vain."""
    handler = _router(
        limits={"competitor-keywords-rows": 3},
        competitor_keywords={"keywords": [{"kw": "k"}], "__hasNextPage": False},
    )
    client = _api(handler)
    await client.fetch(
        mangools_api.FetchSpec(
            locale="en-US", domain="example.com", location_id=2840, limit=100_000
        )
    )
    assert handler.calls[-1].url.params["limit"] == "3"


async def test_pagination_terminates_even_if_the_provider_always_says_there_is_more() -> None:
    handler = _router(
        limits=fixture("limits"),
        competitor_keywords={"keywords": [{"kw": "k"}], "__hasNextPage": True},
    )
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", domain="example.com", location_id=2840)
    )
    assert len(rows) == mangools_api.MAX_PAGES


async def test_pagination_stops_if_the_provider_ignores_the_page_parameter() -> None:
    """Le seul point déduit du parse : le nom du paramètre de page.

    S'il était faux, l'API rendrait la page 1 à chaque tour — on paierait des unités pour
    importer deux fois les mêmes lignes. Le `page` échoé le dit à l'exécution.
    """
    handler = _router(
        limits=fixture("limits"),
        #  L'API répond « page 1 » quoi qu'on demande.
        competitor_keywords={"keywords": [{"kw": "k"}], "page": 1, "__hasNextPage": True},
    )
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", domain="example.com", location_id=2840)
    )
    #  La page 1 est gardée, la page 2 est vue pour ce qu'elle est et jetée : deux appels
    #  payants au lieu de cinq, et aucun doublon.
    assert len(rows) == 1
    assert len(handler.calls) == 3


async def test_every_page_asks_for_the_same_size() -> None:
    """Une taille de page qui varie **fabrique un trou**, et rien ne le signalerait.

    Si l'offset de la page *n* vaut `(n-1) × limit`, demander 1000 puis « ce qu'il me manque »
    (500) fait relire les lignes 500-999 et sauter les 1000-1499. Le `page` échoé vaudrait
    pourtant 2. La seule défense qui tienne quelle que soit la convention d'offset — qui n'a
    pas été capturée — est de ne jamais faire varier le `limit`.
    """
    pages = {
        1: {
            "keywords": [{"kw": f"a{i}", "_id": f"a{i}"} for i in range(4)],
            "page": 1,
            "__hasNextPage": True,
        },
        2: {
            "keywords": [{"kw": f"b{i}", "_id": f"b{i}"} for i in range(4)],
            "page": 2,
            "__hasNextPage": False,
        },
    }

    def competitor(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=pages[int(request.url.params["page"])])

    handler = _router(limits={"competitor-keywords-rows": 10}, competitor_keywords=competitor)
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", domain="example.com", location_id=2840, limit=7)
    )
    asked = [call.url.params["limit"] for call in handler.calls if "competitor" in call.url.path]
    #  Le plafond du plan, deux fois — surtout pas « 10 » puis « 6 ».
    assert asked == ["10", "10"], "le `limit` doit être identique sur toutes les pages"
    #  Huit lignes ramenées pour sept voulues : la troncature est en **sortie**, une fois les
    #  pages alignées, jamais en rognant la requête suivante.
    assert [row.keyword for row in rows] == ["a0", "a1", "a2", "a3", "b0", "b1", "b2"]


async def test_pagination_stops_when_a_page_overlaps_the_previous_one() -> None:
    """L'autre façon de se tromper : l'API comprend `page`, mais pas comme nous.

    Aucun champ de la réponse ne le dirait — le `page` échoé serait juste. Ce qui le dit, ce
    sont les lignes elles-mêmes : un `_id` déjà vu veut dire qu'on a payé une unité pour
    relire, et donc qu'on a sauté ce qu'on croyait avoir lu. Mieux vaut un import court.
    """
    shared = [{"kw": f"k{i}", "_id": f"k{i}"} for i in range(4)]
    pages = {
        1: {"keywords": shared[:3], "page": 1, "__hasNextPage": True},
        #  Page 2 mal alignée : elle recommence à la ligne 2.
        2: {"keywords": shared[2:], "page": 2, "__hasNextPage": True},
    }

    def competitor(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=pages[int(request.url.params["page"])])

    handler = _router(limits=fixture("limits"), competitor_keywords=competitor)
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", domain="example.com", location_id=2840)
    )
    #  Les trois lignes de la page 1, et rien de la page 2 : aucun doublon n'entre.
    assert [row.keyword for row in rows] == ["k0", "k1", "k2"]
    assert len({row.keyword for row in rows}) == len(rows)


async def test_a_page_echoed_as_a_string_does_not_slip_past_the_guard() -> None:
    """`"2"` et `2` disent la même chose : la garde compare des chaînes, pas des types."""
    handler = _router(
        limits=fixture("limits"),
        #  L'API échoe toujours « page 1 », mais en chaîne.
        competitor_keywords={
            "keywords": [{"kw": "k", "_id": "k"}],
            "page": "1",
            "__hasNextPage": True,
        },
    )
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", domain="example.com", location_id=2840)
    )
    assert len(rows) == 1


async def test_enrichment_sends_one_request_per_batch_not_per_keyword() -> None:
    """Le quota se compte en **appels** : découper serait payer plusieurs fois la même chose."""
    wanted = [f"mot {index}" for index in range(mangools_api.KEYWORD_IMPORTS_BATCH + 1)]

    def imports(request: httpx.Request) -> httpx.Response:
        asked = json.loads(request.content)["keywords"]
        return httpx.Response(200, json={"keywords": [{"kw": kw} for kw in asked]})

    handler = _router(limits=fixture("limits"), keyword_imports=imports)
    client = _api(handler)
    rows = await client.fetch(
        mangools_api.FetchSpec(locale="en-US", keywords=tuple(wanted), location_id=2840)
    )
    assert len(rows) == len(wanted)
    #  limits + deux lots (101 mots-clés, lots de 100) = trois requêtes.
    assert len(handler.calls) == 3


async def test_enrichment_pins_the_market_of_the_target() -> None:
    """Tout le cloisonnement pays du chemin API tient dans ce `location_id`."""
    captured: dict[str, Any] = {}

    def imports(request: httpx.Request) -> httpx.Response:
        captured.update(json.loads(request.content))
        return httpx.Response(200, json={"keywords": []})

    handler = _router(
        limits=fixture("limits"), locations=fixture("locations"), keyword_imports=imports
    )
    client = _api(handler)
    await client.fetch(mangools_api.FetchSpec(locale="en-US", keywords=("chaise",)))
    assert captured["location_id"] == 2840


# --- le garde de format et la clé -----------------------------------------------------------


async def test_fetch_refuses_before_any_network_call_when_the_lock_is_closed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Le garde est **en tête** : on ne brûle pas le quota du client pour un JSON illisible."""

    class _Exploding(httpx.AsyncClient):
        async def request(self, *args: Any, **kwargs: Any) -> httpx.Response:
            raise AssertionError("aucun appel réseau ne doit partir tant que le format est inconnu")

    monkeypatch.setattr(mangools_api, "PARSER_WRITTEN", False)
    client = mangools_api.MangoolsApi(KEY, client=_Exploding(), min_request_interval=0.0)
    with pytest.raises(mangools_api.FormatNotCapturedError):
        await client.fetch(mangools_api.FetchSpec(domain="example.com", locale="fr-FR"))


async def test_api_errors_never_leak_the_key() -> None:
    transport = httpx.MockTransport(lambda request: httpx.Response(401, json={"error": "nope"}))
    client = mangools_api.MangoolsApi(KEY, client=httpx.AsyncClient(transport=transport))
    with pytest.raises(mangools_api.MangoolsApiError) as caught:
        await client.check()
    assert KEY not in str(caught.value)


async def test_an_error_body_is_never_echoed_back_to_the_user() -> None:
    """Un corps d'erreur peut renvoyer la requête en écho, en-têtes compris — donc la clé."""
    body = {"error": f"bad token {KEY}", "headers": {mangools_api.AUTH_HEADER: KEY}}
    transport = httpx.MockTransport(lambda request: httpx.Response(500, json=body))
    client = mangools_api.MangoolsApi(KEY, client=httpx.AsyncClient(transport=transport))
    with pytest.raises(mangools_api.MangoolsApiError) as caught:
        await client.check()
    assert KEY not in str(caught.value)
    assert "bad token" not in str(caught.value)


async def test_check_accepts_a_working_key() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.headers[mangools_api.AUTH_HEADER] == KEY
        return httpx.Response(200, json={"quota": {"left": 42}})

    transport = httpx.MockTransport(handler)
    client = mangools_api.MangoolsApi(KEY, client=httpx.AsyncClient(transport=transport))
    assert await client.check() == {"quota": {"left": 42}}


async def test_check_reports_a_rate_limit_and_never_calls_it_a_quota() -> None:
    """Un 429 dit le **débit**, jamais le quota — les endpoints de listes n'en consomment pas.

    Le message d'avant (« Quota Mangools atteint ») envoyait l'utilisateur racheter des crédits
    pour un simple espacement de requêtes trop court, et il finissait dans `imports.error`.
    """
    transport = httpx.MockTransport(lambda request: httpx.Response(429))
    client = mangools_api.MangoolsApi(KEY, client=httpx.AsyncClient(transport=transport))
    with pytest.raises(mangools_api.MangoolsApiError, match="Débit") as raised:
        await client.check()
    assert "uota" not in str(raised.value)


async def test_check_survives_a_non_json_success() -> None:
    transport = httpx.MockTransport(lambda request: httpx.Response(200, text="ok"))
    client = mangools_api.MangoolsApi(KEY, client=httpx.AsyncClient(transport=transport))
    #  Un 200 non-JSON reste un succès d'authentification : on ne prétend pas savoir lire.
    assert await client.check() == {}


# --- le job, de bout en bout ------------------------------------------------------------------


async def test_the_job_writes_the_reservoir_from_a_provider_response(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Le chemin complet : job enfilé, fournisseur bouchonné, réservoir peuplé.

    C'est le test que le verrou rendait impossible à écrire. Le fournisseur est bouchonné —
    aucune clé réelle, aucun quota consommé — mais tout le reste est le vrai chemin : payload,
    déchiffrement, fetch, parse, pipeline d'écriture.
    """
    #  `config.location_id` forcé : la cible du bac à sable est française, et l'extrait de
    #  `/locations` (capturé sur une recherche « united ») ne porte pas la France. C'est
    #  l'échappatoire que le message d'erreur de `resolve_location` recommande — la tester ici
    #  vaut mieux que de fabriquer une entrée `FR` qui ne viendrait d'aucune capture.
    integration = await _integration(db, sandbox.org_id, {"location_id": 2250})
    handler = _router(
        limits=fixture("limits"),
        competitor_keywords={**fixture("competitor_keywords"), "__hasNextPage": False},
    )
    #  Le client est construit **avant** le monkeypatch : le construire dans la lambda le
    #  ferait passer par le nom déjà remplacé, donc par la lambda elle-même.
    stub = _api(handler)
    monkeypatch.setattr(api_pipeline.mangools_api, "MangoolsApi", lambda api_key, **kwargs: stub)

    [(entry, job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id
    )
    finished = await _run(db, job)
    assert finished is not None
    assert finished.status == "done", finished.error

    await db.refresh(entry)
    assert entry.status == "done"
    written = await db.scalar(
        select(func.count()).select_from(Keyword).where(Keyword.org_id == sandbox.org_id)
    )
    assert written == 2


async def test_a_provider_refusal_fails_the_job_with_a_readable_message(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Clé refusée : le job échoue, la ligne `imports` porte le motif, rien n'entre."""
    integration = await _integration(db, sandbox.org_id)
    stub = _api(lambda request: httpx.Response(401))
    monkeypatch.setattr(api_pipeline.mangools_api, "MangoolsApi", lambda api_key, **kwargs: stub)

    [(entry, job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id
    )
    entry_id = entry.id
    finished = await _run(db, job)
    assert finished is not None
    assert finished.status == "failed"
    assert KEY not in (finished.error or "")

    refreshed = await db.scalar(select(Import).where(Import.id == entry_id))
    assert refreshed is not None
    assert refreshed.status == "failed"
    assert refreshed.error

    written = await db.scalar(
        select(func.count()).select_from(Keyword).where(Keyword.org_id == sandbox.org_id)
    )
    assert written == 0


async def test_an_import_without_a_mode_says_what_is_missing(
    db: AsyncSession, sandbox: Fixture
) -> None:
    """Ni domaine ni liste de mots-clés : refus explicite, jamais un `FetchSpec` deviné."""
    integration = await _integration(db, sandbox.org_id)
    spec = api_pipeline.ApiImportSpec(
        import_id=uuid.uuid4(),
        integration_id=integration.id,
        provider="mangools",
        source="mangools",
        site_target_id=sandbox.target_id,
        locale="fr-FR",
        fetched_at=datetime.now(UTC),
        domain="",
    )
    with pytest.raises(ValueError, match="sans mode"):
        await api_pipeline.fetch_rows(db, sandbox.org_id, spec)


async def test_a_payload_without_its_integration_fails_rather_than_guessing(
    db: AsyncSession, sandbox: Fixture
) -> None:
    """Intégration supprimée entre l'enfilage et le run : échec net, jamais un import muet."""
    spec = api_pipeline.ApiImportSpec(
        import_id=uuid.uuid4(),
        integration_id=uuid.uuid4(),
        provider="mangools",
        source="mangools",
        site_target_id=sandbox.target_id,
        locale="fr-FR",
        fetched_at=datetime.now(UTC),
        domain="example.com",
    )
    with pytest.raises(ValueError, match="intégration inconnue"):
        await api_pipeline.fetch_rows(db, sandbox.org_id, spec)


async def test_a_job_cannot_read_another_org_key(db: AsyncSession, sandbox: Fixture) -> None:
    """Le `SELECT` de l'intégration est filtré par `org_id` : pas de clé volée par payload."""
    integration = await _integration(db, sandbox.org_id)
    spec = api_pipeline.ApiImportSpec(
        import_id=uuid.uuid4(),
        integration_id=integration.id,
        provider="mangools",
        source="mangools",
        site_target_id=sandbox.target_id,
        locale="fr-FR",
        fetched_at=datetime.now(UTC),
        domain="example.com",
    )
    with pytest.raises(ValueError, match="intégration inconnue"):
        await api_pipeline.fetch_rows(db, f"org_{uuid.uuid4().hex}", spec)


def test_payload_must_carry_a_timezone() -> None:
    payload = {
        "import_id": str(uuid.uuid4()),
        "integration_id": str(uuid.uuid4()),
        "provider": "mangools",
        "source": "mangools",
        "site_target_id": str(uuid.uuid4()),
        "locale": "fr-FR",
        "fetched_at": "2026-09-02T10:00:00",
    }
    with pytest.raises(ValueError, match="fuseau"):
        api_pipeline.parse_payload(payload)


def test_incomplete_payload_says_what_is_missing() -> None:
    with pytest.raises(ValueError, match="incomplet"):
        api_pipeline.parse_payload({"import_id": str(uuid.uuid4())})


# --- le quota, et le mode porté par le job --------------------------------------------------


def test_resource_quota_reads_the_counter_that_actually_moves() -> None:
    """`related-keywords` est la ressource qui décompte les appels rendant des mots-clés."""
    limit, remaining = mangools_api.resource_quota(fixture("limits"))
    assert (limit, remaining) == (1200, 1192)


@pytest.mark.parametrize(
    "payload",
    [
        None,
        {},
        {"resources": []},
        {"resources": {"related-keywords": "beaucoup"}},
        {"resources": {"serps": {"limit": 10, "remaining": 9}}},
    ],
)
def test_resource_quota_invents_nothing_when_the_shape_surprises(payload: Any) -> None:
    """Un quota inconnu reste inconnu : ni 0 (qui découragerait), ni le plafond (qui mentirait)."""
    assert mangools_api.resource_quota(payload) == (None, None)


def test_a_payload_with_keywords_is_an_enrichment() -> None:
    spec = api_pipeline.parse_payload(
        {
            "import_id": str(uuid.uuid4()),
            "integration_id": str(uuid.uuid4()),
            "provider": "mangools",
            "source": "mangools",
            "site_target_id": str(uuid.uuid4()),
            "locale": "fr-FR",
            "fetched_at": "2026-09-02T10:00:00+00:00",
            "domain": "",
            "keywords": ["chaise de bureau", "  ", "fauteuil"],
        }
    )
    #  Les cases vides d'un payload ne deviennent pas des mots-clés vides : le lot envoyé au
    #  fournisseur les aurait payées, et le rapport les aurait comptées.
    assert spec.keywords == ("chaise de bureau", "fauteuil")
    assert spec.domain == ""


async def test_an_enrichment_never_reaches_the_job_because_it_has_no_destination(
    db: AsyncSession, sandbox: Fixture
) -> None:
    """L'enrichissement est **refusé à l'enfilage**, décision du 2026-09-02.

    L'adaptateur sait toujours le faire (`test_a_payload_with_keywords_is_an_enrichment`,
    `_fetch_keywords`), mais rien ne l'enfile plus : coller des mots-clés ne donne aucun nom de
    liste stable, et un import sans liste de destination écrit des mots-clés que le réservoir
    n'affichera jamais (`app.keywords.service._BELONGS`). Le mode dort, il n'écrit pas dans le
    vide.
    """
    integration = await _integration(db, sandbox.org_id, {"location_id": 2250})
    with pytest.raises(AppError) as raised:
        await service.trigger_import(
            db,
            sandbox.org_id,
            integration.id,
            sandbox.target_id,
            keywords=["chaise de bureau", "fauteuil ergonomique"],
        )
    assert raised.value.status == 422
    written = await db.scalar(
        select(func.count()).select_from(Keyword).where(Keyword.org_id == sandbox.org_id)
    )
    assert written == 0


async def test_a_trigger_domain_beats_the_domain_of_the_site(
    db: AsyncSession, sandbox: Fixture
) -> None:
    """Découverte d'un concurrent : le domaine saisi remplace celui du site, sans le modifier."""
    integration = await _integration(db, sandbox.org_id)
    [(_, job, *_rest)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, domain="concurrent.example"
    )
    assert job.payload["domain"] == "concurrent.example"
    assert "keywords" not in job.payload


# --- les listes enregistrées du compte ----------------------------------------------------
#
# Le mode que l'interface expose depuis la reprise du 2026-09-02. Mesuré le même jour : les deux
# endpoints (`/kwfinder/lists` et `/kwfinder/lists/{id}`) **ne consomment aucune unité de
# quota** — trois lectures de `/kwfinder/limits` encadrant les appels ont rendu vingt compteurs
# `resources` strictement identiques. C'est ce qui rend le recensement des marchés gratuit, donc
# possible avant même de choisir une cible.


LIST_ID = "6a589821eefecfa46ad01f2e"
OTHER_LIST_ID = "6a6986b7a5359704ad73709e"


def _lists_router(detail: Any | None = None) -> Any:
    """Un bouchon qui sert `/kwfinder/lists` et `/kwfinder/lists/{id}`, et compte les appels.

    `_router` indexe par dernier segment de chemin : il ne sait pas distinguer le listing du
    détail, dont le dernier segment est un identifiant. D'où ce second bouchon, qui garde la
    même promesse — compter les requêtes, seule façon de tester ce que coûte un import.
    """
    calls: list[httpx.Request] = []
    body = fixture("list_detail") if detail is None else detail

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        #  `endswith` et non `==` : le chemin réel est préfixé par la version de l'api
        #  (`/v3/kwfinder/lists`), qui vient de la configuration et n'a rien à faire ici.
        if request.url.path.endswith(mangools_api.LISTS_PATH):
            return httpx.Response(200, json=fixture("lists"))
        if f"{mangools_api.LISTS_PATH}/" in request.url.path:
            return httpx.Response(200, json=body)
        return httpx.Response(404, json={"error": "chemin non bouchonné"})

    handler.calls = calls  # type: ignore[attr-defined]
    return handler


def test_the_account_lists_are_parsed_with_their_counts_and_dates() -> None:
    """`_id`, nom, nombre de mots-clés, dates — et les dates sont en **millisecondes**.

    Le schéma publié annonce des secondes ; la réponse réelle porte `1784191009000`. Les lire
    en secondes daterait toutes les listes de 1970 et l'écran trierait n'importe comment.
    """
    parsed = mangools_api.parse_lists(fixture("lists"))
    assert [item.id for item in parsed] == [LIST_ID, OTHER_LIST_ID, "67db8efb553d96b7dff08eef"]
    assert [item.name for item in parsed] == ["Liste courte", "Liste France", "Liste vide"]
    assert [item.keyword_count for item in parsed] == [3, 1, 0]
    assert parsed[0].updated_at == datetime(2026, 7, 16, 8, 36, 49, tzinfo=UTC)


def test_a_list_entry_without_an_identifier_is_skipped_not_guessed() -> None:
    """Sans `_id` la liste n'est pas importable ; sans nom elle n'est pas reconnaissable."""
    entries = [{"name": "sans id"}, {"_id": "abc"}, {"_id": "", "name": "x"}]
    assert mangools_api.parse_lists(entries) == []
    assert mangools_api.parse_lists({"pas": "une liste"}) == []


def test_a_list_row_says_its_own_market() -> None:
    """Le marché d'une ligne de liste est un **objet**, et c'est son libellé qui est résolu."""
    rows = fixture("list_detail")["keywords"]
    assert [mangools_api.row_country_label(row) for row in rows] == [
        "United States",
        "France",
        "New York",
    ]
    assert mangools_api.row_country_label({"kw": "sans marché"}) == ""
    #  Les deux modules doivent nommer la même clé : le pipeline ne connaît aucun fournisseur,
    #  l'adaptateur ne connaît pas le pipeline, et rien d'autre ne les tiendrait d'accord.
    assert mangools_api.LOCATION_FIELD == pipeline.API_COUNTRY_FIELD
    assert mangools_api.LOCATION_LABEL_FIELD == pipeline.API_COUNTRY_LABEL_KEY


async def test_fetching_a_list_asks_one_request_and_no_market() -> None:
    """Un seul appel avec `fields`, et ni `limits` ni `locations` : rien à payer, rien à résoudre.

    C'est la différence de nature avec les deux autres modes : eux passent un `location_id` en
    entrée et doivent donc le résoudre ; une liste se lit telle qu'elle a été enregistrée.
    """
    handler = _lists_router()
    rows = await _api(handler).fetch(mangools_api.FetchSpec(locale="fr-FR", list_id=LIST_ID))
    assert len(handler.calls) == 1
    request = handler.calls[0]
    assert request.url.path.endswith(f"{mangools_api.LISTS_PATH}/{LIST_ID}")
    assert request.url.params["fields"] == mangools_api.LIST_FIELDS
    #  Les trois lignes sont rendues telles quelles : le tri par marché est au pipeline.
    assert [row.keyword for row in rows] == [
        "mot cle americain",
        "mot cle francais",
        "mot cle local",
    ]
    assert rows[0].volume == 5500
    assert rows[0].kd == Decimal(26)
    assert rows[0].cpc == Decimal("5.91")
    assert rows[0].trend is not None
    assert len(rows[0].trend) == 12
    #  Deux mois connus seulement : une sparkline de douze cases ne se fabrique pas.
    assert rows[1].trend is None


async def test_a_list_that_cannot_be_read_refuses_rather_than_importing_nothing() -> None:
    """Une enveloppe sans `keywords` est un refus explicite, pas un import à zéro ligne."""
    client = _api(_lists_router(detail={"list": {"_id": LIST_ID}}))
    with pytest.raises(mangools_api.MangoolsApiError, match="sans liste `keywords`"):
        await client.fetch(mangools_api.FetchSpec(locale="fr-FR", list_id=LIST_ID))


async def test_the_job_imports_a_list_and_counts_what_is_off_target(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Une liste qui mélange les marchés : seule la cible entre, le reste est **compté**.

    La cible du bac à sable est `fr-FR`. La liste porte trois lignes : une américaine, une
    française, une new-yorkaise. La française est importée ; l'américaine est « hors cible » et
    la new-yorkaise « pays non reconnu » — deux motifs distincts, parce que le second est le
    seul signal d'un trou dans la table d'alias.
    """
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(api_pipeline.mangools_api, "MangoolsApi", lambda api_key, **kwargs: stub)
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})

    [(entry, job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=[LIST_ID]
    )
    assert job.payload["list_id"] == LIST_ID
    assert job.payload["domain"] == ""
    assert "keywords" not in job.payload

    finished = await _run(db, job)
    assert finished is not None
    assert finished.status == "done", finished.error
    report = finished.report or {}
    assert report["rows_read"] == 3
    assert report["rows_inserted"] == 1
    assert report["rows_off_target"] == 1
    assert report["rows_country_unresolved"] == 1
    assert report["ignored_by_country"] == {"United States": 1}
    assert report["unresolved_by_country"] == {"New York": 1}
    #  Une ligne d'un autre marché n'est pas une ligne fautive.
    assert report["rows_error"] == 0

    written = list(
        (await db.scalars(select(Keyword.keyword).where(Keyword.org_id == sandbox.org_id))).all()
    )
    assert written == ["mot cle francais"]
    await db.refresh(entry)
    assert entry.status == "done"


async def test_an_imported_list_is_visible_in_the_reservoir_and_reimports_into_itself(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """**Le bug du 2026-09-02, et sa réponse.** Un import API réussi n'apparaissait jamais.

    Les mots-clés et leurs métriques étaient bien écrits, le rapport disait `done`, et l'écran
    disait « Réservoir vide » : le chemin API ne créait aucune appartenance, or le réservoir
    n'affiche que ce qui est membre d'une liste de la cible ou porteur d'une attache d'URL
    (`app.keywords.service._BELONGS`). Des données payées, en base, invisibles pour toujours.

    La réponse actée : **une liste Mangools est une liste chez nous**, portant son nom sur la
    cible choisie. Ce test tient les deux moitiés de la promesse : le premier import la crée et
    le mot-clé se voit dans le réservoir ; le second la **rafraîchit** — même liste, aucun
    mot-clé créé en double, appartenance inchangée. Pas cinquante-deux listes par an.
    """
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(api_pipeline.mangools_api, "MangoolsApi", lambda api_key, **kwargs: stub)
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})

    [(entry, job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=[LIST_ID]
    )
    #  L'appartenance voyage dans le payload : sans elle, le worker écrirait dans le vide.
    assert job.payload["keyword_list_id"]
    finished = await _run(db, job)
    assert finished is not None
    assert finished.status == "done", finished.error
    report = finished.report or {}
    assert report["keywords_created"] == 1
    assert report["list_items_added"] == 1

    await db.refresh(entry)
    destination = await db.scalar(
        select(KeywordList).where(KeywordList.id == entry.keyword_list_id)
    )
    assert destination is not None
    #  Le nom de la liste KWFinder, pas un libellé fabriqué : c'est celui que l'utilisateur
    #  reconnaît dans le filtre « Liste » du réservoir.
    assert destination.name == "Liste courte"
    assert destination.site_target_id == sandbox.target_id
    assert destination.source == "mangools"

    rows, _, _, _, _ = await keywords_service.list_keywords(
        db,
        sandbox.org_id,
        site_target_id=sandbox.target_id,
        view=None,
        filters=keywords_service.Filters(),
        cursor=None,
        limit=50,
    )
    assert [row.keyword for row in rows] == ["mot cle francais"]

    #  Réimport de la même liste : c'est le chemin que suit l'utilisateur pour rattraper les
    #  mots-clés déjà écrits par la version fautive.
    [(again_entry, again_job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=[LIST_ID]
    )
    #  La période `manual` étant la même, il faut relâcher la clé de dédup du premier job :
    #  c'est ce que fait le passage à `done`, déjà acquis ci-dessus.
    assert again_job.id != job.id
    replayed = await _run(db, again_job)
    assert replayed is not None
    assert replayed.status == "done", replayed.error
    second = replayed.report or {}
    #  Le mot-clé existe déjà : rien de créé, et l'appartenance l'était déjà aussi.
    assert second["keywords_created"] == 0
    assert second["list_items_added"] == 0

    await db.refresh(again_entry)
    assert again_entry.keyword_list_id == destination.id
    lists = await db.scalar(
        select(func.count())
        .select_from(KeywordList)
        .where(KeywordList.site_target_id == sandbox.target_id)
    )
    assert lists == 1
    members = await db.scalar(
        select(func.count())
        .select_from(KeywordListItem)
        .where(KeywordListItem.list_id == destination.id)
    )
    assert members == 1


async def test_a_failed_api_import_says_why_on_the_import_row(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Un import `failed` ne doit **jamais** rester muet : le motif descend dans `imports`.

    Sans cette recopie, l'écran affiche « Échec » et « Aucun rapport pour l'instant. », et le
    motif ne vit que dans `jobs.error` — que l'utilisateur ne voit pas. Ce qui descend est le
    message court du socle (`worker._short`), **jamais une trace**, et il est borné : le corps
    d'une erreur du fournisseur peut renvoyer la requête en écho, en-têtes compris.
    """
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})
    #  Le déclenchement lit les listes du compte ; le job, lui, se heurte à une clé refusée.
    refused = _api(lambda request: httpx.Response(401))
    monkeypatch.setattr(api_pipeline.mangools_api, "MangoolsApi", lambda api_key, **kwargs: refused)

    [(entry, job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=[LIST_ID]
    )
    finished = await _run(db, job)
    assert finished is not None
    assert finished.status == "failed"
    assert finished.error

    await db.refresh(entry)
    assert entry.status == "failed"
    assert entry.error == finished.error
    assert "Traceback" not in entry.error
    assert len(entry.error) <= worker.MAX_ERROR_LENGTH


async def test_a_discovery_lands_in_a_list_named_after_the_domain(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Le mode dormant n'a plus le droit d'écrire dans le vide non plus.

    La découverte n'a pas de nom de liste naturel — mais elle a le domaine interrogé, et il est
    **stable d'une exécution à l'autre** : le sweep hebdomadaire rafraîchit la même liste au
    lieu d'en empiler cinquante-deux. C'est aussi ce qui prouve que le domaine du site est bien
    repris quand l'appel n'en nomme aucun.
    """
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})

    [(entry, job, *_)] = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id
    )
    domain = job.payload["domain"]
    assert domain
    assert job.payload["keyword_list_id"]
    await db.refresh(entry)
    destination = await db.scalar(
        select(KeywordList).where(KeywordList.id == entry.keyword_list_id)
    )
    assert destination is not None
    assert destination.name == domain


async def test_a_trigger_launches_one_import_per_list(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Deux listes cochées = deux imports, deux rapports, deux étiquettes reconnaissables.

    Sans la liste dans la clé de dédup, la seconde retomberait sur l'import de la première :
    un seul rapport pour deux listes, et rien à l'écran pour le dire.
    """
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})

    launched = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=[LIST_ID, OTHER_LIST_ID]
    )
    assert len(launched) == 2
    assert [created for _, _, created, _, _ in launched] == [True, True]
    assert [list_id for _, _, _, list_id, _ in launched] == [LIST_ID, OTHER_LIST_ID]
    assert [name for _, _, _, _, name in launched] == ["Liste courte", "Liste France"]
    #  L'étiquette d'affichage de l'onglet Imports, et le pays **retenu**.
    assert [entry.filename for entry, _, _, _, _ in launched] == ["Liste courte", "Liste France"]
    assert {entry.detected_country for entry, _, _, _, _ in launched} == {"FR"}
    assert len({job.dedup_key for _, job, _, _, _ in launched}) == 2
    #  Relevés **avant** le second appel : son chemin de dédup fait un `rollback()`, qui expire
    #  toutes les instances de la session — les relire ensuite déclencherait un chargement
    #  paresseux impossible en asynchrone.
    first_ids = [entry.id for entry, _, _, _, _ in launched]

    #  Rejouer la même sélection ne double rien : les deux imports en vol sont rendus tels quels.
    again = await service.trigger_import(
        db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=[LIST_ID, OTHER_LIST_ID]
    )
    assert [created for _, _, created, _, _ in again] == [False, False]
    assert [entry.id for entry, _, _, _, _ in again] == first_ids


async def test_a_trigger_refuses_a_list_the_account_does_not_have(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Un identifiant inconnu est refusé tout de suite, pas découvert dans le worker."""
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})

    with pytest.raises(AppError) as caught:
        await service.trigger_import(
            db, sandbox.org_id, integration.id, sandbox.target_id, list_ids=["inconnue"]
        )
    assert caught.value.status == 404
    assert "inconnue" in caught.value.detail


async def test_the_preview_censuses_the_markets_really_present(
    db: AsyncSession, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Le recensement, liste par liste et en agrégat — le pendant du preview CSV.

    C'est lui qui évite de lancer un import vide : une liste sans ligne pour la cible visée se
    voit avant d'être lancée, pas après avoir produit un rapport à zéro.
    """
    integration = await _integration(db, sandbox.org_id)
    stub = _api(_lists_router())
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})

    items, census = await service.preview_lists(
        db, sandbox.org_id, integration.id, [LIST_ID], sandbox.target_id
    )
    assert [item.name for item in items] == ["Liste courte"]
    assert items[0].keywords_total == 3
    #  Trié par nombre de lignes décroissant, puis par libellé — l'ordre de l'écran.
    assert [(entry.raw, entry.code, entry.rows) for entry in census] == [
        ("France", "FR", 1),
        ("New York", None, 1),
        ("United States", "US", 1),
    ]
    #  Le libellé n'est jamais nul, par contrat avec le miroir Zod.
    assert [entry.label for entry in census] == ["France", "New York", "États-Unis"]
    #  La cible du bac à sable est française : elle est proposée pour la France, pas ailleurs.
    assert census[0].target_ids == [sandbox.target_id]
    assert census[1].target_ids == []
    assert census[1].suggested_language is None