diff --git a/app/routes/auth.py b/app/routes/auth.py index e3f33e8..6956622 100644 --- a/app/routes/auth.py +++ b/app/routes/auth.py @@ -110,12 +110,4 @@ async def refresh(refresh_data: RefreshTokenDTO): "token_type": "bearer", "user_id": str(current_user["_id"]), "is_admin": current_user["is_admin"] - } - -@auth_router.get( - path="/logout", - responses={401: {"model": HTTPError}} -) -async def logout(current_user: User = Depends(get_current_user)): - return {"message": "Logged out"} - + } \ No newline at end of file diff --git a/tests/image-too-large.jpg b/tests/image-too-large.jpg new file mode 100644 index 0000000..cac0d29 Binary files /dev/null and b/tests/image-too-large.jpg differ diff --git a/tests/test_images.py b/tests/test_images.py index e9e9a5f..4c49c9d 100644 --- a/tests/test_images.py +++ b/tests/test_images.py @@ -1,9 +1,12 @@ from test_main import * from PIL import Image import io +import os +from bson import ObjectId +from app.config import UPLOAD_DIR -def create_test_image(): - img = Image.new('RGB', (100, 100), color='red') +def create_test_image(size=(100, 100)): + img = Image.new('RGB', size, color='red') img_byte_arr = io.BytesIO() img.save(img_byte_arr, format='JPEG') img_byte_arr.seek(0) @@ -205,26 +208,34 @@ def test_image_with_invalid_pin_id(token): ) assert response.status_code == 404 # Le pin n'existe pas -# K.O -# -# def test_image_too_large(token): -# # Créer une grande image -# img = Image.fromarray(np.random.randint(0, 256, (8000, 8000, 3), dtype=np.uint8), 'RGB') -# buf = io.BytesIO() -# img.save(buf, format='JPEG', quality=100) -# buf.seek(0) -# buf.name = "large.jpg" +def test_image_too_large(token): + with open("tests/image-too-large.jpg", "rb") as f: + buf = io.BytesIO(f.read()) + buf.name = "large.jpg" -# # Vérifier la taille de l'image -# image_size = len(buf.getvalue()) -# print(f"Image size: {image_size} bytes ({image_size / (1024*1024):.2f} MB)") - -# response = client.post( -# "/image/pin/null/add", -# files={"image": ("large.jpg", buf, "image/jpeg")}, -# data={"exif_date": "2024-03-20T12:00:00", "caption": "Test caption"}, -# headers={"Authorization": f"Bearer {token}"} -# ) -# print(f"Response status: {response.status_code}") -# print(f"Response body: {response.text}") -# assert response.status_code == 413 \ No newline at end of file + response = client.post( + "/image/pin/null/add", + files={"image": ("large.jpg", buf, "image/jpeg")}, + data={"exif_date": "2024-03-20T12:00:00", "caption": "Test caption"}, + headers={"Authorization": f"Bearer {token}"} + ) + + assert response.status_code == 413 + +def test_get_image_file_not_found(token): + pin_id = create_test_pin(token) + image_id = add_test_image(token, pin_id) + + image_doc = db["images"].find_one({"_id": ObjectId(image_id)}) + image_hash = image_doc["image_hash"] + image_path = os.path.join(UPLOAD_DIR, f"{image_hash}.jpg") + + # On supprime le fichier image physiquement + if os.path.exists(image_path): + os.remove(image_path) + + # On tente de récupérer l'image + response = client.get(f"/image/{image_id}", headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 404 + assert response.json()["detail"] == "Image file not found" + diff --git a/tests/test_push.py b/tests/test_push.py new file mode 100644 index 0000000..982de37 --- /dev/null +++ b/tests/test_push.py @@ -0,0 +1,30 @@ +import pytest +from fastapi.testclient import TestClient +from fastapi import FastAPI +from test_main import * + +# Exemple de fausse souscription push +FAKE_SUB = { + "endpoint": "https://test", + "keys": {"p256dh": "key", "auth": "auth"} +} + +def test_push_subscribe_first(token): + response = client.post("/push/subscribe", json=FAKE_SUB, headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 200 + assert response.json()["message"] == "Push subscription successful" + +def test_push_subscribe_duplicate(token): + response = client.post("/push/subscribe", json=FAKE_SUB, headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 400 + assert "already exists" in response.json()["detail"] + +def test_push_subscribe_second(token): + # On s'abonne avec un nouvel endpoint + sub = { + "endpoint": "https://test2", + "keys": {"p256dh": "key2", "auth": "auth2"} + } + response = client.post("/push/subscribe", json=sub, headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 200 + assert response.json()["message"] == "Push subscription successful" \ No newline at end of file diff --git a/tests/test_push_service.py b/tests/test_push_service.py new file mode 100644 index 0000000..71d2459 --- /dev/null +++ b/tests/test_push_service.py @@ -0,0 +1,125 @@ +import sys +import os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))) + +import pytest +from unittest.mock import patch, MagicMock, AsyncMock +from bson import ObjectId +from api.app.push_service import PushService + +# Fixture pour obtenir une instance du service à tester +@pytest.fixture +def push_service(): + return PushService() + +# Teste l'envoi de notification quand l'utilisateur existe et a des abonnements +@patch('api.app.push_service.users_collection') +@patch('api.app.push_service.PushService.send_notification_to_subscription', new_callable=AsyncMock) +@pytest.mark.asyncio +async def test_send_notification_success(mock_send, mock_users, push_service): + user_id = ObjectId() + # Simule un utilisateur avec une souscription push + mock_users.find_one.return_value = {"push_subscriptions": ["{\"endpoint\": \"https://test\"}"]} + mock_send.return_value = True + + payload = {"msg": "test"} + result = await push_service.send_notification(user_id, payload) + + assert result is None or result is True + mock_send.assert_awaited() + +# Teste le cas où l'utilisateur n'existe pas +@patch('api.app.push_service.users_collection') +@pytest.mark.asyncio +async def test_send_notification_no_user(mock_users, push_service): + user_id = ObjectId() + mock_users.find_one.return_value = None # Aucun utilisateur trouvé + + payload = {"msg": "test"} + result = await push_service.send_notification(user_id, payload) + + assert result is False + +# Teste le cas où l'utilisateur n'a pas de souscriptions push +@patch('api.app.push_service.users_collection') +@pytest.mark.asyncio +async def test_send_notification_no_subs(mock_users, push_service): + user_id = ObjectId() + + mock_users.find_one.return_value = {"push_subscriptions": []} # Pas de souscriptions + + payload = {"msg": "test"} + result = await push_service.send_notification(user_id, payload) + + assert result is False + +# Teste l'envoi d'une notification à une souscription valide +@patch('api.app.push_service.webpush') +@patch('api.app.push_service.get_audience', return_value='https://test') +@pytest.mark.asyncio +async def test_send_notification_to_subscription_success(mock_aud, mock_webpush, push_service): + subscription = '{"endpoint": "https://test"}' + + payload = {"msg": "test"} + user_id = ObjectId() + result = await push_service.send_notification_to_subscription(subscription, payload, user_id) + + assert result is True + mock_webpush.assert_called_once() + +# Teste le cas où webpush lève une exception (échec d'envoi) +@patch('api.app.push_service.webpush', side_effect=Exception('fail')) +@pytest.mark.asyncio +async def test_send_notification_to_subscription_exception(mock_webpush, push_service): + subscription = '{"endpoint": "https://test"}' + + payload = {"msg": "test"} + user_id = ObjectId() + result = await push_service.send_notification_to_subscription(subscription, payload, user_id) + + assert result is False + +# Teste l'envoi à plusieurs souscriptions (succès et échec) +@patch('api.app.push_service.PushService.send_notification', new_callable=AsyncMock) +@pytest.mark.asyncio +async def test_send_notification_to_all(mock_send, push_service): + mock_send.side_effect = [True, False] # Un succès, un échec + + subscriptions = ["sub1", "sub2"] + payload = {"msg": "test"} + result = await push_service.send_notification_to_all(subscriptions, payload) + + assert result["success"] == 1 + assert result["failed"] == 1 + +# Teste la création du payload de notification +def test_create_notification_payload(push_service): + title = "Titre" + body = "Corps" + payload = push_service.create_notification_payload(title, body) + + assert payload["notification"]["title"] == title + assert payload["notification"]["body"] == body + assert "icon" in payload["notification"] + +# Teste la suppression d'une souscription (succès) +@patch('api.app.push_service.users_collection') +@pytest.mark.asyncio +async def test_delete_subscription_success(mock_users, push_service): + mock_users.update_one.return_value = MagicMock() # Simule la suppression + + user_id = ObjectId() + subscription = "sub" + result = await push_service.delete_subscription(subscription, user_id) + + assert result is True + +# Teste la suppression d'une souscription quand une exception est levée (échec) +@patch('api.app.push_service.users_collection.update_one', side_effect=Exception('fail')) +@pytest.mark.asyncio +async def test_delete_subscription_exception(mock_update_one, push_service): + user_id = ObjectId() + subscription = "sub" + result = await push_service.delete_subscription(subscription, user_id) + + assert result is False \ No newline at end of file diff --git a/tests/test_user.py b/tests/test_user.py index d12eab4..687533f 100644 --- a/tests/test_user.py +++ b/tests/test_user.py @@ -70,4 +70,21 @@ def test_get_user_not_found(token): # On utilise un ID qui n'existe probablement pas non_existent_id = "507f1f77bcf86cd799439011" response = client.get(f"/user/{non_existent_id}", headers={"Authorization": f"Bearer {token}"}) - assert response.status_code == 404 \ No newline at end of file + assert response.status_code == 404 + +def test_refresh_token_success(): + login_response = client.post("/login", data={"username": "testuser", "password": "testpassword"}) + refresh_token = login_response.json()["refresh_token"] + + response = client.post("/refresh-token", json={"refresh_token": refresh_token}) + assert response.status_code == 200 + data = response.json() + assert "access_token" in data + assert "refresh_token" in data + assert data["token_type"] == "bearer" + assert data["user_id"] is not None + assert data["is_admin"] is False + +def test_refresh_token_invalid(): + response = client.post("/refresh-token", json={"refresh_token": "invalidtoken"}) + assert response.status_code == 401 \ No newline at end of file