"""API du module Imports : upload, aperçu, création, listings (spec Imports §6, T3).

Ce qui est éprouvé ici, dans l'ordre du tableau §6 :

- l'**upload direct** (décision n°8) — spool disque, sha256 au vol, plafond, dedup soft, et le
  CORS strictement limité à ce seul chemin (le vérifier *en négatif* sur un autre endpoint est
  la moitié utile du test : un middleware global passerait le test positif sans broncher) ;
- l'**aperçu** — détection par signature, lignes canoniques, avertissements, et le fait que
  rien n'est écrit en base ;
- la **création** — payload contractuel du job T2, ligne `imports`, liste de destination,
  409 de doublon franchissable par `force` ;
- les **listings** — curseur UUIDv7, entrelacement des événements `map_reconcile`, isolation
  par organisation (404, jamais 403).
"""

import hashlib
import io
import uuid
from collections.abc import AsyncIterator
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

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

from app.core.errors import AppError
from app.imports import service, upload
from app.jobs import service as jobs_service
from app.jobs.models import Job
from app.keywords import pipeline, sources
from app.keywords.models import Import, KeywordList
from app.sites.models import SiteTarget
from tests.conftest import Fixture

FIXTURES = Path(__file__).parent / "fixtures"
POSITIONS_EN = FIXTURES / "semrush_positions_en.csv"
# Export réel Mangools KWFinder : reconnu par l'adaptateur `mangools` (T7), et — format forcé
# à « CSV générique » — le fichier qui éprouve pour de bon la proposition automatique (§5).
KWFINDER = FIXTURES / "real" / "kwfinder_usa_top.csv"
# Le même export, mais multi-marchés (onze pays et une localité) : c'est lui qui éprouve le
# recensement des pays, parce que six de ses marchés n'apparaissent qu'après la 100ᵉ ligne.
MULTI_COUNTRY = FIXTURES / "real" / "kwfinder_multi_country.csv"
# Export réel Keyword Magic : aucune colonne de pays, la source est **muette**.
SEMRUSH_MAGIC = FIXTURES / "real" / "semrush_magic_us.csv"
FETCHED_AT = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)

Headers = dict[str, str]


@pytest_asyncio.fixture(autouse=True)
async def _clean_jobs(db: AsyncSession) -> AsyncIterator[None]:
    """Les jobs enfilés ici ne sont jamais exécutés : aucun worker ne tourne dans ce module.

    Ils sont purgés quand même, pour qu'un test d'un autre module qui lance un worker ne
    ramasse pas le job d'un import dont le fichier a disparu avec le `tmp_path`.
    """
    yield
    leftovers = select(Job.id).where(Job.type == pipeline.JOB_TYPE)
    await db.execute(delete(Import).where(Import.job_id.in_(leftovers)))
    await db.execute(delete(Job).where(Job.type == pipeline.JOB_TYPE))
    await db.commit()


@pytest.fixture
def headers(auth_headers: Any, sandbox: Fixture) -> Headers:
    return auth_headers(sandbox.org_id)


@pytest.fixture
def content() -> bytes:
    return POSITIONS_EN.read_bytes()


@pytest.fixture
def sha(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


async def _upload(
    client: AsyncClient,
    headers: Headers,
    content: bytes,
    *,
    filename: str = "positions.csv",
    site_target_id: uuid.UUID | None = None,
) -> Any:
    data = {} if site_target_id is None else {"site_target_id": str(site_target_id)}
    return await client.post(
        "/imports/files",
        headers=headers,
        files={"file": (filename, content, "text/csv")},
        data=data,
    )


async def _create(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    sha: str,
    **overrides: Any,
) -> Any:
    body: dict[str, Any] = {
        "sha256": sha,
        "format": "semrush_positions",
        "site_target_id": str(sandbox.target_id),
        "list": {"name": "Semrush — 2026-08-30"},
        "fetched_at": FETCHED_AT.isoformat(),
        "filename": "positions.csv",
    }
    body.update(overrides)
    return await client.post("/imports", headers=headers, json=body)


# --- POST /imports/files -------------------------------------------------------------------


async def test_upload_spools_and_hashes(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    response = await _upload(client, headers, content)
    assert response.status_code == 201
    body = response.json()
    assert body == {
        "sha256": sha,
        "filename": "positions.csv",
        "size": len(content),
        "duplicate_of": None,
    }
    spooled = import_dir / sandbox.org_id / f"{sha}.csv"
    assert spooled.read_bytes() == content
    # Aucun `.part` abandonné : le renommage est atomique, le répertoire ne contient que le
    # fichier définitif.
    assert [p.name for p in spooled.parent.iterdir()] == [f"{sha}.csv"]


async def test_upload_filename_is_never_a_path(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    response = await _upload(client, headers, content, filename="../../etc/passwd")
    assert response.status_code == 201
    assert response.json()["filename"] == "passwd"
    assert (import_dir / sandbox.org_id / f"{sha}.csv").is_file()


async def test_upload_reports_duplicate_for_same_target(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    assert (await _upload(client, headers, content, site_target_id=sandbox.target_id)).json()[
        "duplicate_of"
    ] is None
    created = await _create(client, headers, sandbox, sha)
    assert created.status_code == 202
    import_id = created.json()["import"]["id"]

    again = await _upload(client, headers, content, site_target_id=sandbox.target_id)
    assert again.json()["duplicate_of"] == import_id
    # Sans cible, il n'y a pas de doublon à signaler : la déduplication est « même sha, même
    # cible » (§1.3), jamais « même sha » tout court.
    assert (await _upload(client, headers, content)).json()["duplicate_of"] is None


async def test_upload_rejects_unknown_target(
    client: AsyncClient, headers: Headers, import_dir: Path, content: bytes
) -> None:
    response = await _upload(client, headers, content, site_target_id=uuid.uuid4())
    assert response.status_code == 404


async def test_upload_requires_a_token(
    client: AsyncClient, import_dir: Path, content: bytes
) -> None:
    response = await client.post("/imports/files", files={"file": ("a.csv", content, "text/csv")})
    assert response.status_code == 401


# --- garde ASGI : le corps est refusé AVANT le parsing multipart et avant le JWT -------------
#
# FastAPI lit la requête (`await request.form()`) avant d'appeler l'endpoint et avant de résoudre
# `CurrentOrgDep` : aucun contrôle écrit dans le handler ne peut donc empêcher la réception. Ces
# tests visent le middleware `UploadSizeGuard`, et vérifient à chaque fois **deux** choses — le
# code de statut, et le fait que le handler n'a jamais tourné (`upload.spool` jamais appelé,
# `IMPORT_FILES_DIR` intact).


@pytest.fixture
def never_spooled(monkeypatch: pytest.MonkeyPatch) -> list[str]:
    """Espionne `upload.spool` : s'il est appelé, c'est que la requête a atteint le handler."""
    calls: list[str] = []
    original = upload.spool

    async def _spy(file: Any, org_id: str) -> tuple[str, int]:
        calls.append(org_id)
        return await original(file, org_id)

    monkeypatch.setattr(upload, "spool", _spy)
    return calls


def _multipart(content: bytes, boundary: bytes = b"seoinsightBOUNDARY") -> tuple[bytes, str]:
    """Un corps multipart écrit à la main : c'est le seul moyen d'en contrôler le découpage."""
    body = (
        b"--" + boundary + b"\r\n"
        b'Content-Disposition: form-data; name="file"; filename="gros.csv"\r\n'
        b"Content-Type: text/csv\r\n\r\n" + content + b"\r\n"
        b"--" + boundary + b"--\r\n"
    )
    return body, f"multipart/form-data; boundary={boundary.decode()}"


def _entries(path: Path) -> list[str]:
    """Le contenu d'un répertoire — fonction synchrone pour rester lisible en test async."""
    return sorted(entry.name for entry in path.iterdir())


async def _stream(body: bytes, chunk: int = 64) -> AsyncIterator[bytes]:
    for start in range(0, len(body), chunk):
        yield body[start : start + chunk]


async def test_guard_refuses_a_declared_oversize_body(
    client: AsyncClient,
    headers: Headers,
    import_dir: Path,
    never_spooled: list[str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`Content-Length` au-dessus du plafond : refus sans lire un octet du corps."""
    monkeypatch.setattr(upload, "MAX_UPLOAD_BYTES", 16)
    body, content_type = _multipart(b"x" * 1024)
    response = await client.post(
        "/imports/files",
        headers={**headers, "Content-Type": content_type, "Origin": "http://test-auth.local"},
        content=body,
    )
    assert response.status_code == 413
    assert response.headers["content-type"].startswith("application/problem+json")
    #  Socket fermée, pas recyclée : voir `test_guard_refuses_an_anonymous_body_before_reading_it`.
    assert response.headers["connection"] == "close"
    # Le garde est **intérieur** au CORS : un 413 reste lisible par le navigateur, sans quoi
    # l'utilisateur ne verrait qu'une erreur réseau opaque (ordre des middlewares de `main.py`).
    assert response.headers["access-control-allow-origin"] == "http://test-auth.local"
    assert never_spooled == []
    assert _entries(import_dir) == []


async def test_guard_refuses_a_lying_content_length(
    client: AsyncClient,
    headers: Headers,
    import_dir: Path,
    never_spooled: list[str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`Content-Length` est déclaratif : c'est le cumul des octets reçus qui fait foi.

    Le client annonce 10 octets — sous le plafond — puis en envoie plus de mille. Le garde ne
    peut pas s'en apercevoir à l'en-tête ; il le voit au troisième bloc et coupe.

    **Ce que ce test couvre vraiment.** Derrière un vrai serveur, ce cas précis n'atteint pas le
    garde : `httptools` tronque le corps au `Content-Length` annoncé et l'application ne voit
    que 10 octets. Le scénario reste utile — il exerce le compteur cumulatif sur un chemin où
    l'en-tête ment, ce qui est la situation d'un proxy en amont qui ré-emballerait la requête
    sans revérifier —, mais c'est de la **défense en profondeur**, pas la ligne de front. La
    ligne de front, c'est le test `chunked` ci-dessous : là, aucun en-tête ne borne quoi que ce
    soit et le cumul est le seul rempart.
    """
    monkeypatch.setattr(upload, "MAX_UPLOAD_BYTES", 128)
    body, content_type = _multipart(b"x" * 1024)
    response = await client.post(
        "/imports/files",
        headers={**headers, "Content-Type": content_type, "Content-Length": "10"},
        content=_stream(body),
    )
    assert response.status_code == 413
    assert never_spooled == []
    assert _entries(import_dir) == []


async def test_guard_refuses_a_chunked_body_without_content_length(
    client: AsyncClient,
    headers: Headers,
    import_dir: Path,
    never_spooled: list[str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`Transfer-Encoding: chunked` : aucun en-tête à contrôler, seul le cumul protège."""
    monkeypatch.setattr(upload, "MAX_UPLOAD_BYTES", 128)
    body, content_type = _multipart(b"x" * 1024)
    response = await client.post(
        "/imports/files",
        headers={**headers, "Content-Type": content_type},
        content=_stream(body),
    )
    assert response.status_code == 413
    assert never_spooled == []
    assert _entries(import_dir) == []


async def test_guard_refuses_an_anonymous_body_before_reading_it(
    client: AsyncClient, import_dir: Path, never_spooled: list[str], content: bytes
) -> None:
    """Sans en-tête `Authorization`, le corps n'est jamais reçu (scénario `curl` de 10 Go).

    Ce pré-filtre ne remplace pas l'authentification — `CurrentOrgDep` vérifie toujours la
    signature, l'`iss`, l'`aud` et l'organisation ; il refuse seulement ce qui est manifestement
    anonyme avant que ça ne coûte un disque.
    """
    body, content_type = _multipart(content)
    response = await client.post(
        "/imports/files", headers={"Content-Type": content_type}, content=_stream(body)
    )
    assert response.status_code == 401
    assert response.headers["www-authenticate"] == "Bearer"
    #  La socket est fermée plutôt que recyclée : sans cela, HTTP/1.1 fait drainer au serveur
    #  tout le corps restant *après* la réponse (mesuré sur uvicorn : 820 800 octets avalés).
    assert response.headers["connection"] == "close"
    assert never_spooled == []
    assert _entries(import_dir) == []


async def test_an_invalid_token_is_still_rejected_by_the_real_check(
    client: AsyncClient, import_dir: Path, never_spooled: list[str], content: bytes
) -> None:
    """Le pré-filtre du garde ne regarde que la **présence** de l'en-tête ; le jeton, lui, reste
    vérifié par `CurrentOrgDep` — signature, `iss`, `aud`, `exp`, organisation."""
    response = await _upload(
        client, {"Authorization": "Bearer pas-un-jeton"}, content, filename="x.csv"
    )
    assert response.status_code == 401
    assert never_spooled == []
    assert _entries(import_dir) == []


async def test_guard_lets_a_valid_upload_through(
    client: AsyncClient,
    headers: Headers,
    import_dir: Path,
    never_spooled: list[str],
    content: bytes,
) -> None:
    """Le garde ne doit rien casser : le chemin nominal traverse, même en corps streamé."""
    body, content_type = _multipart(content)
    response = await client.post(
        "/imports/files",
        headers={**headers, "Content-Type": content_type},
        content=_stream(body),
    )
    assert response.status_code == 201
    assert response.json()["size"] == len(content)
    assert never_spooled != []


async def test_guard_ignores_every_other_path(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Le plafond est **scopé** : un corps plus gros que lui passe partout ailleurs."""
    monkeypatch.setattr(upload, "MAX_UPLOAD_BYTES", 4)
    response = await client.post("/imports/preview", headers=headers, json={"sha256": "0" * 64})
    #  404 (fichier inconnu), pas 413 : la requête a bien été routée et son corps lu.
    assert response.status_code == 404


def test_spool_refuses_oversize_stream(import_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Ceinture du module `upload` : `_spool` reste correct appelé hors de la route."""
    monkeypatch.setattr(upload, "MAX_UPLOAD_BYTES", 16)
    with pytest.raises(AppError) as caught:
        upload._spool(io.BytesIO(b"x" * 1024), "org_x")
    assert caught.value.status == 413
    # Rien ne reste sur le disque : le temporaire est supprimé sur le chemin d'erreur.
    assert list((import_dir / "org_x").iterdir()) == []


# --- CORS : ce chemin-là, et lui seul ------------------------------------------------------


async def test_cors_allows_the_browser_on_the_upload_only(client: AsyncClient) -> None:
    origin = "http://test-auth.local"
    preflight = await client.request(
        "OPTIONS",
        "/imports/files",
        headers={
            "Origin": origin,
            "Access-Control-Request-Method": "POST",
            "Access-Control-Request-Headers": "authorization",
        },
    )
    assert preflight.status_code == 200
    assert preflight.headers["access-control-allow-origin"] == origin
    # Non credentialisé : le jeton voyage dans `Authorization`, jamais dans un cookie.
    assert "access-control-allow-credentials" not in preflight.headers


async def test_cors_refuses_a_foreign_origin(client: AsyncClient) -> None:
    preflight = await client.request(
        "OPTIONS",
        "/imports/files",
        headers={"Origin": "https://evil.example", "Access-Control-Request-Method": "POST"},
    )
    assert "access-control-allow-origin" not in preflight.headers


@pytest.mark.parametrize(
    "path", ["/sites", "/keywords", "/jobs/00000000-0000-0000-0000-000000000000"]
)
async def test_no_cors_anywhere_else(client: AsyncClient, headers: Headers, path: str) -> None:
    """Le middleware est **scopé** : aucun autre endpoint ne devient appelable du navigateur."""
    origin = "http://test-auth.local"
    response = await client.get(path, headers={**headers, "Origin": origin})
    assert "access-control-allow-origin" not in response.headers
    preflight = await client.request(
        "OPTIONS", path, headers={"Origin": origin, "Access-Control-Request-Method": "GET"}
    )
    assert "access-control-allow-origin" not in preflight.headers


# --- POST /imports/preview -----------------------------------------------------------------


async def test_preview_detects_by_column_signature(
    client: AsyncClient,
    headers: Headers,
    import_dir: Path,
    content: bytes,
    sha: str,
    db: AsyncSession,
) -> None:
    await _upload(client, headers, content)
    response = await client.post("/imports/preview", headers=headers, json={"sha256": sha})
    assert response.status_code == 200
    body = response.json()
    assert body["format"] == "semrush_positions"
    assert body["language"] == "en"
    assert body["columns"]["keyword"] == "Keyword"
    assert [row["keyword"] for row in body["sample"]][:2] == [
        "chaussures rouges",
        "chaussures rouges",
    ]
    assert body["sample"][0]["url"] == "https://Exemple.FR/Chaussures?utm_source=nl"
    # « on n'échoue jamais en silence » : les colonnes non canoniques sont annoncées.
    ignored = next(w for w in body["warnings"] if w["code"] == "ignored_columns")
    assert "Traffic" in ignored["detail"]
    # Rien n'est persisté par l'aperçu (§6) : ni ligne `imports`, ni job.
    assert (await client.get("/imports", headers=headers)).json()["items"] == []
    assert await db.scalar(select(Job.id).where(Job.type == pipeline.JOB_TYPE)) is None


async def test_preview_proposes_generic_when_unrecognised(
    client: AsyncClient, headers: Headers, import_dir: Path, sha: str
) -> None:
    raw = b"Requete,Vol. mensuel\nchaussures,1300\nbottines,880\n"
    uploaded = (await _upload(client, headers, raw)).json()
    response = await client.post(
        "/imports/preview", headers=headers, json={"sha256": uploaded["sha256"]}
    )
    body = response.json()
    assert body["format"] == "generic_csv"
    assert body["confidence"] == 0.0
    assert body["headers"] == ["Requete", "Vol. mensuel"]
    # Aucun adaptateur ne reconnaît le fichier, mais le mapping est proposé **et appliqué** :
    # l'utilisateur voit des lignes canoniques, jamais un écran mort (décision 2026-09-01).
    assert body["suggested_mapping"] == {"keyword": "Requete", "volume": "Vol. mensuel"}
    assert [row["volume"] for row in body["sample"]] == [1300, 880]
    assert {w["code"] for w in body["warnings"]} >= {"ambiguous_format", "mapping_suggested"}
    assert "no_rows" not in {w["code"] for w in body["warnings"]}

    mapped = await client.post(
        "/imports/preview",
        headers=headers,
        json={
            "sha256": uploaded["sha256"],
            "format": "generic_csv",
            "columns": {"keyword": "Requete", "volume": "Vol. mensuel"},
        },
    )
    assert [row["volume"] for row in mapped.json()["sample"]] == [1300, 880]


async def test_preview_detects_the_real_kwfinder_export(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """Depuis T7, l'export Mangools a son adaptateur : plus de mapping à confirmer (§5).

    Le fichier passait par « CSV générique » et sa proposition automatique ; il est maintenant
    reconnu par ses colonnes, avec ce que seul l'adaptateur sait lire — la tendance, tirée des
    133 colonnes mensuelles, qu'aucun mapping manuel ne pouvait câbler.
    """
    uploaded = (await _upload(client, headers, KWFINDER.read_bytes())).json()
    response = await client.post(
        "/imports/preview", headers=headers, json={"sha256": uploaded["sha256"]}
    )
    body = response.json()
    assert body["format"] == "mangools"
    assert body["suggested_mapping"] is None  # rien à proposer : les colonnes sont connues
    assert {w["code"] for w in body["warnings"]}.isdisjoint(
        {"ambiguous_format", "mapping_suggested", "no_rows"}
    )
    first = body["sample"][0]
    assert (first["keyword"], first["volume"], first["kd"]) == ("manufacturer jewelry", 2300, 17.0)
    assert first["trend"] == [100, 43, 43, 43, 82, 66, 43, 82, 43, 43, 36, 36]
    #  `Location` est désormais lue et **résolue** : « United States » vaut `US`, nommé une
    #  fois par l'api. C'est la décision renversée par le cloisonnement — l'audit ne reste plus
    #  vide au motif que le libellé n'est pas déjà un code ISO.
    assert body["countries"] == [
        {
            "raw": "United States",
            "code": "US",
            "label": "États-Unis",
            "rows": 21,
            "suggested_language": "en",
            # Aucun `site_target_id` dans cette requête : l'api ne connaît pas le site.
            "target_ids": [],
        }
    ]


async def test_preview_maps_a_wide_export_by_itself(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """« Le mapping, nous on le fait » : 147 colonnes, zéro `Select` à câbler (§5).

    C'est le scénario du retour utilisateur : un export sans adaptateur tombait sur neuf champs
    « — aucune — ». Il rend maintenant des lignes canoniques dès le premier aperçu, et le piège
    du fichier — 133 volumes mensuels historiques — ne gagne pas. Le format est demandé
    explicitement parce que ce fichier-là a désormais son adaptateur : c'est la proposition
    automatique qu'on éprouve ici, pas la détection.
    """
    uploaded = (await _upload(client, headers, KWFINDER.read_bytes())).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={"sha256": uploaded["sha256"], "format": "generic_csv"},
    )
    body = response.json()
    assert body["format"] == "generic_csv"
    assert body["suggested_mapping"] == {
        "keyword": "Keyword",
        "volume": "Avg. Search Volume (Last 12 months)",
        "kd": "Keyword Difficulty",
        "cpc": "CPC/USD",
        "intent": "Search Intent",
        "serp_features": "Serp Features",
    }
    # L'aperçu est **recalculé avec la proposition** : des lignes, pas un écran mort.
    assert body["columns"] == body["suggested_mapping"]
    first = body["sample"][0]
    assert (first["keyword"], first["volume"], first["kd"]) == ("manufacturer jewelry", 2300, 17.0)
    assert first["intent"] == "commercial"
    assert first["serp_features"] == ["Map Pack", "People Also Ask"]
    # L'hésitation est dite, pas tranchée en silence : les trois autres moyennes sont nommées.
    ambiguous = next(w for w in body["warnings"] if w["code"] == "ambiguous_column")
    assert "Avg. Search Volume (Last 6 months)" in ambiguous["detail"]
    assert ambiguous["detail"].endswith("ne l'ont pas été.")
    #  `Keyword interest growth (%)` porte **une** croissance, pas les douze valeurs d'une
    #  sparkline : la proposer ferait une colonne vide sur toutes les lignes (arbitrage T7).
    assert "trend" not in body["suggested_mapping"]
    # 140 colonnes ignorées : l'énumération est tronquée, sinon l'avertissement est illisible.
    ignored = next(w for w in body["warnings"] if w["code"] == "ignored_columns")
    assert "… et " in ignored["detail"]


async def test_preview_says_when_no_mapping_can_be_proposed(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """Sans colonne mot-clé, il n'y a rien à proposer — et l'écran doit le dire clairement."""
    uploaded = (await _upload(client, headers, b"Volume,CPC (USD)\n1300,0.42\n")).json()
    response = await client.post(
        "/imports/preview", headers=headers, json={"sha256": uploaded["sha256"]}
    )
    body = response.json()
    assert body["suggested_mapping"] is None
    assert body["sample"] == []
    warning = next(w for w in body["warnings"] if w["code"] == "no_mapping_suggested")
    assert "mot-clé" in warning["detail"]


async def test_preview_never_overrides_an_explicit_mapping(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """L'utilisateur qui a tranché fait autorité : la proposition ne repasse pas derrière lui."""
    raw = b"Keyword,Volume,Search Volume\nchaussures,1300,42\n"
    uploaded = (await _upload(client, headers, raw)).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={
            "sha256": uploaded["sha256"],
            "format": "generic_csv",
            "columns": {"keyword": "Keyword", "volume": "Search Volume"},
        },
    )
    body = response.json()
    assert body["columns"] == {"keyword": "Keyword", "volume": "Search Volume"}
    assert [row["volume"] for row in body["sample"]] == [42]
    assert not any(w["code"] == "mapping_suggested" for w in body["warnings"])


async def test_preview_treats_an_empty_mapping_as_a_choice(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """`columns: {}` est un mapping vide **choisi**, pas une absence de choix.

    Le confondre avec l'absence appliquait la proposition tout en taisant l'avertissement qui
    l'annonce : un mapping appliqué sans être dit. Injoignable depuis le wizard, atteignable en
    appel direct — donc à tenir.
    """
    uploaded = (await _upload(client, headers, b"Keyword,Volume\nchaussures,1300\n")).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={"sha256": uploaded["sha256"], "format": "generic_csv", "columns": {}},
    )
    body = response.json()
    assert body["columns"] == {}
    assert body["sample"] == []
    assert not any(w["code"] == "mapping_suggested" for w in body["warnings"])


async def test_preview_rejects_a_bad_mapping(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    uploaded = (await _upload(client, headers, b"A,B\n1,2\n")).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={
            "sha256": uploaded["sha256"],
            "format": "generic_csv",
            "columns": {"keyword": "Introuvable"},
        },
    )
    assert response.status_code == 422
    assert "Introuvable" in response.json()["detail"]


async def test_preview_censuses_every_country_of_the_file(
    client: AsyncClient, headers: Headers, sandbox: Fixture, import_dir: Path, db: AsyncSession
) -> None:
    """Le recensement porte sur le fichier **entier**, pas sur l'aperçu (§4.2).

    C'est tout l'enjeu : l'aperçu ne lit que vingt lignes et la création cinq, alors que six
    des marchés de cet export n'apparaissent qu'après la centième. Un recensement tronqué
    proposerait de cloisonner sur cinq pays au lieu de onze, sans que rien ne le dise.

    Le site possède déjà une cible `en-GB` : elle est rendue dans `target_ids` du seul pays
    `GB`, pour que le wizard sache quoi réutiliser et quoi créer.
    """
    british = SiteTarget(
        site_id=sandbox.site_id, org_id=sandbox.org_id, country="GB", language="en"
    )
    db.add(british)
    await db.commit()

    uploaded = (await _upload(client, headers, MULTI_COUNTRY.read_bytes())).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={"sha256": uploaded["sha256"], "site_target_id": str(sandbox.target_id)},
    )
    body = response.json()
    assert body["format"] == "mangools"
    #  Trié par nombre de lignes décroissant : l'écran met le marché principal en tête.
    assert [(c["raw"], c["code"], c["rows"]) for c in body["countries"]] == [
        ("United States", "US", 74),
        ("United Kingdom", "GB", 55),
        ("Australia", "AU", 32),
        ("Canada", "CA", 13),
        ("Singapore", "SG", 12),
        ("Hong Kong", "HK", 11),
        ("Germany", "DE", 10),
        ("New Zealand", "NZ", 8),
        ("Spain", "ES", 8),
        ("New York", None, 3),
        ("Israel", "IL", 2),
        ("Japan", "JP", 1),
    ]
    by_code = {c["code"]: c for c in body["countries"]}
    #  Le pays est **nommé par l'api**, en français, une seule fois : le web n'a pas de table
    #  de pays à tenir (elle n'en connaissait que 30, sans AU, SG, HK, NZ, IL ni JP).
    assert by_code["DE"]["label"] == "Allemagne"
    assert by_code["JP"]["label"] == "Japon"
    #  Langue **proposée**, jamais imposée : `he` pour Israël, `ja` pour le Japon.
    assert by_code["IL"]["suggested_language"] == "he"
    assert by_code["JP"]["suggested_language"] == "ja"
    #  « New York » est une localité, pas un pays : rien n'est deviné, et le libellé brut sert
    #  de nom pour que l'écran reste lisible.
    unresolved = next(c for c in body["countries"] if c["code"] is None)
    assert (unresolved["raw"], unresolved["label"]) == ("New York", "New York")
    assert (unresolved["suggested_language"], unresolved["target_ids"]) == (None, [])
    #  La cible existante est rattachée à son pays, et à lui seul.
    assert by_code["GB"]["target_ids"] == [str(british.id)]
    assert [c["raw"] for c in body["countries"] if c["target_ids"]] == ["United Kingdom"]


async def test_preview_census_ignores_the_target_when_it_is_not_given(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """`site_target_id` reste facultatif : le recensement tient, `target_ids` est vide (Q7)."""
    uploaded = (await _upload(client, headers, MULTI_COUNTRY.read_bytes())).json()
    response = await client.post(
        "/imports/preview", headers=headers, json={"sha256": uploaded["sha256"]}
    )
    body = response.json()
    assert len(body["countries"]) == 12
    assert all(c["target_ids"] == [] for c in body["countries"])


async def test_preview_leaves_a_source_without_a_country_column_alone(
    client: AsyncClient, headers: Headers, sandbox: Fixture, import_dir: Path
) -> None:
    """Keyword Magic n'écrit aucun pays : le recensement est **vide**, et c'est correct (Q3).

    Le test qui empêche le cloisonnement de se retourner contre les sources muettes. Sans lui,
    une lecture opportuniste de « la première colonne qui ressemble à un pays » finirait par
    s'inviter et un export Semrush entier partirait « hors cible ».
    """
    uploaded = (await _upload(client, headers, SEMRUSH_MAGIC.read_bytes())).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={"sha256": uploaded["sha256"], "site_target_id": str(sandbox.target_id)},
    )
    body = response.json()
    assert body["format"] == "semrush_magic"
    assert body["countries"] == []


async def test_preview_censuses_the_column_chosen_for_a_generic_csv(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """`generic_csv` : la colonne pays est **désignée**, hors mapping canonique (Q4).

    `columns` n'accepte que des champs canoniques — `country` n'en est pas un, et l'y ajouter
    serait un changement à six branches pour un attribut qui n'est pas une métrique du
    mot-clé. Le canal est donc parallèle, et son absence laisse l'import non cloisonné.
    """
    raw = b"Requete,Vol. mensuel,Marche\nchaussures,1300,France\nbottines,880,uk\nbaskets,10,Lyon\n"
    uploaded = (await _upload(client, headers, raw)).json()
    body = {
        "sha256": uploaded["sha256"],
        "format": "generic_csv",
        "columns": {"keyword": "Requete", "volume": "Vol. mensuel"},
    }
    without = (await client.post("/imports/preview", headers=headers, json=body)).json()
    assert without["countries"] == []

    with_column = (
        await client.post(
            "/imports/preview", headers=headers, json={**body, "country_column": "Marche"}
        )
    ).json()
    assert [(c["raw"], c["code"], c["rows"]) for c in with_column["countries"]] == [
        ("France", "FR", 1),
        ("Lyon", None, 1),
        # « uk » n'est pas lisible par pycountry : c'est la table d'alias qui le rattrape.
        ("uk", "GB", 1),
    ]


async def test_preview_422_on_a_country_column_absent_from_the_file(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    """Une colonne pays choisie et introuvable est un refus, pas un recensement vide.

    Même traitement que le mapping : l'utilisateur a désigné quelque chose qui n'existe pas,
    le lui taire lui ferait croire que son fichier ne porte aucun pays.
    """
    uploaded = (await _upload(client, headers, b"Requete,Vol\nchaussures,1300\n")).json()
    response = await client.post(
        "/imports/preview",
        headers=headers,
        json={
            "sha256": uploaded["sha256"],
            "format": "generic_csv",
            "columns": {"keyword": "Requete"},
            "country_column": "Pays",
        },
    )
    assert response.status_code == 422
    assert "Pays" in response.json()["detail"]


async def test_preview_404_on_unknown_file(
    client: AsyncClient, headers: Headers, import_dir: Path
) -> None:
    assert (
        await client.post("/imports/preview", headers=headers, json={"sha256": "0" * 64})
    ).status_code == 404


async def test_preview_422_on_unknown_format(
    client: AsyncClient, headers: Headers, import_dir: Path, content: bytes, sha: str
) -> None:
    await _upload(client, headers, content)
    response = await client.post(
        "/imports/preview", headers=headers, json={"sha256": sha, "format": "ahrefs"}
    )
    assert response.status_code == 422


# --- POST /imports -------------------------------------------------------------------------


async def test_create_import_enqueues_the_contract_payload(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
    db: AsyncSession,
) -> None:
    await _upload(client, headers, content)
    response = await _create(client, headers, sandbox, sha)
    assert response.status_code == 202
    body = response.json()
    record = body["import"]
    assert record["status"] == "queued"
    assert record["source"] == "semrush_positions"
    assert record["kind"] == "csv"
    assert record["filename"] == "positions.csv"
    assert record["file_sha256"] == sha
    assert record["job_id"] == body["job_id"]

    job = await db.scalar(select(Job).where(Job.id == uuid.UUID(body["job_id"])))
    assert job is not None
    # Le payload doit être **exactement** celui que `pipeline.parse_payload` attend (T2).
    spec = pipeline.parse_payload(job.payload)
    assert spec.import_id == uuid.UUID(record["id"])
    assert spec.file_sha256 == sha
    assert spec.source == "semrush_positions"
    assert spec.site_target_id == sandbox.target_id
    assert spec.locale == "fr-FR"
    #  Le pays de la cible voyage **par le payload** (§4.3) : c'est lui qui cloisonne le
    #  fichier, et le figer ici est ce qui rend le rapport reproductible à la reprise.
    assert job.payload["country"] == "FR"
    assert spec.country == "FR"
    assert spec.fetched_at == FETCHED_AT
    assert spec.keyword_list_id == uuid.UUID(record["keyword_list_id"])
    #  Aucune option ici : ni mapping manuel, ni colonne pays choisie.
    assert "options" not in job.payload

    created_list = await db.scalar(
        select(KeywordList).where(KeywordList.id == spec.keyword_list_id)
    )
    assert created_list is not None
    assert created_list.name == "Semrush — 2026-08-30"
    assert created_list.source == "semrush_positions"
    # Traçabilité : la liste sait de quel import elle est née (§3.4).
    assert created_list.created_from_import_id == uuid.UUID(record["id"])


async def test_create_import_accepts_an_existing_list(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
    db: AsyncSession,
) -> None:
    existing = KeywordList(org_id=sandbox.org_id, site_target_id=sandbox.target_id, name="Manuelle")
    db.add(existing)
    await db.commit()
    await _upload(client, headers, content)
    response = await _create(client, headers, sandbox, sha, list={"id": str(existing.id)})
    assert response.status_code == 202
    assert response.json()["import"]["keyword_list_id"] == str(existing.id)


async def test_create_import_conflicts_on_duplicate_unless_forced(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    await _upload(client, headers, content)
    assert (await _create(client, headers, sandbox, sha)).status_code == 202
    conflict = await _create(client, headers, sandbox, sha, list={"name": "Deuxième"})
    assert conflict.status_code == 409
    forced = await _create(client, headers, sandbox, sha, list={"name": "Deuxième"}, force=True)
    assert forced.status_code == 202


async def test_create_import_rejects_a_taken_list_name(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    await _upload(client, headers, content)
    assert (await _create(client, headers, sandbox, sha)).status_code == 202
    again = await _create(client, headers, sandbox, sha, force=True)
    assert again.status_code == 409


@pytest.mark.parametrize(
    ("overrides", "status"),
    [
        ({"format": "ahrefs"}, 422),
        ({"site_target_id": "11111111-1111-1111-1111-111111111111"}, 404),
        ({"list": {"id": "11111111-1111-1111-1111-111111111111"}}, 404),
        ({"list": {}}, 422),
        ({"list": {"id": "11111111-1111-1111-1111-111111111111", "name": "x"}}, 422),
        ({"fetched_at": "2026-08-30T12:00:00"}, 422),
        ({"columns": {"keyword": "Keyword"}}, 422),
        ({"unexpected": 1}, 422),
    ],
)
async def test_create_import_validation(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
    overrides: dict[str, Any],
    status: int,
) -> None:
    await _upload(client, headers, content)
    assert (await _create(client, headers, sandbox, sha, **overrides)).status_code == status


async def test_create_import_404_when_the_file_is_gone(
    client: AsyncClient, headers: Headers, sandbox: Fixture, import_dir: Path
) -> None:
    assert (await _create(client, headers, sandbox, "a" * 64)).status_code == 404


async def test_create_import_validates_the_generic_mapping_before_enqueuing(
    client: AsyncClient, headers: Headers, sandbox: Fixture, import_dir: Path, db: AsyncSession
) -> None:
    uploaded = (await _upload(client, headers, b"Requete,Vol\nchaussures,10\n")).json()
    bad = await _create(
        client,
        headers,
        sandbox,
        uploaded["sha256"],
        format="generic_csv",
        columns={"keyword": "Absente"},
    )
    assert bad.status_code == 422
    # Rien n'a été enfilé : le mapping est refusé **avant** le job (§5).
    assert await db.scalar(select(Job.id).where(Job.type == pipeline.JOB_TYPE)) is None

    good = await _create(
        client,
        headers,
        sandbox,
        uploaded["sha256"],
        format="generic_csv",
        columns={"keyword": "Requete", "volume": "Vol"},
    )
    assert good.status_code == 202
    job = await db.scalar(select(Job).where(Job.type == pipeline.JOB_TYPE))
    assert job is not None
    assert pipeline.parse_payload(job.payload).options == {
        "columns": {"keyword": "Requete", "volume": "Vol"}
    }


async def test_create_import_carries_the_chosen_country_column_to_the_job(
    client: AsyncClient, headers: Headers, sandbox: Fixture, import_dir: Path, db: AsyncSession
) -> None:
    """La colonne pays désignée à l'aperçu doit suivre jusqu'au job, sinon rien ne cloisonne.

    C'est le pendant de `PreviewIn.country_column` (§4.1, Q4) : sans lui, la capacité pays de
    `generic_csv` existe, l'aperçu la déclenche… et l'import réel l'ignore en silence. Elle
    voyage **hors du mapping canonique** — `country` n'est pas un champ de mot-clé.
    """
    uploaded = (
        await _upload(client, headers, b"Requete,Vol,Marche\nchaussures,10,France\n")
    ).json()
    created = await _create(
        client,
        headers,
        sandbox,
        uploaded["sha256"],
        format="generic_csv",
        columns={"keyword": "Requete", "volume": "Vol"},
        country_column="Marche",
    )
    assert created.status_code == 202
    job = await db.scalar(select(Job).where(Job.type == pipeline.JOB_TYPE))
    assert job is not None
    spec = pipeline.parse_payload(job.payload)
    assert spec.options == {
        "columns": {"keyword": "Requete", "volume": "Vol"},
        "country_column": "Marche",
    }
    #  Et l'adaptateur la retrouve bien : c'est ce que le pipeline interrogera.
    assert sources.get("generic_csv").country_header(spec.options) == "Marche"


async def test_create_import_is_atomic(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
    db: AsyncSession,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Un échec après l'insertion du job ne doit **rien** laisser derrière lui.

    C'est le scénario de l'*import fantôme* : job committé, liste committée, ligne `imports`
    jamais écrite — le worker exécute alors l'import, le réservoir se peuple, la liste se
    remplit, et l'onglet Imports reste vide à jamais.

    **Le point d'injection est choisi, pas commode.** La panne est levée à la construction de
    `Import`, donc *après le retour* de `enqueue_pending`, dans `create_import` lui-même. Une
    panne levée à l'intérieur du double laisserait passer exactement la régression que ce test
    doit attraper : un `await session.commit()` réintroduit entre l'enfilage et la ligne
    `imports` — le job serait alors publié, et le test resterait vert. Ici il tombe.

    `force=True` neutralise la pré-vérification de doublon, seul autre usage de `Import` dans ce
    chemin ; le double n'intercepte donc bien que la création de la ligne.
    """
    await _upload(client, headers, content)
    enqueued: list[str] = []
    original = jobs_service.enqueue_pending

    async def _spy(*args: Any, **kwargs: Any) -> Job:
        job = await original(*args, **kwargs)
        enqueued.append(str(job.id))
        return job

    def _boom(*_args: Any, **_kwargs: Any) -> Import:
        raise RuntimeError("panne simulée entre l'enfilage et la ligne imports")

    monkeypatch.setattr(service.jobs_service, "enqueue_pending", _spy)
    monkeypatch.setattr(service, "Import", _boom)
    with pytest.raises(RuntimeError):
        await _create(client, headers, sandbox, sha, force=True)

    #  Le job **a bien été inséré** avant la panne : sans cette assertion, le test passerait
    #  aussi si `create_import` échouait plus tôt, sans rien prouver de l'atomicité.
    assert len(enqueued) == 1
    #  Et pourtant : ni job, ni liste, ni import. La transaction entière a été annulée.
    assert await db.scalar(select(Job.id).where(Job.org_id == sandbox.org_id)) is None
    assert (
        await db.scalar(select(KeywordList.id).where(KeywordList.org_id == sandbox.org_id)) is None
    )
    assert await db.scalar(select(Import.id).where(Import.org_id == sandbox.org_id)) is None


async def test_create_import_records_the_country_of_the_target(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    db: AsyncSession,
) -> None:
    """`imports.detected_country` dit ce qui a été **gardé**, pas ce que le fichier annonce (§4.5).

    C'est le renversement du cloisonnement. Le fichier annonce `us` sur toutes ses lignes, la
    cible du bac à sable est `FR` : la colonne vaut `FR`, parce que c'est le marché de cet
    import et donc le seul pays que le pipeline y laissera entrer. Un export multi-marchés en
    annoncerait onze — « le pays annoncé par le fichier » n'a plus de sens, et c'est le
    recensement de l'aperçu qui les nomme tous.
    """
    raw = b"Keyword,Database,Volume\nchaussures,us,1300\nbottines,us,880\n"
    uploaded = (await _upload(client, headers, raw)).json()
    response = await _create(
        client,
        headers,
        sandbox,
        uploaded["sha256"],
        format="generic_csv",
        columns={"keyword": "Keyword", "volume": "Volume"},
    )
    assert response.status_code == 202
    assert response.json()["import"]["detected_country"] == "FR"

    record = await db.scalar(
        select(Import).where(Import.id == uuid.UUID(response.json()["import"]["id"]))
    )
    assert record is not None and record.detected_country == "FR"


async def test_create_import_records_the_country_even_for_a_mute_export(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    """Un export sans colonne de pays (Keyword Magic, Positions…) renseigne la colonne aussi.

    Elle ne dépend plus du fichier : la cible suffit à la remplir. `null` ne subsiste donc que
    sur les imports **antérieurs** au cloisonnement, jamais sur un nouveau.
    """
    await _upload(client, headers, content)
    created = await _create(client, headers, sandbox, sha)
    assert created.json()["import"]["detected_country"] == "FR"


# --- listings, détail, errors.csv ------------------------------------------------------------


async def test_listing_paginates_and_interleaves_reconcile_events(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
    db: AsyncSession,
) -> None:
    await _upload(client, headers, content)
    first = (await _create(client, headers, sandbox, sha, list={"name": "A"})).json()["import"][
        "id"
    ]
    second = (await _create(client, headers, sandbox, sha, list={"name": "B"}, force=True)).json()[
        "import"
    ]["id"]
    # Un événement de rattachement, tel que T6 l'enfilera — inséré à la main, aucun type
    # `map_reconcile` n'existe encore au registre.
    event = Job(
        org_id=sandbox.org_id,
        type="map_reconcile",
        status="done",
        payload={"site_target_id": str(sandbox.target_id)},
        report={"attached": 3},
    )
    db.add(event)
    await db.commit()

    page = (await client.get("/imports", headers=headers, params={"limit": 2})).json()
    assert [item["entry"] for item in page["items"]] == ["map_reconcile", "import"]
    assert page["items"][0]["report"] == {"attached": 3}
    assert page["items"][1]["id"] == second
    assert page["next_cursor"] is not None

    tail = (
        await client.get(
            "/imports", headers=headers, params={"limit": 2, "cursor": page["next_cursor"]}
        )
    ).json()
    assert [item["id"] for item in tail["items"]] == [first]
    assert tail["next_cursor"] is None


async def test_listing_filters_by_target(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    await _upload(client, headers, content)
    await _create(client, headers, sandbox, sha)
    other = uuid.uuid4()
    assert (
        await client.get("/imports", headers=headers, params={"site_target_id": str(other)})
    ).json()["items"] == []
    assert (
        len(
            (
                await client.get(
                    "/imports", headers=headers, params={"site_target_id": str(sandbox.target_id)}
                )
            ).json()["items"]
        )
        == 1
    )


async def test_listing_rejects_a_broken_cursor(client: AsyncClient, headers: Headers) -> None:
    response = await client.get("/imports", headers=headers, params={"cursor": "pas-un-curseur"})
    assert response.status_code == 422


async def test_detail_and_errors_csv(
    client: AsyncClient,
    headers: Headers,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
    db: AsyncSession,
) -> None:
    await _upload(client, headers, content)
    import_id = (await _create(client, headers, sandbox, sha)).json()["import"]["id"]

    # Pas de rapport, donc pas de fichier d'erreurs à télécharger.
    assert (
        await client.get(f"/imports/{import_id}/errors.csv", headers=headers)
    ).status_code == 404

    record = await db.scalar(select(Import).where(Import.id == uuid.UUID(import_id)))
    assert record is not None
    record.status = "done"
    record.report = {"rows_read": 7, "rows_error": 1, "errors_csv": f"{sha}.errors.csv"}
    await db.commit()

    detail = (await client.get(f"/imports/{import_id}", headers=headers)).json()
    assert detail["status"] == "done"
    assert detail["report"]["rows_read"] == 7

    # Le rapport annonce un fichier qui n'est plus sur le disque : il a existé, il a été purgé
    # (rétention 30 jours) — 410, pas 404. Les deux cas ne disent pas la même chose à l'écran :
    # « aucune erreur » n'est pas « rapport expiré ».
    assert (
        await client.get(f"/imports/{import_id}/errors.csv", headers=headers)
    ).status_code == 410
    errors_path = pipeline.import_file_path(sandbox.org_id, sha, suffix=".errors.csv")
    errors_path.write_text("line,reason\n6,mot-clé vide\n", encoding="utf-8")
    download = await client.get(f"/imports/{import_id}/errors.csv", headers=headers)
    assert download.status_code == 200
    assert download.headers["content-type"].startswith("text/csv")
    assert download.text.splitlines()[1] == "6,mot-clé vide"


# --- isolation par organisation ---------------------------------------------------------------


async def test_another_org_sees_nothing(
    client: AsyncClient,
    headers: Headers,
    auth_headers: Any,
    sandbox: Fixture,
    import_dir: Path,
    content: bytes,
    sha: str,
) -> None:
    await _upload(client, headers, content)
    import_id = (await _create(client, headers, sandbox, sha)).json()["import"]["id"]
    stranger = auth_headers(f"org_{uuid.uuid4().hex}")

    assert (await client.get("/imports", headers=stranger)).json()["items"] == []
    assert (await client.get(f"/imports/{import_id}", headers=stranger)).status_code == 404
    assert (
        await client.get(f"/imports/{import_id}/errors.csv", headers=stranger)
    ).status_code == 404
    # Le fichier lui-même est rangé sous l'org : une autre organisation ne peut pas le lire.
    assert (
        await client.post("/imports/preview", headers=stranger, json={"sha256": sha})
    ).status_code == 404