PunimTag Web Application - Major Feature Release #1
62
tests/test_api_health.py
Normal file
62
tests/test_api_health.py
Normal file
@ -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)
|
||||
|
||||
72
tests/test_api_jobs.py
Normal file
72
tests/test_api_jobs.py
Normal file
@ -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"
|
||||
|
||||
264
tests/test_api_people.py
Normal file
264
tests/test_api_people.py
Normal file
@ -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
|
||||
|
||||
429
tests/test_api_photos.py
Normal file
429
tests/test_api_photos.py
Normal file
@ -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]
|
||||
|
||||
297
tests/test_api_tags.py
Normal file
297
tests/test_api_tags.py
Normal file
@ -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)
|
||||
|
||||
273
tests/test_api_users.py
Normal file
273
tests/test_api_users.py
Normal file
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user