"""协作台账的店铺字典：增删改查 / 停用 / 改名级联 / 自定义输入自动登记。 背景：新建协作单时"店铺"原来是手打自由文本，同一个店铺会出现多种写法。 现在改成字典（`huya_collab_shops`）：新建时点一下选，也能自定义输入新名字。 """ from __future__ import annotations import pytest from tests.support.db import get_test_engine, make_test_session from web.backend.models import AuditLog, HuyaCollabShop, HuyaExternalOrder, User from web.backend.services import huya_collab_service as collab class TestShopBase: def setup_method(self): self.engine = get_test_engine() self.db = make_test_session(self.engine) self.user = User(username="shop_op", password_hash="h", role="operation") self.db.add(self.user) self.db.commit() def teardown_method(self): self.db.close() def _shop(self, name: str, **kwargs) -> HuyaCollabShop: return collab.create_shop(self.db, self.user, name=name, **kwargs) def _order(self, order_no: str, shop: str = "") -> HuyaExternalOrder: return collab.create_order(self.db, self.user, order_no=order_no, shop=shop) def _shop_audits(self, shop: HuyaCollabShop) -> list[AuditLog]: return ( self.db.query(AuditLog) .filter(AuditLog.target == f"huya_collab_shop:{shop.id}") .order_by(AuditLog.id.asc()) .all() ) class TestShopCrud(TestShopBase): def test_create_and_list_sorted(self): self._shop("小鱼儿", sort_order=2) self._shop("佳佳", sort_order=1) shops = collab.list_shop_records(self.db) assert [shop.name for shop in shops] == ["佳佳", "小鱼儿"] # 按 sort_order assert all(shop.is_active for shop in shops) def test_create_rejects_duplicate_name(self): self._shop("佳佳") with pytest.raises(collab.CollabError) as excinfo: self._shop(" 佳佳 ") # 去空白后同名 assert excinfo.value.status_code == 409 assert excinfo.value.code == "shop_name_conflict" def test_create_rejects_blank_and_too_long_name(self): with pytest.raises(collab.CollabError) as excinfo: self._shop(" ") assert excinfo.value.code == "shop_name_required" with pytest.raises(collab.CollabError) as excinfo: collab.create_shop(self.db, self.user, name="店" * 65) assert excinfo.value.code == "field_too_long" def test_create_writes_audit(self): shop = self._shop("佳佳") assert [row.action for row in self._shop_audits(shop)] == [ "huya:collab:shop_create" ] def test_update_returns_field_diff_and_audits(self): shop = self._shop("佳佳") changes = collab.update_shop( self.db, self.user, shop, {"sort_order": 5, "remark": "熟人开的", "is_active": False}, ) assert set(changes) == {"sort_order", "remark", "is_active"} assert changes["is_active"] == {"old": True, "new": False} assert shop.remark == "熟人开的" assert "huya:collab:shop_update" in [ row.action for row in self._shop_audits(shop) ] def test_update_without_changes_is_a_noop(self): shop = self._shop("佳佳") assert collab.update_shop(self.db, self.user, shop, {"name": "佳佳"}) == {} assert [row.action for row in self._shop_audits(shop)] == [ "huya:collab:shop_create" ] def test_rename_cascades_to_existing_orders(self): """改名默认级联：否则筛选里会堆一堆"改名前的老名字"。""" shop = self._shop("佳佳") order = self._order("S1", shop="佳佳") changes = collab.update_shop(self.db, self.user, shop, {"name": "佳佳优选"}) assert changes["name"] == {"old": "佳佳", "new": "佳佳优选"} self.db.refresh(order) assert order.shop == "佳佳优选" detail = self._shop_audits(shop)[-1].detail assert "renamed_orders" in detail and "1" in detail def test_rename_can_skip_the_cascade(self): shop = self._shop("佳佳") order = self._order("S1", shop="佳佳") collab.update_shop( self.db, self.user, shop, {"name": "佳佳优选"}, rename_orders=False ) self.db.refresh(order) assert order.shop == "佳佳" # 历史单子保持原名 def test_rename_rejects_name_taken_by_another_shop(self): self._shop("佳佳") other = self._shop("小鱼儿") with pytest.raises(collab.CollabError) as excinfo: collab.update_shop(self.db, self.user, other, {"name": "佳佳"}) assert excinfo.value.code == "shop_name_conflict" def test_delete_only_removes_the_dictionary_row(self): shop = self._shop("佳佳") order = self._order("S1", shop="佳佳") collab.delete_shop(self.db, self.user, shop) assert self.db.query(HuyaCollabShop).count() == 0 self.db.refresh(order) assert order.shop == "佳佳" # 订单不动 def test_get_missing_shop_is_404(self): with pytest.raises(collab.CollabError) as excinfo: collab.get_shop(self.db, 999999) assert excinfo.value.status_code == 404 assert excinfo.value.code == "shop_not_found" class TestShopAutoRegister(TestShopBase): def test_creating_an_order_registers_its_shop(self): """自定义输入一个新店铺名 → 自动进字典，下次下拉里就能点。""" self._order("S1", shop="新店") assert [shop.name for shop in collab.list_shop_records(self.db)] == ["新店"] assert collab.list_shops(self.db) == ["新店"] def test_order_without_shop_does_not_touch_the_dictionary(self): self._order("S1", shop="") assert collab.list_shop_records(self.db) == [] def test_editing_an_order_to_a_new_shop_registers_it(self): order = self._order("S1", shop="") collab.update_order(self.db, self.user, order, {"shop": "随手填的店"}) assert [shop.name for shop in collab.list_shop_records(self.db)] == [ "随手填的店" ] def test_import_registers_all_shops_and_keeps_partial_failure_isolated(self): result = collab.import_orders( self.db, self.user, "店铺A\tS1\tshop_op\t用户1\t礼包\t\n店铺B\tS2\tshop_op\t用户2\t礼包\t", owner_by_name={"shop_op": self.user.id}, ) assert (result["created"], result["errors"]) == (2, []) assert collab.list_shops(self.db) == ["店铺A", "店铺B"] def test_registering_twice_does_not_duplicate(self): self._order("S1", shop="佳佳") self._order("S2", shop="佳佳") assert collab.list_shops(self.db) == ["佳佳"] def test_auto_registered_shop_is_recorded_in_audit(self): self._order("S1", shop="新店") rows = ( self.db.query(AuditLog).filter(AuditLog.target == "huya_collab_shops").all() ) assert [row.action for row in rows] == ["huya:collab:shop_create"] assert "新店" in rows[0].detail def test_ensure_shops_registered_ignores_blank_and_oversized_names(self): created = collab.ensure_shops_registered( self.db, self.user, ["", " ", "店" * 65, "正常店"] ) assert created == ["正常店"] assert collab.list_shops(self.db) == ["正常店"] def test_registered_shop_can_be_picked_by_the_filter(self): self._order("S1", shop="佳佳") rows, total = collab.list_orders(self.db, shop="佳佳") assert total == 1 and rows[0].order_no == "S1"