import os
from urllib.parse import urlsplit

# La base de test par défaut. Un agent qui doit lancer la suite **en parallèle** d'un autre
# en demande une à lui via `PYTEST_DATABASE_URL` : deux suites sur la même base se volent
# leurs jobs et leurs lignes (cf. `_purge_leftover_test_jobs`). La variable est délibérément
# distincte de `DATABASE_URL` — une variable d'environnement qui traîne dans un shell ne doit
# jamais pouvoir faire tourner la suite, migrations comprises, sur la base de développement.
_DEFAULT_TEST_DATABASE_URL = (
    "postgresql+asyncpg://seoinsight:seoinsight@localhost:5432/seoinsight_data_test"
)


def _test_database_url() -> str:
    url = os.environ.get("PYTEST_DATABASE_URL", "").strip()
    if not url:
        return _DEFAULT_TEST_DATABASE_URL
    parsed = urlsplit(url)
    if parsed.hostname not in ("localhost", "127.0.0.1", "::1"):
        raise RuntimeError(
            f"PYTEST_DATABASE_URL doit viser une base locale, reçu {parsed.hostname!r}"
        )
    name = parsed.path.lstrip("/")
    if not name.startswith("seoinsight_data_test"):
        raise RuntimeError(
            f"PYTEST_DATABASE_URL doit viser une base seoinsight_data_test*, reçu {name!r} — "
            "la suite migre et purge la base qu'on lui donne."
        )
    return url


os.environ["DATABASE_URL"] = _test_database_url()
os.environ["AUTH_BASE_URL"] = "http://test-auth.local"
# Clé Fernet fixe de test : le chiffrement des secrets BYOK doit être *actif* dans la suite
# (un round-trip est testé), sans quoi les tests d'intégrations passeraient sur un chemin que
# la prod n'emprunte pas. Elle ne protège rien — elle est publique par construction.
os.environ["CREDENTIALS_KEY"] = "AjKMDR-BzYqbEXFYbcE_Bg1sxvFhwZ4EZjLwSMxDXwE="
# Aucune requête sortante dans la suite : ce qui pointerait vers api.mangools.com pointe ici,
# et tout test qui appelle vraiment le fournisseur bouchonne son client.
os.environ["MANGOOLS_API_BASE_URL"] = "http://mangools.invalid/v3"

import time  # noqa: E402
import uuid  # noqa: E402
from collections.abc import AsyncIterator, Awaitable, Callable, Iterator  # noqa: E402
from dataclasses import dataclass  # noqa: E402
from pathlib import Path  # noqa: E402

import jwt  # noqa: E402
import pytest  # noqa: E402
import pytest_asyncio  # noqa: E402
from alembic import command  # noqa: E402
from alembic.config import Config  # noqa: E402
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey  # noqa: E402
from fastapi import APIRouter  # noqa: E402
from httpx import ASGITransport, AsyncClient  # noqa: E402
from jwt import PyJWK  # noqa: E402
from sqlalchemy import delete, text  # noqa: E402
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker  # noqa: E402

from app.core.auth import CurrentOrgDep, set_jwks_client  # noqa: E402
from app.core.config import get_settings  # noqa: E402
from app.core.errors import AppError  # noqa: E402
from app.db.engine import dispose_engine, get_engine  # noqa: E402
from app.integrations.models import Integration  # noqa: E402
from app.jobs.models import Job  # noqa: E402
from app.keywords.models import Keyword, OrgSettings, Tag  # noqa: E402
from app.main import create_app  # noqa: E402
from app.sites.models import Site, SiteTarget  # noqa: E402


# Doit rester sync : alembic/env.py appelle asyncio.run() et ne peut pas tourner dans une boucle.
@pytest.fixture(scope="session", autouse=True)
def _migrate_test_db() -> None:
    cfg = Config("alembic.ini")
    command.upgrade(cfg, "head")


@pytest_asyncio.fixture(scope="session", autouse=True)
async def _purge_leftover_test_jobs(_migrate_test_db: None) -> None:
    """Les jobs de test qu'un lancement **tué** a laissés derrière lui, avant de commencer.

    `data_test` est partagée par tous les worktrees, et `claim()` prend n'importe quel job en
    attente (par design : le worker ne connaît pas les organisations). Un pytest interrompu
    avant son teardown laisse donc des lignes `jobs` que le lancement suivant — dans
    **n'importe quel** worktree — réclame et compte : `test_jobs_worker.py` échoue alors de
    façon fantôme, sur des tests que personne n'a touchés. Constaté (une ligne
    `type='test_ok'` orpheline, 11 à 19 échecs selon l'ordre).

    Le filtre ne vise que les types de test (`test_*`, ceux de `test_jobs_worker.py`) : aucun
    type de production ne commence par là, et `DELETE` est idempotent. Le jour où plusieurs
    agents lanceront la suite **en parallèle** sur la même base, il faudra une base par
    worktree — ceci ne protège que du cadavre, pas de la course.
    """
    async with get_engine().begin() as connection:
        await connection.execute(text(r"DELETE FROM jobs WHERE type LIKE 'test\_%'"))


@pytest_asyncio.fixture(scope="session", autouse=True)
async def _dispose_engine() -> AsyncIterator[None]:
    yield
    await dispose_engine()


_test_router = APIRouter(prefix="/_test")


@_test_router.get("/boom")
async def boom() -> None:
    raise AppError(418, "Teapot", "Je suis une théière")


@_test_router.get("/echo")
async def echo(n: int) -> dict[str, int]:
    return {"n": n}


@_test_router.get("/org")
async def org(user: CurrentOrgDep) -> dict[str, str]:
    return {"org_id": user.org_id}


@pytest.fixture(scope="session")
def test_private_key() -> Ed25519PrivateKey:
    return Ed25519PrivateKey.generate()


@pytest.fixture(scope="session")
def auth_headers(test_private_key: Ed25519PrivateKey) -> Callable[[str], dict[str, str]]:
    """En-têtes Authorization pour une organisation donnée (JWT EdDSA signé par la clé de test)."""

    def _for(org_id: str) -> dict[str, str]:
        now = int(time.time())
        token = jwt.encode(
            {
                "sub": f"user_{org_id}",
                "email": "dev@test.local",
                "org_id": org_id,
                "iat": now,
                "exp": now + 900,
                "iss": "http://test-auth.local",
                "aud": "http://test-auth.local",
            },
            test_private_key,
            algorithm="EdDSA",
            headers={"kid": "test-key"},
        )
        return {"Authorization": f"Bearer {token}"}

    return _for


class _FakeJwksClient:
    def __init__(self, private_key: Ed25519PrivateKey) -> None:
        self._key = PyJWK.from_json(
            jwt.algorithms.OKPAlgorithm.to_jwk(private_key.public_key()), algorithm="EdDSA"
        )

    def get_signing_key_from_jwt(self, _token: str) -> PyJWK:
        return self._key


@dataclass(frozen=True)
class Fixture:
    """Une organisation jetable avec un site et une cible — le socle des tests de modèle."""

    org_id: str
    site_id: uuid.UUID
    target_id: uuid.UUID


# `make_target(country, language) -> id de la cible` — voir la fixture du même nom.
MakeTarget = Callable[[str, str], Awaitable[uuid.UUID]]


@pytest_asyncio.fixture
async def db() -> AsyncIterator[AsyncSession]:
    """Session SQLAlchemy directe : les tests de DDL parlent à la base, pas à l'API."""
    async with async_sessionmaker(get_engine(), expire_on_commit=False)() as session:
        yield session


@pytest_asyncio.fixture
async def sandbox(db: AsyncSession) -> AsyncIterator[Fixture]:
    """Org / site / cible neufs à chaque test, nettoyés ensuite.

    `data_test` n'est pas vidée entre deux lancements : l'org est unique par test (aucune
    interférence possible) et le teardown supprime tout ce qui lui appartient — y compris
    ce que les CASCADE ne couvrent pas volontairement (le réservoir survit à ses cibles,
    décision n°4).
    """
    org_id = f"org_{uuid.uuid4().hex}"
    site = Site(org_id=org_id, domain=f"{uuid.uuid4().hex}.example", name="Test")
    db.add(site)
    await db.flush()
    target = SiteTarget(site_id=site.id, org_id=org_id, country="FR", language="fr")
    db.add(target)
    await db.commit()
    yield Fixture(org_id=org_id, site_id=site.id, target_id=target.id)
    await db.rollback()
    #  `Integration` n'est pas rattachée au site : elle vit au niveau de l'organisation, donc
    #  aucun CASCADE ne l'emporte avec `Site` — sans cette ligne, l'UNIQUE (org_id, provider)
    #  survivrait au test et ferait échouer le suivant.
    for table in (Site, Integration, Keyword, Tag, OrgSettings, Job):
        await db.execute(delete(table).where(table.org_id == org_id))
    await db.commit()


@pytest_asyncio.fixture
async def other_sandbox(db: AsyncSession) -> AsyncIterator[Fixture]:
    """Une **seconde** organisation, complètement indépendante de `sandbox`.

    `sandbox` seul ne peut pas prouver qu'un traitement transverse — le sweep du
    planificateur, qui balaie toutes les organisations — n'en laisse pas une empoisonner
    l'autre. Il en faut deux, et le teardown est le même que celui de `sandbox`.
    """
    org_id = f"org_{uuid.uuid4().hex}"
    site = Site(org_id=org_id, domain=f"{uuid.uuid4().hex}.example", name="Test 2")
    db.add(site)
    await db.flush()
    target = SiteTarget(site_id=site.id, org_id=org_id, country="FR", language="fr")
    db.add(target)
    await db.commit()
    yield Fixture(org_id=org_id, site_id=site.id, target_id=target.id)
    await db.rollback()
    for table in (Site, Integration, Keyword, Tag, OrgSettings, Job):
        await db.execute(delete(table).where(table.org_id == org_id))
    await db.commit()


@pytest_asyncio.fixture
async def make_target(db: AsyncSession, sandbox: Fixture) -> MakeTarget:
    """Une cible supplémentaire sur le **même site** que le bac à sable.

    `sandbox` n'en crée qu'une, `FR`/`fr`, ce qui suffit à presque tout. Le cloisonnement par
    pays, lui, ne se teste qu'à deux marchés au moins : le même fichier importé sur `en-US`
    puis sur `en-GB` doit donner deux imports disjoints. Rien à nettoyer — `site_targets.site_id`
    est en `ON DELETE CASCADE` et le teardown de `sandbox` supprime le site.
    """

    async def _make(country: str, language: str) -> uuid.UUID:
        target = SiteTarget(
            site_id=sandbox.site_id, org_id=sandbox.org_id, country=country, language=language
        )
        db.add(target)
        await db.commit()
        return target.id

    return _make


@pytest.fixture
def import_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]:
    """`IMPORT_FILES_DIR` isolé par test (spec Imports §3.12).

    `get_settings` est mis en cache : le cache est vidé des deux côtés du test, sans quoi le
    réglage fuirait vers les suivants — ou, pire, le précédent tiendrait encore ici.
    """
    monkeypatch.setenv("IMPORT_FILES_DIR", str(tmp_path))
    get_settings.cache_clear()
    yield tmp_path
    get_settings.cache_clear()


@pytest_asyncio.fixture(scope="session")
async def client(test_private_key: Ed25519PrivateKey) -> AsyncIterator[AsyncClient]:
    set_jwks_client(_FakeJwksClient(test_private_key))
    app = create_app()
    app.include_router(_test_router)
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c