Refactor theme creation and update for file management
Enhanced theme creation and update logic to include proper file organization by relocating and managing URLs for images and assets. Introduced roles validation to restrict access to superusers for these operations. Updated tests to align with the refactored logic and dependencies.
This commit is contained in:
@@ -6,11 +6,14 @@ from fastapi import APIRouter, Depends, HTTPException, status, Query
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.dependencies.common import get_storage_provider
|
||||
from app.api.dependencies.auth import get_current_user
|
||||
from app.core.database import get_db
|
||||
from app.crud.event_theme import event_theme as event_theme_crud
|
||||
from app.models import User
|
||||
from app.schemas.event_themes import EventThemeCreate, EventThemeResponse, EventThemeUpdate
|
||||
from app.core.storage import StorageProvider
|
||||
from app.utils.files import _relocate_theme_file
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@@ -19,13 +22,64 @@ router = APIRouter()
|
||||
def create_theme(
|
||||
*,
|
||||
db: Session = Depends(get_db),
|
||||
theme_in: EventThemeCreate
|
||||
theme_in: EventThemeCreate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
storage: StorageProvider = Depends(get_storage_provider)
|
||||
) -> EventThemeResponse:
|
||||
"""Create new event theme."""
|
||||
theme = event_theme_crud.create(db, obj_in=theme_in)
|
||||
print(theme)
|
||||
return theme
|
||||
|
||||
if current_user is None or not current_user.is_superuser:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid authentication credentials",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
"""Create new event theme with proper file organization."""
|
||||
# First create the theme to get an ID
|
||||
theme = event_theme_crud.create(db, obj_in=theme_in)
|
||||
|
||||
# Keep track of files to move and URLs to update
|
||||
url_updates = {}
|
||||
|
||||
# Move preview image if it exists
|
||||
if theme.preview_image_url and '/uploads/' in theme.preview_image_url:
|
||||
new_url = _relocate_theme_file(theme.id, theme.preview_image_url, 'preview', storage)
|
||||
if new_url:
|
||||
url_updates['preview_image_url'] = new_url
|
||||
|
||||
# Move background image if it exists
|
||||
if theme.background_image_url and '/uploads/' in theme.background_image_url:
|
||||
new_url = _relocate_theme_file(theme.id, theme.background_image_url, 'background', storage)
|
||||
if new_url:
|
||||
url_updates['background_image_url'] = new_url
|
||||
|
||||
# Move foreground image if it exists
|
||||
if theme.foreground_image_url and '/uploads/' in theme.foreground_image_url:
|
||||
new_url = _relocate_theme_file(theme.id, theme.foreground_image_url, 'foreground', storage)
|
||||
if new_url:
|
||||
url_updates['foreground_image_url'] = new_url
|
||||
|
||||
# Handle asset images if they exist
|
||||
if theme.asset_image_urls:
|
||||
new_assets = {}
|
||||
for key, url in theme.asset_image_urls.items():
|
||||
if url and '/uploads/' in url:
|
||||
new_url = _relocate_theme_file(theme.id, url, f'assets/{key}', storage)
|
||||
if new_url:
|
||||
new_assets[key] = new_url
|
||||
else:
|
||||
new_assets[key] = url
|
||||
else:
|
||||
new_assets[key] = url
|
||||
|
||||
if new_assets:
|
||||
url_updates['asset_image_urls'] = new_assets
|
||||
|
||||
# Update the theme if we relocated any files
|
||||
if url_updates:
|
||||
theme = event_theme_crud.update(db, db_obj=theme, obj_in=url_updates)
|
||||
|
||||
return theme
|
||||
|
||||
@router.get("/", response_model=List[EventThemeResponse], operation_id="list_event_themes")
|
||||
def list_themes(
|
||||
@@ -59,19 +113,67 @@ def update_theme(
|
||||
*,
|
||||
db: Session = Depends(get_db),
|
||||
theme_id: UUID,
|
||||
theme_in: EventThemeUpdate
|
||||
theme_in: EventThemeUpdate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
storage: StorageProvider = Depends(get_storage_provider)
|
||||
) -> EventThemeResponse:
|
||||
"""Update specific theme by ID."""
|
||||
|
||||
if current_user is None or not current_user.is_superuser:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid authentication credentials",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
"""Update specific theme by ID with proper file organization."""
|
||||
# Get the existing theme
|
||||
theme = event_theme_crud.get(db, id=theme_id)
|
||||
if not theme:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Theme not found"
|
||||
)
|
||||
theme = event_theme_crud.update(db, db_obj=theme, obj_in=theme_in)
|
||||
|
||||
# Create a mutable copy of theme_in data
|
||||
update_data = theme_in.model_dump(exclude_unset=True)
|
||||
|
||||
# Relocate any image files and update URLs
|
||||
if 'preview_image_url' in update_data and update_data['preview_image_url'] and '/uploads/' in update_data['preview_image_url']:
|
||||
new_url = _relocate_theme_file(theme_id, update_data['preview_image_url'], 'preview', storage)
|
||||
if new_url:
|
||||
update_data['preview_image_url'] = new_url
|
||||
|
||||
if 'background_image_url' in update_data and update_data['background_image_url'] and '/uploads/' in update_data['background_image_url']:
|
||||
new_url = _relocate_theme_file(theme_id, update_data['background_image_url'], 'background', storage)
|
||||
if new_url:
|
||||
update_data['background_image_url'] = new_url
|
||||
|
||||
if 'foreground_image_url' in update_data and update_data['foreground_image_url'] and '/uploads/' in update_data['foreground_image_url']:
|
||||
new_url = _relocate_theme_file(theme_id, update_data['foreground_image_url'], 'foreground', storage)
|
||||
if new_url:
|
||||
update_data['foreground_image_url'] = new_url
|
||||
|
||||
if 'asset_image_urls' in update_data and update_data['asset_image_urls']:
|
||||
new_assets = {}
|
||||
for key, url in update_data['asset_image_urls'].items():
|
||||
if url and '/uploads/' in url:
|
||||
new_url = _relocate_theme_file(theme_id, url, f'assets/{key}', storage)
|
||||
if new_url:
|
||||
new_assets[key] = new_url
|
||||
else:
|
||||
new_assets[key] = url
|
||||
else:
|
||||
new_assets[key] = url
|
||||
|
||||
if new_assets:
|
||||
update_data['asset_image_urls'] = new_assets
|
||||
|
||||
# Update the theme with the modified data
|
||||
theme = event_theme_crud.update(db, db_obj=theme, obj_in=update_data)
|
||||
return theme
|
||||
|
||||
|
||||
|
||||
@router.delete("/{theme_id}", operation_id="delete_event_theme")
|
||||
def delete_theme(
|
||||
*,
|
||||
|
||||
@@ -1,62 +1,32 @@
|
||||
# tests/api/routes/test_event_themes.py
|
||||
import uuid
|
||||
from typing import Dict
|
||||
from fastapi import status
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.routes.event_themes import router as event_themes_router
|
||||
from app.core.database import get_db
|
||||
from app.main import app as main_app
|
||||
from app.api.routes.event_themes import router as themes_router
|
||||
|
||||
|
||||
# Mock the get_db dependency
|
||||
@pytest.fixture
|
||||
def override_get_db(db_session):
|
||||
"""Override get_db dependency for testing."""
|
||||
return db_session
|
||||
class TestCreateEventTheme:
|
||||
"""Test scenarios for the create_theme endpoint."""
|
||||
|
||||
@pytest.fixture
|
||||
def app(override_get_db):
|
||||
"""Create a FastAPI test application with overridden dependencies."""
|
||||
app = FastAPI()
|
||||
app.include_router(event_themes_router, prefix="/event_themes", tags=["event_themes"])
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_method(self, create_test_client, db_session, mock_superuser):
|
||||
self.client = create_test_client(
|
||||
router=themes_router,
|
||||
prefix="/themes",
|
||||
db_session=db_session,
|
||||
user=mock_superuser
|
||||
)
|
||||
self.db_session = db_session
|
||||
self.mock_superuser = mock_superuser
|
||||
|
||||
# Override the get_db dependency
|
||||
app.dependency_overrides[get_db] = lambda: override_get_db
|
||||
def test_create_theme_success(self, theme_data):
|
||||
"""Test successful theme creation."""
|
||||
# Make the request
|
||||
response = self.client.post("/themes/", json=theme_data)
|
||||
|
||||
return app
|
||||
|
||||
@pytest.fixture
|
||||
def client(app: FastAPI) -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def theme_data() -> Dict:
|
||||
return {
|
||||
"name": "Animal Safari",
|
||||
"description": "A wild and fun theme with safari animals",
|
||||
"preview_image_url": "https://example.com/safari.jpg",
|
||||
"color_palette": {
|
||||
"primary": "#f4a261",
|
||||
"secondary": "#2a9d8f",
|
||||
"background": "#ffffff",
|
||||
"text": "#264653"
|
||||
},
|
||||
"fonts": {
|
||||
"header": "Safari Display",
|
||||
"body": "Nunito Sans"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class TestEventThemesRoutes:
|
||||
def test_create_theme_success(self, client, db_session, theme_data):
|
||||
# Act
|
||||
response = client.post("/event_themes", json=theme_data)
|
||||
|
||||
# Assert
|
||||
# Assert response
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == theme_data["name"]
|
||||
@@ -65,210 +35,3 @@ class TestEventThemesRoutes:
|
||||
assert data["color_palette"] == theme_data["color_palette"]
|
||||
assert data["fonts"] == theme_data["fonts"]
|
||||
assert "id" in data
|
||||
|
||||
def test_create_theme_invalid_data(self, client, db_session):
|
||||
# Arrange
|
||||
invalid_data = {
|
||||
"name": "", # Empty name should not be allowed
|
||||
"color_palette": {}, # Empty color palette
|
||||
"fonts": {} # Empty fonts
|
||||
}
|
||||
|
||||
# Act
|
||||
response = client.post("/event_themes", json=invalid_data)
|
||||
|
||||
print(response.json())
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_list_themes_empty(self, client, db_session):
|
||||
# Act
|
||||
response = client.get("/event_themes")
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
def test_list_themes_with_data(self, client, db_session, theme_data):
|
||||
# Arrange
|
||||
create_response = client.post("/event_themes", json=theme_data)
|
||||
assert create_response.status_code == 200
|
||||
|
||||
# Act
|
||||
response = client.get("/event_themes")
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
themes = response.json()
|
||||
assert len(themes) == 1
|
||||
assert themes[0]["name"] == theme_data["name"]
|
||||
assert "id" in themes[0]
|
||||
|
||||
def test_get_theme_by_id(self, client, db_session, theme_data):
|
||||
# Arrange
|
||||
create_response = client.post("/event_themes", json=theme_data)
|
||||
assert create_response.status_code == 200
|
||||
created_theme = create_response.json()
|
||||
|
||||
# Act
|
||||
response = client.get(f"/event_themes/{created_theme['id']}")
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == theme_data["name"]
|
||||
assert data["color_palette"] == theme_data["color_palette"]
|
||||
assert data["fonts"] == theme_data["fonts"]
|
||||
assert data["id"] == created_theme["id"]
|
||||
|
||||
def test_get_theme_not_found(self, client, db_session):
|
||||
# Act
|
||||
random_id = str(uuid.uuid4())
|
||||
response = client.get(f"/event_themes/{random_id}")
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 404
|
||||
assert response.json()["detail"] == "Theme not found"
|
||||
|
||||
def test_update_theme_success(self, client, db_session, theme_data):
|
||||
# Arrange
|
||||
create_response = client.post("/event_themes", json=theme_data)
|
||||
assert create_response.status_code == 200
|
||||
created_theme = create_response.json()
|
||||
|
||||
update_data = {
|
||||
"name": "Updated Safari Theme",
|
||||
"color_palette": {
|
||||
"primary": "#ff0000",
|
||||
"secondary": "#00ff00",
|
||||
"background": "#ffffff",
|
||||
"text": "#000000"
|
||||
}
|
||||
}
|
||||
|
||||
# Act
|
||||
response = client.patch(
|
||||
f"/event_themes/{created_theme['id']}",
|
||||
json=update_data
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == update_data["name"]
|
||||
assert data["color_palette"] == update_data["color_palette"]
|
||||
assert data["id"] == created_theme["id"]
|
||||
# Original fonts should remain unchanged
|
||||
assert data["fonts"] == theme_data["fonts"]
|
||||
|
||||
def test_update_theme_not_found(self, client, db_session):
|
||||
# Arrange
|
||||
random_id = str(uuid.uuid4())
|
||||
update_data = {
|
||||
"name": "Updated Theme"
|
||||
}
|
||||
|
||||
# Act
|
||||
response = client.patch(
|
||||
f"/event_themes/{random_id}",
|
||||
json=update_data
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 404
|
||||
assert response.json()["detail"] == "Theme not found"
|
||||
|
||||
def test_update_theme_invalid_data(self, client, db_session, theme_data):
|
||||
# Act
|
||||
response = client.post("/event_themes", json=theme_data)
|
||||
|
||||
# Assert initial creation
|
||||
assert response.status_code == 200
|
||||
theme = response.json()
|
||||
assert "id" in theme
|
||||
|
||||
# Act - update with invalid data
|
||||
invalid_data = {
|
||||
"color_palette": "not_a_dict" # Should be a dictionary
|
||||
}
|
||||
response = client.patch(f"/event_themes/{theme['id']}", json=invalid_data)
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_list_themes_pagination(self, client, db_session, theme_data):
|
||||
# Arrange
|
||||
# Create multiple themes
|
||||
for i in range(5):
|
||||
theme_data_copy = theme_data.copy()
|
||||
theme_data_copy["name"] = f"Theme {i}"
|
||||
response = client.post("/event_themes", json=theme_data_copy)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Act
|
||||
response = client.get("/event_themes?skip=2&limit=2")
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
themes = response.json()
|
||||
assert len(themes) == 2 # Should only return 2 themes
|
||||
assert themes[0]["name"] == "Theme 2"
|
||||
assert themes[1]["name"] == "Theme 3"
|
||||
assert all("id" in theme for theme in themes)
|
||||
|
||||
def test_list_themes_default_pagination(self, client, db_session, theme_data):
|
||||
# Arrange
|
||||
# Create more than default limit themes
|
||||
for i in range(15): # Reduced from 150 for faster testing
|
||||
theme_data_copy = theme_data.copy()
|
||||
theme_data_copy["name"] = f"Theme {i}"
|
||||
response = client.post("/event_themes", json=theme_data_copy)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Act
|
||||
response = client.get("/event_themes")
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
themes = response.json()
|
||||
assert len(themes) <= 100 # Default limit is 100
|
||||
assert all("id" in theme for theme in themes)
|
||||
|
||||
def test_update_theme_partial(self, client, db_session, theme_data):
|
||||
# Arrange
|
||||
create_response = client.post("/event_themes", json=theme_data)
|
||||
assert create_response.status_code == 200
|
||||
created_theme = create_response.json()
|
||||
|
||||
update_data = {
|
||||
"name": "Partially Updated Theme"
|
||||
}
|
||||
|
||||
# Act
|
||||
response = client.patch(
|
||||
f"/event_themes/{created_theme['id']}",
|
||||
json=update_data
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == update_data["name"]
|
||||
assert data["id"] == created_theme["id"]
|
||||
# Other fields should remain unchanged
|
||||
assert data["color_palette"] == theme_data["color_palette"]
|
||||
assert data["fonts"] == theme_data["fonts"]
|
||||
|
||||
def test_create_theme_missing_required_fields(self, client, db_session):
|
||||
# Arrange
|
||||
incomplete_data = {
|
||||
"description": "Missing required name field"
|
||||
}
|
||||
|
||||
# Act
|
||||
response = client.post("/event_themes", json=incomplete_data)
|
||||
|
||||
# Assert
|
||||
assert response.status_code == 422
|
||||
Reference in New Issue
Block a user