office_translator/tests/test_auth_logout.py

"""
Tests pour POST /api/v1/auth/logout
Couvre les AC 1-6 de la story 1.4 : Logout Utilisateur
"""

import pytest
from pathlib import Path
from fastapi.testclient import TestClient


LOGOUT_URL = "/api/v1/auth/logout"
LOGIN_URL = "/api/v1/auth/login"
REGISTER_URL = "/api/v1/auth/register"

VALID_USER = {
    "email": "logout@example.com",
    "password": "Password123!",
    "name": "Logout User",
}

VALID_CREDENTIALS = {
    "email": "logout@example.com",
    "password": "Password123!",
}


@pytest.fixture()
def client(tmp_path: Path, monkeypatch):
    """TestClient avec stockage JSON isolé, blocklist réinitialisée et rate limiting désactivé."""
    import services.auth_service as auth_svc

    monkeypatch.setattr(auth_svc, "USERS_FILE", tmp_path / "users.json")
    monkeypatch.setattr(auth_svc, "USE_DATABASE", False)
    monkeypatch.setattr(auth_svc, "DATABASE_AVAILABLE", False)
    monkeypatch.setattr(auth_svc, "_revoked_jtis", {})

    from middleware.rate_limiting import RateLimitManager

    async def _allow(self, request):
        return True, "ok", "test"

    async def _allow_translation(self, request, file_size_mb=0):
        return True, "ok"

    monkeypatch.setattr(RateLimitManager, "check_request", _allow)
    monkeypatch.setattr(RateLimitManager, "check_translation", _allow_translation)

    from main import app

    return TestClient(app, raise_server_exceptions=True)


@pytest.fixture()
def tokens(client):
    """Enregistre un utilisateur et retourne ses tokens access + refresh."""
    client.post(REGISTER_URL, json=VALID_USER)
    resp = client.post(LOGIN_URL, json=VALID_CREDENTIALS)
    assert resp.status_code == 200
    data = resp.json()["data"]
    return data["access_token"], data["refresh_token"]


# ---------------------------------------------------------------------------
# AC1 & AC4 : Endpoint existe et retourne 200 avec message succès
# ---------------------------------------------------------------------------


class TestLogoutSuccess:
    """AC1 & AC4 : POST /logout valide → 200 + message Déconnexion réussie"""

    def test_returns_200_on_success(self, client, tokens):
        access_token, _ = tokens
        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
        )
        assert response.status_code == 200

    def test_response_structure(self, client, tokens):
        access_token, _ = tokens
        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
        )
        body = response.json()
        assert "data" in body
        assert body["data"]["message"] == "Déconnexion réussie"
        assert "meta" in body
        assert body["meta"] == {}

    def test_logout_with_refresh_token_in_body(self, client, tokens):
        access_token, refresh_token = tokens
        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
            json={"refresh_token": refresh_token},
        )
        assert response.status_code == 200


# ---------------------------------------------------------------------------
# AC3 : Après logout, l'access token est révoqué
# ---------------------------------------------------------------------------


class TestAccessTokenRevoked:
    """AC3 : access token révoqué après logout → 401 à la réutilisation"""

    def test_reusing_access_token_after_logout_returns_401(self, client, tokens):
        access_token, _ = tokens
        logout_resp = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
        )
        assert logout_resp.status_code == 200

        reuse_resp = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
        )
        assert reuse_resp.status_code == 401


# ---------------------------------------------------------------------------
# AC2 : Après logout, le refresh token est révoqué
# ---------------------------------------------------------------------------


class TestRefreshTokenRevoked:
    """AC2 : refresh token révoqué après logout → ne peut plus obtenir de nouvel access token"""

    def test_refresh_token_revoked_after_logout(self, client, tokens):
        access_token, refresh_token = tokens
        logout_resp = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {access_token}"},
            json={"refresh_token": refresh_token},
        )
        assert logout_resp.status_code == 200

        # AC2: utiliser le refresh token révoqué doit retourner 401
        refresh_resp = client.post(
            "/api/v1/auth/refresh",
            json={"refresh_token": refresh_token},
        )
        assert refresh_resp.status_code == 401, (
            "Le refresh token révoqué ne doit plus permettre d'obtenir un access token"
        )


# ---------------------------------------------------------------------------
# AC5 : Token manquant → 401 TOKEN_MISSING
# ---------------------------------------------------------------------------


class TestTokenMissing:
    """AC5 : Appel sans Authorization header → 401 TOKEN_MISSING"""

    def test_no_auth_header_returns_401(self, client):
        response = client.post(LOGOUT_URL)
        assert response.status_code == 401

    def test_no_auth_header_error_code(self, client):
        response = client.post(LOGOUT_URL)
        body = response.json()
        assert body["error"] == "TOKEN_MISSING"
        assert "Token d'authentification requis" in body["message"]

    def test_non_bearer_auth_returns_401(self, client):
        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": "Basic dXNlcjpwYXNz"},
        )
        assert response.status_code == 401
        assert response.json()["error"] == "TOKEN_MISSING"


# ---------------------------------------------------------------------------
# AC6 : Token invalide/expiré → 401 TOKEN_INVALID
# ---------------------------------------------------------------------------


class TestTokenInvalid:
    """AC6 : Token invalide ou malformé → 401 TOKEN_INVALID"""

    def test_malformed_token_returns_401(self, client):
        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": "Bearer this.is.not.a.valid.jwt"},
        )
        assert response.status_code == 401

    def test_malformed_token_error_code(self, client):
        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": "Bearer invalid"},
        )
        body = response.json()
        assert body["error"] == "TOKEN_INVALID"
        assert "Token invalide ou expiré" in body["message"]

    def test_expired_token_returns_401(self, client, monkeypatch):
        """Un token créé avec une expiry passée doit retourner 401."""
        import jwt as pyjwt
        import services.auth_service as auth_svc
        from datetime import datetime, timedelta, timezone

        expired_payload = {
            "sub": "some-user-id",
            "type": "access",
            "exp": datetime.now(timezone.utc) - timedelta(minutes=5),
            "jti": "expired-jti-123",
        }
        expired_token = pyjwt.encode(
            expired_payload, auth_svc.SECRET_KEY, algorithm=auth_svc.ALGORITHM
        )

        response = client.post(
            LOGOUT_URL,
            headers={"Authorization": f"Bearer {expired_token}"},
        )
        assert response.status_code == 401
        assert response.json()["error"] == "TOKEN_INVALID"


# ---------------------------------------------------------------------------
# Backward Compatibility : tokens sans JTI restent valides
# ---------------------------------------------------------------------------


class TestBackwardCompatibility:
    """Tokens créés sans JTI (avant cette story) ne sont pas révocables mais fonctionnent."""

    def test_token_without_jti_is_valid(self, client):
        """Un token sans JTI doit être accepté (verify_token retourne le payload)."""
        import jwt as pyjwt
        import services.auth_service as auth_svc
        from datetime import datetime, timedelta, timezone

        payload_no_jti = {
            "sub": "some-user-id",
            "type": "access",
            "exp": datetime.now(timezone.utc) + timedelta(minutes=15),
        }
        token_no_jti = pyjwt.encode(
            payload_no_jti, auth_svc.SECRET_KEY, algorithm=auth_svc.ALGORITHM
        )

        result = auth_svc.verify_token(token_no_jti)
        assert result is not None, "Token sans JTI doit rester valide"
        assert result.get("jti") is None