6245159e7c
- 新增后台登录图形验证码功能，完善登录安全防护 - 新增登录rememberMe参数，控制Refresh Token的会话持久化策略 - 实现OSS私有桶媒体URL自动签名，统一处理图片资源的临时访问签名 - 新增素材库数据库表与上传API，规范媒体资源管理流程 - 统一前端UI图标使用@element-plus/icons-vue，重构布局图标组件 - 登录页新增验证码输入、刷新功能，添加账号记忆与记住密码逻辑 - 更新全套文档，补充API契约、技术决策记录与集成流程说明 - 修复多个业务页面的图标展示问题，新增认证流程相关测试用例
658 lines
21 KiB
Python
658 lines
21 KiB
Python
from datetime import datetime
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from pydantic import ValidationError
|
||
|
||
from app.auth import create_customer_token, hash_password, require_admin
|
||
from app.database import get_db
|
||
from app.main import create_app
|
||
from app.models import AdminUser, Customer, HeroSlide, Lead, MediaAsset, VehicleOption
|
||
from app.routers import admin as admin_router
|
||
from app.routers import public as public_router
|
||
from app.routers.shared import site_config
|
||
from app.redis_session import InMemoryAdminSessionStore, get_admin_session_store
|
||
from app.schemas import LeadCreateIn, LeadQuery
|
||
|
||
|
||
class FakeScalarResult:
|
||
def __init__(self, items):
|
||
self.items = items
|
||
|
||
def unique(self):
|
||
return self
|
||
|
||
def all(self):
|
||
return self.items
|
||
|
||
def first(self):
|
||
return self.items[0] if self.items else None
|
||
|
||
def one(self):
|
||
return self.items[0]
|
||
|
||
|
||
class FakeExecuteResult:
|
||
def __init__(self, rows):
|
||
self.rows = rows
|
||
|
||
def all(self):
|
||
return self.rows
|
||
|
||
|
||
class FakeDb:
|
||
def __init__(self, *, scalar_results=None, scalar_values=None, execute_results=None, get_result=None):
|
||
self.scalar_results = list(scalar_results or [])
|
||
self.scalar_values = list(scalar_values or [])
|
||
self.execute_results = list(execute_results or [])
|
||
self.scalar_statements = []
|
||
self.execute_statements = []
|
||
self.get_result = get_result
|
||
self.added = []
|
||
self.deleted = []
|
||
self.committed = False
|
||
|
||
def scalars(self, stmt):
|
||
self.scalar_statements.append(stmt)
|
||
return FakeScalarResult(self.scalar_results.pop(0) if self.scalar_results else [])
|
||
|
||
def scalar(self, stmt):
|
||
self.scalar_statements.append(stmt)
|
||
return self.scalar_values.pop(0) if self.scalar_values else None
|
||
|
||
def execute(self, stmt):
|
||
self.execute_statements.append(stmt)
|
||
return FakeExecuteResult(self.execute_results.pop(0) if self.execute_results else [])
|
||
|
||
def get(self, _model, _item_id):
|
||
return self.get_result
|
||
|
||
def add(self, item):
|
||
self.added.append(item)
|
||
|
||
def delete(self, item):
|
||
self.deleted.append(item)
|
||
|
||
def flush(self):
|
||
for item in self.added:
|
||
if not getattr(item, "id", None):
|
||
item.id = f"{item.__class__.__name__.lower()}-test-id"
|
||
|
||
def commit(self):
|
||
self.committed = True
|
||
|
||
def refresh(self, item):
|
||
if not getattr(item, "id", None):
|
||
item.id = "lead-test-id"
|
||
|
||
|
||
def make_hero_slide(**overrides):
|
||
return HeroSlide(
|
||
id=overrides.get("id", "slide-test"),
|
||
title=overrides.get("title", "测试轮播"),
|
||
kicker=overrides.get("kicker", "测试副标题"),
|
||
image=overrides.get("image", "/assets/slide.jpg"),
|
||
targetType=overrides.get("targetType"),
|
||
targetValue=overrides.get("targetValue"),
|
||
sortOrder=overrides.get("sortOrder", 1),
|
||
isActive=overrides.get("isActive", True),
|
||
createdAt=datetime(2026, 1, 1),
|
||
updatedAt=datetime(2026, 1, 2),
|
||
)
|
||
|
||
|
||
def make_vehicle_option(**overrides):
|
||
return VehicleOption(
|
||
id=overrides.get("id", "vehicle-option-test"),
|
||
title=overrides.get("title", "5座舒适用车"),
|
||
description=overrides.get("description", "2-8人小团，按人数和行李匹配车型"),
|
||
image=overrides.get("image", "/assets/vehicle.jpg"),
|
||
sortOrder=overrides.get("sortOrder", 2),
|
||
isActive=overrides.get("isActive", True),
|
||
createdAt=datetime(2026, 1, 1),
|
||
updatedAt=datetime(2026, 1, 2),
|
||
)
|
||
|
||
|
||
def make_admin_user():
|
||
return AdminUser(
|
||
id="admin-test",
|
||
email="admin@example.test",
|
||
name="Admin",
|
||
role="admin",
|
||
passwordHash="not-used",
|
||
isActive=True,
|
||
)
|
||
|
||
|
||
def authenticated_app(fake_db):
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
app.dependency_overrides[require_admin] = make_admin_user
|
||
return app
|
||
|
||
|
||
|
||
def test_retired_domain_endpoints_are_not_registered():
|
||
client = TestClient(create_app())
|
||
|
||
assert client.get("/api/public/products").status_code == 404
|
||
assert client.get("/api/public/products/retired").status_code == 404
|
||
assert client.get("/api/admin/products").status_code == 404
|
||
assert client.get("/api/public/destinations").status_code == 404
|
||
assert client.get("/api/admin/destinations").status_code == 404
|
||
|
||
|
||
def test_lead_schema_accepts_date_only_string():
|
||
lead = LeadCreateIn(phone=" contact handle ", travelDate="2027-01-01")
|
||
|
||
assert lead.phone == "contact handle"
|
||
assert lead.travelDate == datetime(2027, 1, 1)
|
||
|
||
|
||
def test_query_status_validation_rejects_unknown_values():
|
||
with pytest.raises(ValidationError):
|
||
LeadQuery(status="closed")
|
||
|
||
|
||
def test_public_leads_endpoint_accepts_date_only_and_returns_minimal_response():
|
||
fake_db = FakeDb()
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
|
||
try:
|
||
response = TestClient(app).post(
|
||
"/api/public/leads",
|
||
json={
|
||
"phone": " contact handle ",
|
||
"travelDate": "2027-01-01",
|
||
"peopleCount": 2,
|
||
"budgetMin": 0,
|
||
"budgetMax": 100,
|
||
},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 201
|
||
assert response.json() == {
|
||
"code": 201,
|
||
"msg": "success",
|
||
"data": {"id": "lead-test-id", "status": "new"},
|
||
}
|
||
assert fake_db.added[0].phone == "contact handle"
|
||
assert fake_db.added[0].travelDate == datetime(2027, 1, 1)
|
||
|
||
|
||
def test_public_phone_login_creates_customer_and_returns_masked_session(monkeypatch):
|
||
monkeypatch.setattr(public_router, "exchange_phone_code", lambda code: "10000000000", raising=False)
|
||
fake_db = FakeDb(scalar_values=[None])
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
|
||
try:
|
||
response = TestClient(app).post("/api/public/auth/phone-login", json={"code": "phone-code"})
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()["data"]
|
||
assert body["token"]
|
||
assert body["customer"] == {"id": "customer-test-id", "phoneMasked": "100****0000"}
|
||
assert fake_db.added[0].phone == "10000000000"
|
||
assert fake_db.committed
|
||
|
||
|
||
def test_public_phone_login_requires_wechat_configuration(monkeypatch):
|
||
def missing_config(_code):
|
||
raise public_router.WechatConfigError("微信小程序登录未配置")
|
||
|
||
monkeypatch.setattr(public_router, "exchange_phone_code", missing_config, raising=False)
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: FakeDb()
|
||
|
||
try:
|
||
response = TestClient(app).post("/api/public/auth/phone-login", json={"code": "phone-code"})
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 503
|
||
assert response.json() == {
|
||
"code": 503,
|
||
"msg": "微信小程序登录未配置",
|
||
"data": None,
|
||
}
|
||
|
||
|
||
def test_public_me_returns_customer_for_customer_token():
|
||
customer = Customer(id="customer-test", phone="10000000000")
|
||
fake_db = FakeDb(get_result=customer)
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
token = create_customer_token(customer)
|
||
|
||
try:
|
||
response = TestClient(app).get("/api/public/auth/me", headers={"Authorization": f"Bearer {token}"})
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
assert response.json() == {
|
||
"code": 200,
|
||
"msg": "success",
|
||
"data": {"id": "customer-test", "phoneMasked": "100****0000"},
|
||
}
|
||
|
||
|
||
def test_public_me_requires_customer_token():
|
||
response = TestClient(create_app()).get("/api/public/auth/me")
|
||
|
||
assert response.status_code == 401
|
||
assert response.json() == {"code": 401, "msg": "请先登录", "data": None}
|
||
|
||
|
||
def test_admin_requires_auth_for_protected_endpoint():
|
||
response = TestClient(create_app()).get("/api/admin/dashboard")
|
||
|
||
assert response.status_code == 401
|
||
assert response.json() == {"code": 401, "msg": "请先登录后台", "data": None}
|
||
|
||
|
||
def test_admin_dynamic_routers_requires_admin_authentication():
|
||
response = TestClient(create_app()).get("/api/admin/system/routers")
|
||
|
||
assert response.status_code == 401
|
||
assert response.json() == {"code": 401, "msg": "请先登录后台", "data": None}
|
||
|
||
|
||
def test_admin_login_requires_valid_captcha():
|
||
admin_user = AdminUser(
|
||
id="admin-test",
|
||
email="admin@example.com",
|
||
name="Admin",
|
||
role="admin",
|
||
passwordHash=hash_password("ChangeMe123!", rounds=4),
|
||
isActive=True,
|
||
)
|
||
fake_db = FakeDb(scalar_values=[admin_user])
|
||
store = InMemoryAdminSessionStore()
|
||
store.create_captcha("captcha-test", "ABCD", 120)
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||
|
||
try:
|
||
response = TestClient(app).post(
|
||
"/api/admin/auth/login",
|
||
json={
|
||
"email": "admin@example.com",
|
||
"password": "ChangeMe123!",
|
||
"captchaId": "captcha-test",
|
||
"captchaCode": "WRONG",
|
||
},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 401
|
||
assert response.json()["data"] is None
|
||
|
||
|
||
def test_admin_login_returns_token_and_user():
|
||
admin_user = AdminUser(
|
||
id="admin-test",
|
||
email="admin@example.com",
|
||
name="Admin",
|
||
role="admin",
|
||
passwordHash=hash_password("ChangeMe123!", rounds=4),
|
||
isActive=True,
|
||
)
|
||
fake_db = FakeDb(scalar_values=[admin_user])
|
||
store = InMemoryAdminSessionStore()
|
||
store.create_captcha("captcha-contract", "ABCD", 120)
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||
|
||
try:
|
||
response = TestClient(app).post(
|
||
"/api/admin/auth/login",
|
||
json={
|
||
"email": "admin@example.com",
|
||
"password": "ChangeMe123!",
|
||
"captchaId": "captcha-contract",
|
||
"captchaCode": "ABCD",
|
||
"rememberMe": True,
|
||
},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()["data"]
|
||
assert body["token"]
|
||
assert body["user"] == {
|
||
"id": "admin-test",
|
||
"email": "admin@example.com",
|
||
"name": "Admin",
|
||
"role": "admin",
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("module", "payload", "expected"),
|
||
[
|
||
("heroSlides", {"title": "新轮播", "image": None, "targetType": None}, {"title": "新轮播", "image": None}),
|
||
("vehicleOptions", {"title": "5座舒适用车", "description": "用车文案", "image": None}, {"title": "5座舒适用车", "description": "用车文案", "image": None}),
|
||
],
|
||
)
|
||
def test_site_config_create_supported_modules_defaults_fields_and_audits(module, payload, expected):
|
||
fake_db = FakeDb(scalar_values=[4])
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).post(f"/api/admin/site-config/{module}", json=payload)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 201
|
||
body = response.json()["data"]
|
||
for key, value in expected.items():
|
||
assert body[key] == value
|
||
assert body["isActive"] is True
|
||
assert body["sortOrder"] == 5
|
||
assert fake_db.committed
|
||
assert fake_db.added[-1].action == "create"
|
||
|
||
|
||
def test_site_config_invalid_module_returns_structured_error():
|
||
fake_db = FakeDb()
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).post("/api/admin/site-config/unknown", json={"title": "测试"})
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 400
|
||
assert response.json() == {
|
||
"code": 400,
|
||
"msg": "模块不存在或无权限操作",
|
||
"data": None,
|
||
"errorCode": "MODULE_CONFIG_FORBIDDEN",
|
||
"details": {"module": "unknown"},
|
||
}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("module", "payload"),
|
||
[
|
||
("themes", {"label": "已移除模块"}),
|
||
("map", {"image": "/assets/removed-map.webp"}),
|
||
("destinationRegions", {"title": "已移除热门区域"}),
|
||
],
|
||
)
|
||
def test_site_config_create_rejects_removed_module(module, payload):
|
||
fake_db = FakeDb()
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).post(f"/api/admin/site-config/{module}", json=payload)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 400
|
||
assert response.json()["errorCode"] == "MODULE_CONFIG_FORBIDDEN"
|
||
|
||
|
||
def test_admin_site_config_hero_slides_use_dedicated_contract_without_targets():
|
||
hero_slide = make_hero_slide(targetType="search", targetValue="search-test")
|
||
fake_db = FakeDb(scalar_results=[[hero_slide], [], [], [], [], [], [], [], [], [], [], []])
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).get("/api/admin/site-config")
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
hero = response.json()["data"]["heroSlides"][0]
|
||
assert hero["title"] == "测试轮播"
|
||
assert hero["image"] == "/assets/slide.jpg"
|
||
assert "targetType" not in hero
|
||
assert "targetValue" not in hero
|
||
|
||
|
||
def test_site_config_create_hero_slide_ignores_target_fields_and_returns_dedicated_contract():
|
||
fake_db = FakeDb(scalar_values=[0])
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).post(
|
||
"/api/admin/site-config/heroSlides",
|
||
json={
|
||
"title": " 新轮播 ",
|
||
"kicker": "",
|
||
"image": None,
|
||
"targetType": "search",
|
||
"targetValue": "ignored",
|
||
},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 201
|
||
body = response.json()["data"]
|
||
assert body["title"] == "新轮播"
|
||
assert body["image"] is None
|
||
assert "targetType" not in body
|
||
assert "targetValue" not in body
|
||
created = fake_db.added[0]
|
||
assert created.targetType is None
|
||
assert created.targetValue is None
|
||
|
||
|
||
def test_site_config_patch_hero_slide_ignores_target_fields_and_returns_dedicated_contract():
|
||
hero_slide = make_hero_slide(targetType="search", targetValue="search-test")
|
||
fake_db = FakeDb(get_result=hero_slide)
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).patch(
|
||
"/api/admin/site-config/heroSlides/slide-test",
|
||
json={
|
||
"title": "夏日贵州小包团",
|
||
"image": None,
|
||
"targetType": "search",
|
||
"targetValue": "ignored",
|
||
},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()["data"]
|
||
assert body["title"] == "夏日贵州小包团"
|
||
assert body["image"] is None
|
||
assert "targetType" not in body
|
||
assert "targetValue" not in body
|
||
assert hero_slide.targetType == "search"
|
||
assert hero_slide.targetValue == "search-test"
|
||
|
||
|
||
def test_site_config_reorder_reassigns_sort_order_and_returns_items():
|
||
items = [
|
||
make_hero_slide(id="slide-1", title="第一张", sortOrder=0),
|
||
make_hero_slide(id="slide-2", title="第二张", sortOrder=1),
|
||
make_hero_slide(id="slide-3", title="第三张", sortOrder=2),
|
||
]
|
||
fake_db = FakeDb(scalar_results=[items])
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).patch(
|
||
"/api/admin/site-config/heroSlides/reorder",
|
||
json={"itemIds": ["slide-2", "slide-1", "slide-3"]},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
assert [item.id for item in items] == ["slide-1", "slide-2", "slide-3"]
|
||
assert {item.id: item.sortOrder for item in items} == {"slide-2": 0, "slide-1": 1, "slide-3": 2}
|
||
assert [item["id"] for item in response.json()["data"]["items"]] == ["slide-2", "slide-1", "slide-3"]
|
||
assert [item["sortOrder"] for item in response.json()["data"]["items"]] == [0, 1, 2]
|
||
assert fake_db.committed
|
||
assert fake_db.added[-1].action == "reorder"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"item_ids",
|
||
[
|
||
["slide-1", "slide-1", "slide-2"],
|
||
["slide-1", "slide-2"],
|
||
["slide-1", "slide-2", "other-module-id"],
|
||
],
|
||
)
|
||
def test_site_config_reorder_rejects_duplicate_missing_and_unknown_ids(item_ids):
|
||
items = [
|
||
make_hero_slide(id="slide-1"),
|
||
make_hero_slide(id="slide-2"),
|
||
make_hero_slide(id="slide-3"),
|
||
]
|
||
fake_db = FakeDb(scalar_results=[items])
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).patch("/api/admin/site-config/heroSlides/reorder", json={"itemIds": item_ids})
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 400
|
||
assert response.json()["errorCode"] == "MODULE_CONFIG_REORDER_INVALID"
|
||
assert not fake_db.committed
|
||
|
||
|
||
def test_admin_media_upload_streams_image_to_oss_records_asset_and_audits(monkeypatch):
|
||
uploaded = {}
|
||
|
||
def fake_upload(file_obj, key, mime_type, size_bytes):
|
||
uploaded["key"] = key
|
||
uploaded["mimeType"] = mime_type
|
||
uploaded["sizeBytes"] = size_bytes
|
||
uploaded["body"] = file_obj.read()
|
||
return f"https://cdn.example.test/{key}"
|
||
|
||
monkeypatch.setattr(admin_router, "upload_image_to_oss", fake_upload)
|
||
fake_db = FakeDb()
|
||
app = authenticated_app(fake_db)
|
||
content = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
|
||
|
||
try:
|
||
response = TestClient(app).post(
|
||
"/api/admin/media-assets/upload",
|
||
data={"group": "heroSlides"},
|
||
files={"file": ("hero.png", content, "image/png")},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 201
|
||
body = response.json()["data"]
|
||
assert body["url"].startswith("https://cdn.example.test/admin/heroSlides/")
|
||
assert body["name"] == "hero.png"
|
||
assert body["mimeType"] == "image/png"
|
||
assert body["sizeBytes"] == len(content)
|
||
assert body["group"] == "heroSlides"
|
||
assert uploaded["body"] == content
|
||
assert uploaded["key"].endswith(".png")
|
||
assert isinstance(fake_db.added[0], MediaAsset)
|
||
assert fake_db.added[-1].entity == "media_asset"
|
||
assert fake_db.committed
|
||
|
||
|
||
def test_admin_media_upload_rejects_non_image_file():
|
||
fake_db = FakeDb()
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).post(
|
||
"/api/admin/media-assets/upload",
|
||
data={"group": "docs"},
|
||
files={"file": ("note.txt", b"hello", "text/plain")},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 400
|
||
assert response.json()["errorCode"] == "MEDIA_UPLOAD_INVALID_TYPE"
|
||
assert not fake_db.added
|
||
assert not fake_db.committed
|
||
|
||
|
||
def test_admin_leads_endpoint_accepts_source_keyword_and_created_range_filters():
|
||
fake_db = FakeDb(scalar_results=[[]])
|
||
app = authenticated_app(fake_db)
|
||
|
||
try:
|
||
response = TestClient(app).get(
|
||
"/api/admin/leads",
|
||
params={
|
||
"status": "new",
|
||
"sourcePage": "demand_page",
|
||
"keyword": "荔波",
|
||
"createdFrom": "2026-01-01",
|
||
"createdTo": "2026-01-31",
|
||
"take": 50,
|
||
},
|
||
)
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
assert response.json() == {
|
||
"code": 200,
|
||
"msg": "success",
|
||
"data": {"items": [], "total": 0, "offset": 0, "take": 50},
|
||
}
|
||
stmt = str(fake_db.scalar_statements[0])
|
||
assert '"Lead".status' in stmt
|
||
assert '"Lead"."sourcePage"' in stmt
|
||
assert '"Lead"."createdAt"' in stmt
|
||
assert '"Lead".phone' in stmt
|
||
assert '"Lead".destination' in stmt
|
||
assert '"Lead".note' in stmt
|
||
|
||
|
||
def test_admin_lead_status_update_returns_updated_status_and_audits():
|
||
lead = Lead(
|
||
id="lead-test",
|
||
destination="测试目的地",
|
||
phone="contact handle",
|
||
status="new",
|
||
createdAt=datetime(2026, 1, 1),
|
||
updatedAt=datetime(2026, 1, 2),
|
||
)
|
||
fake_db = FakeDb(get_result=lead)
|
||
app = create_app()
|
||
app.dependency_overrides[get_db] = lambda: fake_db
|
||
app.dependency_overrides[require_admin] = lambda: AdminUser(
|
||
id="admin-test",
|
||
email="admin@example.test",
|
||
name="Admin",
|
||
role="admin",
|
||
passwordHash="not-used",
|
||
isActive=True,
|
||
)
|
||
|
||
try:
|
||
response = TestClient(app).patch("/api/admin/leads/lead-test/status", json={"status": "contacted"})
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
assert response.status_code == 200
|
||
assert response.json()["data"]["status"] == "contacted"
|
||
assert lead.status == "contacted"
|
||
assert fake_db.committed
|
||
assert fake_db.added[-1].entity == "lead"
|