Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2e93bf437c | |||
| e5bfb0cefc |
@@ -20,6 +20,7 @@ from app.schemas.cover_template import (
|
||||
UpdateCoverTemplateRequest,
|
||||
)
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from sqlalchemy.exc import OperationalError, ProgrammingError
|
||||
|
||||
from packages.domain.cover_template import CoverTemplate
|
||||
|
||||
@@ -35,10 +36,17 @@ def list_cover_templates(
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
repo: Any = Depends(get_cover_template_repository),
|
||||
) -> ListCoverTemplatesResponse:
|
||||
"""列出当前用户可见的封面模板(系统模板 + 用户自定义模板)。"""
|
||||
"""列出当前用户可见的封面模板(系统模板 + 用户自定义模板)。
|
||||
|
||||
当数据库表不存在时(迁移未执行),降级返回空列表而非 500。
|
||||
"""
|
||||
user_id = authenticated_user.user.id
|
||||
items = repo.list_for_user(user_id, skip=skip, limit=limit)
|
||||
total = repo.count_for_user(user_id)
|
||||
try:
|
||||
items = repo.list_for_user(user_id, skip=skip, limit=limit)
|
||||
total = repo.count_for_user(user_id)
|
||||
except (OperationalError, ProgrammingError) as exc:
|
||||
logger.warning("cover_templates 表查询失败(可能未迁移),返回空列表: %s", exc)
|
||||
return ListCoverTemplatesResponse(items=[], total=0)
|
||||
return ListCoverTemplatesResponse(
|
||||
items=[
|
||||
CoverTemplateResponse(
|
||||
@@ -70,7 +78,11 @@ def create_cover_template(
|
||||
config=config_dict,
|
||||
thumbnail_url=request.thumbnail_url,
|
||||
)
|
||||
created = repo.create(template)
|
||||
try:
|
||||
created = repo.create(template)
|
||||
except (OperationalError, ProgrammingError) as exc:
|
||||
logger.warning("cover_templates 表不可用(可能未迁移): %s", exc)
|
||||
raise HTTPException(status_code=503, detail="封面模板服务暂不可用,请稍后重试") from None
|
||||
return CoverTemplateResponse(
|
||||
id=created.id,
|
||||
name=created.name,
|
||||
@@ -90,7 +102,11 @@ def update_cover_template(
|
||||
) -> CoverTemplateResponse:
|
||||
"""更新封面模板(仅允许更新自己的模板)。"""
|
||||
user_id = authenticated_user.user.id
|
||||
template = repo.get(template_id)
|
||||
try:
|
||||
template = repo.get(template_id)
|
||||
except (OperationalError, ProgrammingError) as exc:
|
||||
logger.warning("cover_templates 表不可用: %s", exc)
|
||||
raise HTTPException(status_code=503, detail="封面模板服务暂不可用,请稍后重试") from None
|
||||
if template is None:
|
||||
raise HTTPException(status_code=404, detail="模板不存在")
|
||||
if template.is_system:
|
||||
@@ -124,7 +140,11 @@ def delete_cover_template(
|
||||
) -> None:
|
||||
"""删除用户自定义封面模板(系统模板不可删除)。"""
|
||||
user_id = authenticated_user.user.id
|
||||
template = repo.get(template_id)
|
||||
try:
|
||||
template = repo.get(template_id)
|
||||
except (OperationalError, ProgrammingError) as exc:
|
||||
logger.warning("cover_templates 表不可用: %s", exc)
|
||||
raise HTTPException(status_code=503, detail="封面模板服务暂不可用,请稍后重试") from None
|
||||
if template is None:
|
||||
raise HTTPException(status_code=404, detail="模板不存在")
|
||||
if template.is_system:
|
||||
|
||||
@@ -307,3 +307,96 @@ class TestInMemoryCoverTemplateRepository:
|
||||
system = repo.list_system_templates()
|
||||
assert len(system) == 2
|
||||
assert all(t.is_system for t in system)
|
||||
|
||||
|
||||
class TestCoverTemplatesErrorHandling:
|
||||
"""测试 API 错误处理加固 - 数据库表不存在时降级处理。"""
|
||||
|
||||
def test_list_raises_operational_error_when_table_missing(self):
|
||||
"""当 cover_templates 表不存在时,repository 应抛出 OperationalError。"""
|
||||
from sqlalchemy.exc import OperationalError
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.cover_template_repository import SQLAlchemyCoverTemplateRepository
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.query.side_effect = OperationalError(
|
||||
"statement", {}, Exception('relation "cover_templates" does not exist')
|
||||
)
|
||||
repo = SQLAlchemyCoverTemplateRepository(mock_session)
|
||||
|
||||
with pytest.raises(OperationalError):
|
||||
repo.list_for_user("user-1")
|
||||
|
||||
def test_route_catches_operational_error_on_list(self):
|
||||
"""路由层应捕获 OperationalError 并返回空列表。"""
|
||||
from app.api.routes.cover_templates import list_cover_templates
|
||||
from app.schemas.cover_template import ListCoverTemplatesResponse
|
||||
from sqlalchemy.exc import OperationalError
|
||||
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.list_for_user.side_effect = OperationalError(
|
||||
"statement", {}, Exception('relation "cover_templates" does not exist')
|
||||
)
|
||||
|
||||
mock_user = MagicMock()
|
||||
mock_user.user.id = "user-1"
|
||||
|
||||
# 调用路由函数,验证它不会抛出异常
|
||||
result = list_cover_templates(
|
||||
skip=0,
|
||||
limit=100,
|
||||
authenticated_user=mock_user,
|
||||
repo=mock_repo,
|
||||
)
|
||||
assert isinstance(result, ListCoverTemplatesResponse)
|
||||
assert result.items == []
|
||||
assert result.total == 0
|
||||
|
||||
def test_route_catches_programming_error_on_list(self):
|
||||
"""路由层应捕获 ProgrammingError 并返回空列表。"""
|
||||
from app.api.routes.cover_templates import list_cover_templates
|
||||
from app.schemas.cover_template import ListCoverTemplatesResponse
|
||||
from sqlalchemy.exc import ProgrammingError
|
||||
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.list_for_user.side_effect = ProgrammingError(
|
||||
"statement", {}, Exception("no such table: cover_templates")
|
||||
)
|
||||
|
||||
mock_user = MagicMock()
|
||||
mock_user.user.id = "user-1"
|
||||
|
||||
result = list_cover_templates(
|
||||
skip=0,
|
||||
limit=100,
|
||||
authenticated_user=mock_user,
|
||||
repo=mock_repo,
|
||||
)
|
||||
assert isinstance(result, ListCoverTemplatesResponse)
|
||||
assert result.items == []
|
||||
assert result.total == 0
|
||||
|
||||
def test_create_returns_503_when_table_missing(self):
|
||||
"""创建模板时,如果表不存在应抛出 HTTPException(503)。"""
|
||||
from app.api.routes.cover_templates import create_cover_template
|
||||
from app.schemas.cover_template import CreateCoverTemplateRequest
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy.exc import OperationalError
|
||||
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.create.side_effect = OperationalError(
|
||||
"statement", {}, Exception('relation "cover_templates" does not exist')
|
||||
)
|
||||
|
||||
mock_user = MagicMock()
|
||||
mock_user.user.id = "user-1"
|
||||
|
||||
request = CreateCoverTemplateRequest(name="test")
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
create_cover_template(
|
||||
request=request,
|
||||
authenticated_user=mock_user,
|
||||
repo=mock_repo,
|
||||
)
|
||||
assert exc_info.value.status_code == 503
|
||||
|
||||
Reference in New Issue
Block a user