"""S4 回归：账号操作的事件发射与审计关联。 S4 之前的状态（都是「定义了却从没发过」）： - `assigned` / `tag_changed` 两个事件类型**定义了、前端也渲染了，但没有任何代码发**； - 分配/改标签的审计**不带 `account_id`**（`target` 是自由文本，账号改名就关联不上）； - 清 CK 不发事件 —— 而这是**不可逆**操作，时间线里却不留痕； - `ck_invalid` **每轮巡检都发**：注释写着「避免每轮刷屏」，代码却是一个失效一周的 账号攒 28 条重复事件。 本文件守住修复后的行为。 """ from datetime import UTC, datetime import pytest from tests.support.db import get_test_engine, make_test_session from web.backend.models import Account, AccountEvent, AuditLog, User from web.backend.services import account_events_service as svc NOW = datetime(2026, 10, 4, 12, 0, 0, tzinfo=UTC) class _Base: def setup_method(self): self.engine = get_test_engine() self.session = make_test_session(self.engine) self.admin = User( username="act-admin", password_hash="h", role="super_admin", is_active=True ) self.support = User( username="act-kf", password_hash="h", role="support", is_active=True ) self.session.add_all([self.admin, self.support]) self.session.commit() def teardown_method(self): self.session.close() def make_account(self, name="acc", **kwargs) -> Account: account = Account( username=name, password="p", email=f"{name}@e.com", email_password="ep", cookie=kwargs.pop("cookie", "acf_uid=1"), ck_check_status=kwargs.pop("ck_check_status", "valid"), **kwargs, ) self.session.add(account) self.session.commit() return account def events(self, account_id: int, event_type: str) -> list[AccountEvent]: return ( self.session.query(AccountEvent) .filter( AccountEvent.account_id == account_id, AccountEvent.event_type == event_type, ) .order_by(AccountEvent.id.asc()) .all() ) def latest_audit(self, action: str) -> AuditLog | None: return ( self.session.query(AuditLog) .filter(AuditLog.action == action) .order_by(AuditLog.id.desc()) .first() ) class TestAssignEmitsEvent(_Base): def test_single_assign_writes_event_and_links_audit(self): from web.backend.routers.accounts import assign_account from web.backend.schemas import AccountAssign account = self.make_account("a") assign_account( account.id, AccountAssign(assigned_to=self.support.id), db=self.session, current=self.admin, ) self.session.refresh(account) assert account.assigned_to == self.support.id events = self.events(account.id, svc.EVENT_ASSIGNED) assert len(events) == 1 assert events[0].category == svc.CATEGORY_OPS assert "act-kf" in events[0].title assert events[0].actor_type == "user" assert events[0].actor_id == self.admin.id # 审计也要能反查到账号（以前只写自由文本 target） audit = self.latest_audit("account:assign") assert audit is not None and audit.account_id == account.id def test_unassign_writes_event(self): from web.backend.routers.accounts import assign_account from web.backend.schemas import AccountAssign account = self.make_account("a", assigned_to=self.support.id) assign_account( account.id, AccountAssign(assigned_to=None), db=self.session, current=self.admin, ) events = self.events(account.id, svc.EVENT_ASSIGNED) assert len(events) == 1 assert "取消分配" in events[0].title def test_batch_assign_emits_one_event_per_account(self): from web.backend.routers.accounts import batch_assign_accounts from web.backend.schemas import BatchAssign accounts = [self.make_account(f"a{i}") for i in range(3)] batch_assign_accounts( BatchAssign( account_ids=[a.id for a in accounts], assigned_to=self.support.id ), db=self.session, current=self.admin, ) for account in accounts: self.session.refresh(account) assert account.assigned_to == self.support.id assert len(self.events(account.id, svc.EVENT_ASSIGNED)) == 1 class TestTagEmitsEvent(_Base): def test_single_tag_writes_event_with_from_to(self): from web.backend.routers.accounts import set_account_tag from web.backend.schemas import AccountTag account = self.make_account("a", tag="旧标签") set_account_tag( account.id, AccountTag(tag="新标签"), db=self.session, current=self.admin ) self.session.refresh(account) assert account.tag == "新标签" events = self.events(account.id, svc.EVENT_TAG_CHANGED) assert len(events) == 1 assert events[0].detail == {"from": "旧标签", "to": "新标签"} assert events[0].category == svc.CATEGORY_OPS audit = self.latest_audit("account:tag") assert audit is not None and audit.account_id == account.id def test_batch_tag_emits_per_account_events(self): from web.backend.routers.accounts import batch_tag from web.backend.schemas import AccountTag accounts = [self.make_account(f"a{i}") for i in range(3)] batch_tag( AccountTag(tag="批量标签", account_ids=[a.id for a in accounts]), db=self.session, current=self.admin, ) for account in accounts: self.session.refresh(account) assert account.tag == "批量标签" events = self.events(account.id, svc.EVENT_TAG_CHANGED) assert len(events) == 1 assert events[0].detail["to"] == "批量标签" class TestCkCleared(_Base): def test_single_clear_emits_event_and_audit_link(self): from web.backend.routers.cookies import delete_cookie account = self.make_account("a") delete_cookie(account.id, db=self.session, current=self.admin) self.session.refresh(account) assert account.cookie == "" events = self.events(account.id, svc.EVENT_CK_CLEARED) assert len(events) == 1 assert events[0].category == svc.CATEGORY_AUTH audit = self.latest_audit("cookie:delete") assert audit is not None and audit.account_id == account.id def test_batch_clear_emits_event_per_account(self): from web.backend.routers.cookies import delete_cookies_batch accounts = [self.make_account(f"a{i}") for i in range(3)] delete_cookies_batch( account_ids=",".join(str(a.id) for a in accounts), db=self.session, current=self.admin, ) for account in accounts: self.session.refresh(account) assert account.cookie == "" assert len(self.events(account.id, svc.EVENT_CK_CLEARED)) == 1 class TestCkInvalidOnlyOnTransition(_Base): """`ck_invalid` 是**一次状态转变**，不是每轮读数。 旧实现每轮巡检只要判定 invalid 就发一条（源码注释却写着「避免每轮刷屏」）—— 6 小时一轮、失效一周 = 28 条一模一样的事件，把时间线淹掉。 """ def _job(self): from web.backend.models import AutomationJob from web.backend.services.automation.jobs import ck_health return AutomationJob( key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 ) def _run(self, message="鱼丸接口: 账户登陆异常，请重新登陆"): """跑一轮巡检。 ⚠️ 先清掉 `ck_checked_at`：候选查询会过滤「距上次检测超过作业间隔」， interval 最小 60 秒，同一轮里刚查过的账号**根本不会被再次选中** —— 不重置的话这个用例是假绿的（跑三轮其实只跑了一轮）。 """ from web.backend.services.automation.jobs import ck_health self.session.query(Account).update( {Account.ck_checked_at: None}, synchronize_session=False ) self.session.commit() original = ck_health.check_douyu_cookie ck_health.check_douyu_cookie = lambda _c: { "valid": False, "message": message, "fish_ball": None, "nickname": None, "level": None, "checked_at": NOW, } try: return ck_health.run_ck_health(self.session, self._job()) finally: ck_health.check_douyu_cookie = original def test_repeated_invalid_rounds_emit_only_once(self): account = self.make_account("a", ck_check_status="valid") self._run() self._run() self._run() assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 1 def test_recovering_then_failing_again_emits_a_new_event(self): """由好变坏 → 恢复 → 再变坏，应当各记一条（是两次转变）。""" from web.backend.services.account_lifecycle import on_ck_verdict account = self.make_account("a", ck_check_status="valid") self._run() assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 1 # 恢复（巡检判 valid） from web.backend.services.automation.jobs import ck_health original = ck_health.check_douyu_cookie ck_health.check_douyu_cookie = lambda _c: { "valid": True, "message": "ok", "fish_ball": 1, "nickname": "n", "level": 1, "checked_at": NOW, } self.session.query(Account).update( {Account.ck_checked_at: None}, synchronize_session=False ) self.session.commit() try: ck_health.run_ck_health(self.session, self._job()) finally: ck_health.check_douyu_cookie = original # 再失效 self._run() assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 2 assert account.id # 避免未使用告警 # error（查不到）不改变结论、也不发事件 on_ck_verdict(self.session, account, "error", detail={}, occurred_at=NOW) self.session.commit() assert len(self.events(account.id, svc.EVENT_CK_INVALID)) == 2 class TestEmitBulk(_Base): def test_writes_many_and_returns_count(self): accounts = [self.make_account(f"a{i}") for i in range(5)] written = svc.emit_bulk( self.session, [ { "account_id": acc.id, "event_type": svc.EVENT_TAG_CHANGED, "title": f"批量 {acc.username}", "category": svc.CATEGORY_OPS, } for acc in accounts ], ) self.session.commit() assert written == 5 for acc in accounts: assert len(self.events(acc.id, svc.EVENT_TAG_CHANGED)) == 1 def test_skips_invalid_rows_and_never_raises(self): account = self.make_account("a") written = svc.emit_bulk( self.session, [ {"account_id": None, "event_type": "x", "title": "无账号"}, {"account_id": account.id, "event_type": "", "title": "无类型"}, { "account_id": account.id, "event_type": svc.EVENT_TAG_CHANGED, "title": "有效", }, ], ) self.session.commit() assert written == 1 def test_empty_input_is_noop(self): assert svc.emit_bulk(self.session, []) == 0 class TestEventsCategoryFilter(_Base): def test_filter_by_category(self): from web.backend.routers.accounts import account_events account = self.make_account("a") svc.emit( self.session, account_id=account.id, event_type=svc.EVENT_TAG_CHANGED, title="改标签", category=svc.CATEGORY_OPS, ) svc.emit( self.session, account_id=account.id, event_type=svc.EVENT_CK_INVALID, title="失效", category=svc.CATEGORY_AUTH, ) self.session.commit() def titles(**kwargs): page = account_events( account.id, page=1, page_size=50, db=self.session, current=self.admin, **kwargs, ) return [item["title"] for item in page["items"]] assert set(titles(category="")) == {"改标签", "失效"} assert titles(category=svc.CATEGORY_OPS) == ["改标签"] assert titles(category=svc.CATEGORY_AUTH) == ["失效"] # 事件类型筛选仍然可用 assert titles(event_type=svc.EVENT_CK_INVALID) == ["失效"] def test_response_carries_ledger_fields(self): from web.backend.routers.accounts import account_events account = self.make_account("a") svc.emit( self.session, account_id=account.id, event_type=svc.EVENT_EXCHANGE, title="兑换 兔兔白日梦", category=svc.CATEGORY_BUSINESS, subject_type="commodity", subject_key="sku-1", actor_type="user", actor_id=self.admin.id, ) self.session.commit() page = account_events( account.id, page=1, page_size=50, db=self.session, current=self.admin ) item = page["items"][0] assert item["category"] == svc.CATEGORY_BUSINESS assert item["subject_key"] == "sku-1" assert item["actor_id"] == self.admin.id class TestBatchDeleteRecordsIds(_Base): def test_delete_audit_keeps_account_ids_for_forensics(self): """删账号会**连带删掉它的事件时间线**，所以审计是唯一痕迹。""" from web.backend.routers.accounts import batch_delete_accounts accounts = [self.make_account(f"a{i}") for i in range(3)] ids = [a.id for a in accounts] batch_delete_accounts( account_ids=",".join(str(i) for i in ids), db=self.session, current=self.admin, ) audit = self.latest_audit("account:delete") assert audit is not None assert audit.detail is not None for account_id in ids: assert str(account_id) in audit.detail class TestRemarkEndpoint(_Base): """改备注：备注是自由文本，走单个接口（标签那种短枚举才有批量）。""" def _set(self, account_id: int, remark, user=None): from web.backend.routers.accounts import set_account_remark from web.backend.schemas import AccountRemark return set_account_remark( account_id, AccountRemark(remark=remark), db=self.session, current=user or self.admin, ) def test_sets_remark_and_emits_event_with_from_to(self): account = self.make_account("a", remark="老备注") res = self._set(account.id, "新备注") self.session.refresh(account) assert res["success"] and account.remark == "新备注" events = self.events(account.id, svc.EVENT_REMARK_CHANGED) assert len(events) == 1 assert events[0].category == svc.CATEGORY_OPS assert events[0].detail == {"from": "老备注", "to": "新备注"} assert events[0].actor_id == self.admin.id audit = self.latest_audit("account:remark") assert audit is not None and audit.account_id == account.id def test_unchanged_remark_is_noop(self): """没变就别写事件 —— 否则时间线会被「保存了一下但没改」刷满。""" account = self.make_account("a", remark="一样") res = self._set(account.id, "一样") assert res["success"] is True assert self.events(account.id, svc.EVENT_REMARK_CHANGED) == [] assert self.latest_audit("account:remark") is None def test_empty_string_clears_remark_and_is_recorded(self): account = self.make_account("a", remark="要清掉") self._set(account.id, " ") self.session.refresh(account) assert account.remark == "" events = self.events(account.id, svc.EVENT_REMARK_CHANGED) assert len(events) == 1 assert events[0].detail == {"from": "要清掉", "to": ""} def test_remark_is_trimmed_and_truncated(self): account = self.make_account("a") self._set(account.id, " " + "字" * 400 + " ") self.session.refresh(account) assert account.remark == "字" * 256 def test_missing_account_returns_404(self): from fastapi import HTTPException with pytest.raises(HTTPException) as exc: self._set(999999, "x") assert exc.value.status_code == 404 def test_permission_granted_to_operation_and_support(self): """新权限点必须在角色默认里 —— 否则加完接口没人能用。 注：`custom_permissions` 若被显式设过会覆盖角色默认，本次确认过库里 admin / kf1_test 的 `custom_permissions` 是 JSON null（Python 侧为 None）， 走的是角色默认，所以新增权限点不会把它们锁掉。 """ from web.backend.permissions import ROLE_PERMISSIONS for role in ("super_admin", "operation", "support"): assert "account:remark" in ROLE_PERMISSIONS[role], ( f"{role} 缺少 account:remark" )