106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
"""充值审计日志的权限、查询和脱敏测试。"""
|
|
|
|
import json
|
|
import os
|
|
import unittest
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
os.environ.setdefault("APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=")
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from web.backend.database import Base
|
|
from web.backend.models import Account, AuditLog, DouyuTask, User
|
|
from web.backend.routers.audit import list_audit_logs, require_super_admin
|
|
from web.backend.services.audit_service import record_audit
|
|
|
|
|
|
class AuditLogTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(self.engine)
|
|
self.session = sessionmaker(bind=self.engine)()
|
|
self.admin = User(username="admin", password_hash="hash", role="super_admin")
|
|
self.operator = User(
|
|
username="operator", password_hash="hash", role="operation",
|
|
custom_permissions=["audit:view"],
|
|
)
|
|
self.session.add_all([self.admin, self.operator])
|
|
self.session.commit()
|
|
|
|
def tearDown(self):
|
|
self.session.close()
|
|
Base.metadata.drop_all(self.engine)
|
|
self.engine.dispose()
|
|
|
|
def test_only_super_admin_can_read_audit_logs(self):
|
|
with self.assertRaises(HTTPException) as context:
|
|
require_super_admin(self.operator)
|
|
self.assertEqual(context.exception.status_code, 403)
|
|
self.assertIs(require_super_admin(self.admin), self.admin)
|
|
|
|
def test_list_audit_logs_supports_pagination_and_result_filter(self):
|
|
for index in range(3):
|
|
record_audit(
|
|
self.session, self.admin, action="recharge:yyb:payment",
|
|
target=f"yyb_task:{index}", detail={"task_id": index}, success=index != 1,
|
|
)
|
|
self.session.commit()
|
|
|
|
result = list_audit_logs(
|
|
page=1, page_size=1, username=None, action="recharge:yyb:payment",
|
|
keyword=None, success=True, start_time=None, end_time=None,
|
|
db=self.session, _=self.admin,
|
|
)
|
|
|
|
self.assertEqual(result["total"], 2)
|
|
self.assertEqual(len(result["items"]), 1)
|
|
self.assertTrue(result["items"][0]["success"])
|
|
|
|
def test_audit_detail_masks_sensitive_fields(self):
|
|
record_audit(
|
|
self.session, self.admin, action="recharge:yyb:payment",
|
|
detail={"token": "token-value", "nested": {"cookie": "cookie-value"}, "order_id": "order-1"},
|
|
)
|
|
self.session.commit()
|
|
|
|
entry = self.session.query(AuditLog).one()
|
|
detail = json.loads(entry.detail)
|
|
self.assertEqual(detail["token"], "[已脱敏]")
|
|
self.assertEqual(detail["nested"]["cookie"], "[已脱敏]")
|
|
self.assertEqual(detail["order_id"], "order-1")
|
|
|
|
def test_old_douyu_recharge_log_is_enriched_with_account_identity(self):
|
|
account = Account(
|
|
username="douyu-login", password="password", email="mail@example.com",
|
|
email_password="mail-password", uid="10001", nickname="斗鱼昵称",
|
|
)
|
|
self.session.add(account)
|
|
self.session.commit()
|
|
self.session.add(DouyuTask(
|
|
batch_id="batch-old", account_id=account.id, task_type="create_gold_qr",
|
|
handbook_scope="elite", created_by=self.admin.id,
|
|
))
|
|
self.session.add(AuditLog(
|
|
user_id=self.admin.id, username=self.admin.username,
|
|
action="recharge:douyu:create", target="douyu_batch:batch-old", detail="{}", success=True,
|
|
))
|
|
self.session.commit()
|
|
|
|
result = list_audit_logs(
|
|
page=1, page_size=50, username=None, action=None, keyword=None, success=None,
|
|
start_time=None, end_time=None, db=self.session, _=self.admin,
|
|
)
|
|
|
|
detail = json.loads(result["items"][0]["detail"])
|
|
self.assertEqual(detail["recharge_accounts"][0]["username"], "douyu-login")
|
|
self.assertEqual(detail["recharge_accounts"][0]["douyu_uid"], "10001")
|
|
self.assertEqual(detail["recharge_channel"], "wechat_qr")
|
|
self.assertEqual(detail["payment_method"], "微信扫码支付")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|