"""Endpoints des intégrations BYOK (spec Imports T8) : CRUD, plannings, import manuel.

La propriété que ce module défend avant tout : **la clé n'est jamais renvoyée en clair**.
Elle est cherchée dans chaque réponse du module, y compris celle qui vient de l'écrire.
"""

import asyncio
import json
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

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

from app.core import crypto
from app.integrations import service
from app.integrations.models import ImportSchedule, Integration
from app.integrations.pipeline import JOB_TYPE
from app.jobs.models import Job
from app.keywords.models import Import
from app.keywords.sources import mangools_api
from tests.conftest import Fixture

KEY = "mangools-secret-key-9f3a"
OTHER_KEY = "mangools-autre-cle-4b7c"


def headers(auth_headers: Any, sandbox: Fixture) -> dict[str, str]:
    return auth_headers(sandbox.org_id)


async def _create(client: AsyncClient, auth: Any, sandbox: Fixture, key: str = KEY) -> dict:
    response = await client.post(
        "/integrations",
        json={"provider": "mangools", "api_key": key},
        headers=headers(auth, sandbox),
    )
    assert response.status_code == 201, response.text
    return response.json()


# --- CRUD -------------------------------------------------------------------------------------


async def test_create_stores_the_key_encrypted_and_never_returns_it(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    body = await _create(client, auth_headers, sandbox)
    assert KEY not in response_text(body)
    assert body["credentials_hint"] == "••••••9f3a"
    assert body["status"] == "unverified"
    assert "credentials" not in body
    assert "api_key" not in body

    stored = await db.scalar(select(Integration).where(Integration.org_id == sandbox.org_id))
    assert stored is not None
    #  En base : chiffré, et illisible sans la clé Fernet.
    assert KEY not in stored.credentials
    assert crypto.decrypt(stored.credentials) == KEY


def response_text(body: Any) -> str:
    return str(body)


async def test_listing_and_reading_never_show_the_key(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    listing = await client.get("/integrations", headers=headers(auth_headers, sandbox))
    assert listing.status_code == 200
    assert KEY not in listing.text
    assert [item["id"] for item in listing.json()["items"]] == [created["id"]]

    detail = await client.get(
        f"/integrations/{created['id']}", headers=headers(auth_headers, sandbox)
    )
    assert detail.status_code == 200
    assert KEY not in detail.text


async def test_a_second_integration_for_the_same_provider_is_a_conflict(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    await _create(client, auth_headers, sandbox)
    again = await client.post(
        "/integrations",
        json={"provider": "mangools", "api_key": OTHER_KEY},
        headers=headers(auth_headers, sandbox),
    )
    assert again.status_code == 409
    assert again.headers["content-type"].startswith("application/problem+json")


async def test_replacing_the_key_resets_the_status(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    created = await _create(client, auth_headers, sandbox)
    #  Statut posé à la main : la vérification réelle appelle le fournisseur, testée ailleurs.
    record = await db.scalar(select(Integration).where(Integration.id == uuid.UUID(created["id"])))
    assert record is not None
    record.status = "connected"
    record.checked_at = datetime.now(UTC)
    await db.commit()

    patched = await client.patch(
        f"/integrations/{created['id']}",
        json={"api_key": OTHER_KEY},
        headers=headers(auth_headers, sandbox),
    )
    assert patched.status_code == 200
    body = patched.json()
    assert body["credentials_hint"] == "••••••4b7c"
    #  Une clé neuve n'hérite jamais du « connectée » de l'ancienne.
    assert body["status"] == "unverified"
    assert body["checked_at"] is None
    assert OTHER_KEY not in patched.text

    await db.refresh(record)
    assert crypto.decrypt(record.credentials) == OTHER_KEY


async def test_unknown_provider_is_refused(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    response = await client.post(
        "/integrations",
        json={"provider": "dataforseo", "api_key": KEY},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422


async def test_an_empty_or_tiny_key_is_refused(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    response = await client.post(
        "/integrations",
        json={"provider": "mangools", "api_key": "x"},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422


async def test_another_org_sees_nothing(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    stranger = auth_headers(f"org_{uuid.uuid4().hex}")

    listing = await client.get("/integrations", headers=stranger)
    assert listing.json()["items"] == []
    #  404 et non 403 : l'intégration d'une autre organisation n'existe pas.
    assert (await client.get(f"/integrations/{created['id']}", headers=stranger)).status_code == 404
    assert (
        await client.patch(f"/integrations/{created['id']}", json={"config": {}}, headers=stranger)
    ).status_code == 404
    assert (
        await client.delete(f"/integrations/{created['id']}", headers=stranger)
    ).status_code == 404


async def test_delete_removes_the_integration(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    gone = await client.delete(
        f"/integrations/{created['id']}", headers=headers(auth_headers, sandbox)
    )
    assert gone.status_code == 204
    listing = await client.get("/integrations", headers=headers(auth_headers, sandbox))
    assert listing.json()["items"] == []


async def test_endpoints_require_a_token(client: AsyncClient) -> None:
    assert (await client.get("/integrations")).status_code == 401


# --- vérification de la clé -------------------------------------------------------------------


class _Refusing:
    """Bouchon de client fournisseur : la clé est refusée. Aucun appel réseau dans la suite."""

    def __init__(self, api_key: str, *, timeout: float | None = None) -> None:
        self.api_key = api_key
        self.timeout = timeout

    async def check(self) -> dict:
        raise mangools_api.MangoolsApiError("Clé Mangools refusée par le fournisseur (401/403).")


class _Accepting:
    #  Retient le `timeout` reçu : c'est lui que vérifie le test de l'invariant de frontière.
    seen_timeout: float | None = None

    def __init__(self, api_key: str, *, timeout: float | None = None) -> None:
        self.api_key = api_key
        type(self).seen_timeout = timeout

    async def check(self) -> dict:
        return {"quota": 1000}


async def test_check_records_a_refusal_as_a_status_not_an_error(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setitem(service._CLIENTS, "mangools", _Refusing)

    response = await client.post(
        f"/integrations/{created['id']}/check", headers=headers(auth_headers, sandbox)
    )
    #  Un refus du fournisseur n'est pas une erreur de *notre* api : c'est un état affichable.
    assert response.status_code == 200
    body = response.json()
    assert body["status"] == "error"
    assert "refus" in body["status_error"].lower()
    assert body["checked_at"] is not None
    assert KEY not in response.text


async def test_check_marks_a_working_key_connected(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setitem(service._CLIENTS, "mangools", _Accepting)

    response = await client.post(
        f"/integrations/{created['id']}/check", headers=headers(auth_headers, sandbox)
    )
    assert response.status_code == 200
    assert response.json()["status"] == "connected"
    assert response.json()["status_error"] is None


async def test_check_waits_less_than_the_browser_does(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Invariant de frontière : le plafond côté fournisseur reste **sous** celui d'`apiFetch`.

    `web/lib/api.ts` avorte tout appel à l'api au bout de 10 s, et un `TimeoutError` n'est pas
    une `ApiError` : `relay` le relance, la Server Action rejette, et **tout l'écran
    Intégrations** part dans son `error.tsx`. L'utilisateur verrait un crash là où
    `check_integration` promet un badge « En erreur ». Ce test fige le sens de l'inégalité.
    """
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setitem(service._CLIENTS, "mangools", _Accepting)

    await client.post(
        f"/integrations/{created['id']}/check", headers=headers(auth_headers, sandbox)
    )
    assert _Accepting.seen_timeout == service.CHECK_TIMEOUT_SECONDS
    #  10 s est le budget d'`apiFetch` ; la marge couvre le temps de notre propre traitement.
    assert service.CHECK_TIMEOUT_SECONDS < 10.0


async def test_a_provider_that_never_answers_is_a_status_not_a_crash(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Un fournisseur muet est un état de l'intégration, pas une erreur HTTP de notre api."""
    created = await _create(client, auth_headers, sandbox)

    class _Silent:
        def __init__(self, api_key: str, *, timeout: float | None = None) -> None:
            self.api_key = api_key

        async def check(self) -> dict:
            #  Ce que lève httpx quand le tiers ne répond pas ; `MangoolsApi.check` le traduit
            #  déjà, on vérifie ici que le service le laisse devenir un statut.
            raise mangools_api.MangoolsApiError("API Mangools injoignable : timeout")

    monkeypatch.setitem(service._CLIENTS, "mangools", _Silent)
    response = await client.post(
        f"/integrations/{created['id']}/check", headers=headers(auth_headers, sandbox)
    )
    assert response.status_code == 200
    assert response.json()["status"] == "error"
    assert "injoignable" in response.json()["status_error"]


# --- plannings ---------------------------------------------------------------------------------


async def test_schedule_upsert_is_idempotent_and_sets_the_appointment(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    payload = {"site_target_id": str(sandbox.target_id), "sync_mode": "weekly"}

    first = await client.put(
        f"/integrations/{created['id']}/schedules",
        json=payload,
        headers=headers(auth_headers, sandbox),
    )
    assert first.status_code == 200
    body = first.json()
    assert body["sync_mode"] == "weekly"
    assert body["next_run_at"] is not None

    second = await client.put(
        f"/integrations/{created['id']}/schedules",
        json=payload,
        headers=headers(auth_headers, sandbox),
    )
    assert second.status_code == 200
    #  Même ligne, et **la même échéance** : réenregistrer ne repousse pas la synchro.
    assert second.json()["id"] == body["id"]
    assert second.json()["next_run_at"] == body["next_run_at"]


async def test_changing_the_cadence_keeps_enabled_and_config(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    """Changer de cadence ne remet à zéro ni l'interrupteur, ni les réglages de la cible.

    L'écran n'envoie que `sync_mode`. Si `enabled` et `config` avaient un défaut concret, chaque
    passage hebdomadaire → mensuel réactiverait silencieusement une synchro suspendue et
    effacerait ses réglages — un piège qui ne se déclenche que le jour où ces champs servent.
    """
    created = await _create(client, auth_headers, sandbox)
    await client.put(
        f"/integrations/{created['id']}/schedules",
        json={
            "site_target_id": str(sandbox.target_id),
            "sync_mode": "weekly",
            "enabled": False,
            "config": {"domain": "exemple.test"},
        },
        headers=headers(auth_headers, sandbox),
    )

    #  Le corps que l'écran envoie réellement : la seule cadence.
    changed = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "monthly"},
        headers=headers(auth_headers, sandbox),
    )
    assert changed.status_code == 200
    body = changed.json()
    assert body["sync_mode"] == "monthly"
    assert body["enabled"] is False
    assert body["config"] == {"domain": "exemple.test"}

    stored = await db.scalar(select(ImportSchedule).where(ImportSchedule.org_id == sandbox.org_id))
    assert stored is not None
    assert stored.enabled is False
    assert stored.config == {"domain": "exemple.test"}


async def test_switching_to_manual_clears_the_appointment(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "monthly"},
        headers=headers(auth_headers, sandbox),
    )
    back = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "manual"},
        headers=headers(auth_headers, sandbox),
    )
    assert back.status_code == 200
    assert back.json()["sync_mode"] == "manual"
    #  La contrainte de base garantit qu'un manuel n'a pas d'échéance fantôme.
    assert back.json()["next_run_at"] is None


async def test_schedule_on_an_unknown_target_is_a_404(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    response = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(uuid.uuid4()), "sync_mode": "weekly"},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 404


async def test_a_non_integer_location_id_is_refused_at_write(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    """`config.location_id` est validé **à l'écriture**, pas découvert par un job trois jours après.

    Sans ce 422, le `PUT` répondait 200 et la panne surgissait bien plus tard, dans le worker,
    sur un rendez-vous que personne ne regardait — et elle y bloquait le planificateur entier.
    """
    created = await _create(client, auth_headers, sandbox)
    for bad in ("abc", 12.5, True, [], {"x": 1}):
        response = await client.put(
            f"/integrations/{created['id']}/schedules",
            json={
                "site_target_id": str(sandbox.target_id),
                "sync_mode": "weekly",
                "config": {"location_id": bad},
            },
            headers=headers(auth_headers, sandbox),
        )
        assert response.status_code == 422, f"{bad!r} aurait dû être refusé"
        assert "location_id" in response.text


async def test_an_integer_location_id_is_accepted(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    response = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={
            "site_target_id": str(sandbox.target_id),
            "sync_mode": "weekly",
            "config": {"location_id": 2250, "domain": "exemple.test"},
        },
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 200
    assert response.json()["config"] == {"location_id": 2250, "domain": "exemple.test"}


async def test_a_non_integer_location_id_is_refused_on_the_integration_too(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    """Le `config` de l'intégration alimente le même `int()` : même validation, même 422."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.patch(
        f"/integrations/{created['id']}",
        json={"config": {"location_id": "abc"}},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422
    assert "location_id" in response.text

    refused = await client.post(
        "/integrations",
        json={"provider": "mangools", "api_key": OTHER_KEY, "config": {"location_id": "abc"}},
        headers=auth_headers(f"org_{uuid.uuid4().hex}"),
    )
    assert refused.status_code == 422


async def test_invalid_sync_mode_is_refused(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    response = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "daily"},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422


async def test_listing_schedules_carries_the_last_report(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    created = await _create(client, auth_headers, sandbox)
    await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "weekly"},
        headers=headers(auth_headers, sandbox),
    )
    #  Un import terminé, comme le worker l'aurait laissé.
    trigger = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(sandbox.target_id)},
        headers=headers(auth_headers, sandbox),
    )
    assert trigger.status_code == 202
    entry = await db.scalar(
        select(Import).where(Import.id == uuid.UUID(trigger.json()["items"][0]["import_id"]))
    )
    assert entry is not None
    entry.status = "done"
    entry.report = {"rows_read": 12, "rows_inserted": 12}
    await db.commit()

    listing = await client.get(
        f"/integrations/{created['id']}/schedules", headers=headers(auth_headers, sandbox)
    )
    assert listing.status_code == 200
    items = listing.json()["items"]
    assert len(items) == 1
    assert items[0]["last_import"]["status"] == "done"
    assert items[0]["last_import"]["report"]["rows_read"] == 12


async def test_the_last_report_is_reachable_without_any_schedule(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    """Décision « pas de synchro récurrente » (2026-09-02) : plus aucun planning n'est créé.

    L'écran ne pose plus de cadence, donc plus de ligne `import_schedules` — et le dernier
    rapport doit rester visible. Passer par `/schedules` le cacherait : une cible sans
    planning n'y figure pas, et l'écran afficherait « aucun import » juste après en avoir
    lancé un.
    """
    created = await _create(client, auth_headers, sandbox)
    trigger = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(sandbox.target_id)},
        headers=headers(auth_headers, sandbox),
    )
    assert trigger.status_code == 202
    entry = await db.scalar(
        select(Import).where(Import.id == uuid.UUID(trigger.json()["items"][0]["import_id"]))
    )
    assert entry is not None
    entry.status = "done"
    entry.report = {"rows_read": 7, "rows_inserted": 7}
    await db.commit()

    #  Aucun `PUT /schedules` n'a eu lieu : la liste des plannings est vide…
    schedules = await client.get(
        f"/integrations/{created['id']}/schedules", headers=headers(auth_headers, sandbox)
    )
    assert schedules.json()["items"] == []

    #  …et le rapport est là quand même.
    listing = await client.get(
        f"/integrations/{created['id']}/imports", headers=headers(auth_headers, sandbox)
    )
    assert listing.status_code == 200
    items = listing.json()["items"]
    assert len(items) == 1
    assert items[0]["site_target_id"] == str(sandbox.target_id)
    assert items[0]["last_import"]["report"]["rows_read"] == 7


async def test_last_reports_stay_inside_the_organisation(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, other_sandbox: Fixture
) -> None:
    """Deux organisations, deux listings — c'est le test que `target_ids=None` rend nécessaire.

    Le listing ne reçoit plus de liste de cibles à filtrer : il balaie **tous** les imports API
    de l'organisation. Le seul garde-fou est donc le `WHERE org_id`, et un test qui se
    contenterait du 404 sur l'intégration d'autrui ne le prouverait pas — il prouverait la
    résolution de l'intégration, pas le contenu du listing.
    """
    mine = await _create(client, auth_headers, sandbox)
    theirs = await _create(client, auth_headers, other_sandbox, key=OTHER_KEY)
    for owner, integration in ((sandbox, mine), (other_sandbox, theirs)):
        triggered = await client.post(
            f"/integrations/{integration['id']}/import",
            json={"site_target_id": str(owner.target_id)},
            headers=headers(auth_headers, owner),
        )
        assert triggered.status_code == 202

    listing = await client.get(
        f"/integrations/{mine['id']}/imports", headers=headers(auth_headers, sandbox)
    )
    assert listing.status_code == 200
    seen = {item["site_target_id"] for item in listing.json()["items"]}
    assert seen == {str(sandbox.target_id)}
    assert str(other_sandbox.target_id) not in seen

    #  Et l'intégration d'autrui n'existe pas : 404, jamais 403.
    other = await client.get(
        f"/integrations/{theirs['id']}/imports", headers=headers(auth_headers, sandbox)
    )
    assert other.status_code == 404


async def test_delete_schedule(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    created = await _create(client, auth_headers, sandbox)
    posted = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "weekly"},
        headers=headers(auth_headers, sandbox),
    )
    removed = await client.delete(
        f"/integrations/{created['id']}/schedules/{posted.json()['id']}",
        headers=headers(auth_headers, sandbox),
    )
    assert removed.status_code == 204
    left = (
        await db.scalars(select(ImportSchedule).where(ImportSchedule.org_id == sandbox.org_id))
    ).all()
    assert list(left) == []


# --- import manuel -------------------------------------------------------------------------


async def test_trigger_creates_an_import_and_a_job(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(sandbox.target_id)},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 202
    body = response.json()
    assert len(body["items"]) == 1
    assert body["items"][0]["created"] is True

    job = await db.scalar(select(Job).where(Job.id == uuid.UUID(body["items"][0]["job_id"])))
    assert job is not None
    assert job.type == JOB_TYPE
    assert job.status == "queued"
    #  La clé n'entre jamais dans le payload d'un job, lisible par `GET /jobs/{id}`.
    assert KEY not in str(job.payload)

    entry = await db.scalar(
        select(Import).where(Import.id == uuid.UUID(body["items"][0]["import_id"]))
    )
    assert entry is not None
    assert entry.kind == "api"
    assert entry.source == "mangools"
    assert entry.file_sha256 is None


async def test_trigger_twice_returns_the_same_import(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    payload = {"site_target_id": str(sandbox.target_id)}
    first = await client.post(
        f"/integrations/{created['id']}/import",
        json=payload,
        headers=headers(auth_headers, sandbox),
    )
    second = await client.post(
        f"/integrations/{created['id']}/import",
        json=payload,
        headers=headers(auth_headers, sandbox),
    )
    assert second.status_code == 202
    assert second.json()["items"][0]["created"] is False
    assert second.json()["items"][0]["import_id"] == first.json()["items"][0]["import_id"]


async def test_trigger_on_another_org_target_is_a_404(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(uuid.uuid4())},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 404


# --- le mode de l'import : découverte ou enrichissement ---------------------------------------


async def test_a_trigger_without_a_mode_still_discovers_the_site_domain(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    """Le contrat d'avant le mode reste valide : c'est ce qu'envoie l'écran Intégrations."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(sandbox.target_id)},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 202
    job = await db.scalar(
        select(Job).where(Job.id == uuid.UUID(response.json()["items"][0]["job_id"]))
    )
    assert job is not None
    assert job.payload["domain"]
    assert "keywords" not in job.payload


async def test_an_enrichment_is_refused_because_it_has_nowhere_to_land(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, db: AsyncSession
) -> None:
    """L'enrichissement par mots-clés n'a **pas de liste de destination**, donc pas d'import.

    Un mot-clé n'est visible dans le réservoir que s'il appartient à une liste de la cible ou
    porte une attache d'URL. L'enrichissement ne produit ni l'une ni l'autre et n'a aucun nom
    stable à donner à une liste — la découverte a le domaine, le mode liste a le nom KWFinder.
    L'enfiler écrirait des mots-clés que rien ne rattache à la cible, c'est-à-dire invisibles
    pour toujours (bug du 2026-09-02). Refus à l'entrée, avec le motif.
    """
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={
            "site_target_id": str(sandbox.target_id),
            #  Doublon et case vide compris : c'est ce que produit un collage humain.
            "keywords": ["chaise de bureau", " fauteuil ", "chaise de bureau", "  "],
        },
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422
    assert response.headers["content-type"].startswith("application/problem+json")
    #  Rien n'a été enfilé : aucun job d'import API.
    assert await db.scalar(select(func.count()).select_from(Job).where(Job.type == JOB_TYPE)) == 0


async def test_a_trigger_refuses_to_be_both_modes_at_once(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    """L'ambiguïté est refusée à l'entrée, pas arbitrée par l'ordre des `if` du service."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={
            "site_target_id": str(sandbox.target_id),
            "domain": "concurrent.example",
            "keywords": ["chaise"],
        },
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422
    assert response.headers["content-type"].startswith("application/problem+json")


async def test_a_list_of_nothing_is_not_an_enrichment(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    """Une liste vide ferait un appel payé pour rien : elle est refusée avant tout enfilage."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(sandbox.target_id), "keywords": ["   ", ""]},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422


# --- le quota du récapitulatif ------------------------------------------------------------


class _WithQuota:
    """Bouchon rendant la forme réelle de `/kwfinder/limits`. Aucun appel réseau."""

    def __init__(self, api_key: str, *, timeout: float | None = None) -> None:
        self.api_key = api_key

    async def check(self) -> dict:
        return {"resources": {"related-keywords": {"limit": 1200, "remaining": 1192}}}


async def test_quota_reports_what_is_left_of_the_resource_that_counts(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setitem(service._CLIENTS, "mangools", _WithQuota)

    response = await client.get(
        f"/integrations/{created['id']}/quota", headers=headers(auth_headers, sandbox)
    )
    assert response.status_code == 200
    body = response.json()
    assert body["resource"] == mangools_api.QUOTA_RESOURCE
    assert body["limit"] == 1200
    assert body["remaining"] == 1192
    assert body["error"] is None
    assert KEY not in response.text


async def test_an_unreachable_provider_makes_the_quota_unknown_not_an_error(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Le récapitulatif doit pouvoir dire « quota indisponible » et laisser lancer l'import."""
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setitem(service._CLIENTS, "mangools", _Refusing)

    response = await client.get(
        f"/integrations/{created['id']}/quota", headers=headers(auth_headers, sandbox)
    )
    assert response.status_code == 200
    body = response.json()
    assert body["remaining"] is None
    assert "refus" in body["error"].lower()


async def test_reading_the_quota_does_not_touch_the_status(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Un `GET` ne repositionne rien : le statut de la clé a son propre endpoint."""
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setitem(service._CLIENTS, "mangools", _Refusing)

    await client.get(f"/integrations/{created['id']}/quota", headers=headers(auth_headers, sandbox))
    detail = await client.get(
        f"/integrations/{created['id']}", headers=headers(auth_headers, sandbox)
    )
    assert detail.json()["status"] == "unverified"
    assert detail.json()["checked_at"] is None


async def test_the_quota_of_another_org_does_not_exist(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    stranger = auth_headers(f"org_{uuid.uuid4().hex}")
    response = await client.get(f"/integrations/{created['id']}/quota", headers=stranger)
    #  404 et non 403 : l'intégration d'une autre organisation n'existe pas — et le quota du
    #  fournisseur n'est donc jamais interrogé avec sa clé.
    assert response.status_code == 404


async def test_next_run_at_is_a_week_ahead(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture
) -> None:
    created = await _create(client, auth_headers, sandbox)
    posted = await client.put(
        f"/integrations/{created['id']}/schedules",
        json={"site_target_id": str(sandbox.target_id), "sync_mode": "weekly"},
        headers=headers(auth_headers, sandbox),
    )
    due = datetime.fromisoformat(posted.json()["next_run_at"])
    #  Poser une cadence ne déclenche pas d'import immédiat : le quota du client est à lui.
    assert due > datetime.now(UTC) + timedelta(days=6)


# --- les listes enregistrées du compte, à la frontière HTTP -------------------------------------
#
# Ce que ces tests défendent, c'est la **forme du contrat** — celle que recopie en Zod le module
# `web/lib/integrations.ts`. Le comportement de l'adaptateur et du cloisonnement pays est prouvé
# ailleurs (`tests/test_mangools_api_adapter.py`), sur les extraits de réponses réelles.

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

LIST_ID = "6a589821eefecfa46ad01f2e"


def _lists_stub() -> Any:
    """Un fournisseur bouchonné : le listing du compte et le détail d'une liste, sans réseau."""
    lists = json.loads((FIXTURES / "lists.json").read_text())
    detail = json.loads((FIXTURES / "list_detail.json").read_text())

    def handler(request: Any) -> Any:
        if f"{mangools_api.LISTS_PATH}/" in request.url.path:
            return httpx.Response(200, json=detail)
        return httpx.Response(200, json=lists)

    return mangools_api.MangoolsApi(
        KEY,
        client=AsyncClient(transport=httpx.MockTransport(handler)),
        min_request_interval=0.0,
    )


@pytest.fixture
def stubbed_provider(monkeypatch: pytest.MonkeyPatch) -> None:
    stub = _lists_stub()
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})


async def test_the_account_lists_are_listed_with_their_size_and_dates(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    """L'étape 1 du wizard : nom, nombre de mots-clés, dernière mise à jour."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.get(
        f"/integrations/{created['id']}/lists", headers=headers(auth_headers, sandbox)
    )
    assert response.status_code == 200, response.text
    items = response.json()["items"]
    assert [item["name"] for item in items] == ["Liste courte", "Liste France", "Liste vide"]
    assert items[0]["id"] == LIST_ID
    assert items[0]["keyword_count"] == 3
    #  Une date ISO exploitable par l'écran, pas un entier de millisecondes brut.
    assert datetime.fromisoformat(items[0]["updated_at"]).year == 2026


async def test_the_lists_of_another_org_do_not_exist(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    created = await _create(client, auth_headers, sandbox)
    stranger = auth_headers(f"org_{uuid.uuid4().hex}")
    response = await client.get(f"/integrations/{created['id']}/lists", headers=stranger)
    #  404 et non 403 : l'intégration d'une autre organisation n'existe pas — sa clé n'est donc
    #  jamais présentée au fournisseur.
    assert response.status_code == 404


async def test_the_preview_returns_the_markets_of_the_selection(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    """Le recensement, par liste et en agrégat — même forme que `PreviewOut.countries`."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": [LIST_ID], "site_target_id": str(sandbox.target_id)},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 200, response.text
    body = response.json()
    assert [item["list_id"] for item in body["items"]] == [LIST_ID]
    assert body["items"][0]["keywords_total"] == 3
    assert [(entry["raw"], entry["code"]) for entry in body["countries"]] == [
        ("France", "FR"),
        ("New York", None),
        ("United States", "US"),
    ]
    #  La cible du bac à sable est française : elle est proposée pour la France.
    assert body["countries"][0]["target_ids"] == [str(sandbox.target_id)]


async def test_a_preview_without_any_list_is_a_422(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": []},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422


async def test_reading_a_list_gets_the_long_timeout_not_the_key_check_one(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """**Mesuré le 2026-09-02** : la plus grosse liste du compte met 17,68 s à revenir.

    Le recensement empruntait le plafond de la vérification de clé (6 s) : cette liste-là — 2106
    mots-clés, 1,9 Mo — était coupée avant la fin et rendue « injoignable », donc impossible à
    importer. Le plafond du détail est celui d'une lecture, pas celui d'un ping.
    """
    created = await _create(client, auth_headers, sandbox)
    seen: list[float | None] = []
    stub = _lists_stub()

    def build(api_key: str, **kwargs: Any) -> Any:
        seen.append(kwargs.get("timeout"))
        return stub

    monkeypatch.setattr(service, "_CLIENTS", {"mangools": build})
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": [LIST_ID]},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 200, response.text
    #  Le client qui ouvre les listes a le plafond long ; celui du listing du compte garde le
    #  court, parce qu'il répond derrière un `apiFetch` qui abandonne à dix secondes.
    assert service.LIST_TIMEOUT_SECONDS in seen
    assert service.LIST_TIMEOUT_SECONDS > 17.68
    assert service.CHECK_TIMEOUT_SECONDS in seen


async def test_a_census_that_overruns_its_budget_says_to_pick_fewer_lists(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Le budget est **global à la sélection**, pas par liste — et il répond avant le web.

    `LIST_TIMEOUT_SECONDS` borne *une* lecture ; quatre listes de la taille mesurée (17,7 s
    chacune, espacées du débit) font ~81 s, soit plus que les 60 s qu'attend la Server Action.
    Elle abandonnait alors sur « réessayez » pendant que l'api continuait à brûler le débit, et
    le réessai prenait des 429 : un dépassement déterministe que rien n'expliquait. L'api rend
    maintenant un 504 qui dit quoi faire.
    """

    class _Slow:
        def __init__(self, api_key: str, **kwargs: Any) -> None:
            self.api_key = api_key

        async def lists(self) -> list[Any]:
            return mangools_api.parse_lists(json.loads((FIXTURES / "lists.json").read_text()))

        async def list_keywords(self, list_id: str) -> list[Any]:
            await asyncio.sleep(5)  # jamais atteint : le budget est ramené à ~0 juste en dessous
            return []

    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": _Slow})
    #  Budget réduit au strict minimum : on teste le mécanisme, pas la patience.
    monkeypatch.setattr(service, "PREVIEW_BUDGET_SECONDS", 0.05)
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": [LIST_ID]},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 504
    assert response.headers["content-type"].startswith("application/problem+json")
    assert "cochez moins de listes" in response.json()["detail"]


async def test_a_refused_key_at_click_time_queues_nothing_at_all(
    client: AsyncClient,
    auth_headers: Any,
    sandbox: Fixture,
    db: AsyncSession,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Une clé refusée **au moment du clic** refuse le déclenchement, elle ne l'enfile pas.

    Avant, `_list_names` ravalait l'erreur et enfilait « sans nom » : N imports partaient, tous
    échouaient un par un dans le worker, et l'utilisateur découvrait N lignes rouges au lieu
    d'un message. Depuis que le nom **nomme la liste de destination**, la tolérance serait pire
    encore — elle fabriquerait des listes « Liste <id> » à côté de celles qu'il reconnaît.
    """

    class _RefusingLists:
        def __init__(self, api_key: str, **kwargs: Any) -> None:
            self.api_key = api_key

        async def lists(self) -> list[Any]:
            raise mangools_api.MangoolsApiError("Clé Mangools refusée par le fournisseur (401).")

    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": _RefusingLists})
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={"site_target_id": str(sandbox.target_id), "list_ids": [LIST_ID]},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 502
    assert response.headers["content-type"].startswith("application/problem+json")
    #  La clé ne repart jamais dans une réponse, même en relayant un refus du fournisseur.
    assert KEY not in response.text
    assert await db.scalar(select(func.count()).select_from(Job).where(Job.type == JOB_TYPE)) == 0


async def test_replacing_the_key_forgets_the_lists_of_the_previous_account(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Le listing du compte est mémorisé cinq minutes — **pas au travers d'un changement de clé**.

    Sans l'oubli, une clé remplacée continuerait de proposer les listes de l'ancien compte, et
    un import lancé dessus prendrait un 404 chez le fournisseur.
    """
    created = await _create(client, auth_headers, sandbox)
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: _lists_stub()})
    first = await client.get(
        f"/integrations/{created['id']}/lists", headers=headers(auth_headers, sandbox)
    )
    assert first.status_code == 200
    assert service._LISTS_CACHE.get(uuid.UUID(created["id"])) is not None

    patched = await client.patch(
        f"/integrations/{created['id']}",
        json={"api_key": OTHER_KEY},
        headers=headers(auth_headers, sandbox),
    )
    assert patched.status_code == 200
    assert service._LISTS_CACHE.get(uuid.UUID(created["id"])) is None


async def test_a_preview_on_another_org_integration_does_not_exist(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    """404 et non 403 : la clé d'une autre organisation n'est jamais présentée au fournisseur."""
    created = await _create(client, auth_headers, sandbox)
    stranger = auth_headers(f"org_{uuid.uuid4().hex}")
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": [LIST_ID]},
        headers=stranger,
    )
    assert response.status_code == 404


async def test_a_preview_of_an_unknown_list_is_a_404_at_the_boundary(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    """Un identifiant que le compte ne connaît pas est refusé **ici**, pas dans le worker."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": ["6a0000000000000000000000"]},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 404
    assert response.headers["content-type"].startswith("application/problem+json")


async def test_a_provider_failure_is_a_502_and_never_echoes_its_body(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Un tiers en panne est un 502 lisible — **jamais** son corps, qui peut renvoyer la clé.

    Le corps d'une erreur Mangools renvoie la requête en écho, en-têtes compris : le recopier
    remettrait la clé dans une réponse HTTP, puis dans `imports.error`, que l'écran affiche.
    """
    created = await _create(client, auth_headers, sandbox)
    stub = mangools_api.MangoolsApi(
        KEY,
        client=AsyncClient(
            transport=httpx.MockTransport(
                lambda request: httpx.Response(
                    500, json={"sent": {"headers": {"x-access-token": KEY}}}
                )
            )
        ),
        min_request_interval=0.0,
    )
    monkeypatch.setattr(service, "_CLIENTS", {"mangools": lambda api_key, **kwargs: stub})
    response = await client.post(
        f"/integrations/{created['id']}/lists/preview",
        json={"list_ids": [LIST_ID]},
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 502
    assert response.headers["content-type"].startswith("application/problem+json")
    assert KEY not in response.text


async def test_a_trigger_can_name_the_lists_to_import(
    client: AsyncClient,
    auth_headers: Any,
    sandbox: Fixture,
    db: AsyncSession,
    stubbed_provider: None,
) -> None:
    """Deux listes = deux imports, chacun avec son nom : l'onglet Imports les distingue."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={
            "site_target_id": str(sandbox.target_id),
            #  Doublon compris : c'est ce que produit une case cochée deux fois.
            "list_ids": [LIST_ID, "6a6986b7a5359704ad73709e", LIST_ID],
        },
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 202, response.text
    items = response.json()["items"]
    assert len(items) == 2
    assert [item["list_name"] for item in items] == ["Liste courte", "Liste France"]
    assert all(item["created"] for item in items)

    job = await db.scalar(select(Job).where(Job.id == uuid.UUID(items[0]["job_id"])))
    assert job is not None
    assert job.payload["list_id"] == LIST_ID
    assert job.payload["domain"] == ""
    #  La clé n'entre jamais dans le payload d'un job, lisible par `GET /jobs/{id}`.
    assert KEY not in str(job.payload)


async def test_a_trigger_refuses_a_list_and_a_domain_at_once(
    client: AsyncClient, auth_headers: Any, sandbox: Fixture, stubbed_provider: None
) -> None:
    """L'ambiguïté est refusée à l'entrée, pas arbitrée par l'ordre des `if` du service."""
    created = await _create(client, auth_headers, sandbox)
    response = await client.post(
        f"/integrations/{created['id']}/import",
        json={
            "site_target_id": str(sandbox.target_id),
            "list_ids": [LIST_ID],
            "domain": "concurrent.example",
        },
        headers=headers(auth_headers, sandbox),
    )
    assert response.status_code == 422