selfrelease
fad458b2a7
- UIUX 文档：填充 19 个缺口（多主体画像/健康度/AI+看板/增长域/洞察域/创始人端/OODA/助推/商密） - UIUX 文档：插入 6 个新章节（十四~十九），旧章节重编号为二十~三十一，更新目录和交叉引用 - 作业指导书 x5：导航改为 6 域分组，新增 Context Bar/工作模式/Insight Rail/决策线程/多工作区等 UI 概念 - 新建 docs/2-task-uiux.md：50 个代码落地开发任务，按 P0-P6 分优先级 + 8 Sprint 规划 - 后端/前端：大量新增模型、路由、组件（来自之前 Phase 开发）
127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""T3.13 协作闭环 + 任务管理测试。
|
|
|
|
测试通知服务、任务 CRUD 和评论 CRUD。
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.services.notification import send_email, send_in_app_notification, send_webhook
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_headers(client: TestClient):
|
|
"""注册并登录，返回认证头。"""
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "task_test@example.com",
|
|
"password": "password123",
|
|
"name": "任务测试用户",
|
|
"tenant_name": "任务测试机构",
|
|
"role": "investor",
|
|
},
|
|
)
|
|
resp = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": "task_test@example.com", "password": "password123"},
|
|
)
|
|
token = resp.json()["data"]["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture
|
|
def company_id(client: TestClient, auth_headers: dict):
|
|
"""创建测试企业，返回企业 ID。"""
|
|
resp = client.post(
|
|
"/api/v1/companies",
|
|
json={"name": "任务测试公司", "industry": "AI"},
|
|
headers=auth_headers,
|
|
)
|
|
return resp.json()["data"]["id"]
|
|
|
|
|
|
class TestNotificationService:
|
|
"""通知服务测试。"""
|
|
|
|
async def test_send_email(self):
|
|
"""发送邮件通知。"""
|
|
result = await send_email("test@example.com", "测试邮件", "邮件内容")
|
|
assert result is True
|
|
|
|
async def test_send_in_app_notification(self):
|
|
"""发送站内信通知。"""
|
|
result = await send_in_app_notification("user-1", "测试通知", "通知内容")
|
|
assert result is True
|
|
|
|
async def test_send_webhook(self):
|
|
"""发送 Webhook 通知。"""
|
|
result = await send_webhook("https://example.com/webhook", {"event": "test"})
|
|
assert result is True
|
|
|
|
|
|
class TestTaskCRUD:
|
|
"""任务 CRUD 测试。"""
|
|
|
|
def test_create_task(self, client: TestClient, auth_headers: dict, company_id: str):
|
|
"""创建任务。"""
|
|
resp = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "跟进融资进度", "company_id": company_id, "priority": "high"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["code"] == 0
|
|
|
|
def test_list_tasks(self, client: TestClient, auth_headers: dict):
|
|
"""查询任务列表。"""
|
|
resp = client.get("/api/v1/tasks", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert isinstance(resp.json()["data"], list)
|
|
|
|
def test_update_task(self, client: TestClient, auth_headers: dict, company_id: str):
|
|
"""更新任务状态。"""
|
|
create_resp = client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "待更新任务", "company_id": company_id},
|
|
headers=auth_headers,
|
|
)
|
|
task_id = create_resp.json()["data"]["id"]
|
|
resp = client.put(
|
|
f"/api/v1/tasks/{task_id}",
|
|
json={"status": "done"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["code"] == 0
|
|
|
|
def test_no_auth(self, client: TestClient):
|
|
"""无认证返回 401。"""
|
|
resp = client.get("/api/v1/tasks")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
class TestCommentCRUD:
|
|
"""评论 CRUD 测试。"""
|
|
|
|
def test_create_and_list_comment(self, client: TestClient, auth_headers: dict, company_id: str):
|
|
"""创建并查询评论。"""
|
|
client.post(
|
|
"/api/v1/comments",
|
|
json={"target_type": "company", "target_id": company_id, "content": "测试评论"},
|
|
headers=auth_headers,
|
|
)
|
|
resp = client.get(
|
|
f"/api/v1/comments?target_type=company&target_id={company_id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert len(data) >= 1
|
|
assert data[0]["content"] == "测试评论"
|
|
|
|
def test_no_auth(self, client: TestClient):
|
|
"""无认证返回 401。"""
|
|
resp = client.get("/api/v1/comments?target_type=company&target_id=xxx")
|
|
assert resp.status_code == 401
|