44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
"""对外元数据必须过滤历史原异常和未知报告字段。"""
|
|
|
|
from uuid import uuid4
|
|
|
|
from app.models import TPlusSyncError
|
|
from app.schemas.tplus import TPlusCreateIn, TPlusErrorQuery
|
|
from app.services.tplus import query_service, task_service
|
|
from app.services.tplus.relationship_audit import public_report
|
|
|
|
|
|
def test_历史报告只开放白名单并保持兼容待验():
|
|
result = public_report(
|
|
{
|
|
"source_version": "16.0",
|
|
"compatibility_status": "PASSED",
|
|
"password": "sensitive",
|
|
"sales_links": {"row_count": 1, "payload": "secret"},
|
|
"state_enums": [{"name": "password=hidden", "extra": "secret"}],
|
|
}
|
|
)
|
|
assert result["compatibility_status"] == "PENDING"
|
|
assert "password" not in result and "payload" not in result["sales_links"]
|
|
assert "hidden" not in str(result) and "secret" not in str(result)
|
|
|
|
|
|
def test_异常不返回手机号或原始错误文本(tplus_session):
|
|
batch = task_service.create_task(
|
|
tplus_session, TPlusCreateIn(entity_types=["personnel"], request_key=uuid4())
|
|
)
|
|
tplus_session.add(
|
|
TPlusSyncError(
|
|
batch_no=batch.batch_no,
|
|
entity_type="personnel",
|
|
source_table="AA_Person",
|
|
source_id="13800000000",
|
|
error_code="READ_ERROR",
|
|
error_message="password=private",
|
|
)
|
|
)
|
|
tplus_session.commit()
|
|
page = query_service.list_errors(tplus_session, batch.batch_no, TPlusErrorQuery())
|
|
assert page.records[0].source_id is None
|
|
assert "private" not in page.model_dump_json()
|