qzl
18b5e876ee
- test_auth_routes: monkeypatch 环境为 production 使 phone-session 限速生效 - test_schedule_items_routes: 补充必填 timezone 字段 - test_llm_pricing_service: 更新 deepseek-chat 费率期望值匹配实际 catalog - test_sse_flow_live: 补充 runId 查询参数、改用附件上传 API、修复 history 响应断言、独立 DB session 避免跨事件循环崩溃 - test_agent_prompt: 移除已删除的 project_cli_defaults 断言 - test_toolkit: 更新 action card 断言匹配 module/method 格式
222 lines
7.0 KiB
Python
222 lines
7.0 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Callable
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import app
|
|
from core.auth.models import CurrentUser
|
|
from v1.schedule_items.dependencies import get_schedule_item_service
|
|
from v1.schedule_items.schemas import (
|
|
ScheduleItemCreateRequest,
|
|
ScheduleItemListRequest,
|
|
ScheduleItemResponse,
|
|
ScheduleItemSourceType,
|
|
ScheduleItemStatus,
|
|
ScheduleItemUpdateRequest,
|
|
)
|
|
from v1.schedule_items.service import ScheduleItemService
|
|
|
|
|
|
class FakeScheduleItemService:
|
|
def __init__(self, item: ScheduleItemResponse | None) -> None:
|
|
self._item = item
|
|
|
|
async def create(self, request: ScheduleItemCreateRequest) -> ScheduleItemResponse:
|
|
if not self._item:
|
|
raise HTTPException(status_code=503, detail="Store unavailable")
|
|
return self._item
|
|
|
|
async def get_by_id(self, item_id: UUID) -> ScheduleItemResponse:
|
|
if not self._item or str(self._item.id) != str(item_id):
|
|
raise HTTPException(status_code=404, detail="Schedule item not found")
|
|
return self._item
|
|
|
|
async def update(
|
|
self, item_id: UUID, request: ScheduleItemUpdateRequest
|
|
) -> ScheduleItemResponse:
|
|
if not self._item or str(self._item.id) != str(item_id):
|
|
raise HTTPException(status_code=404, detail="Schedule item not found")
|
|
return self._item
|
|
|
|
async def delete(self, item_id: UUID) -> None:
|
|
if not self._item or str(self._item.id) != str(item_id):
|
|
raise HTTPException(status_code=404, detail="Schedule item not found")
|
|
|
|
async def list_by_date_range(
|
|
self, request: ScheduleItemListRequest
|
|
) -> list[ScheduleItemResponse]:
|
|
return [self._item] if self._item else []
|
|
|
|
|
|
def _override_schedule_item_service(
|
|
service: FakeScheduleItemService,
|
|
) -> Callable[[], ScheduleItemService]:
|
|
def _get_service() -> ScheduleItemService:
|
|
return service # type: ignore[return-value]
|
|
|
|
return _get_service
|
|
|
|
|
|
def _override_current_user(user_id: UUID) -> Callable[[], CurrentUser]:
|
|
def _get_user() -> CurrentUser:
|
|
return CurrentUser(id=user_id)
|
|
|
|
return _get_user
|
|
|
|
|
|
def test_create_schedule_item_returns_201() -> None:
|
|
item = ScheduleItemResponse(
|
|
id=uuid4(),
|
|
owner_id=uuid4(),
|
|
title="Test Event",
|
|
start_at=datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc),
|
|
timezone="UTC",
|
|
status=ScheduleItemStatus.ACTIVE,
|
|
source_type=ScheduleItemSourceType.MANUAL,
|
|
created_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
updated_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
permission=15,
|
|
is_owner=True,
|
|
)
|
|
|
|
app.dependency_overrides[get_schedule_item_service] = (
|
|
_override_schedule_item_service(FakeScheduleItemService(item))
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.post(
|
|
"/api/v1/schedule-items",
|
|
json={
|
|
"title": "Test Event",
|
|
"start_at": "2026-02-28T16:00:00Z",
|
|
"timezone": "UTC",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_list_schedule_items_returns_200() -> None:
|
|
item = ScheduleItemResponse(
|
|
id=uuid4(),
|
|
owner_id=uuid4(),
|
|
title="Test Event",
|
|
start_at=datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc),
|
|
timezone="UTC",
|
|
status=ScheduleItemStatus.ACTIVE,
|
|
source_type=ScheduleItemSourceType.MANUAL,
|
|
created_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
updated_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
permission=15,
|
|
is_owner=True,
|
|
)
|
|
|
|
app.dependency_overrides[get_schedule_item_service] = (
|
|
_override_schedule_item_service(FakeScheduleItemService(item))
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get(
|
|
"/api/v1/schedule-items",
|
|
params={
|
|
"start_at": "2026-02-01T00:00:00Z",
|
|
"end_at": "2026-02-28T23:59:59Z",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert isinstance(response.json(), list)
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_get_schedule_item_returns_200() -> None:
|
|
item_id = uuid4()
|
|
item = ScheduleItemResponse(
|
|
id=item_id,
|
|
owner_id=uuid4(),
|
|
title="Test Event",
|
|
start_at=datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc),
|
|
timezone="UTC",
|
|
status=ScheduleItemStatus.ACTIVE,
|
|
source_type=ScheduleItemSourceType.MANUAL,
|
|
created_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
updated_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
permission=15,
|
|
is_owner=True,
|
|
)
|
|
|
|
app.dependency_overrides[get_schedule_item_service] = (
|
|
_override_schedule_item_service(FakeScheduleItemService(item))
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get(f"/api/v1/schedule-items/{item_id}")
|
|
assert response.status_code == 200
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_update_schedule_item_returns_200() -> None:
|
|
item_id = uuid4()
|
|
item = ScheduleItemResponse(
|
|
id=item_id,
|
|
owner_id=uuid4(),
|
|
title="Updated Event",
|
|
start_at=datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc),
|
|
timezone="UTC",
|
|
status=ScheduleItemStatus.ACTIVE,
|
|
source_type=ScheduleItemSourceType.MANUAL,
|
|
created_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
updated_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
permission=15,
|
|
is_owner=True,
|
|
)
|
|
|
|
app.dependency_overrides[get_schedule_item_service] = (
|
|
_override_schedule_item_service(FakeScheduleItemService(item))
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.patch(
|
|
f"/api/v1/schedule-items/{item_id}",
|
|
json={"title": "Updated Event"},
|
|
)
|
|
assert response.status_code == 200
|
|
finally:
|
|
app.dependency_overrides = {}
|
|
|
|
|
|
def test_delete_schedule_item_returns_204() -> None:
|
|
item_id = uuid4()
|
|
item = ScheduleItemResponse(
|
|
id=item_id,
|
|
owner_id=uuid4(),
|
|
title="Test Event",
|
|
start_at=datetime(2026, 2, 28, 16, 0, 0, tzinfo=timezone.utc),
|
|
timezone="UTC",
|
|
status=ScheduleItemStatus.ACTIVE,
|
|
source_type=ScheduleItemSourceType.MANUAL,
|
|
created_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
updated_at=datetime(2026, 2, 27, 10, 0, 0, tzinfo=timezone.utc),
|
|
permission=15,
|
|
is_owner=True,
|
|
)
|
|
|
|
app.dependency_overrides[get_schedule_item_service] = (
|
|
_override_schedule_item_service(FakeScheduleItemService(item))
|
|
)
|
|
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.delete(f"/api/v1/schedule-items/{item_id}")
|
|
assert response.status_code == 204
|
|
finally:
|
|
app.dependency_overrides = {}
|