"""虎牙协作礼包字典：CRUD、改名级联、积分快照边界与权限。""" from __future__ import annotations import json import pytest from tests.support.db import get_test_engine, make_test_session from web.backend.models import AuditLog, HuyaCollabGiftCatalog, HuyaExternalOrder, User from web.backend.permissions import ROLE_PERMISSIONS from web.backend.routers.huya import collab as collab_api from web.backend.schemas import HuyaCollabGiftCatalogOut from web.backend.services import huya_collab_service as collab class TestHuyaCollabGiftCatalog: def setup_method(self): self.engine = get_test_engine() self.db = make_test_session(self.engine) self.user = User(username="gift_op", password_hash="h", role="operation") self.support = User(username="gift_support", password_hash="h", role="support") self.db.add_all([self.user, self.support]) self.db.commit() def teardown_method(self): self.db.close() def _gift(self, name: str, points: int = 8000, **kwargs) -> HuyaCollabGiftCatalog: return collab.create_gift( self.db, self.user, name=name, points=points, **kwargs ) def _order(self, order_no: str, gift_name: str = "", **kwargs) -> HuyaExternalOrder: return collab.create_order( self.db, self.user, order_no=order_no, gift_name=gift_name, **kwargs, ) def _audits(self, gift: HuyaCollabGiftCatalog) -> list[AuditLog]: return ( self.db.query(AuditLog) .filter(AuditLog.target == f"huya_collab_gift:{gift.id}") .order_by(AuditLog.id.asc()) .all() ) def test_create_list_and_serialize(self): self._gift("乙礼包", points=900, sort_order=2) self._gift("甲礼包", points=800, sort_order=1) rows = collab.list_gift_records(self.db) assert [row.name for row in rows] == ["甲礼包", "乙礼包"] dumped = HuyaCollabGiftCatalogOut.model_validate( collab.serialize_gift( rows[0], users={int(self.user.id): self.user.username} ) ).model_dump() assert dumped["name"] == "甲礼包" assert dumped["points"] == 800 assert dumped["is_active"] is True assert dumped["created_by_name"] == self.user.username assert [row.action for row in self._audits(rows[0])] == [ "huya:collab:gift_create" ] def test_create_rejects_duplicate_blank_and_invalid_points(self): self._gift("甲礼包") with pytest.raises(collab.CollabError) as duplicate: self._gift(" 甲礼包 ") assert duplicate.value.status_code == 409 assert duplicate.value.code == "gift_name_conflict" with pytest.raises(collab.CollabError) as blank: self._gift(" ") assert blank.value.code == "gift_name_required" with pytest.raises(collab.CollabError) as invalid: self._gift("乙礼包", points="not-a-number") assert invalid.value.code == "required_points_invalid" def test_create_order_uses_catalog_points_when_points_are_omitted(self): gift = self._gift("标准礼包", points=4321) order = self._order("GIFT-POINTS-1", gift_name=gift.name) assert order.required_points == 4321 def test_changing_order_gift_name_uses_new_catalog_points(self): first = self._gift("礼包A", points=200) second = self._gift("礼包B", points=300) order = self._order("GIFT-POINTS-2", gift_name=first.name) changes = collab.update_order( self.db, self.user, order, {"gift_name": second.name} ) assert changes["required_points"] == {"old": 200, "new": 300} assert order.required_points == 300 def test_rename_cascades_but_points_change_does_not_backfill_orders(self): gift = self._gift("旧礼包名", points=8000) order = self._order("GIFT-RENAME-1", gift_name=gift.name) changes = collab.update_gift( self.db, self.user, gift, {"name": "新礼包名", "points": 9000} ) self.db.refresh(order) assert changes["name"] == {"old": "旧礼包名", "new": "新礼包名"} assert changes["points"] == {"old": 8000, "new": 9000} assert order.gift_name == "新礼包名" assert order.required_points == 8000 audit = self._audits(gift)[-1] assert json.loads(audit.detail)["renamed_orders"] == 1 def test_delete_is_blocked_while_orders_use_the_gift(self): gift = self._gift("在用的礼包") self._order("GIFT-DELETE-1", gift_name=gift.name) with pytest.raises(collab.CollabError) as excinfo: collab.delete_gift(self.db, self.user, gift) assert excinfo.value.code == "gift_in_use" assert self.db.query(HuyaCollabGiftCatalog).count() == 1 def test_unused_gift_can_be_deleted_with_audit(self): gift = self._gift("可删礼包") snapshot = collab.delete_gift(self.db, self.user, gift) assert snapshot["name"] == "可删礼包" assert self.db.query(HuyaCollabGiftCatalog).count() == 0 audit = ( self.db.query(AuditLog) .filter(AuditLog.action == "huya:collab:gift_delete") .one() ) assert "可删礼包" in audit.detail def test_inactive_gift_stays_in_catalog_for_management(self): gift = self._gift("准备停用") collab.update_gift(self.db, self.user, gift, {"is_active": False}) rows = collab.list_gift_records(self.db) assert len(rows) == 1 assert rows[0].is_active is False def test_write_permission_is_operation_only_by_default(self): assert "huya:collab:manage" in ROLE_PERMISSIONS["operation"] assert "huya:collab:manage" not in ROLE_PERMISSIONS["support"] assert "huya:collab:manage" in ROLE_PERMISSIONS["super_admin"] def test_routes_are_registered(self): from web.backend.main import app paths = app.openapi()["paths"] assert "/api/huya/collab-gifts" in paths assert "get" in paths["/api/huya/collab-gifts"] assert "post" in paths["/api/huya/collab-gifts"] assert "/api/huya/collab-gifts/{gift_id}" in paths assert "put" in paths["/api/huya/collab-gifts/{gift_id}"] assert "delete" in paths["/api/huya/collab-gifts/{gift_id}"] def test_router_response_keeps_all_catalog_fields(self): gift = self._gift("接口礼包", points=6600, remark="按兑换记录标准名") payload = collab_api.create_collab_gift( req=collab_api.HuyaCollabGiftCatalogCreate(name="另一个礼包", points=7700), db=self.db, current=self.user, ) assert payload["name"] == "另一个礼包" assert payload["points"] == 7700 assert payload["created_by_name"] == self.user.username listed = collab_api.list_collab_gifts(db=self.db, _current=self.user) by_name = {item["name"]: item for item in listed} assert by_name[gift.name]["remark"] == "按兑换记录标准名"