ac4ba7514b
将 feat/settings-rating-brand（基于 927592b）合并到最新 main（d03ddad）： - 冲突收口：系统设置服务/契约、集成测试夹具、布局/路由/登录页、客户列表 - 迁移重编号：本分支未发布的三个迁移压成 V11（客户评级 + 站点信息 + 版本号）， down_revision 接到 main 的 90653814a421，db/migration 相应改为 V11 - 站点图标上传与 logo 上传统一为同步 def，超限先按 file.size 拦截
481 lines
20 KiB
Python
481 lines
20 KiB
Python
"""客户管理接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是客户归属的一致性 —— 归属变更必须「主表更新 + 留痕新增」同事务
|
||
完成，拆开就会出现「主表已换人、留痕缺失」的静默不一致，事后追溯无据；
|
||
编辑端点若能触归属，就会出现「改了归属却无留痕」的旁路，破坏留痕的
|
||
append-only 完整性；「公司名称+负责人」组合唯一缺失会放行重复客户。
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
|
||
数据卫生：公司名与归属人员姓名一律 ``t_`` 前缀，由 conftest 兜底清理。
|
||
"""
|
||
|
||
import uuid
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from app.core.constants import (
|
||
ACTION_CREATE,
|
||
BUSINESS_ROLE_SALESPERSON,
|
||
CUSTOMER_TYPE_ENTERPRISE,
|
||
CUSTOMER_TYPE_PERSONAL,
|
||
MODULE_CUSTOMER,
|
||
PRODUCT_CATEGORY_MUNICIPAL_WATER,
|
||
)
|
||
from app.core.db import get_session_factory
|
||
from app.models import BusinessRole, OperationLog, OwnershipChangeLog
|
||
from app.models.customer import OWNER_TYPE_CUSTOMER
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_PERSONNEL = "/api/v1/base/personnel"
|
||
_CUSTOMERS = "/api/v1/base/customers"
|
||
_BUSINESSES = "/api/v1/business/businesses"
|
||
|
||
|
||
def _company() -> str:
|
||
return f"t_公司_{uuid.uuid4().hex[:8]}"
|
||
|
||
|
||
def _role_id() -> int:
|
||
"""种子岗位（营销员）：归属人员的合法岗位引用。"""
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
return int(
|
||
session.execute(
|
||
select(BusinessRole.id).where(
|
||
BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON,
|
||
BusinessRole.status == 1,
|
||
)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
def _personnel(client: TestClient, headers: dict) -> dict:
|
||
"""建一名 t_ 前缀的在职营销人员（归属变更的被引用方）。"""
|
||
resp = client.post(
|
||
_PERSONNEL,
|
||
json={
|
||
"personnel_name": f"t_人员_{uuid.uuid4().hex[:8]}",
|
||
"business_role_id": _role_id(),
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _cpayload(company: str, **overrides: object) -> dict:
|
||
body: dict = {
|
||
"company_name": company,
|
||
"owner_name": "张老板",
|
||
"customer_type": CUSTOMER_TYPE_PERSONAL,
|
||
"address": "本溪市平山区",
|
||
}
|
||
body.update(overrides)
|
||
return body
|
||
|
||
|
||
def _create_customer(client: TestClient, headers: dict, **overrides: object) -> dict:
|
||
resp = client.post(_CUSTOMERS, json=_cpayload(_company(), **overrides), headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _logs_of(username: str) -> list[OperationLog]:
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
return list(
|
||
session.execute(
|
||
select(OperationLog)
|
||
.where(OperationLog.username == username)
|
||
.order_by(OperationLog.id.desc())
|
||
).scalars()
|
||
)
|
||
|
||
|
||
def _latest_log(customer_id: int) -> OwnershipChangeLog | None:
|
||
"""取该客户最新一条归属留痕（按 id 倒序）。"""
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
return (
|
||
session.execute(
|
||
select(OwnershipChangeLog)
|
||
.where(OwnershipChangeLog.customer_id == customer_id)
|
||
.order_by(OwnershipChangeLog.id.desc())
|
||
)
|
||
.scalars()
|
||
.first()
|
||
)
|
||
|
||
|
||
class TestAuth:
|
||
"""鉴权：登录后接口，裸访问一律 401。"""
|
||
|
||
def test_未登录访问客户列表_401(self, client: TestClient) -> None:
|
||
resp = client.get(_CUSTOMERS)
|
||
assert resp.status_code == 401, resp.text
|
||
|
||
|
||
class TestCrudFlow:
|
||
"""正向链路与组合唯一 ——「公司名称+负责人」是业务唯一标识。"""
|
||
|
||
def test_新增客户带初始归属_回显归属人员姓名(self, client: TestClient, normal: dict) -> None:
|
||
owner = _personnel(client, normal["headers"])
|
||
data = _create_customer(client, normal["headers"], owner_personnel_id=owner["id"])
|
||
assert data["owner_personnel_id"] == owner["id"]
|
||
assert data["owner_personnel_name"] == owner["personnel_name"]
|
||
assert data["customer_type"] == CUSTOMER_TYPE_PERSONAL
|
||
|
||
def test_同公司名不同负责人创建成功(self, client: TestClient, normal: dict) -> None:
|
||
company = _company()
|
||
first = _create_customer(client, normal["headers"], company_name=company)
|
||
second = client.post(
|
||
_CUSTOMERS,
|
||
json=_cpayload(company, owner_name="李老板"),
|
||
headers=normal["headers"],
|
||
)
|
||
assert second.status_code == 200, second.text
|
||
assert second.json()["data"]["id"] != first["id"]
|
||
|
||
def test_重名同负责人_400(self, client: TestClient, normal: dict) -> None:
|
||
company = _company()
|
||
_create_customer(client, normal["headers"], company_name=company)
|
||
dup = client.post(
|
||
_CUSTOMERS,
|
||
json=_cpayload(company, owner_name="张老板"),
|
||
headers=normal["headers"],
|
||
)
|
||
assert dup.status_code == 400, dup.text
|
||
assert "该公司名称与负责人的客户已存在" in dup.json()["message"]
|
||
|
||
def test_删除被业务引用的客户_400引导处理单据(self, client: TestClient, normal: dict) -> None:
|
||
"""业务表就位后删除保护补齐：FK RESTRICT 不能以 500 暴露。"""
|
||
headers = normal["headers"]
|
||
customer = _create_customer(client, headers)
|
||
resp = client.post(
|
||
_BUSINESSES,
|
||
json={
|
||
"customer_id": customer["id"],
|
||
"items": [
|
||
{
|
||
"product_name": "t_被引用产品",
|
||
"product_category": PRODUCT_CATEGORY_MUNICIPAL_WATER,
|
||
"quantity": "1",
|
||
"sale_price": "1",
|
||
}
|
||
],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
blocked = client.delete(f"{_CUSTOMERS}/{customer['id']}", headers=headers)
|
||
assert blocked.status_code == 400, blocked.text
|
||
assert "已存在关联业务" in blocked.json()["message"]
|
||
|
||
def test_客户分类非法_400(self, client: TestClient, normal: dict) -> None:
|
||
resp = client.post(
|
||
_CUSTOMERS,
|
||
json=_cpayload(_company(), customer_type="not_a_type"),
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "客户分类不合法" in resp.json()["message"]
|
||
|
||
def test_编辑其他字段不触归属_塞归属字段也被忽略(
|
||
self, client: TestClient, normal: dict
|
||
) -> None:
|
||
"""归属只有 /ownership 一条写路径 —— 编辑端点防「改归属不留痕」旁路。"""
|
||
headers = normal["headers"]
|
||
owner = _personnel(client, headers)
|
||
stranger = _personnel(client, headers)
|
||
customer = _create_customer(client, headers, owner_personnel_id=owner["id"])
|
||
|
||
updated = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}",
|
||
json={
|
||
"company_name": customer["company_name"],
|
||
"owner_name": customer["owner_name"],
|
||
"customer_type": CUSTOMER_TYPE_ENTERPRISE,
|
||
"address": "沈阳市沈河区",
|
||
# 试图经编辑端点改归属：schema 未声明该字段，必须被忽略
|
||
"ownerPersonnelId": stranger["id"],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert updated.status_code == 200, updated.text
|
||
body = updated.json()["data"]
|
||
assert body["customer_type"] == CUSTOMER_TYPE_ENTERPRISE
|
||
assert body["address"] == "沈阳市沈河区"
|
||
assert body["owner_personnel_id"] == owner["id"], "编辑端点不得改动归属"
|
||
assert body["owner_personnel_name"] == owner["personnel_name"]
|
||
|
||
detail = client.get(f"{_CUSTOMERS}/{customer['id']}", headers=headers)
|
||
assert detail.json()["data"]["owner_personnel_id"] == owner["id"]
|
||
assert _latest_log(int(customer["id"])) is None, "编辑不应产生归属留痕"
|
||
|
||
|
||
class TestOwnership:
|
||
"""归属变更：主表更新与留痕同事务；校验失败不留脏数据。"""
|
||
|
||
def test_归属变更_主表更新且留痕新增(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
old = _personnel(client, headers)
|
||
new = _personnel(client, headers)
|
||
customer = _create_customer(client, headers, owner_personnel_id=old["id"])
|
||
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": new["id"], "reason": "区域调整"},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()["data"]
|
||
assert body["owner_personnel_id"] == new["id"]
|
||
assert body["owner_personnel_name"] == new["personnel_name"]
|
||
|
||
log = _latest_log(int(customer["id"]))
|
||
assert log is not None, "归属变更未写留痕"
|
||
assert log.owner_type == OWNER_TYPE_CUSTOMER
|
||
assert log.customer_id == int(customer["id"])
|
||
assert log.old_owner_personnel_id == int(old["id"])
|
||
assert log.new_owner_personnel_id == int(new["id"])
|
||
assert log.change_reason == "区域调整"
|
||
assert log.operator_name == normal["username"]
|
||
|
||
def test_变更后归属人员已离职_400(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
target = _personnel(client, headers)
|
||
client.put(f"{_PERSONNEL}/{target['id']}/status", json={"status": 0}, headers=headers)
|
||
customer = _create_customer(client, headers)
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": target["id"]},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "归属人员不存在或已离职" in resp.json()["message"]
|
||
assert _latest_log(int(customer["id"])) is None, "失败变更不应留痕"
|
||
|
||
def test_新旧归属相同_400(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
owner = _personnel(client, headers)
|
||
customer = _create_customer(client, headers, owner_personnel_id=owner["id"])
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": owner["id"]},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "归属未变化" in resp.json()["message"]
|
||
|
||
def test_双方均为空_400(self, client: TestClient, normal: dict) -> None:
|
||
"""无归属客户再清一次归属也属「未变化」——不产生空转留痕。"""
|
||
customer = _create_customer(client, normal["headers"])
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": None},
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "归属未变化" in resp.json()["message"]
|
||
|
||
def test_变更为空即清空归属_主表与留痕一致(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
owner = _personnel(client, headers)
|
||
customer = _create_customer(client, headers, owner_personnel_id=owner["id"])
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": None, "reason": "暂无负责人"},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()["data"]
|
||
assert body["owner_personnel_id"] is None
|
||
assert body["owner_personnel_name"] is None
|
||
|
||
log = _latest_log(int(customer["id"]))
|
||
assert log is not None
|
||
assert log.old_owner_personnel_id == int(owner["id"])
|
||
assert log.new_owner_personnel_id is None
|
||
assert log.change_reason == "暂无负责人"
|
||
|
||
|
||
class TestOwnershipLogs:
|
||
"""留痕分页：新旧归属人姓名由联表带出，倒序可追溯变更轨迹。"""
|
||
|
||
def test_留痕分页带新旧归属人姓名并倒序(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
first = _personnel(client, headers)
|
||
second = _personnel(client, headers)
|
||
customer = _create_customer(client, headers, owner_personnel_id=first["id"])
|
||
|
||
r1 = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": second["id"], "reason": "第一次调整"},
|
||
headers=headers,
|
||
)
|
||
assert r1.status_code == 200, r1.text
|
||
r2 = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": None, "reason": "清空归属"},
|
||
headers=headers,
|
||
)
|
||
assert r2.status_code == 200, r2.text
|
||
|
||
resp = client.get(f"{_CUSTOMERS}/{customer['id']}/ownership-logs", headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()["data"]
|
||
assert body["total"] == 2
|
||
latest, earlier = body["list"][0], body["list"][1]
|
||
assert latest["change_reason"] == "清空归属"
|
||
assert latest["old_owner_personnel_id"] == int(second["id"])
|
||
assert latest["old_owner_name"] == second["personnel_name"]
|
||
assert latest["new_owner_personnel_id"] is None
|
||
assert latest["new_owner_name"] is None
|
||
assert earlier["old_owner_name"] == first["personnel_name"]
|
||
assert earlier["new_owner_name"] == second["personnel_name"]
|
||
assert earlier["new_owner_personnel_id"] == int(second["id"])
|
||
|
||
|
||
class TestAudit:
|
||
"""写操作（含归属变更）留痕，摘要要能定位到客户。"""
|
||
|
||
def test_新增客户留审计(self, client: TestClient, normal: dict) -> None:
|
||
data = _create_customer(client, normal["headers"])
|
||
logs = [
|
||
log
|
||
for log in _logs_of(normal["username"])
|
||
if log.module == MODULE_CUSTOMER and log.action == ACTION_CREATE
|
||
]
|
||
assert logs, "新增客户未留操作日志"
|
||
latest = logs[0]
|
||
assert latest.success == 1
|
||
assert data["company_name"] in latest.summary
|
||
|
||
def test_归属变更审计摘要含客户名(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
old = _personnel(client, headers)
|
||
new = _personnel(client, headers)
|
||
customer = _create_customer(client, headers, owner_personnel_id=old["id"])
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer['id']}/ownership",
|
||
json={"new_owner_personnel_id": new["id"], "reason": "区域调整"},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
logs = [
|
||
log
|
||
for log in _logs_of(normal["username"])
|
||
if log.module == MODULE_CUSTOMER and log.success == 1
|
||
]
|
||
assert logs, "归属变更未留操作日志"
|
||
latest = logs[0]
|
||
assert customer["company_name"] in latest.summary, "摘要应含客户名"
|
||
assert "→" in latest.summary, "摘要应体现新旧归属走向"
|
||
|
||
|
||
# ==================== 客户评级列（V11） ====================
|
||
|
||
_RATING_KEYS = (
|
||
"score_ship_amount",
|
||
"score_receipt_rate",
|
||
"score_profit_rate",
|
||
"score_contract_cooperation",
|
||
"score_site_construction",
|
||
)
|
||
|
||
|
||
def _rate(client: TestClient, headers: dict, customer_id: int, scores: list[str]) -> dict:
|
||
"""保存客户评级（五项 0-20），返回响应 data。"""
|
||
resp = client.put(
|
||
f"{_CUSTOMERS}/{customer_id}/rating",
|
||
json=dict(zip(_RATING_KEYS, scores, strict=True)),
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _list_by_keyword(client: TestClient, headers: dict, keyword: str, **params) -> list[dict]:
|
||
query: dict = {"keyword": keyword, "pageNum": 1, "pageSize": 50, **params}
|
||
resp = client.get(_CUSTOMERS, params=query, headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]["list"]
|
||
|
||
|
||
class TestRatingColumn:
|
||
"""客户评级列：列表带出评级快照，列头排序走服务端（未评级恒在末尾）。
|
||
|
||
列表是服务端分页，排序必须在 SQL 层完成 —— 前端只传
|
||
``order_field=rating`` + ``order_type``；MySQL 无 NULLS LAST，
|
||
未评级行用布尔键模拟，正序/倒序都在末尾。
|
||
"""
|
||
|
||
def _seed_three(self, client: TestClient, headers: dict) -> tuple[str, dict, dict, dict]:
|
||
"""建三名同批次客户：90 分（A）、0 分（D）、未评级。"""
|
||
token = uuid.uuid4().hex[:8]
|
||
keyword = f"t_公司_{token}"
|
||
high = _create_customer(client, headers, company_name=f"{keyword}_甲")
|
||
low = _create_customer(client, headers, company_name=f"{keyword}_乙")
|
||
none = _create_customer(client, headers, company_name=f"{keyword}_丙")
|
||
_rate(client, headers, high["id"], ["18"] * 5)
|
||
_rate(client, headers, low["id"], ["0"] * 5)
|
||
return keyword, high, low, none
|
||
|
||
def test_列表带出评级字段(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
keyword, high, low, none = self._seed_three(client, headers)
|
||
rows = _list_by_keyword(client, headers, keyword)
|
||
by_id = {row["id"]: row for row in rows}
|
||
|
||
assert Decimal(by_id[high["id"]]["rating_total"]) == Decimal("90")
|
||
assert by_id[high["id"]]["rating_grade_name"] == "A"
|
||
assert Decimal(by_id[low["id"]]["rating_total"]) == Decimal("0")
|
||
assert by_id[low["id"]]["rating_grade_name"] == "D"
|
||
assert by_id[none["id"]]["rating_total"] is None
|
||
assert by_id[none["id"]]["rating_grade_name"] is None
|
||
|
||
def test_评级列正序倒序_未评级恒末尾(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
keyword, high, low, none = self._seed_three(client, headers)
|
||
|
||
desc = _list_by_keyword(client, headers, keyword, order_field="rating", order_type="desc")
|
||
assert [row["id"] for row in desc] == [high["id"], low["id"], none["id"]]
|
||
|
||
asc = _list_by_keyword(client, headers, keyword, order_field="rating", order_type="asc")
|
||
assert [row["id"] for row in asc] == [low["id"], high["id"], none["id"]]
|
||
|
||
def test_非法排序字段_400(self, client: TestClient, normal: dict) -> None:
|
||
"""排序字段是白名单（Literal）：不透传给 SQL，非法值 400。"""
|
||
resp = client.get(
|
||
_CUSTOMERS,
|
||
params={"order_field": "company_name", "order_type": "asc"},
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
def test_非法排序方向_400(self, client: TestClient, normal: dict) -> None:
|
||
resp = client.get(
|
||
_CUSTOMERS,
|
||
params={"order_field": "rating", "order_type": "sideways"},
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
def test_删除客户联动删评级行(self, client: TestClient, normal: dict) -> None:
|
||
"""评级行 FK RESTRICT：删除客户须先清评级行，否则 500。"""
|
||
headers = normal["headers"]
|
||
customer = _create_customer(client, headers)
|
||
_rate(client, headers, customer["id"], ["20"] * 5)
|
||
|
||
resp = client.delete(f"{_CUSTOMERS}/{customer['id']}", headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
assert (
|
||
client.get(f"{_CUSTOMERS}/{customer['id']}/rating", headers=headers).status_code == 404
|
||
)
|