102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""统一任务接口响应结构回归测试。"""
|
||
from pathlib import Path
|
||
import sys
|
||
from types import ModuleType, SimpleNamespace
|
||
|
||
from flask import Flask
|
||
import jwt
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "api"))
|
||
|
||
from insurance.generation import task_service
|
||
from insurance.generation.routes import workspace_bp
|
||
|
||
|
||
def test_ppt_draft_revision_conflict_returns_current_revision(monkeypatch):
|
||
session = SimpleNamespace(draft_revision=1)
|
||
|
||
class SessionQuery:
|
||
def filter_by(self, **_kwargs):
|
||
return self
|
||
|
||
def first(self):
|
||
return session
|
||
|
||
model_module = ModuleType("insurance.models.ppt_session")
|
||
model_module.PptSession = SimpleNamespace(query=SessionQuery())
|
||
compat_module = ModuleType("insurance.db.compat")
|
||
compat_module.db = SimpleNamespace(session=SimpleNamespace())
|
||
monkeypatch.setitem(sys.modules, "insurance.models.ppt_session", model_module)
|
||
monkeypatch.setitem(sys.modules, "insurance.db.compat", compat_module)
|
||
|
||
app = Flask(__name__)
|
||
app.config["JWT_SECRET"] = "test-secret"
|
||
app.config["GUEST_MODE"] = True
|
||
app.register_blueprint(workspace_bp, url_prefix="/insurance/workspace")
|
||
token = jwt.encode({"user_id": "user-1"}, "test-secret", algorithm="HS256")
|
||
|
||
response = app.test_client().post(
|
||
"/insurance/workspace/ppt/workspaces/session-1/draft",
|
||
json={"expected_revision": 3, "draft_options": {"templateId": "template-1"}},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
)
|
||
|
||
assert response.status_code == 409
|
||
body = response.get_json()
|
||
assert body["code"] == 1001
|
||
assert body["data"] == {"current_revision": 1, "requested_revision": 3}
|
||
|
||
|
||
|
||
def test_task_detail_returns_task_as_direct_data(monkeypatch):
|
||
"""任务详情不能重复包裹 code/data，否则前端轮询读不到 status。"""
|
||
monkeypatch.setattr(
|
||
task_service,
|
||
"get_task",
|
||
lambda task_id, user_id: {
|
||
"code": 0,
|
||
"data": {"id": task_id, "userId": user_id, "status": "running"},
|
||
},
|
||
)
|
||
|
||
app = Flask(__name__)
|
||
app.config["JWT_SECRET"] = "test-secret"
|
||
app.config["GUEST_MODE"] = True
|
||
app.register_blueprint(workspace_bp, url_prefix="/insurance/workspace")
|
||
token = jwt.encode({"user_id": "user-1"}, "test-secret", algorithm="HS256")
|
||
|
||
response = app.test_client().get(
|
||
"/insurance/workspace/tasks/task-1",
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.get_json()
|
||
assert body["data"]["status"] == "running"
|
||
assert "data" not in body["data"]
|
||
|
||
|
||
def test_generation_tasks_reject_guest_identity(monkeypatch):
|
||
"""生成物是账号私有数据，访客令牌不能读取任务。"""
|
||
called = False
|
||
|
||
def fake_get_task(_task_id, _user_id):
|
||
nonlocal called
|
||
called = True
|
||
return {"code": 0, "data": {}}
|
||
|
||
monkeypatch.setattr(task_service, "get_task", fake_get_task)
|
||
|
||
app = Flask(__name__)
|
||
app.config["GUEST_MODE"] = True
|
||
app.register_blueprint(workspace_bp, url_prefix="/insurance/workspace")
|
||
|
||
response = app.test_client().get(
|
||
"/insurance/workspace/tasks/task-1",
|
||
headers={"Authorization": "Bearer guest_test"},
|
||
)
|
||
|
||
assert response.status_code == 401
|
||
assert response.get_json()["message"] == "请先登录账号"
|
||
assert called is False
|