import uuid from pathlib import Path from datetime import datetime, timezone, timedelta import pytest from fastapi.testclient import TestClient from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app.core.security import create_access_token from app.db.database import get_async_session from app.main import app from app.models.base import Base from app.models.utilisateur import Utilisateur, Session as UserSession from app.models.role import Role # ── Fixtures ────────────────────────────────────────────────────────────────── @pytest.fixture(scope="function") async def test_db(): """Database setup for tests using in-memory SQLite""" engine = create_async_engine( "sqlite+aiosqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False} ) async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # Create all tables async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # Insert test role async with async_session() as session: role = Role( id=uuid.uuid4(), nom="UTILISATEUR", description="Utilisateur standard", permissions=[] ) session.add(role) await session.commit() # Override dependency async def override_get_async_session(): async with async_session() as session: yield session app.dependency_overrides[get_async_session] = override_get_async_session yield async_session # Cleanup await engine.dispose() app.dependency_overrides.clear() @pytest.fixture async def test_user(test_db): """Create a test user""" async with test_db() as session: user = Utilisateur( id=uuid.uuid4(), email="test@example.com", mot_de_passe_hash="hashed_password", nom="Test", prenom="User", type_utilisateur="UTILISATEUR", est_actif=True, ) session.add(user) await session.commit() await session.refresh(user) return user @pytest.fixture def client(): """FastAPI test client""" class PrefixedClient(TestClient): def request(self, method: str, url: str, *args, **kwargs): if url.startswith("/postes"): url = f"/api/v1{url}" return super().request(method, url, *args, **kwargs) return PrefixedClient(app) # ── Helper Functions ────────────────────────────────────────────────────────── async def get_auth_header(test_db, user_id: str): """Get Authorization header with valid JWT token and register session""" token, jti = create_access_token(user_id=user_id, role="UTILISATEUR") async with test_db() as session: user_session = UserSession( utilisateur_id=uuid.UUID(user_id) if isinstance(user_id, str) else user_id, jti=jti, est_actif=True, date_expiration=datetime.now(timezone.utc) + timedelta(days=1) ) session.add(user_session) await session.commit() return {"Authorization": f"Bearer {token}"} # ── Tests: Postes ───────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_creer_poste(client: TestClient, test_user, test_db): """Test creating a new post""" headers = await get_auth_header(test_db, str(test_user.id)) body = { "titre": "Mon premier poste", "contenu": "Ceci est un test", "categorie": "TEST", "tags": ["python", "test"] } response = client.post("/postes", json=body, headers=headers) assert response.status_code == 201 assert "id" in response.json()["data"] assert response.json()["data"]["titre"] == "Mon premier poste" @pytest.mark.asyncio async def test_list_postes(client: TestClient, test_user, test_db): """Test listing posts with pagination""" headers = await get_auth_header(test_db, str(test_user.id)) # Create a post first body = {"titre": "Test", "contenu": "Content", "categorie": "TEST"} client.post("/postes", json=body, headers=headers) # List posts response = client.get("/postes?page=0&size=10", headers=headers) assert response.status_code == 200 assert len(response.json()["data"]["items"]) >= 1 @pytest.mark.asyncio async def test_list_mes_postes(client: TestClient, test_user, test_db): """Test listing user's own posts""" headers = await get_auth_header(test_db, str(test_user.id)) # Create a post first body = {"titre": "Mon Post", "contenu": "Content", "categorie": "TEST"} client.post("/postes", json=body, headers=headers) # Get my posts response = client.get("/postes/me?page=0&size=10", headers=headers) assert response.status_code == 200 assert len(response.json()["data"]["items"]) == 1 assert response.json()["data"]["items"][0]["titre"] == "Mon Post" @pytest.mark.asyncio async def test_get_poste(client: TestClient, test_user, test_db): """Test getting a single post""" headers = await get_auth_header(test_db, str(test_user.id)) # Create a post body = {"titre": "Test Get", "contenu": "Content"} create_resp = client.post("/postes", json=body, headers=headers) poste_id = create_resp.json()["data"]["id"] # Get the post response = client.get(f"/postes/{poste_id}", headers=headers) assert response.status_code == 200 assert response.json()["data"]["id"] == poste_id assert response.json()["data"]["titre"] == "Test Get" @pytest.mark.asyncio async def test_supprimer_poste(client: TestClient, test_user, test_db): """Test deleting a post""" headers = await get_auth_header(test_db, str(test_user.id)) # Create a post body = {"titre": "Delete Me", "contenu": "Content"} create_resp = client.post("/postes", json=body, headers=headers) poste_id = create_resp.json()["data"]["id"] # Delete it delete_resp = client.delete(f"/postes/{poste_id}", headers=headers) assert delete_resp.status_code == 204 # Verify it's gone get_resp = client.get(f"/postes/{poste_id}", headers=headers) assert get_resp.status_code == 404 # ── Tests: Commentaires ─────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_ajouter_commentaire(client: TestClient, test_user, test_db): """Test adding a comment to a post""" headers = await get_auth_header(test_db, str(test_user.id)) # Create a post poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers) poste_id = poste.json()["data"]["id"] # Add comment comment_body = {"contenu": "Super post!"} response = client.post(f"/postes/{poste_id}/commentaires", json=comment_body, headers=headers) assert response.status_code == 201 assert response.json()["data"]["contenu"] == "Super post!" assert response.json()["data"]["posteId"] == poste_id @pytest.mark.asyncio async def test_list_commentaires(client: TestClient, test_user, test_db): """Test listing comments on a post""" headers = await get_auth_header(test_db, str(test_user.id)) # Create post and comments poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers) poste_id = poste.json()["data"]["id"] client.post(f"/postes/{poste_id}/commentaires", json={"contenu": "Comment 1"}, headers=headers) client.post(f"/postes/{poste_id}/commentaires", json={"contenu": "Comment 2"}, headers=headers) # List comments response = client.get(f"/postes/{poste_id}/commentaires", headers=headers) assert response.status_code == 200 assert len(response.json()["data"]["items"]) == 2 @pytest.mark.asyncio async def test_supprimer_commentaire(client: TestClient, test_user, test_db): """Test deleting a comment""" headers = await get_auth_header(test_db, str(test_user.id)) # Create post and comment poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers) poste_id = poste.json()["data"]["id"] comment_resp = client.post(f"/postes/{poste_id}/commentaires", json={"contenu": "Test"}, headers=headers) comment_id = comment_resp.json()["data"]["id"] # Delete comment delete_resp = client.delete(f"/postes/{poste_id}/commentaires/{comment_id}", headers=headers) assert delete_resp.status_code == 204 # Verify it's gone list_resp = client.get(f"/postes/{poste_id}/commentaires", headers=headers) assert len(list_resp.json()["data"]["items"]) == 0 # ── Tests: Réactions ────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_ajouter_reaction(client: TestClient, test_user, test_db): """Test adding a reaction to a post""" headers = await get_auth_header(test_db, str(test_user.id)) # Create post poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers) poste_id = poste.json()["data"]["id"] # Add reaction reaction_body = {"type": "LIKE"} response = client.post(f"/postes/{poste_id}/reactions", json=reaction_body, headers=headers) assert response.status_code == 201 assert response.json()["data"]["type"] == "LIKE" @pytest.mark.asyncio async def test_list_reactions(client: TestClient, test_user, test_db): """Test listing reactions on a post""" headers = await get_auth_header(test_db, str(test_user.id)) # Create post poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers) poste_id = poste.json()["data"]["id"] # Add reactions client.post(f"/postes/{poste_id}/reactions", json={"type": "LIKE"}, headers=headers) client.post(f"/postes/{poste_id}/reactions", json={"type": "HEART"}, headers=headers) # List reactions response = client.get(f"/postes/{poste_id}/reactions", headers=headers) assert response.status_code == 200 assert len(response.json()["data"]) == 2 @pytest.mark.asyncio async def test_supprimer_reaction(client: TestClient, test_user, test_db): """Test deleting a reaction""" headers = await get_auth_header(test_db, str(test_user.id)) # Create post and reaction poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers) poste_id = poste.json()["data"]["id"] reaction_resp = client.post(f"/postes/{poste_id}/reactions", json={"type": "LIKE"}, headers=headers) reaction_id = reaction_resp.json()["data"]["id"] # Delete reaction delete_resp = client.delete(f"/postes/{poste_id}/reactions/{reaction_id}", headers=headers) assert delete_resp.status_code == 204 # Verify it's gone list_resp = client.get(f"/postes/{poste_id}/reactions", headers=headers) assert len(list_resp.json()["data"]) == 0 # ── Tests: Permissions ──────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_cannot_delete_others_comment(client: TestClient, test_user, test_db): """Test that users can't delete comments from others""" headers1 = await get_auth_header(test_db, str(test_user.id)) user2 = Utilisateur( id=uuid.uuid4(), email="test2@example.com", mot_de_passe_hash="hashed", nom="User", prenom="Two", type_utilisateur="UTILISATEUR", est_actif=True, ) async with test_db() as session: session.add(user2) await session.commit() headers2 = await get_auth_header(test_db, str(user2.id)) # User 1 creates post poste = client.post("/postes", json={"titre": "Test", "contenu": "Content"}, headers=headers1) poste_id = poste.json()["data"]["id"] # User 2 adds comment comment_resp = client.post(f"/postes/{poste_id}/commentaires", json={"contenu": "Test"}, headers=headers2) comment_id = comment_resp.json()["data"]["id"] # User 1 tries to delete (should fail) delete_resp = client.delete(f"/postes/{poste_id}/commentaires/{comment_id}", headers=headers1) assert delete_resp.status_code == 403 @pytest.mark.asyncio async def test_admin_can_delete_others_content(client: TestClient, test_user, test_db): """Test that admins can delete content from other users""" pass