import uuid from collections.abc import Callable from typing import Any import pytest from httpx import AsyncClient from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.db.engine import get_engine from app.sites import service from app.sites.models import SiteTarget Headers = Callable[[str], dict[str, str]] def new_org() -> str: # data_test n'est pas vidée entre deux lancements de pytest : chaque test travaille # dans une organisation neuve pour rester rejouable. return f"org_{uuid.uuid4().hex[:12]}" PAYLOAD: dict[str, Any] = { "domain": "https://www.Exemple.FR/page?x=1", "targets": [ {"country": "FR", "language": "fr", "is_default": True}, {"country": "BE", "language": "fr"}, ], } async def create_site( client: AsyncClient, headers: dict[str, str], **overrides: Any ) -> dict[str, Any]: resp = await client.post("/sites", json={**PAYLOAD, **overrides}, headers=headers) assert resp.status_code == 201, resp.text return resp.json() async def _count_targets(site_id: str) -> int: async with AsyncSession(get_engine()) as session: count = await session.scalar( select(func.count()) .select_from(SiteTarget) .where(SiteTarget.site_id == uuid.UUID(site_id)) ) return count or 0 async def test_create_site_normalizes_and_sets_default( client: AsyncClient, auth_headers: Headers ) -> None: body = await create_site(client, auth_headers(new_org())) assert body["domain"] == "www.exemple.fr" # normalisé, www conservé assert body["name"] == "www.exemple.fr" # défaut applicatif : le domaine assert len(body["targets"]) == 2 defaults = [t for t in body["targets"] if t["is_default"]] assert len(defaults) == 1 and defaults[0]["country"] == "FR" async def test_create_site_with_explicit_name(client: AsyncClient, auth_headers: Headers) -> None: body = await create_site(client, auth_headers(new_org()), name="Boutique") assert body["name"] == "Boutique" async def test_create_duplicate_domain_is_409(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) await create_site(client, headers) resp = await client.post("/sites", json=PAYLOAD, headers=headers) assert resp.status_code == 409 assert resp.headers["content-type"].startswith("application/problem+json") assert "www.exemple.fr" in resp.json()["detail"] async def test_create_invalid_domain_is_422(client: AsyncClient, auth_headers: Headers) -> None: resp = await client.post( "/sites", json={**PAYLOAD, "domain": "192.168.1.10"}, headers=auth_headers(new_org()), ) assert resp.status_code == 422 assert resp.headers["content-type"].startswith("application/problem+json") async def test_create_two_defaults_is_422(client: AsyncClient, auth_headers: Headers) -> None: targets = [ {"country": "FR", "language": "fr", "is_default": True}, {"country": "BE", "language": "fr", "is_default": True}, ] resp = await client.post( "/sites", json={**PAYLOAD, "targets": targets}, headers=auth_headers(new_org()) ) assert resp.status_code == 422 async def test_create_zero_default_is_422(client: AsyncClient, auth_headers: Headers) -> None: targets = [{"country": "FR", "language": "fr"}] resp = await client.post( "/sites", json={**PAYLOAD, "targets": targets}, headers=auth_headers(new_org()) ) assert resp.status_code == 422 async def test_create_duplicate_couple_is_422(client: AsyncClient, auth_headers: Headers) -> None: targets = [ {"country": "FR", "language": "fr", "is_default": True}, {"country": "FR", "language": "fr"}, ] resp = await client.post( "/sites", json={**PAYLOAD, "targets": targets}, headers=auth_headers(new_org()) ) assert resp.status_code == 422 async def test_create_more_than_20_targets_is_422( client: AsyncClient, auth_headers: Headers ) -> None: targets = [ {"country": f"{chr(65 + i)}A", "language": "aa", "is_default": i == 0} for i in range(21) ] resp = await client.post( "/sites", json={**PAYLOAD, "targets": targets}, headers=auth_headers(new_org()) ) assert resp.status_code == 422 async def test_get_site_returns_targets(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) resp = await client.get(f"/sites/{created['id']}", headers=headers) assert resp.status_code == 200 assert resp.json() == created async def test_get_unknown_site_is_404(client: AsyncClient, auth_headers: Headers) -> None: resp = await client.get(f"/sites/{uuid.uuid4()}", headers=auth_headers(new_org())) assert resp.status_code == 404 assert resp.headers["content-type"].startswith("application/problem+json") async def test_rename_site(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) resp = await client.patch( f"/sites/{created['id']}", json={"name": "Nouveau nom"}, headers=headers ) assert resp.status_code == 200 assert resp.json()["name"] == "Nouveau nom" assert resp.json()["domain"] == created["domain"] async def test_patch_domain_is_422(client: AsyncClient, auth_headers: Headers) -> None: # Le domaine est immuable : absent du schéma PATCH (extra="forbid"). headers = auth_headers(new_org()) created = await create_site(client, headers) resp = await client.patch( f"/sites/{created['id']}", json={"name": "x", "domain": "autre.fr"}, headers=headers, ) assert resp.status_code == 422 async def test_delete_site_removes_it(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) resp = await client.delete(f"/sites/{created['id']}", headers=headers) assert resp.status_code == 204 resp = await client.get(f"/sites/{created['id']}", headers=headers) assert resp.status_code == 404 async def test_delete_site_cascades_targets(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) assert await _count_targets(created["id"]) == 2 await client.delete(f"/sites/{created['id']}", headers=headers) assert await _count_targets(created["id"]) == 0 async def test_add_target(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) resp = await client.post( f"/sites/{created['id']}/targets", json={"country": "CH", "language": "fr"}, headers=headers, ) assert resp.status_code == 201 body = resp.json() assert (body["country"], body["language"], body["is_default"]) == ("CH", "fr", False) async def test_add_duplicate_target_is_409(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) resp = await client.post( f"/sites/{created['id']}/targets", json={"country": "BE", "language": "fr"}, headers=headers, ) assert resp.status_code == 409 async def test_add_target_on_unknown_site_is_404( client: AsyncClient, auth_headers: Headers ) -> None: resp = await client.post( f"/sites/{uuid.uuid4()}/targets", json={"country": "FR", "language": "fr"}, headers=auth_headers(new_org()), ) assert resp.status_code == 404 async def test_add_target_beyond_cap_is_422(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) targets = [ {"country": f"{chr(65 + i)}A", "language": "aa", "is_default": i == 0} for i in range(20) ] created = await create_site(client, headers, targets=targets) resp = await client.post( f"/sites/{created['id']}/targets", json={"country": "ZZ", "language": "zz"}, headers=headers, ) assert resp.status_code == 422 assert "20" in resp.json()["detail"] async def test_promote_target_demotes_previous_default( client: AsyncClient, auth_headers: Headers ) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) secondary = next(t for t in created["targets"] if not t["is_default"]) resp = await client.patch( f"/sites/{created['id']}/targets/{secondary['id']}", json={"is_default": True}, headers=headers, ) assert resp.status_code == 200 assert resp.json()["is_default"] is True site = (await client.get(f"/sites/{created['id']}", headers=headers)).json() defaults = [t["id"] for t in site["targets"] if t["is_default"]] assert defaults == [secondary["id"]] # l'ancienne défaut est démise async def test_promote_with_false_is_422(client: AsyncClient, auth_headers: Headers) -> None: # On promeut une autre cible, on ne « dé-défaute » pas (spec §5). headers = auth_headers(new_org()) created = await create_site(client, headers) default = next(t for t in created["targets"] if t["is_default"]) resp = await client.patch( f"/sites/{created['id']}/targets/{default['id']}", json={"is_default": False}, headers=headers, ) assert resp.status_code == 422 async def test_delete_target(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) secondary = next(t for t in created["targets"] if not t["is_default"]) resp = await client.delete(f"/sites/{created['id']}/targets/{secondary['id']}", headers=headers) assert resp.status_code == 204 site = (await client.get(f"/sites/{created['id']}", headers=headers)).json() assert len(site["targets"]) == 1 async def test_delete_default_target_is_409(client: AsyncClient, auth_headers: Headers) -> None: headers = auth_headers(new_org()) created = await create_site(client, headers) default = next(t for t in created["targets"] if t["is_default"]) resp = await client.delete(f"/sites/{created['id']}/targets/{default['id']}", headers=headers) assert resp.status_code == 409 async def test_delete_target_promoted_concurrently_is_409( client: AsyncClient, auth_headers: Headers, monkeypatch: pytest.MonkeyPatch ) -> None: """La cible est promue par défaut entre son chargement et sa suppression. Le pré-chargement la croit secondaire (l'instance déjà en session garde `is_default` à False) : seule la clause `NOT is_default` du DELETE, évaluée par la base, ferme la fenêtre. Le test le prouve en faisant commiter une promotion concurrente juste après le chargement du site. """ org_id = new_org() headers = auth_headers(org_id) created = await create_site(client, headers) secondary = next(t for t in created["targets"] if not t["is_default"]) site_id, target_id = uuid.UUID(created["id"]), uuid.UUID(secondary["id"]) real_get_site = service.get_site raced = False async def racing_get_site(session: AsyncSession, org: str, sid: uuid.UUID) -> Any: nonlocal raced site = await real_get_site(session, org, sid) if not raced: # une seule fois : `promote_target` repasse par `get_site` raced = True # Transaction concurrente, committée avant le DELETE. async with AsyncSession(get_engine()) as other: await service.promote_target(other, org, site_id, target_id) return site monkeypatch.setattr(service, "get_site", racing_get_site) resp = await client.delete(f"/sites/{created['id']}/targets/{secondary['id']}", headers=headers) assert resp.status_code == 409, resp.text assert resp.headers["content-type"].startswith("application/problem+json") assert await _count_targets(created["id"]) == 2 # la nouvelle défaut a survécu async def test_unknown_target_is_404(client: AsyncClient, auth_headers: Headers) -> None: # Cible inexistante, et cible d'un autre site de la *même* org : dans les deux cas le # couple (site, cible) n'existe pas — 404 « Cible inconnue », jamais un 500 ni une # promotion qui démettrait la défaut de l'autre site. headers = auth_headers(new_org()) created = await create_site(client, headers) other = await create_site(client, headers, domain="autre-exemple.fr") foreign = next(t for t in other["targets"] if not t["is_default"]) for target_id in (str(uuid.uuid4()), foreign["id"]): resp = await client.patch( f"/sites/{created['id']}/targets/{target_id}", json={"is_default": True}, headers=headers, ) assert resp.status_code == 404, resp.text assert resp.json()["detail"] == "Cible inconnue" resp = await client.delete(f"/sites/{created['id']}/targets/{target_id}", headers=headers) assert resp.status_code == 404, resp.text # La cible par défaut de l'autre site n'a pas bougé. refreshed = (await client.get(f"/sites/{other['id']}", headers=headers)).json() assert [t["id"] for t in refreshed["targets"] if t["is_default"]] == [ next(t["id"] for t in other["targets"] if t["is_default"]) ]