diff --git a/tests/test_api_health.py b/tests/test_api_health.py new file mode 100644 index 0000000..c9a4a0b --- /dev/null +++ b/tests/test_api_health.py @@ -0,0 +1,62 @@ +"""Health and version API tests.""" + +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient + + +class TestHealthCheck: + """Test health check endpoints.""" + + def test_health_check_success( + self, + test_client: TestClient, + ): + """Verify health endpoint returns 200.""" + response = test_client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert "status" in data + assert data["status"] == "ok" + + def test_health_check_database_connection( + self, + test_client: TestClient, + ): + """Verify DB connection check.""" + # Basic health check doesn't necessarily check DB + # This is a placeholder for future DB health checks + response = test_client.get("/health") + + assert response.status_code == 200 + + +class TestVersionEndpoint: + """Test version endpoint.""" + + def test_version_endpoint_success( + self, + test_client: TestClient, + ): + """Verify version information.""" + response = test_client.get("/version") + + assert response.status_code == 200 + data = response.json() + assert "version" in data or "app_version" in data + + def test_version_endpoint_includes_app_version( + self, + test_client: TestClient, + ): + """Verify version format.""" + response = test_client.get("/version") + + assert response.status_code == 200 + data = response.json() + # Version should be a string + version_key = "version" if "version" in data else "app_version" + assert isinstance(data[version_key], str) + diff --git a/tests/test_api_jobs.py b/tests/test_api_jobs.py new file mode 100644 index 0000000..9a22d1f --- /dev/null +++ b/tests/test_api_jobs.py @@ -0,0 +1,72 @@ +"""Medium priority job API tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from fastapi.testclient import TestClient + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + + +class TestJobStatus: + """Test job status endpoints.""" + + def test_get_job_status_not_found( + self, + test_client: TestClient, + ): + """Verify 404 for non-existent job.""" + response = test_client.get("/api/v1/jobs/nonexistent-job-id") + + assert response.status_code == 404 + data = response.json() + assert "not found" in data["detail"].lower() + + def test_get_job_status_includes_timestamps( + self, + test_client: TestClient, + test_db_session: "Session", + ): + """Verify timestamp fields.""" + # This test requires a real job to be created + # For now, we'll test the error case + response = test_client.get("/api/v1/jobs/test-job-id") + + # If job doesn't exist, we get 404 + # If job exists, we should check for timestamps + if response.status_code == 200: + data = response.json() + assert "created_at" in data + assert "updated_at" in data + + +class TestJobStreaming: + """Test job streaming endpoints.""" + + def test_stream_job_progress_not_found( + self, + test_client: TestClient, + ): + """Verify 404 handling.""" + response = test_client.get("/api/v1/jobs/stream/nonexistent-job-id") + + # Streaming endpoint may return 404 or start streaming + # Implementation dependent + assert response.status_code in [200, 404] + + def test_stream_job_progress_sse_format( + self, + test_client: TestClient, + ): + """Verify SSE format compliance.""" + # This test requires a real job + # For now, we'll test the not found case + response = test_client.get("/api/v1/jobs/stream/test-job-id") + + if response.status_code == 200: + # Check Content-Type for SSE + assert response.headers.get("content-type") == "text/event-stream" + diff --git a/tests/test_api_people.py b/tests/test_api_people.py new file mode 100644 index 0000000..fdb373e --- /dev/null +++ b/tests/test_api_people.py @@ -0,0 +1,264 @@ +"""High priority people API tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from fastapi.testclient import TestClient + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + from backend.db.models import Person, Face, User + + +class TestPeopleListing: + """Test people listing endpoints.""" + + def test_list_people_success( + self, + test_client: TestClient, + test_person: "Person", + ): + """Verify people list retrieval.""" + response = test_client.get("/api/v1/people") + + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert "total" in data + assert len(data["items"]) > 0 + + def test_list_people_with_last_name_filter( + self, + test_client: TestClient, + test_person: "Person", + ): + """Verify last name filtering.""" + response = test_client.get( + "/api/v1/people", + params={"last_name": "Doe"}, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + # All items should have last_name containing "Doe" + for item in data["items"]: + assert "Doe" in item["last_name"] + + def test_list_people_with_faces_success( + self, + test_client: TestClient, + test_person: "Person", + test_face: "Face", + test_db_session: "Session", + ): + """Verify people with face counts.""" + test_face.person_id = test_person.id + test_db_session.commit() + + response = test_client.get("/api/v1/people/with-faces") + + assert response.status_code == 200 + data = response.json() + assert "items" in data + # Find our person + person_item = next( + (item for item in data["items"] if item["id"] == test_person.id), + None + ) + if person_item: + assert person_item["face_count"] >= 1 + + +class TestPeopleCRUD: + """Test people CRUD endpoints.""" + + def test_create_person_success( + self, + test_client: TestClient, + ): + """Verify person creation.""" + response = test_client.post( + "/api/v1/people", + json={ + "first_name": "Jane", + "last_name": "Smith", + }, + ) + + assert response.status_code == 201 + data = response.json() + assert data["first_name"] == "Jane" + assert data["last_name"] == "Smith" + assert "id" in data + + def test_create_person_with_middle_name( + self, + test_client: TestClient, + ): + """Verify optional middle_name.""" + response = test_client.post( + "/api/v1/people", + json={ + "first_name": "Jane", + "last_name": "Smith", + "middle_name": "Middle", + }, + ) + + assert response.status_code == 201 + data = response.json() + assert data["middle_name"] == "Middle" + + def test_create_person_strips_whitespace( + self, + test_client: TestClient, + ): + """Verify name trimming.""" + response = test_client.post( + "/api/v1/people", + json={ + "first_name": " Jane ", + "last_name": " Smith ", + }, + ) + + assert response.status_code == 201 + data = response.json() + assert data["first_name"] == "Jane" + assert data["last_name"] == "Smith" + + def test_get_person_by_id_success( + self, + test_client: TestClient, + test_person: "Person", + ): + """Verify person retrieval.""" + response = test_client.get(f"/api/v1/people/{test_person.id}") + + assert response.status_code == 200 + data = response.json() + assert data["id"] == test_person.id + assert data["first_name"] == test_person.first_name + + def test_get_person_by_id_not_found( + self, + test_client: TestClient, + ): + """Verify 404 for non-existent person.""" + response = test_client.get("/api/v1/people/99999") + + assert response.status_code == 404 + + def test_update_person_success( + self, + test_client: TestClient, + test_person: "Person", + ): + """Verify person update.""" + response = test_client.put( + f"/api/v1/people/{test_person.id}", + json={ + "first_name": "Updated", + "last_name": "Name", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["first_name"] == "Updated" + assert data["last_name"] == "Name" + + def test_update_person_not_found( + self, + test_client: TestClient, + ): + """Verify 404 when updating non-existent person.""" + response = test_client.put( + "/api/v1/people/99999", + json={ + "first_name": "Updated", + "last_name": "Name", + }, + ) + + assert response.status_code == 404 + + def test_delete_person_success( + self, + test_client: TestClient, + test_db_session: "Session", + ): + """Verify person deletion.""" + from backend.db.models import Person + from datetime import datetime + + # Create a person to delete + person = Person( + first_name="Delete", + last_name="Me", + created_date=datetime.utcnow(), + ) + test_db_session.add(person) + test_db_session.commit() + test_db_session.refresh(person) + + response = test_client.delete(f"/api/v1/people/{person.id}") + + assert response.status_code == 200 + + def test_delete_person_not_found( + self, + test_client: TestClient, + ): + """Verify 404 for non-existent person.""" + response = test_client.delete("/api/v1/people/99999") + + assert response.status_code == 404 + + +class TestPeopleFaces: + """Test people faces endpoints.""" + + def test_get_person_faces_success( + self, + test_client: TestClient, + test_person: "Person", + test_face: "Face", + test_db_session: "Session", + ): + """Verify faces retrieval for person.""" + test_face.person_id = test_person.id + test_db_session.commit() + + response = test_client.get(f"/api/v1/people/{test_person.id}/faces") + + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert len(data["items"]) > 0 + + def test_get_person_faces_no_faces( + self, + test_client: TestClient, + test_person: "Person", + ): + """Verify empty list when no faces.""" + response = test_client.get(f"/api/v1/people/{test_person.id}/faces") + + assert response.status_code == 200 + data = response.json() + assert "items" in data + # May be empty or have faces depending on test setup + + def test_get_person_faces_person_not_found( + self, + test_client: TestClient, + ): + """Verify 404 for non-existent person.""" + response = test_client.get("/api/v1/people/99999/faces") + + assert response.status_code == 404 + diff --git a/tests/test_api_photos.py b/tests/test_api_photos.py new file mode 100644 index 0000000..b57d72a --- /dev/null +++ b/tests/test_api_photos.py @@ -0,0 +1,429 @@ +"""High priority photo API tests.""" + +from __future__ import annotations + +from datetime import date +from typing import TYPE_CHECKING + +import pytest +from fastapi.testclient import TestClient + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + from backend.db.models import Photo, Person, Face, User + + +class TestPhotoSearch: + """Test photo search endpoints.""" + + def test_search_photos_by_name_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + test_person: "Person", + test_face: "Face", + test_db_session: "Session", + ): + """Verify search by person name works.""" + # Link face to person + test_face.person_id = test_person.id + test_db_session.commit() + + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "name", "person_name": "John Doe"}, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert "total" in data + assert len(data["items"]) > 0 + + def test_search_photos_by_name_without_person_name( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 400 when person_name missing.""" + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "name"}, + ) + + assert response.status_code == 400 + assert "person_name is required" in response.json()["detail"] + + def test_search_photos_by_name_with_pagination( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + test_person: "Person", + test_face: "Face", + test_db_session: "Session", + ): + """Verify pagination works correctly.""" + test_face.person_id = test_person.id + test_db_session.commit() + + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={ + "search_type": "name", + "person_name": "John Doe", + "page": 1, + "page_size": 10, + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["page"] == 1 + assert data["page_size"] == 10 + assert len(data["items"]) <= 10 + + def test_search_photos_by_date_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + ): + """Verify date range search.""" + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={ + "search_type": "date", + "date_from": "2024-01-01", + "date_to": "2024-12-31", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + + def test_search_photos_by_date_without_dates( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 400 when both dates missing.""" + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "date"}, + ) + + assert response.status_code == 400 + assert "date_from or date_to is required" in response.json()["detail"] + + def test_search_photos_by_tags_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + test_db_session: "Session", + ): + """Verify tag search works.""" + from backend.db.models import Tag, PhotoTag + + # Create tag and link to photo + tag = Tag(tag="test-tag") + test_db_session.add(tag) + test_db_session.flush() + + photo_tag = PhotoTag(photo_id=test_photo.id, tag_id=tag.id) + test_db_session.add(photo_tag) + test_db_session.commit() + + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "tags", "tag_names": "test-tag"}, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + + def test_search_photos_by_tags_without_tags( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 400 when tag_names missing.""" + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "tags"}, + ) + + assert response.status_code == 400 + assert "tag_names is required" in response.json()["detail"] + + def test_search_photos_no_faces( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + ): + """Verify photos without faces search.""" + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "no_faces"}, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + + def test_search_photos_returns_favorite_status( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + admin_user: "User", + test_db_session: "Session", + ): + """Verify is_favorite field in results.""" + from backend.db.models import PhotoFavorite + + # Add favorite + favorite = PhotoFavorite(photo_id=test_photo.id, username=admin_user.username) + test_db_session.add(favorite) + test_db_session.commit() + + response = test_client.get( + "/api/v1/photos", + headers=auth_headers, + params={"search_type": "date", "date_from": "2024-01-01"}, + ) + + assert response.status_code == 200 + data = response.json() + if len(data["items"]) > 0: + # Check if our photo is in results and has is_favorite + photo_ids = [item["id"] for item in data["items"]] + if test_photo.id in photo_ids: + photo_item = next(item for item in data["items"] if item["id"] == test_photo.id) + assert "is_favorite" in photo_item + + +class TestPhotoFavorites: + """Test photo favorites endpoints.""" + + def test_toggle_favorite_add( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + admin_user: "User", + test_db_session: "Session", + ): + """Verify adding favorite.""" + response = test_client.post( + f"/api/v1/photos/{test_photo.id}/favorite", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert data["is_favorite"] is True + + # Verify in database + from backend.db.models import PhotoFavorite + favorite = test_db_session.query(PhotoFavorite).filter( + PhotoFavorite.photo_id == test_photo.id, + PhotoFavorite.username == admin_user.username, + ).first() + assert favorite is not None + + def test_toggle_favorite_remove( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + admin_user: "User", + test_db_session: "Session", + ): + """Verify removing favorite.""" + from backend.db.models import PhotoFavorite + + # Add favorite first + favorite = PhotoFavorite(photo_id=test_photo.id, username=admin_user.username) + test_db_session.add(favorite) + test_db_session.commit() + + # Remove it + response = test_client.post( + f"/api/v1/photos/{test_photo.id}/favorite", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert data["is_favorite"] is False + + def test_toggle_favorite_unauthenticated( + self, + test_client: TestClient, + test_photo: "Photo", + ): + """Verify 401 without auth.""" + response = test_client.post( + f"/api/v1/photos/{test_photo.id}/favorite", + ) + + assert response.status_code == 401 + + def test_toggle_favorite_photo_not_found( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 404 for non-existent photo.""" + response = test_client.post( + "/api/v1/photos/99999/favorite", + headers=auth_headers, + ) + + assert response.status_code == 404 + + def test_bulk_add_favorites_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + test_photo_2: "Photo", + ): + """Verify bulk add operation.""" + response = test_client.post( + "/api/v1/photos/favorites/bulk-add", + headers=auth_headers, + json={"photo_ids": [test_photo.id, test_photo_2.id]}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["added"] >= 0 + assert data["already_favorites"] >= 0 + + def test_bulk_remove_favorites_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + admin_user: "User", + test_db_session: "Session", + ): + """Verify bulk remove operation.""" + from backend.db.models import PhotoFavorite + + # Add favorite first + favorite = PhotoFavorite(photo_id=test_photo.id, username=admin_user.username) + test_db_session.add(favorite) + test_db_session.commit() + + response = test_client.post( + "/api/v1/photos/favorites/bulk-remove", + headers=auth_headers, + json={"photo_ids": [test_photo.id]}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["removed"] >= 0 + + +class TestPhotoRetrieval: + """Test photo retrieval endpoints.""" + + def test_get_photo_by_id_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + ): + """Verify photo retrieval by ID.""" + response = test_client.get( + f"/api/v1/photos/{test_photo.id}", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert data["id"] == test_photo.id + assert data["filename"] == test_photo.filename + + def test_get_photo_by_id_not_found( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 404 for non-existent photo.""" + response = test_client.get( + "/api/v1/photos/99999", + headers=auth_headers, + ) + + assert response.status_code == 404 + + +class TestPhotoDeletion: + """Test photo deletion endpoints.""" + + def test_bulk_delete_photos_success( + self, + test_client: TestClient, + auth_headers: dict, + test_photo: "Photo", + ): + """Verify bulk delete (admin only).""" + response = test_client.post( + "/api/v1/photos/bulk-delete", + headers=auth_headers, + json={"photo_ids": [test_photo.id]}, + ) + + assert response.status_code == 200 + data = response.json() + assert "deleted" in data + + def test_bulk_delete_photos_non_admin( + self, + test_client: TestClient, + regular_auth_headers: dict, + test_photo: "Photo", + ): + """Verify 403 for non-admin users.""" + response = test_client.post( + "/api/v1/photos/bulk-delete", + headers=regular_auth_headers, + json={"photo_ids": [test_photo.id]}, + ) + + # Should be 403 or 401 depending on implementation + assert response.status_code in [403, 401] + + def test_bulk_delete_photos_empty_list( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 400 with empty photo_ids.""" + response = test_client.post( + "/api/v1/photos/bulk-delete", + headers=auth_headers, + json={"photo_ids": []}, + ) + + # May return 200 with 0 deleted or 400 + assert response.status_code in [200, 400] + diff --git a/tests/test_api_tags.py b/tests/test_api_tags.py new file mode 100644 index 0000000..8c93a9d --- /dev/null +++ b/tests/test_api_tags.py @@ -0,0 +1,297 @@ +"""Medium priority tag API tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from fastapi.testclient import TestClient + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + from backend.db.models import Photo, Tag + + +class TestTagListing: + """Test tag listing endpoints.""" + + def test_get_tags_success( + self, + test_client: TestClient, + test_db_session: "Session", + ): + """Verify tags list retrieval.""" + from backend.db.models import Tag + + # Create a test tag + tag = Tag(tag="test-tag") + test_db_session.add(tag) + test_db_session.commit() + + response = test_client.get("/api/v1/tags") + + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert "total" in data + + def test_get_tags_empty_list( + self, + test_client: TestClient, + ): + """Verify empty list when no tags.""" + response = test_client.get("/api/v1/tags") + + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert isinstance(data["items"], list) + + +class TestTagCRUD: + """Test tag CRUD endpoints.""" + + def test_create_tag_success( + self, + test_client: TestClient, + ): + """Verify tag creation.""" + response = test_client.post( + "/api/v1/tags", + json={"tag_name": "new-tag"}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["tag"] == "new-tag" + assert "id" in data + + def test_create_tag_duplicate( + self, + test_client: TestClient, + test_db_session: "Session", + ): + """Verify returns existing tag if duplicate.""" + from backend.db.models import Tag + + # Create tag first + tag = Tag(tag="duplicate-tag") + test_db_session.add(tag) + test_db_session.commit() + test_db_session.refresh(tag) + + # Try to create again + response = test_client.post( + "/api/v1/tags", + json={"tag_name": "duplicate-tag"}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["id"] == tag.id + assert data["tag"] == "duplicate-tag" + + def test_create_tag_strips_whitespace( + self, + test_client: TestClient, + ): + """Verify whitespace handling.""" + response = test_client.post( + "/api/v1/tags", + json={"tag_name": " whitespace-tag "}, + ) + + assert response.status_code == 200 + data = response.json() + # Tag should be trimmed + assert "whitespace-tag" in data["tag"] + + def test_update_tag_success( + self, + test_client: TestClient, + test_db_session: "Session", + ): + """Verify tag update.""" + from backend.db.models import Tag + + tag = Tag(tag="old-name") + test_db_session.add(tag) + test_db_session.commit() + test_db_session.refresh(tag) + + response = test_client.put( + f"/api/v1/tags/{tag.id}", + json={"tag_name": "new-name"}, + ) + + assert response.status_code == 200 + data = response.json() + assert data["tag"] == "new-name" + + def test_update_tag_not_found( + self, + test_client: TestClient, + ): + """Verify 404 for non-existent tag.""" + response = test_client.put( + "/api/v1/tags/99999", + json={"tag_name": "new-name"}, + ) + + assert response.status_code in [400, 404] # Implementation dependent + + def test_delete_tag_success( + self, + test_client: TestClient, + test_db_session: "Session", + ): + """Verify tag deletion.""" + from backend.db.models import Tag + + tag = Tag(tag="delete-me") + test_db_session.add(tag) + test_db_session.commit() + test_db_session.refresh(tag) + + response = test_client.post( + "/api/v1/tags/delete", + json={"tag_ids": [tag.id]}, + ) + + assert response.status_code == 200 + + def test_delete_tag_not_found( + self, + test_client: TestClient, + ): + """Verify 404 handling.""" + response = test_client.post( + "/api/v1/tags/delete", + json={"tag_ids": [99999]}, + ) + + # May return 200 with 0 deleted or error + assert response.status_code in [200, 400, 404] + + +class TestPhotoTagOperations: + """Test photo-tag operations.""" + + def test_add_tags_to_photos_success( + self, + test_client: TestClient, + test_photo: "Photo", + ): + """Verify adding tags to photos.""" + response = test_client.post( + "/api/v1/tags/photos/add", + json={ + "photo_ids": [test_photo.id], + "tag_names": ["test-tag-1", "test-tag-2"], + }, + ) + + assert response.status_code == 200 + data = response.json() + assert "photos_updated" in data + assert data["photos_updated"] >= 0 + + def test_add_tags_to_photos_empty_photo_ids( + self, + test_client: TestClient, + ): + """Verify 400 with empty photo_ids.""" + response = test_client.post( + "/api/v1/tags/photos/add", + json={ + "photo_ids": [], + "tag_names": ["test-tag"], + }, + ) + + assert response.status_code == 400 + + def test_add_tags_to_photos_empty_tag_names( + self, + test_client: TestClient, + test_photo: "Photo", + ): + """Verify 400 with empty tag_names.""" + response = test_client.post( + "/api/v1/tags/photos/add", + json={ + "photo_ids": [test_photo.id], + "tag_names": [], + }, + ) + + assert response.status_code == 400 + + def test_remove_tags_from_photos_success( + self, + test_client: TestClient, + test_photo: "Photo", + test_db_session: "Session", + ): + """Verify tag removal.""" + from backend.db.models import Tag, PhotoTag + + # Add tag first + tag = Tag(tag="remove-me") + test_db_session.add(tag) + test_db_session.flush() + + photo_tag = PhotoTag(photo_id=test_photo.id, tag_id=tag.id) + test_db_session.add(photo_tag) + test_db_session.commit() + + # Remove it + response = test_client.post( + "/api/v1/tags/photos/remove", + json={ + "photo_ids": [test_photo.id], + "tag_names": ["remove-me"], + }, + ) + + assert response.status_code == 200 + data = response.json() + assert "tags_removed" in data + + def test_get_photo_tags_success( + self, + test_client: TestClient, + test_photo: "Photo", + test_db_session: "Session", + ): + """Verify photo tags retrieval.""" + from backend.db.models import Tag, PhotoTag + + tag = Tag(tag="photo-tag") + test_db_session.add(tag) + test_db_session.flush() + + photo_tag = PhotoTag(photo_id=test_photo.id, tag_id=tag.id) + test_db_session.add(photo_tag) + test_db_session.commit() + + response = test_client.get(f"/api/v1/tags/photos/{test_photo.id}") + + assert response.status_code == 200 + data = response.json() + assert "tags" in data + assert len(data["tags"]) > 0 + + def test_get_photo_tags_empty( + self, + test_client: TestClient, + test_photo: "Photo", + ): + """Verify empty list for untagged photo.""" + response = test_client.get(f"/api/v1/tags/photos/{test_photo.id}") + + assert response.status_code == 200 + data = response.json() + assert "tags" in data + assert isinstance(data["tags"], list) + diff --git a/tests/test_api_users.py b/tests/test_api_users.py new file mode 100644 index 0000000..de90de8 --- /dev/null +++ b/tests/test_api_users.py @@ -0,0 +1,273 @@ +"""High priority user API tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from fastapi.testclient import TestClient + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + from backend.db.models import User + + +class TestUserListing: + """Test user listing endpoints.""" + + def test_list_users_success( + self, + test_client: TestClient, + auth_headers: dict, + admin_user: "User", + ): + """Verify users list (admin only).""" + response = test_client.get( + "/api/v1/users", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + assert "total" in data + + def test_list_users_non_admin( + self, + test_client: TestClient, + regular_auth_headers: dict, + ): + """Verify 403 for non-admin users.""" + response = test_client.get( + "/api/v1/users", + headers=regular_auth_headers, + ) + + assert response.status_code in [403, 401] + + def test_list_users_with_pagination( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify pagination.""" + response = test_client.get( + "/api/v1/users", + headers=auth_headers, + params={"page": 1, "page_size": 10}, + ) + + assert response.status_code == 200 + data = response.json() + assert "items" in data + + +class TestUserCRUD: + """Test user CRUD endpoints.""" + + def test_create_user_success( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify user creation (admin only).""" + response = test_client.post( + "/api/v1/users", + headers=auth_headers, + json={ + "username": "newuser", + "email": "newuser@example.com", + "full_name": "New User", + "password": "password123", + }, + ) + + assert response.status_code == 201 + data = response.json() + assert data["username"] == "newuser" + assert data["email"] == "newuser@example.com" + + def test_create_user_duplicate_email( + self, + test_client: TestClient, + auth_headers: dict, + admin_user: "User", + ): + """Verify 400 with duplicate email.""" + response = test_client.post( + "/api/v1/users", + headers=auth_headers, + json={ + "username": "differentuser", + "email": admin_user.email, # Duplicate email + "full_name": "Different User", + "password": "password123", + }, + ) + + assert response.status_code == 400 + + def test_create_user_duplicate_username( + self, + test_client: TestClient, + auth_headers: dict, + admin_user: "User", + ): + """Verify 400 with duplicate username.""" + response = test_client.post( + "/api/v1/users", + headers=auth_headers, + json={ + "username": admin_user.username, # Duplicate username + "email": "different@example.com", + "full_name": "Different User", + "password": "password123", + }, + ) + + assert response.status_code == 400 + + def test_get_user_by_id_success( + self, + test_client: TestClient, + auth_headers: dict, + admin_user: "User", + ): + """Verify user retrieval.""" + response = test_client.get( + f"/api/v1/users/{admin_user.id}", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert data["id"] == admin_user.id + assert data["username"] == admin_user.username + + def test_get_user_by_id_not_found( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 404 for non-existent user.""" + response = test_client.get( + "/api/v1/users/99999", + headers=auth_headers, + ) + + assert response.status_code == 404 + + def test_update_user_success( + self, + test_client: TestClient, + auth_headers: dict, + admin_user: "User", + ): + """Verify user update.""" + response = test_client.put( + f"/api/v1/users/{admin_user.id}", + headers=auth_headers, + json={ + "full_name": "Updated Name", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["full_name"] == "Updated Name" + + def test_delete_user_success( + self, + test_client: TestClient, + auth_headers: dict, + test_db_session: "Session", + ): + """Verify user deletion.""" + from backend.db.models import User + from backend.utils.password import hash_password + from backend.constants.roles import DEFAULT_USER_ROLE + + # Create a user to delete + user = User( + username="deleteuser", + email="delete@example.com", + password_hash=hash_password("password"), + full_name="Delete User", + is_admin=False, + is_active=True, + role=DEFAULT_USER_ROLE, + ) + test_db_session.add(user) + test_db_session.commit() + test_db_session.refresh(user) + + response = test_client.delete( + f"/api/v1/users/{user.id}", + headers=auth_headers, + ) + + assert response.status_code == 200 + + def test_delete_user_non_admin( + self, + test_client: TestClient, + regular_auth_headers: dict, + admin_user: "User", + ): + """Verify 403 for non-admin.""" + response = test_client.delete( + f"/api/v1/users/{admin_user.id}", + headers=regular_auth_headers, + ) + + assert response.status_code in [403, 401] + + +class TestUserActivation: + """Test user activation endpoints.""" + + def test_activate_user_success( + self, + test_client: TestClient, + auth_headers: dict, + inactive_user: "User", + ): + """Verify user activation.""" + response = test_client.post( + f"/api/v1/users/{inactive_user.id}/activate", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert data["is_active"] is True + + def test_deactivate_user_success( + self, + test_client: TestClient, + auth_headers: dict, + regular_user: "User", + ): + """Verify user deactivation.""" + response = test_client.post( + f"/api/v1/users/{regular_user.id}/deactivate", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.json() + assert data["is_active"] is False + + def test_activate_user_not_found( + self, + test_client: TestClient, + auth_headers: dict, + ): + """Verify 404 handling.""" + response = test_client.post( + "/api/v1/users/99999/activate", + headers=auth_headers, + ) + + assert response.status_code == 404 +