101 lines
4.3 KiB
Python
101 lines
4.3 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import Mock
|
||
|
|
|
||
|
|
from inventory_sync.feishu import FeishuClient, FeishuFieldValidationError
|
||
|
|
from inventory_sync.service import CHANNEL_INVENTORY_SCHEMA_ALERT_CHAT_ID, sync_brand_target
|
||
|
|
from inventory_sync.specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME
|
||
|
|
|
||
|
|
|
||
|
|
def valid_fields() -> list[dict[str, object]]:
|
||
|
|
fields: list[dict[str, object]] = [
|
||
|
|
{"field_name": PRIMARY_FIELD_NAME, "is_primary": True, "type": 1},
|
||
|
|
]
|
||
|
|
for spec in FIELD_SPECS:
|
||
|
|
field: dict[str, object] = {
|
||
|
|
"field_name": spec.feishu_name,
|
||
|
|
"is_primary": False,
|
||
|
|
"type": spec.field_type,
|
||
|
|
}
|
||
|
|
if spec.formatter is not None:
|
||
|
|
field["property"] = {"formatter": spec.formatter}
|
||
|
|
fields.append(field)
|
||
|
|
return fields
|
||
|
|
|
||
|
|
|
||
|
|
class ChannelInventorySchemaTests(unittest.TestCase):
|
||
|
|
def setUp(self) -> None:
|
||
|
|
self.client = FeishuClient(None) # type: ignore[arg-type]
|
||
|
|
self.client.list_fields = Mock() # type: ignore[method-assign]
|
||
|
|
|
||
|
|
def test_allows_fields_in_any_order(self) -> None:
|
||
|
|
fields = valid_fields()
|
||
|
|
self.client.list_fields.return_value = [fields[0], *reversed(fields[1:])]
|
||
|
|
|
||
|
|
available_fields = self.client.ensure_target_fields("token", "app", "table")
|
||
|
|
|
||
|
|
self.assertEqual(available_fields, [field["field_name"] for field in [fields[0], *reversed(fields[1:])]])
|
||
|
|
|
||
|
|
def test_rejects_missing_or_renamed_required_field(self) -> None:
|
||
|
|
fields = valid_fields()
|
||
|
|
fields = [field for field in fields if field["field_name"] != "品牌"]
|
||
|
|
self.client.list_fields.return_value = fields
|
||
|
|
|
||
|
|
with self.assertRaisesRegex(FeishuFieldValidationError, "缺少必需字段或字段已改名：“品牌”"):
|
||
|
|
self.client.ensure_target_fields("token", "app", "table")
|
||
|
|
|
||
|
|
def test_rejects_wrong_type_and_formatter(self) -> None:
|
||
|
|
fields = valid_fields()
|
||
|
|
price_field = next(field for field in fields if field["field_name"] == "零售价")
|
||
|
|
price_field["type"] = 1
|
||
|
|
price_field["property"] = {"formatter": "0"}
|
||
|
|
self.client.list_fields.return_value = fields
|
||
|
|
|
||
|
|
with self.assertRaisesRegex(FeishuFieldValidationError, "字段类型错误：“零售价”") as context:
|
||
|
|
self.client.ensure_target_fields("token", "app", "table")
|
||
|
|
|
||
|
|
self.assertIn("字段格式错误：“零售价”", str(context.exception))
|
||
|
|
|
||
|
|
def test_sends_group_text_message(self) -> None:
|
||
|
|
request_json = Mock(return_value={"code": 0})
|
||
|
|
self.client.request_json = request_json # type: ignore[method-assign]
|
||
|
|
|
||
|
|
self.client.send_text_message("token", "oc_example", "字段校验失败")
|
||
|
|
|
||
|
|
url = request_json.call_args.args[0]
|
||
|
|
payload = request_json.call_args.kwargs["payload"]
|
||
|
|
self.assertEqual(url, "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id")
|
||
|
|
self.assertEqual(payload["receive_id"], "oc_example")
|
||
|
|
self.assertEqual(json.loads(payload["content"]), {"text": "字段校验失败"})
|
||
|
|
|
||
|
|
def test_schema_failure_notifies_group_before_records_are_changed(self) -> None:
|
||
|
|
class SchemaFailureClient:
|
||
|
|
def __init__(self) -> None:
|
||
|
|
self.notifications: list[tuple[str, str, str]] = []
|
||
|
|
|
||
|
|
def ensure_target_fields(self, *_args: object) -> list[str]:
|
||
|
|
raise FeishuFieldValidationError("缺少必需字段或字段已改名：“品牌”")
|
||
|
|
|
||
|
|
def send_text_message(self, access_token: str, chat_id: str, text: str) -> None:
|
||
|
|
self.notifications.append((access_token, chat_id, text))
|
||
|
|
|
||
|
|
client = SchemaFailureClient()
|
||
|
|
target = BRAND_SYNC_TARGETS[0]
|
||
|
|
|
||
|
|
with self.assertRaises(FeishuFieldValidationError):
|
||
|
|
sync_brand_target(
|
||
|
|
client, # type: ignore[arg-type]
|
||
|
|
"token",
|
||
|
|
"app",
|
||
|
|
[{"name": target.table_name, "table_id": "table"}],
|
||
|
|
target,
|
||
|
|
[],
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(len(client.notifications), 1)
|
||
|
|
_, chat_id, text = client.notifications[0]
|
||
|
|
self.assertEqual(chat_id, CHANNEL_INVENTORY_SCHEMA_ALERT_CHAT_ID)
|
||
|
|
self.assertIn("不会删除或重建字段，也不会删除或写入记录", text)
|