diff --git a/.githooks/pre-push b/.githooks/pre-push new file mode 100755 index 0000000..cbeaa60 --- /dev/null +++ b/.githooks/pre-push @@ -0,0 +1,43 @@ +#!/usr/bin/env bash +set -euo pipefail + +cd "$(git rev-parse --show-toplevel)" + +push_lines="$(cat)" +while read -r local_ref local_sha remote_ref remote_sha; do + if [ "$local_sha" = "0000000000000000000000000000000000000000" ]; then + echo "[pre-push] 分支删除推送，跳过测试" + exit 0 + fi +done <<< "$push_lines" + +if [ "${SKIP_TESTS:-0}" = "1" ]; then + echo "[pre-push] SKIP_TESTS=1，跳过测试" + exit 0 +fi + +python_files=() +while read -r local_ref local_sha remote_ref remote_sha; do + [ -z "$local_sha" ] && continue + if [ "$local_sha" = "0000000000000000000000000000000000000000" ]; then + continue + fi + if [ "$remote_sha" = "0000000000000000000000000000000000000000" ]; then + changed_paths="$(git diff-tree --no-commit-id --name-only -r --root "$local_sha" -- '*.py')" + else + changed_paths="$(git diff --name-only "$remote_sha..$local_sha" -- '*.py')" + fi + while IFS= read -r path; do + [ -n "$path" ] && python_files+=("$path") + done <<< "$changed_paths" +done <<< "$push_lines" + +if [ "${#python_files[@]}" -gt 0 ]; then + echo "[pre-push] 检查 Python 格式..." + uv run --group dev ruff format --check "${python_files[@]}" + echo "[pre-push] Python 格式通过" +fi + +echo "[pre-push] 运行 pytest..." +./dev.sh test -q +echo "[pre-push] pytest 通过" diff --git a/.gitignore b/.gitignore index f459a6b..1a207f3 100644 --- a/.gitignore +++ b/.gitignore @@ -27,6 +27,7 @@ evidence/ # ── 逆向实验脚本（保留业务脚本 migrate_sqlite_to_mysql.py） scripts/* !scripts/migrate_sqlite_to_mysql.py +!scripts/install-git-hooks.sh # ── 逆向工具集（frida hook、unidbg hydev、虎牙协议复刻脚本等） tools/ @@ -37,4 +38,4 @@ web/frontend/dist/ .reasonix/ .tmp_reverse/ apks/ -work/ \ No newline at end of file +work/ diff --git a/README.md b/README.md index 5894086..1fae1cf 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,10 @@ uv run --group dev pytest # 或 ./dev.sh test # 只跑单个文件 / 用例 uv run --group dev pytest tests/test_cookie_operations.py +# 格式化 / 仅检查格式 +./dev.sh format tests conftest.py +./dev.sh format-check tests conftest.py + # 容器内回归（镜像内为构建时代码，改动后需重新构建） docker compose --profile test run --rm test ``` @@ -125,8 +129,14 @@ docker compose --profile test run --rm test 编译 + 空库整链 `upgrade head` 后与模型元数据表/列/索引对齐；MySQL 专属 DDL 由迁移脚本内的方言防护跳过，真实执行仍以部署流程 `docker-entrypoint.sh` / `./deploy.sh` / `./dev.sh` 的 MySQL 为准）。 -已安装本地 pre-push 钩子：推送前自动跑全量测试，失败阻止推送 -（紧急绕过：`git push --no-verify` 或 `SKIP_TESTS=1 git push`）。 +启用本地 pre-push 门禁（Ruff 格式检查 + pytest）： + +```bash +./scripts/install-git-hooks.sh +``` + +推送前自动跑全量测试，失败阻止推送（紧急绕过：`git push --no-verify` +或 `SKIP_TESTS=1 git push`）。 ## 技术栈 diff --git a/conftest.py b/conftest.py index c0b6621..c594f20 100644 --- a/conftest.py +++ b/conftest.py @@ -9,4 +9,6 @@ web/backend 的模块在 import 阶段就会读取 DATABASE_URL 创建全局引 import os os.environ.setdefault("DATABASE_URL", "sqlite://") -os.environ.setdefault("APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=") \ No newline at end of file +os.environ.setdefault( + "APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=" +) diff --git a/dev.sh b/dev.sh index 41a5edb..dc8ba90 100755 --- a/dev.sh +++ b/dev.sh @@ -61,6 +61,16 @@ if [ "${1:-}" = "test" ]; then exec uv run --group dev pytest "$@" fi +if [ "${1:-}" = "format" ]; then + shift + exec uv run --group dev ruff format "$@" +fi + +if [ "${1:-}" = "format-check" ]; then + shift + exec uv run --group dev ruff format --check "$@" +fi + BACKEND_PID="" FRONTEND_PID="" WORKER_PID="" diff --git a/pyproject.toml b/pyproject.toml index 064fd4b..4949e6d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,4 +41,5 @@ testpaths = ["tests"] [dependency-groups] dev = [ "pytest>=9.1.1", + "ruff>=0.12.0", ] diff --git a/scripts/install-git-hooks.sh b/scripts/install-git-hooks.sh new file mode 100755 index 0000000..c2ef33d --- /dev/null +++ b/scripts/install-git-hooks.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "$0")/.." && pwd)" +git -C "$ROOT_DIR" config core.hooksPath .githooks +echo "已启用 .githooks（Ruff 格式检查 + pytest pre-push 门禁）" diff --git a/tests/test_account_sensitive_fields.py b/tests/test_account_sensitive_fields.py index 9f16389..86203ff 100644 --- a/tests/test_account_sensitive_fields.py +++ b/tests/test_account_sensitive_fields.py @@ -1,4 +1,4 @@ -import unittest +import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @@ -10,22 +10,24 @@ from web.backend.routers.accounts import list_accounts, set_account_tag from web.backend.schemas import AccountTag -class AccountSensitiveFieldsTests(unittest.TestCase): - def setUp(self): +class TestAccountSensitiveFields: + def setup_method(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.session.add(self.admin) - self.session.add(Account( - username="account", - password="account-password", - email="account@example.com", - email_password="email-password", - )) + self.session.add( + Account( + username="account", + password="account-password", + email="account@example.com", + email_password="email-password", + ) + ) self.session.commit() - def tearDown(self): + def teardown_method(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() @@ -44,9 +46,9 @@ class AccountSensitiveFieldsTests(unittest.TestCase): ) item = result["items"][0] - self.assertIsNone(item.password) - self.assertIsNone(item.email) - self.assertIsNone(item.email_password) + assert item.password is None + assert item.email is None + assert item.email_password is None def test_admin_can_explicitly_request_sensitive_fields(self): result = list_accounts( @@ -62,34 +64,42 @@ class AccountSensitiveFieldsTests(unittest.TestCase): ) item = result["items"][0] - self.assertEqual(item.password, "account-password") - self.assertEqual(item.email, "account@example.com") - self.assertEqual(item.email_password, "email-password") + assert item.password == "account-password" + assert item.email == "account@example.com" + assert item.email_password == "email-password" def test_support_can_only_change_tags_on_assigned_accounts(self): support = User(username="support", password_hash="hash", role="support") - other_support = User(username="other-support", password_hash="hash", role="support") + other_support = User( + username="other-support", password_hash="hash", role="support" + ) self.session.add_all([support, other_support]) self.session.commit() assigned = Account( - username="assigned", password="password", email="assigned@example.com", - email_password="mail-password", assigned_to=support.id, + username="assigned", + password="password", + email="assigned@example.com", + email_password="mail-password", + assigned_to=support.id, ) other = Account( - username="other", password="password", email="other@example.com", - email_password="mail-password", assigned_to=other_support.id, + username="other", + password="password", + email="other@example.com", + email_password="mail-password", + assigned_to=other_support.id, ) self.session.add_all([assigned, other]) self.session.commit() - set_account_tag(assigned.id, AccountTag(tag="客服组"), db=self.session, current=support) + set_account_tag( + assigned.id, AccountTag(tag="客服组"), db=self.session, current=support + ) self.session.refresh(assigned) - self.assertEqual(assigned.tag, "客服组") + assert assigned.tag == "客服组" - with self.assertRaises(HTTPException) as context: - set_account_tag(other.id, AccountTag(tag="越权"), db=self.session, current=support) - self.assertEqual(context.exception.status_code, 404) - - -if __name__ == "__main__": - unittest.main() + with pytest.raises(HTTPException) as context: + set_account_tag( + other.id, AccountTag(tag="越权"), db=self.session, current=support + ) + assert context.value.status_code == 404 diff --git a/tests/test_audit_logs.py b/tests/test_audit_logs.py index e3a0c10..a1abefc 100644 --- a/tests/test_audit_logs.py +++ b/tests/test_audit_logs.py @@ -1,7 +1,8 @@ """充值审计日志的权限、查询和脱敏测试。""" +import pytest + import json -import unittest from fastapi import HTTPException from sqlalchemy import create_engine @@ -13,89 +14,127 @@ 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): +class TestAuditLog: + def setup_method(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", + 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): + def teardown_method(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: + with pytest.raises(HTTPException) as context: require_super_admin(self.operator) - self.assertEqual(context.exception.status_code, 403) - self.assertIs(require_super_admin(self.admin), self.admin) + assert context.value.status_code == 403 + assert require_super_admin(self.admin) is 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, + 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, + 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"]) + assert result["total"] == 2 + assert len(result["items"]) == 1 + assert 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, + 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") + assert detail["token"] == "[已脱敏]" + assert detail["nested"]["cookie"] == "[已脱敏]" + assert 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="斗鱼昵称", + 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.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, + 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() + assert detail["recharge_accounts"][0]["username"] == "douyu-login" + assert detail["recharge_accounts"][0]["douyu_uid"] == "10001" + assert detail["recharge_channel"] == "wechat_qr" + assert detail["payment_method"] == "微信扫码支付" diff --git a/tests/test_cookie_custom_order.py b/tests/test_cookie_custom_order.py index 58123bb..abbb9e3 100644 --- a/tests/test_cookie_custom_order.py +++ b/tests/test_cookie_custom_order.py @@ -1,4 +1,3 @@ -import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine @@ -9,8 +8,8 @@ from web.backend.models import Account, LoginTask, User from web.backend.routers.cookies import _order_cookie_tasks -class CustomCookieOrderTests(unittest.TestCase): - def setUp(self): +class TestCustomCookieOrder: + def setup_method(self): self.engine = create_engine("sqlite://") Base.metadata.create_all(self.engine) self.db = Session(self.engine) @@ -27,41 +26,32 @@ class CustomCookieOrderTests(unittest.TestCase): self.db.flush() # 完成时间特意与用户输入顺序相反。 for index, name in enumerate(("account-a", "account-b", "account-c")): - self.db.add(LoginTask( - batch_id=f"batch-{name}", - account_id=accounts[name].id, - status="success", - created_by=self.user.id, - finished_at=now + timedelta(minutes=index), - )) + self.db.add( + LoginTask( + batch_id=f"batch-{name}", + account_id=accounts[name].id, + status="success", + created_by=self.user.id, + finished_at=now + timedelta(minutes=index), + ) + ) self.db.commit() - def tearDown(self): + def teardown_method(self): self.db.close() Base.metadata.drop_all(self.engine) self.engine.dispose() def test_custom_cookie_order_follows_input_not_finished_time(self): selected_names = ["account-c", "account-a", "account-b"] - tasks = ( - _order_cookie_tasks( - self.db.query(LoginTask) - .join(Account, LoginTask.account_id == Account.id) - .filter(Account.username.in_(selected_names)), - selected_names, - ) - .all() - ) + tasks = _order_cookie_tasks( + self.db.query(LoginTask) + .join(Account, LoginTask.account_id == Account.id) + .filter(Account.username.in_(selected_names)), + selected_names, + ).all() account_names = { - account.id: account.username - for account in self.db.query(Account).all() + account.id: account.username for account in self.db.query(Account).all() } - self.assertEqual( - [account_names[task.account_id] for task in tasks], - selected_names, - ) - - -if __name__ == "__main__": - unittest.main() + assert [account_names[task.account_id] for task in tasks] == selected_names diff --git a/tests/test_cookie_operations.py b/tests/test_cookie_operations.py index d2a55fe..68ab5e2 100644 --- a/tests/test_cookie_operations.py +++ b/tests/test_cookie_operations.py @@ -1,4 +1,4 @@ -import unittest +import pytest from types import SimpleNamespace from unittest.mock import patch @@ -7,7 +7,10 @@ from sqlalchemy.orm import sessionmaker from web.backend.database import Base from web.backend.models import Account, LoginTask, User -from web.backend.services.login_service import LoginBatchRunner, cleanup_orphan_relogin_tasks +from web.backend.services.login_service import ( + LoginBatchRunner, + cleanup_orphan_relogin_tasks, +) from web.backend.services.cookie_check_service import check_douyu_cookie from web.backend.routers.cookies import ( check_cookie_operations, @@ -19,14 +22,16 @@ from web.backend.routers.cookies import ( ) -class CookieOperationTests(unittest.TestCase): - def setUp(self): +class TestCookieOperation: + def setup_method(self): self.engine = create_engine("sqlite://") Base.metadata.create_all(self.engine) self.session = sessionmaker(bind=self.engine)() self.support = User(username="support", password_hash="hash", role="support") - self.other_support = User(username="other", password_hash="hash", role="support") + self.other_support = User( + username="other", password_hash="hash", role="support" + ) self.session.add_all([self.support, self.other_support]) self.session.commit() @@ -46,15 +51,33 @@ class CookieOperationTests(unittest.TestCase): ) self.session.add_all([owned, other]) self.session.commit() - self.session.add_all([ - LoginTask(batch_id="owned", account_id=owned.id, created_by=self.support.id, status="success", cookie="owned-secret"), - LoginTask(batch_id="other", account_id=other.id, created_by=self.other_support.id, status="success", cookie="other-secret"), - ]) + self.session.add_all( + [ + LoginTask( + batch_id="owned", + account_id=owned.id, + created_by=self.support.id, + status="success", + cookie="owned-secret", + ), + LoginTask( + batch_id="other", + account_id=other.id, + created_by=self.other_support.id, + status="success", + cookie="other-secret", + ), + ] + ) self.session.commit() - self.owned_task = self.session.query(LoginTask).filter(LoginTask.batch_id == "owned").one() - self.other_task = self.session.query(LoginTask).filter(LoginTask.batch_id == "other").one() + self.owned_task = ( + self.session.query(LoginTask).filter(LoginTask.batch_id == "owned").one() + ) + self.other_task = ( + self.session.query(LoginTask).filter(LoginTask.batch_id == "other").one() + ) - def tearDown(self): + def teardown_method(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() @@ -69,32 +92,38 @@ class CookieOperationTests(unittest.TestCase): current=self.support, ) - self.assertEqual(result["total"], 1) - self.assertEqual(result["items"], [{ - "id": self.owned_task.id, - "account_username": "owned-account", - "tag": "", - "ck_check_status": "", - "ck_checked_at": None, - "created_at": None, - "relogin_status": "", - "relogin_message": "", - "relogin_batch_id": "", - }]) - self.assertNotIn("owned-secret", str(result)) - self.assertNotIn("owned-password", str(result)) + assert result["total"] == 1 + assert result["items"] == [ + { + "id": self.owned_task.id, + "account_username": "owned-account", + "tag": "", + "ck_check_status": "", + "ck_checked_at": None, + "created_at": None, + "relogin_status": "", + "relogin_message": "", + "relogin_batch_id": "", + } + ] + assert "owned-secret" not in str(result) + assert "owned-password" not in str(result) def test_support_cannot_use_cookie_management_endpoints(self): - with self.assertRaisesRegex(Exception, "cookie:view"): + with pytest.raises(Exception, match="cookie:view"): list_cookies(db=self.session, current=self.support) - with self.assertRaisesRegex(Exception, "cookie:view"): + with pytest.raises(Exception, match="cookie:view"): get_cookie(self.owned_task.id, db=self.session, current=self.support) def test_cookie_list_filters_by_account_tag(self): admin = User(username="admin", password_hash="hash", role="super_admin") self.session.add(admin) - self.session.query(Account).filter(Account.id == self.owned_task.account_id).update({"tag": "A组"}) - self.session.query(Account).filter(Account.id == self.other_task.account_id).update({"tag": "B组"}) + self.session.query(Account).filter( + Account.id == self.owned_task.account_id + ).update({"tag": "A组"}) + self.session.query(Account).filter( + Account.id == self.other_task.account_id + ).update({"tag": "B组"}) self.session.commit() result = list_cookies( @@ -108,12 +137,16 @@ class CookieOperationTests(unittest.TestCase): current=admin, ) - self.assertEqual(result["total"], 1) - self.assertEqual([item["id"] for item in result["items"]], [self.owned_task.id]) + assert result["total"] == 1 + assert [item["id"] for item in result["items"]] == [self.owned_task.id] def test_support_operation_tag_filter_and_list_stay_scoped(self): - self.session.query(Account).filter(Account.id == self.owned_task.account_id).update({"tag": "我的标签"}) - self.session.query(Account).filter(Account.id == self.other_task.account_id).update({"tag": "别人的标签"}) + self.session.query(Account).filter( + Account.id == self.owned_task.account_id + ).update({"tag": "我的标签"}) + self.session.query(Account).filter( + Account.id == self.other_task.account_id + ).update({"tag": "别人的标签"}) self.session.commit() tags = list_cookie_operation_tags(db=self.session, current=self.support) @@ -126,8 +159,8 @@ class CookieOperationTests(unittest.TestCase): current=self.support, ) - self.assertEqual(tags, ["我的标签"]) - self.assertEqual(result["total"], 0) + assert tags == ["我的标签"] + assert result["total"] == 0 @patch("web.backend.routers.cookies._check_one_cookie") def test_support_can_check_only_assigned_cookie(self, check_one_cookie): @@ -147,11 +180,11 @@ class CookieOperationTests(unittest.TestCase): current=self.support, ) - self.assertEqual([item["id"] for item in result["results"]], [self.owned_task.id]) + assert [item["id"] for item in result["results"]] == [self.owned_task.id] self.session.refresh(self.owned_task) self.session.refresh(self.other_task) - self.assertEqual(self.owned_task.ck_check_status, "valid") - self.assertEqual(self.other_task.ck_check_status, "") + assert self.owned_task.ck_check_status == "valid" + assert self.other_task.ck_check_status == "" def test_operation_permission_does_not_allow_unrelated_users(self): no_permission_user = SimpleNamespace( @@ -160,7 +193,7 @@ class CookieOperationTests(unittest.TestCase): role="support", custom_permissions=[], ) - with self.assertRaisesRegex(Exception, "cookie:operate"): + with pytest.raises(Exception, match="cookie:operate"): list_cookie_operations( search="", tag="", @@ -176,12 +209,17 @@ class CookieOperationTests(unittest.TestCase): self.session.commit() result = list_cookie_operations( - search="", tag="", page=1, page_size=20, db=self.session, current=self.support, + search="", + tag="", + page=1, + page_size=20, + db=self.session, + current=self.support, ) - self.assertEqual(result["total"], 1) - self.assertEqual(result["items"][0]["relogin_status"], "relogin_failed") - self.assertIn("旧 Cookie 已保留", result["items"][0]["relogin_message"]) + assert result["total"] == 1 + assert result["items"][0]["relogin_status"] == "relogin_failed" + assert "旧 Cookie 已保留" in result["items"][0]["relogin_message"] def test_service_restart_cleans_orphan_relogin_tasks(self): self.owned_task.status = "relogin_running" @@ -189,12 +227,12 @@ class CookieOperationTests(unittest.TestCase): cleaned = cleanup_orphan_relogin_tasks(self.session) - self.assertEqual(cleaned, 1) + assert cleaned == 1 self.session.refresh(self.owned_task) self.session.refresh(self.other_task) - self.assertEqual(self.owned_task.status, "relogin_failed") - self.assertIn("服务重启", self.owned_task.message) - self.assertEqual(self.other_task.status, "success") + assert self.owned_task.status == "relogin_failed" + assert "服务重启" in self.owned_task.message + assert self.other_task.status == "success" def test_runner_copies_proxy_config_before_background_execution(self): proxy = SimpleNamespace( @@ -219,40 +257,46 @@ class CookieOperationTests(unittest.TestCase): proxy_dict, _ = runner._resolve_static_proxy() - self.assertEqual(proxy_dict, { + assert proxy_dict == { "http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080", - }) + } @patch("web.backend.services.cookie_check_service.requests.get") def test_cookie_check_uses_new_cookie_result(self, mock_get): mock_get.side_effect = [ SimpleNamespace(json=lambda: {"error": 0, "data": {"count": 9}}), - SimpleNamespace(json=lambda: {"error": 0, "data": {"nn": "new-name", "lv": 12}}), + SimpleNamespace( + json=lambda: {"error": 0, "data": {"nn": "new-name", "lv": 12}} + ), ] result = check_douyu_cookie("new-cookie") - self.assertTrue(result["valid"]) - self.assertEqual(result["fish_ball"], 9) - self.assertEqual(result["nickname"], "new-name") - self.assertEqual(result["level"], 12) + assert result["valid"] + assert result["fish_ball"] == 9 + assert result["nickname"] == "new-name" + assert result["level"] == 12 @patch("web.backend.routers.cookies._start_relogin_tasks") def test_relogin_invalid_only_targets_support_visible_accounts(self, start_relogin): self.owned_task.ck_check_status = "invalid" self.other_task.ck_check_status = "invalid" self.session.commit() - start_relogin.return_value = {"batch_id": "batch", "count": 1, "skipped": 0, "success": True} + start_relogin.return_value = { + "batch_id": "batch", + "count": 1, + "skipped": 0, + "success": True, + } result = relogin_invalid_cookie_operations( - search="", tag="", db=self.session, current=self.support, + search="", + tag="", + db=self.session, + current=self.support, ) - self.assertTrue(result["success"]) + assert result["success"] selected_tasks = start_relogin.call_args.args[0] - self.assertEqual([task.id for task in selected_tasks], [self.owned_task.id]) - - -if __name__ == "__main__": - unittest.main() + assert [task.id for task in selected_tasks] == [self.owned_task.id] diff --git a/tests/test_douyu_elite_exchange_flow.py b/tests/test_douyu_elite_exchange_flow.py index 29b5bb7..c486aa2 100644 --- a/tests/test_douyu_elite_exchange_flow.py +++ b/tests/test_douyu_elite_exchange_flow.py @@ -1,17 +1,18 @@ """精英手册兑换：csrf 复用、新链路接口与浏览器状态机路由的测试。""" -import unittest +import pytest + from unittest.mock import Mock, call from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError -class CsrfReuseTests(unittest.TestCase): +class TestCsrfReuse: def test_reuses_cookie_token_without_any_http_call(self): client = DouyuActivityClient("acf_uid=100; cvl_csrf_token=existing-token") client._request_json = Mock() - self.assertEqual(client.csrf_token(), "existing-token") + assert client.csrf_token() == "existing-token" client._request_json.assert_not_called() def test_generates_csrf_when_cookie_missing(self): @@ -24,7 +25,7 @@ class CsrfReuseTests(unittest.TestCase): client._request_json = Mock(side_effect=fake_generate) - self.assertEqual(client.csrf_token(), "new-token") + assert client.csrf_token() == "new-token" client._request_json.assert_called_once() def test_force_refresh_rotates_token(self): @@ -32,45 +33,51 @@ class CsrfReuseTests(unittest.TestCase): client._request_json = Mock(return_value={"error": 0}) client._set_cookie_value("cvl_csrf_token", "rotated-token") - self.assertEqual(client.csrf_token(force_refresh=True), "rotated-token") + assert client.csrf_token(force_refresh=True) == "rotated-token" def test_refreshes_csrf_once_on_token_error(self): client = DouyuActivityClient("acf_uid=100") client.csrf_token = Mock(side_effect=["bad-token", "good-token"]) - client._request_json = Mock(side_effect=[ - {"error": 1, "msg": "csrfToken 校验失败"}, - {"error": 0, "data": {"orderId": "9", "expireSeconds": "300"}}, - ]) + client._request_json = Mock( + side_effect=[ + {"error": 1, "msg": "csrfToken 校验失败"}, + {"error": 0, "data": {"orderId": "9", "expireSeconds": "300"}}, + ] + ) result = client.create_exchange_order(manual_id="m", rid="r", commodity_id="c") - self.assertEqual(result["order_id"], "9") - self.assertEqual(client.csrf_token.call_args_list, [call(), call(force_refresh=True)]) + assert result["order_id"] == "9" + assert client.csrf_token.call_args_list == [call(), call(force_refresh=True)] def test_no_refresh_on_regular_error(self): client = DouyuActivityClient("acf_uid=100") client.csrf_token = Mock(return_value="t") - client._request_json = Mock(return_value={"error": 1, "msg": "需要开通精英手册，才可以兑换本道具哦"}) + client._request_json = Mock( + return_value={"error": 1, "msg": "需要开通精英手册，才可以兑换本道具哦"} + ) - with self.assertRaisesRegex(DouyuActivityError, "需要开通精英手册"): + with pytest.raises(DouyuActivityError, match="需要开通精英手册"): client.create_exchange_order(manual_id="m", rid="r", commodity_id="c") client.csrf_token.assert_called_once_with() def test_pay_referer_includes_room_id_like_browser(self): client = DouyuActivityClient("acf_uid=100") client.csrf_token = Mock(return_value="t") - client._request_json = Mock(return_value={"error": 0, "data": {"exchangeId": "11098933"}}) + client._request_json = Mock( + return_value={"error": 0, "data": {"exchangeId": "11098933"}} + ) client.pay_exchange_order(manual_id="m", order_id="5874", rid="9263298") headers = client._request_json.call_args.kwargs["headers"] - self.assertEqual( - headers["Referer"], - "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId=9263298", + assert ( + headers["Referer"] + == "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId=9263298" ) -class ExchangePlanTests(unittest.TestCase): +class TestExchangePlan: """对齐浏览器兑换按钮状态机（$e / Ve / Ge）的路由判定。""" @staticmethod @@ -92,72 +99,70 @@ class ExchangePlanTests(unittest.TestCase): def test_normal_item_uses_classic_flow(self): plan = DouyuActivityClient.resolve_exchange_plan(self.detail()) - self.assertEqual(plan["action"], "classic") - self.assertEqual(plan["max_num"], 1) + assert plan["action"] == "classic" + assert plan["max_num"] == 1 def test_pre_exchange_order_still_classic(self): - plan = DouyuActivityClient.resolve_exchange_plan(self.detail(preExchange={"isExists": 1})) - self.assertEqual(plan["action"], "classic") + plan = DouyuActivityClient.resolve_exchange_plan( + self.detail(preExchange={"isExists": 1}) + ) + assert plan["action"] == "classic" def test_reached_exchange_limit_blocked(self): plan = DouyuActivityClient.resolve_exchange_plan(self.detail(status=1)) - self.assertEqual(plan["action"], "blocked") - self.assertIn("上限", plan["text"]) + assert plan["action"] == "blocked" + assert "上限" in plan["text"] def test_no_stock_blocked(self): plan = DouyuActivityClient.resolve_exchange_plan(self.detail(storeStatus=0)) - self.assertEqual(plan["action"], "blocked") - self.assertIn("无库存", plan["text"]) + assert plan["action"] == "blocked" + assert "无库存" in plan["text"] def test_room_limited_and_not_in_room_blocked(self): plan = DouyuActivityClient.resolve_exchange_plan( self.detail(roomLimited=1, isLimitRoom=0) ) - self.assertEqual(plan["action"], "blocked") - self.assertIn("限定房间", plan["text"]) + assert plan["action"] == "blocked" + assert "限定房间" in plan["text"] def test_not_open_subscribe_item_subscribes(self): plan = DouyuActivityClient.resolve_exchange_plan( self.detail(openStatus=0, subscribeType=1) ) - self.assertEqual(plan["action"], "subscribe") + assert plan["action"] == "subscribe" def test_not_open_pre_exchange_item_pre_exchanges(self): plan = DouyuActivityClient.resolve_exchange_plan( self.detail(openStatus=0, subscribeType=2) ) - self.assertEqual(plan["action"], "pre_exchange") + assert plan["action"] == "pre_exchange" def test_not_open_plain_item_blocked(self): plan = DouyuActivityClient.resolve_exchange_plan(self.detail(openStatus=0)) - self.assertEqual(plan["action"], "blocked") - self.assertIn("未开启", plan["text"]) + assert plan["action"] == "blocked" + assert "未开启" in plan["text"] def test_elite_limited_requires_manual(self): plan = DouyuActivityClient.resolve_exchange_plan( self.detail(eliteLimited=1), manual_type=0 ) - self.assertEqual(plan["action"], "blocked") - self.assertIn("精英手册", plan["text"]) + assert plan["action"] == "blocked" + assert "精英手册" in plan["text"] # 已开通手册(manualType=1)不受精英专享限制 plan = DouyuActivityClient.resolve_exchange_plan( self.detail(eliteLimited=1), manual_type=1 ) - self.assertEqual(plan["action"], "classic") + assert plan["action"] == "classic" def test_pre_exchanged_waits_for_open(self): plan = DouyuActivityClient.resolve_exchange_plan( self.detail(subscribeType=2, exchangeInfo={"userStatus": 1}) ) - self.assertEqual(plan["action"], "wait") + assert plan["action"] == "wait" def test_batch_exchange_clamps_num(self): plan = DouyuActivityClient.resolve_exchange_plan( self.detail(batchExchange=1), batch_num_limit="5" ) - self.assertEqual(plan["max_num"], 5) - self.assertTrue(plan["batch_exchange"]) - - -if __name__ == "__main__": - unittest.main() \ No newline at end of file + assert plan["max_num"] == 5 + assert plan["batch_exchange"] diff --git a/tests/test_douyu_elite_lock_exchange.py b/tests/test_douyu_elite_lock_exchange.py index 122b420..5aab18d 100644 --- a/tests/test_douyu_elite_lock_exchange.py +++ b/tests/test_douyu_elite_lock_exchange.py @@ -1,4 +1,4 @@ -import unittest +import pytest from unittest.mock import Mock import requests @@ -6,8 +6,8 @@ import requests from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError -class EliteLockExchangeTests(unittest.TestCase): - def setUp(self): +class TestEliteLockExchange: + def setup_method(self): self.client = DouyuActivityClient("acf_uid=100") self.client.csrf_token = Mock(side_effect=["csrf-create", "csrf-pay"]) @@ -16,33 +16,35 @@ class EliteLockExchangeTests(unittest.TestCase): return {key: value[1] for key, value in files.items()} def test_exchange_locks_then_pays_with_fresh_csrf_tokens(self): - self.client._request_json = Mock(side_effect=[ - { - "error": 0, - "msg": "请求正常", - "data": { - "orderId": "5874", - "commodityName": "套装-兔兔白日梦", - "commodityImage": "https://example.test/item.png", - "score": 1280, - "num": 1, - "expireSeconds": "300", + self.client._request_json = Mock( + side_effect=[ + { + "error": 0, + "msg": "请求正常", + "data": { + "orderId": "5874", + "commodityName": "套装-兔兔白日梦", + "commodityImage": "https://example.test/item.png", + "score": 1280, + "num": 1, + "expireSeconds": "300", + }, }, - }, - { - "error": 0, - "msg": "请求正常", - "data": { - "orderId": "5874", - "exchangeId": "11098933", - "commodityType": 3, - "commodityImage": "https://example.test/item.png", - "exchangeNum": 1, - "sType": 1, - "gType": 2, + { + "error": 0, + "msg": "请求正常", + "data": { + "orderId": "5874", + "exchangeId": "11098933", + "commodityType": 3, + "commodityImage": "https://example.test/item.png", + "exchangeNum": 1, + "sType": 1, + "gType": 2, + }, }, - }, - ]) + ] + ) result = self.client.exchange_goods( manual_id="G4KA4Qnz4LDp7", @@ -50,34 +52,38 @@ class EliteLockExchangeTests(unittest.TestCase): commodity_id="20260120QYOOB_bag10", ) - self.assertEqual(result["order_id"], "5874") - self.assertEqual(result["exchange_id"], "11098933") - self.assertEqual(self.client.csrf_token.call_count, 2) + assert result["order_id"] == "5874" + assert result["exchange_id"] == "11098933" + assert self.client.csrf_token.call_count == 2 create_call, pay_call = self.client._request_json.call_args_list - self.assertEqual(create_call.args[:3], ( - "post", self.client.CREATE_EXCHANGE_ORDER_API, "锁定兑换商品", - )) - self.assertEqual(self.multipart_values(create_call.kwargs["files"]), { + assert create_call.args[:3] == ( + "post", + self.client.CREATE_EXCHANGE_ORDER_API, + "锁定兑换商品", + ) + assert self.multipart_values(create_call.kwargs["files"]) == { "manualID": "G4KA4Qnz4LDp7", "rid": "9263298", "commodityID": "20260120QYOOB_bag10", "num": "1", "csrfToken": "csrf-create", - }) - self.assertNotIn("Content-Type", create_call.kwargs["headers"]) - self.assertEqual(pay_call.args[:3], ( - "post", self.client.PAY_EXCHANGE_ORDER_API, "支付锁单", - )) - self.assertEqual(self.multipart_values(pay_call.kwargs["files"]), { + } + assert "Content-Type" not in create_call.kwargs["headers"] + assert pay_call.args[:3] == ( + "post", + self.client.PAY_EXCHANGE_ORDER_API, + "支付锁单", + ) + assert self.multipart_values(pay_call.kwargs["files"]) == { "orderID": "5874", "manualID": "G4KA4Qnz4LDp7", "csrfToken": "csrf-pay", - }) + } def test_lock_requires_order_id_in_success_response(self): self.client._request_json = Mock(return_value={"error": 0, "data": {}}) - with self.assertRaisesRegex(DouyuActivityError, "没有 orderId"): + with pytest.raises(DouyuActivityError, match="没有 orderId"): self.client.create_exchange_order( manual_id="manual", rid="9263298", @@ -85,32 +91,42 @@ class EliteLockExchangeTests(unittest.TestCase): ) def test_locked_order_list_uses_captured_query_shape(self): - self.client._request_json = Mock(return_value={ - "error": 0, - "data": {"total": 1, "list": [{"orderId": "5874"}]}, - }) + self.client._request_json = Mock( + return_value={ + "error": 0, + "data": {"total": 1, "list": [{"orderId": "5874"}]}, + } + ) result = self.client.exchange_orders(manual_id="manual", page=2, page_size=20) - self.assertEqual(result["total"], 1) - self.assertEqual(result["orders"], [{"orderId": "5874"}]) + assert result["total"] == 1 + assert result["orders"] == [{"orderId": "5874"}] call = self.client._request_json.call_args - self.assertEqual(call.args[:3], ( - "get", self.client.EXCHANGE_ORDER_LIST_API, "查询锁单", - )) - self.assertEqual(call.kwargs["params"], { - "page": 2, "pageSize": 20, "manualID": "manual", - }) + assert call.args[:3] == ( + "get", + self.client.EXCHANGE_ORDER_LIST_API, + "查询锁单", + ) + assert call.kwargs["params"] == { + "page": 2, + "pageSize": 20, + "manualID": "manual", + } def test_redirect_is_not_followed_and_is_reported_without_query_values(self): response = requests.Response() response.status_code = 302 response.url = "https://www.douyu.com/japi/example?request_secret=hidden" - response.headers["Location"] = "https://www.douyu.com/member/login?ticket=secret" + response.headers["Location"] = ( + "https://www.douyu.com/member/login?ticket=secret" + ) self.client.session.request = Mock(return_value=response) self.client.logger = Mock() - with self.assertRaisesRegex(DouyuActivityError, "Cookie 已失效，请重新登录") as caught: + with pytest.raises( + DouyuActivityError, match="Cookie 已失效，请重新登录" + ) as caught: self.client._request( "get", "https://www.douyu.com/japi/example?request_secret=hidden", @@ -118,15 +134,11 @@ class EliteLockExchangeTests(unittest.TestCase): ) self.client.session.request.assert_called_once() - self.assertFalse(self.client.session.request.call_args.kwargs["allow_redirects"]) - message = str(caught.exception) - self.assertIn("method=GET", message) - self.assertIn("url=https://www.douyu.com/japi/example", message) - self.assertIn("location=https://www.douyu.com/member/login", message) - self.assertNotIn("request_secret", message) - self.assertNotIn("ticket=secret", message) - self.assertIn("已拦截重定向", self.client.logger.call_args.args[0]) - - -if __name__ == "__main__": - unittest.main() + assert not self.client.session.request.call_args.kwargs["allow_redirects"] + message = str(caught.value) + assert "method=GET" in message + assert "url=https://www.douyu.com/japi/example" in message + assert "location=https://www.douyu.com/member/login" in message + assert "request_secret" not in message + assert "ticket=secret" not in message + assert "已拦截重定向" in self.client.logger.call_args.args[0] diff --git a/tests/test_douyu_gold_recharge_channel.py b/tests/test_douyu_gold_recharge_channel.py index d06b87d..66d03d4 100644 --- a/tests/test_douyu_gold_recharge_channel.py +++ b/tests/test_douyu_gold_recharge_channel.py @@ -2,8 +2,7 @@ import asyncio import os -import unittest -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import ANY, AsyncMock, Mock, patch from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @@ -17,8 +16,8 @@ from web.backend.services.audit_service import record_audit from core.douyu import FishFinRechargeClient, FishFinRechargeConfig -class DouyuGoldRechargeChannelTests(unittest.TestCase): - def setUp(self): +class TestDouyuGoldRechargeChannel: + def setup_method(self): self.engine = create_engine("sqlite://") Base.metadata.create_all(self.engine) self.session = sessionmaker(bind=self.engine)() @@ -36,8 +35,12 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): self.session.add(self.account) self.session.commit() self.task = DouyuTask( - batch_id="batch", account_id=self.account.id, task_type="create_gold_qr", - handbook_scope="elite", status="running", created_by=user.id, + batch_id="batch", + account_id=self.account.id, + task_type="create_gold_qr", + handbook_scope="elite", + status="running", + created_by=user.id, ) self.session.add(self.task) self.session.commit() @@ -46,13 +49,15 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): os.environ["FISH_FIN_RECHARGE_APP_ID"] = "test-app" os.environ["FISH_FIN_RECHARGE_APP_SECRET"] = "test-secret" - def tearDown(self): + def teardown_method(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() @patch("web.backend.services.douyu_runner_gold.FishFinRechargeClient") - def test_supplier_channel_creates_order_using_nickname_and_finishes_on_success(self, client_class): + def test_supplier_channel_creates_order_using_nickname_and_finishes_on_success( + self, client_class + ): supplier = Mock() supplier.create_order.return_value = { "code": 200, @@ -69,32 +74,36 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): self.task.result = {"payload": {"amount": 10}} self.session.commit() self.runner._execute_create_gold_qr( - self.session, self.task, self.account, "acf_uid=10001", config, + self.session, + self.task, + self.account, + "acf_uid=10001", + config, ) expected_order_id = self.runner._supplier_out_order_id(self.task) supplier.create_order.assert_called_once_with( buy_num=10, - pay_amount=unittest.mock.ANY, + pay_amount=ANY, out_order_id=expected_order_id, product_id="gold-product", recharge_arg=[{"templateName": "斗鱼账号", "templateVal": "罗炅729"}], order_type=0, - notify_url=unittest.mock.ANY, + notify_url=ANY, ) - self.assertEqual(str(supplier.create_order.call_args.kwargs["pay_amount"]), "10") + assert str(supplier.create_order.call_args.kwargs["pay_amount"]) == "10" self.session.refresh(self.task) - self.assertEqual(self.task.status, "success") - self.assertEqual(self.task.result["recharge_channel"], "supplier_api") - self.assertEqual(self.task.result["supplier_order_status"], 2) - self.assertEqual(self.task.result["buy_num"], 10) - self.assertEqual(self.task.result["out_order_id"], expected_order_id) - self.assertEqual(self.task.result["recharge_account"], "罗炅729") - self.assertEqual(self.task.result["douyu_uid"], "10001") - self.assertEqual(self.task.result["order_id"], "supplier-001") - self.assertEqual(self.task.result["pay_amount"], "10") - self.assertNotIn("pay_url", self.task.result) - self.assertNotIn("sign", self.task.result["supplier_order"]) + assert self.task.status == "success" + assert self.task.result["recharge_channel"] == "supplier_api" + assert self.task.result["supplier_order_status"] == 2 + assert self.task.result["buy_num"] == 10 + assert self.task.result["out_order_id"] == expected_order_id + assert self.task.result["recharge_account"] == "罗炅729" + assert self.task.result["douyu_uid"] == "10001" + assert self.task.result["order_id"] == "supplier-001" + assert self.task.result["pay_amount"] == "10" + assert "pay_url" not in self.task.result + assert "sign" not in self.task.result["supplier_order"] @patch("web.backend.services.douyu_runner_gold.FishFinRechargeClient") def test_supplier_channel_rejects_account_without_nickname(self, client_class): @@ -107,20 +116,21 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): } self.runner._execute_create_gold_qr( - self.session, self.task, self.account, "acf_uid=10001", config, + self.session, + self.task, + self.account, + "acf_uid=10001", + config, ) client_class.assert_not_called() self.session.refresh(self.task) - self.assertEqual(self.task.status, "failed") - self.assertIn("斗鱼昵称", self.task.message) + assert self.task.status == "failed" + assert "斗鱼昵称" in self.task.message def test_supplier_out_order_id_reuses_persisted_value(self): self.task.supplier_out_order_id = "DYGFHISTORICAL-001" - self.assertEqual( - self.runner._supplier_out_order_id(self.task), - "DYGFHISTORICAL-001", - ) + assert self.runner._supplier_out_order_id(self.task) == "DYGFHISTORICAL-001" def test_recharge_channel_can_be_read_without_full_config(self): config = ensure_douyu_config(self.session) @@ -132,7 +142,7 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): current=self.session.get(User, self.task.created_by), ) - self.assertEqual(channel, {"gold_recharge_channel": "supplier_api"}) + assert channel == {"gold_recharge_channel": "supplier_api"} def test_recharge_audit_detail_includes_safe_account_identity(self): record_audit( @@ -140,22 +150,26 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): self.session.get(User, self.task.created_by), action="recharge:douyu:create", detail={ - "recharge_accounts": [{ - "username": self.account.username, - "douyu_uid": self.account.uid, - "douyu_nickname": self.account.nickname, - }], + "recharge_accounts": [ + { + "username": self.account.username, + "douyu_uid": self.account.uid, + "douyu_nickname": self.account.nickname, + } + ], }, ) self.session.commit() audit = self.session.query(AuditLog).one() - self.assertIn("douyu_uid", audit.detail) - self.assertIn("10001", audit.detail) - self.assertNotIn("password", audit.detail) + assert "douyu_uid" in audit.detail + assert "10001" in audit.detail + assert "password" not in audit.detail @patch("web.backend.services.douyu_runner_gold.FishFinRechargeClient") - def test_supplier_channel_marks_non_200_create_response_as_failed(self, client_class): + def test_supplier_channel_marks_non_200_create_response_as_failed( + self, client_class + ): supplier = Mock() supplier.create_order.return_value = {"code": 400, "msg": "商品已下架"} client_class.return_value = supplier @@ -166,16 +180,22 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): } self.runner._execute_create_gold_qr( - self.session, self.task, self.account, "acf_uid=10001", config, + self.session, + self.task, + self.account, + "acf_uid=10001", + config, ) self.session.refresh(self.task) - self.assertEqual(self.task.status, "failed") - self.assertEqual(self.task.message, "商品已下架") + assert self.task.status == "failed" + assert self.task.message == "商品已下架" @patch("web.backend.services.douyu_runner.check_douyu_cookie") @patch("web.backend.services.douyu_runner.SessionLocal") - def test_invalid_cookie_skips_activity_and_marks_relogin_required(self, session_local, check_cookie): + def test_invalid_cookie_skips_activity_and_marks_relogin_required( + self, session_local, check_cookie + ): login_task = LoginTask( batch_id="login-batch", account_id=self.account.id, @@ -206,20 +226,30 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): try: task = verify_session.get(DouyuTask, task_id) checked_login = verify_session.get(LoginTask, login_task_id) - self.assertEqual(task.status, "failed") - self.assertIn("Cookie 已失效，请重新登录", task.message) - self.assertEqual(checked_login.ck_check_status, "invalid") - self.assertEqual(checked_login.ck_check_result["message"], "鱼丸接口: 未登录；等级接口: 未登录") + assert task.status == "failed" + assert "Cookie 已失效，请重新登录" in task.message + assert checked_login.ck_check_status == "invalid" + assert ( + checked_login.ck_check_result["message"] + == "鱼丸接口: 未登录；等级接口: 未登录" + ) finally: verify_session.close() def test_supplier_callback_verifies_signature_and_updates_terminal_task(self): self.task.supplier_out_order_id = f"DYGFBATCHT{self.task.id}" - self.task.result = {"recharge_channel": "supplier_api", "out_order_id": self.task.supplier_out_order_id} + self.task.result = { + "recharge_channel": "supplier_api", + "out_order_id": self.task.supplier_out_order_id, + } self.session.commit() - client = FishFinRechargeClient(FishFinRechargeConfig( - base_url="https://supplier.example", app_id="test-app", app_secret="test-secret", - )) + client = FishFinRechargeClient( + FishFinRechargeConfig( + base_url="https://supplier.example", + app_id="test-app", + app_secret="test-secret", + ) + ) payload = { "app_id": "test-app", "timestamp": 1786629000, @@ -235,25 +265,25 @@ class DouyuGoldRechargeChannelTests(unittest.TestCase): response = asyncio.run(supplier_recharge_callback(request, self.session)) - self.assertEqual(response["code"], 200) - self.assertEqual(response["message"], "success") - self.assertTrue(client.verify_response_sign(response, "POST")) + assert response["code"] == 200 + assert response["message"] == "success" + assert client.verify_response_sign(response, "POST") self.session.refresh(self.task) - self.assertEqual(self.task.status, "success") - self.assertEqual(self.task.message, "供应商直充成功（异步通知）") + assert self.task.status == "success" + assert self.task.message == "供应商直充成功（异步通知）" self.session.refresh(self.account) - self.assertEqual(self.account.bind_status, "gold_recharged") - self.assertEqual(self.task.result["order_id"], "supplier-001") - self.assertEqual(self.task.result["supplier_order_status"], 2) - audit = self.session.query(AuditLog).filter(AuditLog.action == "recharge:douyu:callback").one() - self.assertEqual(audit.username, "supplier") - self.assertTrue(audit.success) - self.assertNotIn("sign", audit.detail) + assert self.account.bind_status == "gold_recharged" + assert self.task.result["order_id"] == "supplier-001" + assert self.task.result["supplier_order_status"] == 2 + audit = ( + self.session.query(AuditLog) + .filter(AuditLog.action == "recharge:douyu:callback") + .one() + ) + assert audit.username == "supplier" + assert audit.success + assert "sign" not in audit.detail # 同一终态回调可重复投递，不会改写已结束任务。 asyncio.run(supplier_recharge_callback(request, self.session)) self.session.refresh(self.task) - self.assertEqual(self.task.status, "success") - - -if __name__ == "__main__": - unittest.main() + assert self.task.status == "success" diff --git a/tests/test_douyu_mobile_bind_skip.py b/tests/test_douyu_mobile_bind_skip.py index 02681dd..5fc64b3 100644 --- a/tests/test_douyu_mobile_bind_skip.py +++ b/tests/test_douyu_mobile_bind_skip.py @@ -1,4 +1,3 @@ -import unittest from types import SimpleNamespace from unittest.mock import Mock @@ -6,61 +5,68 @@ from core.douyu.login import DouyuLogin from core.douyu.login_api_wgapi import WgapiLoginAPI -class MobileBindSkipTests(unittest.TestCase): - def setUp(self): +class TestMobileBindSkip: + def setup_method(self): self.login = DouyuLogin.__new__(DouyuLogin) self.login.api = WgapiLoginAPI() - self.login.account = SimpleNamespace(username='test-user', password='test-password') + self.login.account = SimpleNamespace( + username="test-user", password="test-password" + ) self.login._request_json = Mock() def test_second_login_skips_mobile_bind_with_unique_key(self): self.login._request_json.side_effect = [ { - 'error': 130014, - 'msg': '需要进行手机号绑定', - 'data': {'mobileBind': {'uniqueKey': 'one-time-key'}}, + "error": 130014, + "msg": "需要进行手机号绑定", + "data": {"mobileBind": {"uniqueKey": "one-time-key"}}, }, { - 'error': 0, - 'data': {'url': '//www.douyu.com/api/passport/login?code=callback-code'}, + "error": 0, + "data": { + "url": "//www.douyu.com/api/passport/login?code=callback-code" + }, }, ] next_step, login_url = self.login._second_login( - 'gt', 'challenge', 'validate', 'validate|jordan', 'code-token', + "gt", + "challenge", + "validate", + "validate|jordan", + "code-token", ) - self.assertEqual(next_step, 'mobile_bind_skip') - self.assertEqual( - login_url, - 'https://www.douyu.com/api/passport/login?code=callback-code', + assert next_step == "mobile_bind_skip" + assert ( + login_url == "https://www.douyu.com/api/passport/login?code=callback-code" ) - self.assertEqual(self.login._request_json.call_count, 2) + assert self.login._request_json.call_count == 2 skip_call = self.login._request_json.call_args_list[1] - self.assertEqual(skip_call.args[:3], ( - 'post', - 'https://passport.douyu.com/wgapi/member/passport/login', - '跳过手机号绑定接口', - )) - self.assertEqual(skip_call.kwargs['data'], { - 'type': '3', - 'uniqueKey': 'one-time-key', - 'biz_type': '1', - }) + assert skip_call.args[:3] == ( + "post", + "https://passport.douyu.com/wgapi/member/passport/login", + "跳过手机号绑定接口", + ) + assert skip_call.kwargs["data"] == { + "type": "3", + "uniqueKey": "one-time-key", + "biz_type": "1", + } def test_second_login_preserves_remote_email_flow_without_unique_key(self): self.login._request_json.return_value = { - 'error': 130014, - 'data': {'remoteLogin': {'code': 'remote-code'}}, + "error": 130014, + "data": {"remoteLogin": {"code": "remote-code"}}, } next_step, remote_code = self.login._second_login( - 'gt', 'challenge', 'validate', 'validate|jordan', 'code-token', + "gt", + "challenge", + "validate", + "validate|jordan", + "code-token", ) - self.assertEqual((next_step, remote_code), ('remote_email', 'remote-code')) - self.assertEqual(self.login._request_json.call_count, 1) - - -if __name__ == '__main__': - unittest.main() + assert (next_step, remote_code) == ("remote_email", "remote-code") + assert self.login._request_json.call_count == 1 diff --git a/tests/test_douyu_runner_proxy.py b/tests/test_douyu_runner_proxy.py index bb1a1ea..f61a793 100644 --- a/tests/test_douyu_runner_proxy.py +++ b/tests/test_douyu_runner_proxy.py @@ -1,11 +1,13 @@ """斗鱼任务代理接入三模式行为测试（静态 / API / 关闭）+ 写读分离。""" -import unittest from unittest.mock import Mock from core.douyu.activity_client import DouyuActivityClient from web.backend.services.douyu_runner import DouyuBatchRunner -from web.backend.services.douyu_runner_core import DouyuBatchRunnerCore, DOUYU_PROXY_TASK_TYPES +from web.backend.services.douyu_runner_core import ( + DouyuBatchRunnerCore, + DOUYU_PROXY_TASK_TYPES, +) from web.backend.models import ProxyConfig @@ -36,90 +38,104 @@ def _runner(db_cfg: ProxyConfig, task_type: str = "query_points"): return runner -class DouyuProxyModeTests(unittest.TestCase): +class TestDouyuProxyMode: def test_disabled_mode_goes_direct(self): runner = _runner(_cfg(enabled=False), task_type="exchange_goods") - self.assertIsNone(runner._proxies_for_task()) - self.assertIsNone(runner._proxy_fetcher) + assert runner._proxies_for_task() is None + assert runner._proxy_fetcher is None client = runner._client("acf_uid=100") - self.assertEqual(client.session.proxies, {}) + assert client.session.proxies == {} # 未开启代理时 execute 客户端不带 proxies 参数 - self.assertIsInstance(client, DouyuActivityClient) + assert isinstance(client, DouyuActivityClient) def test_static_mode_uses_configured_proxies_for_write_tasks(self): - runner = _runner(_cfg(enabled=True, http="http://10.0.0.1:8080"), task_type="exchange_goods") + runner = _runner( + _cfg(enabled=True, http="http://10.0.0.1:8080"), task_type="exchange_goods" + ) proxies = runner._proxies_for_task() - self.assertEqual(proxies, {"http": "http://10.0.0.1:8080", "https": "http://10.0.0.1:8080"}) + assert proxies == { + "http": "http://10.0.0.1:8080", + "https": "http://10.0.0.1:8080", + } client = runner._client("acf_uid=100") - self.assertEqual(client.session.proxies["https"], "http://10.0.0.1:8080") + assert client.session.proxies["https"] == "http://10.0.0.1:8080" # 每次任务都取同一静态代理 - self.assertEqual(runner._proxies_for_task(), proxies) + assert runner._proxies_for_task() == proxies def test_api_mode_fetches_new_proxy_per_write_task(self): cfg = _cfg(enabled=True, api_url="https://proxy-api.example/fetch") runner = _runner(cfg, task_type="exchange_goods") - self.assertIsNotNone(runner._proxy_fetcher) + assert runner._proxy_fetcher is not None fake = Mock() fake.fetch_new_proxy.return_value = "http://10.0.0.2:3128" runner._proxy_fetcher = fake proxies = runner._proxies_for_task() - self.assertEqual(proxies, {"http": "http://10.0.0.2:3128", "https": "http://10.0.0.2:3128"}) + assert proxies == { + "http": "http://10.0.0.2:3128", + "https": "http://10.0.0.2:3128", + } client = runner._client("acf_uid=100") - self.assertEqual(client.session.proxies["http"], "http://10.0.0.2:3128") + assert client.session.proxies["http"] == "http://10.0.0.2:3128" # 每任务取新代理：_client 也会再 fetch 一次（客户端构造即取代理） fake.fetch_new_proxy.return_value = "http://10.0.0.3:3128" - self.assertEqual(runner._proxies_for_task()["http"], "http://10.0.0.3:3128") - self.assertEqual(fake.fetch_new_proxy.call_count, 3) + assert runner._proxies_for_task()["http"] == "http://10.0.0.3:3128" + assert fake.fetch_new_proxy.call_count == 3 def test_api_mode_failure_falls_back_to_direct(self): - runner = _runner(_cfg(enabled=True, api_url="https://proxy-api.example/fetch"), task_type="exchange_goods") + runner = _runner( + _cfg(enabled=True, api_url="https://proxy-api.example/fetch"), + task_type="exchange_goods", + ) runner._proxy_fetcher = Mock() runner._proxy_fetcher.fetch_new_proxy.return_value = None runner._push_log = Mock() - self.assertIsNone(runner._proxies_for_task()) + assert runner._proxies_for_task() is None client = runner._client("acf_uid=100") - self.assertEqual(client.session.proxies, {}) + assert client.session.proxies == {} runner._push_log.assert_called() def test_read_tasks_stay_direct_even_with_proxy_enabled(self): """写读分离：查询/刷新/绑定等读任务即使代理开启也直连，不消耗代理额度。""" - runner = _runner(_cfg(enabled=True, http="http://10.0.0.1:8080"), task_type="query_points") - self.assertIsNotNone(runner._static_proxies) - self.assertIsNone(runner._proxies_for_task()) + runner = _runner( + _cfg(enabled=True, http="http://10.0.0.1:8080"), task_type="query_points" + ) + assert runner._static_proxies is not None + assert runner._proxies_for_task() is None client = runner._client("acf_uid=100") - self.assertEqual(client.session.proxies, {}) + assert client.session.proxies == {} # API 模式同理：读任务不触发 fetch - runner = _runner(_cfg(enabled=True, api_url="https://proxy-api.example/fetch"), task_type="query_exchange_records") + runner = _runner( + _cfg(enabled=True, api_url="https://proxy-api.example/fetch"), + task_type="query_exchange_records", + ) runner._proxy_fetcher = Mock() - runner._proxy_fetcher.fetch_new_proxy.side_effect = AssertionError("读任务不应取代理") - self.assertIsNone(runner._proxies_for_task()) + runner._proxy_fetcher.fetch_new_proxy.side_effect = AssertionError( + "读任务不应取代理" + ) + assert runner._proxies_for_task() is None def test_proxy_task_types_covers_all_write_tasks(self): """白名单集合应覆盖全部会消耗/下单的写操作任务，且不含查询类。""" - self.assertIn("exchange_goods", DOUYU_PROXY_TASK_TYPES) - self.assertIn("lock_goods", DOUYU_PROXY_TASK_TYPES) - self.assertIn("pay_locked_order", DOUYU_PROXY_TASK_TYPES) - self.assertIn("exchange_esports_goods", DOUYU_PROXY_TASK_TYPES) - self.assertIn("exchange_xpd_goods", DOUYU_PROXY_TASK_TYPES) - self.assertIn("create_gold_qr", DOUYU_PROXY_TASK_TYPES) - self.assertIn("donate_elite_gift", DOUYU_PROXY_TASK_TYPES) - self.assertNotIn("query_points", DOUYU_PROXY_TASK_TYPES) - self.assertNotIn("refresh_goods", DOUYU_PROXY_TASK_TYPES) - self.assertNotIn("get_bind_qr", DOUYU_PROXY_TASK_TYPES) + assert "exchange_goods" in DOUYU_PROXY_TASK_TYPES + assert "lock_goods" in DOUYU_PROXY_TASK_TYPES + assert "pay_locked_order" in DOUYU_PROXY_TASK_TYPES + assert "exchange_esports_goods" in DOUYU_PROXY_TASK_TYPES + assert "exchange_xpd_goods" in DOUYU_PROXY_TASK_TYPES + assert "create_gold_qr" in DOUYU_PROXY_TASK_TYPES + assert "donate_elite_gift" in DOUYU_PROXY_TASK_TYPES + assert "query_points" not in DOUYU_PROXY_TASK_TYPES + assert "refresh_goods" not in DOUYU_PROXY_TASK_TYPES + assert "get_bind_qr" not in DOUYU_PROXY_TASK_TYPES def test_activity_client_accepts_proxies_at_session_level(self): client = DouyuActivityClient( "acf_uid=100", proxies={"http": "http://10.0.0.9:7890", "https": "http://10.0.0.9:7890"}, ) - self.assertEqual(client.session.proxies["http"], "http://10.0.0.9:7890") + assert client.session.proxies["http"] == "http://10.0.0.9:7890" # trust_env=False 维持不变：不受环境变量代理影响 - self.assertFalse(client.session.trust_env) - - -if __name__ == "__main__": - unittest.main() \ No newline at end of file + assert not client.session.trust_env diff --git a/tests/test_douyu_workbench_scopes.py b/tests/test_douyu_workbench_scopes.py index 8f8021a..20a0df9 100644 --- a/tests/test_douyu_workbench_scopes.py +++ b/tests/test_douyu_workbench_scopes.py @@ -1,19 +1,30 @@ -import unittest +import pytest from types import SimpleNamespace from sqlalchemy import create_engine, event from sqlalchemy.orm import sessionmaker from web.backend.database import Base -from web.backend.models import Account, DouyuTask, DouyuWorkbench, DouyuWorkbenchAccount, LoginTask, User +from web.backend.models import ( + Account, + DouyuTask, + DouyuWorkbench, + DouyuWorkbenchAccount, + LoginTask, + User, +) from web.backend.routers.accounts import delete_account -from web.backend.routers.douyu import list_tasks, list_workbench_accounts, update_workbench_accounts +from web.backend.routers.douyu import ( + list_tasks, + list_workbench_accounts, + update_workbench_accounts, +) from web.backend.schemas import DouyuWorkbenchAccountsUpdate from web.backend.services.douyu_service import create_douyu_planned_tasks -class DouyuWorkbenchScopeTests(unittest.TestCase): - def setUp(self): +class TestDouyuWorkbenchScope: + def setup_method(self): self.engine = create_engine("sqlite://") @event.listens_for(self.engine, "connect") @@ -25,65 +36,113 @@ class DouyuWorkbenchScopeTests(unittest.TestCase): self.user = User(username="operator", password_hash="hash", role="super_admin") self.session.add(self.user) self.session.commit() - self.account = Account(username="douyu-user", password="password", email="mail@example.com", email_password="mail-password") + self.account = Account( + username="douyu-user", + password="password", + email="mail@example.com", + email_password="mail-password", + ) self.session.add(self.account) self.session.commit() - self.session.add(LoginTask(batch_id="login", account_id=self.account.id, created_by=self.user.id, status="success", cookie="acf_uid=1")) + self.session.add( + LoginTask( + batch_id="login", + account_id=self.account.id, + created_by=self.user.id, + status="success", + cookie="acf_uid=1", + ) + ) self.session.commit() - self.current = SimpleNamespace(id=self.user.id, username=self.user.username, role=self.user.role, custom_permissions=None) + self.current = SimpleNamespace( + id=self.user.id, + username=self.user.username, + role=self.user.role, + custom_permissions=None, + ) - def tearDown(self): + def teardown_method(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() def test_workbench_accounts_are_per_user_and_scope(self): update_workbench_accounts( - DouyuWorkbenchAccountsUpdate(handbook_scope="elite", account_ids=[self.account.id]), + DouyuWorkbenchAccountsUpdate( + handbook_scope="elite", account_ids=[self.account.id] + ), db=self.session, current=self.current, ) elite = list_workbench_accounts("elite", db=self.session, current=self.current) - esports = list_workbench_accounts("esports", db=self.session, current=self.current) - self.assertEqual(elite["account_ids"], [self.account.id]) - self.assertEqual(esports["account_ids"], []) - self.assertTrue(elite["configured"]) - self.assertFalse(esports["configured"]) - self.assertEqual(self.session.query(DouyuWorkbenchAccount).count(), 1) + esports = list_workbench_accounts( + "esports", db=self.session, current=self.current + ) + assert elite["account_ids"] == [self.account.id] + assert esports["account_ids"] == [] + assert elite["configured"] + assert not esports["configured"] + assert self.session.query(DouyuWorkbenchAccount).count() == 1 update_workbench_accounts( DouyuWorkbenchAccountsUpdate(handbook_scope="elite", account_ids=[]), db=self.session, current=self.current, ) - empty_elite = list_workbench_accounts("elite", db=self.session, current=self.current) - self.assertEqual(empty_elite["account_ids"], []) - self.assertTrue(empty_elite["configured"]) - self.assertEqual(self.session.query(DouyuWorkbench).count(), 1) + empty_elite = list_workbench_accounts( + "elite", db=self.session, current=self.current + ) + assert empty_elite["account_ids"] == [] + assert empty_elite["configured"] + assert self.session.query(DouyuWorkbench).count() == 1 def test_task_scope_is_persisted_and_filters_shared_task_types(self): elite_batch, _ = create_douyu_planned_tasks( - self.session, [self.account.id], "create_gold_qr", "elite", self.user.id, + self.session, + [self.account.id], + "create_gold_qr", + "elite", + self.user.id, ) esports_batch, _ = create_douyu_planned_tasks( - self.session, [self.account.id], "create_gold_qr", "esports", self.user.id, + self.session, + [self.account.id], + "create_gold_qr", + "esports", + self.user.id, ) elite = list_tasks( - batch_id=None, handbook_scope="elite", include_detail=False, page=None, page_size=100, - db=self.session, current=self.current, + batch_id=None, + handbook_scope="elite", + include_detail=False, + page=None, + page_size=100, + db=self.session, + current=self.current, ) esports = list_tasks( - batch_id=None, handbook_scope="esports", include_detail=False, page=None, page_size=100, - db=self.session, current=self.current, + batch_id=None, + handbook_scope="esports", + include_detail=False, + page=None, + page_size=100, + db=self.session, + current=self.current, ) - self.assertEqual({item.batch_id for item in elite}, {elite_batch}) - self.assertEqual({item.batch_id for item in esports}, {esports_batch}) - self.assertEqual({task.handbook_scope for task in self.session.query(DouyuTask).all()}, {"elite", "esports"}) + assert {item.batch_id for item in elite} == {elite_batch} + assert {item.batch_id for item in esports} == {esports_batch} + assert { + task.handbook_scope for task in self.session.query(DouyuTask).all() + } == {"elite", "esports"} def test_rejects_task_from_wrong_workbench(self): - with self.assertRaisesRegex(ValueError, "不属于当前工作台"): + with pytest.raises(ValueError, match="不属于当前工作台"): create_douyu_planned_tasks( - self.session, [self.account.id], "get_xpd_bind_qr", "elite", self.user.id, + self.session, + [self.account.id], + "get_xpd_bind_qr", + "elite", + self.user.id, ) def test_elite_workbench_accepts_lock_goods_task(self): @@ -96,10 +155,12 @@ class DouyuWorkbenchScopeTests(unittest.TestCase): {"commodity_id": "goods-1"}, ) - task = self.session.query(DouyuTask).filter(DouyuTask.batch_id == batch_id).one() - self.assertEqual(count, 1) - self.assertEqual(task.task_type, "lock_goods") - self.assertEqual(task.result, {"payload": {"commodity_id": "goods-1"}}) + task = ( + self.session.query(DouyuTask).filter(DouyuTask.batch_id == batch_id).one() + ) + assert count == 1 + assert task.task_type == "lock_goods" + assert task.result == {"payload": {"commodity_id": "goods-1"}} _, pay_count = create_douyu_planned_tasks( self.session, @@ -109,20 +170,18 @@ class DouyuWorkbenchScopeTests(unittest.TestCase): self.user.id, {"order_id": "5874", "commodity_id": "goods-1"}, ) - self.assertEqual(pay_count, 1) + assert pay_count == 1 def test_account_deletion_removes_workbench_membership(self): - self.session.add(DouyuWorkbenchAccount( - user_id=self.user.id, - handbook_scope="elite", - account_id=self.account.id, - )) + self.session.add( + DouyuWorkbenchAccount( + user_id=self.user.id, + handbook_scope="elite", + account_id=self.account.id, + ) + ) self.session.commit() delete_account(self.account.id, db=self.session, current=self.current) - self.assertEqual(self.session.query(DouyuWorkbenchAccount).count(), 0) - - -if __name__ == "__main__": - unittest.main() + assert self.session.query(DouyuWorkbenchAccount).count() == 0 diff --git a/tests/test_douyu_xpd_exchange.py b/tests/test_douyu_xpd_exchange.py index 36073fb..943284c 100644 --- a/tests/test_douyu_xpd_exchange.py +++ b/tests/test_douyu_xpd_exchange.py @@ -1,202 +1,253 @@ +import pytest import json import re -import unittest from types import SimpleNamespace from unittest.mock import Mock from core.douyu.activity_client import DouyuActivityClient, DouyuActivityError -class XpdExchangeTests(unittest.TestCase): - def setUp(self): +class TestXpdExchange: + def setup_method(self): self.client = DouyuActivityClient( - 'acf_uid=100; acf_stk=stk; acf_ltkid=ltkid', + "acf_uid=100; acf_stk=stk; acf_ltkid=ltkid", ) def test_exchange_uses_fresh_embed_authorization_and_point_price(self): - self.client._request = Mock(return_value=SimpleNamespace( - text='var buyInfo={"ret":"0","msg":"ok","serial":"order-1","newBalance":88};', - )) + self.client._request = Mock( + return_value=SimpleNamespace( + text='var buyInfo={"ret":"0","msg":"ok","serial":"order-1","newBalance":88};', + ) + ) result = self.client.xpd_exchange_goods( embed_query={ - 'gameId': 'cjm', 'actId': '46195', 'appId': 'bp_cf', - 'livePlatId': 'douyu', 'code': 'fresh-code', - 'timestamp': '123', 'v': '2.0', 'sig': 'fresh-signature', + "gameId": "cjm", + "actId": "46195", + "appId": "bp_cf", + "livePlatId": "douyu", + "code": "fresh-code", + "timestamp": "123", + "v": "2.0", + "sig": "fresh-signature", }, - act_id='46195', - rid='9263298', - commodity_id='goods-1', + act_id="46195", + rid="9263298", + commodity_id="goods-1", price=1200, - picture='https://cdn.example/item.jpg', + picture="https://cdn.example/item.jpg", pay_type=1, - action_id='46195', + action_id="46195", ) - self.assertEqual(result['serial'], 'order-1') - self.assertEqual(result['new_balance'], 88) + assert result["serial"] == "order-1" + assert result["new_balance"] == 88 call = self.client._request.call_args - self.assertEqual(call.args[:2], ('get', self.client.XPD_LIVE_BUY_API)) - params = call.kwargs['params'] - self.assertEqual(params['code'], 'fresh-code') - self.assertEqual(params['sig'], 'fresh-signature') - self.assertEqual(params['propid'], 'goods-1') - self.assertEqual(params['paytype'], '1') - self.assertEqual(json.loads(params['appext']), { - 'user_price': 1200, - 'sGoodsPic': 'https://cdn.example/item.jpg', - 'paytype': 1, - }) + assert call.args[:2] == ("get", self.client.XPD_LIVE_BUY_API) + params = call.kwargs["params"] + assert params["code"] == "fresh-code" + assert params["sig"] == "fresh-signature" + assert params["propid"] == "goods-1" + assert params["paytype"] == "1" + assert json.loads(params["appext"]) == { + "user_price": 1200, + "sGoodsPic": "https://cdn.example/item.jpg", + "paytype": 1, + } def test_role_and_balance_preserve_ios_platform_zero(self): - self.client._request = Mock(return_value=SimpleNamespace( - text=( - 'var info={"gameOpenId":"openid-ios","roleId":"role-ios",' - '"roleName":"iOS角色","type":"wx","platId":0,"area":1};' - ), - )) - role = self.client.xpd_get_role(embed_query={}, act_id='46195', rid='9263298') - self.assertEqual(role['plat_id'], '0') - self.assertEqual(role['area'], '1') - - self.client._request = Mock(return_value=SimpleNamespace( - text='var banlanceInfo={"ret":0,"balance":2199};', - )) - result = self.client.xpd_balance( - embed_query={}, act_id='46195', openid='openid-ios', - roleid='role-ios', plat='0', areaid='1', + self.client._request = Mock( + return_value=SimpleNamespace( + text=( + 'var info={"gameOpenId":"openid-ios","roleId":"role-ios",' + '"roleName":"iOS角色","type":"wx","platId":0,"area":1};' + ), + ) ) - params = self.client._request.call_args.kwargs['params'] - self.assertEqual(params['plat'], '0') - self.assertTrue(re.fullmatch(r'\d{13}', params['_time'])) - self.assertEqual(result['balance'], 2199) + role = self.client.xpd_get_role(embed_query={}, act_id="46195", rid="9263298") + assert role["plat_id"] == "0" + assert role["area"] == "1" + + self.client._request = Mock( + return_value=SimpleNamespace( + text='var banlanceInfo={"ret":0,"balance":2199};', + ) + ) + result = self.client.xpd_balance( + embed_query={}, + act_id="46195", + openid="openid-ios", + roleid="role-ios", + plat="0", + areaid="1", + ) + params = self.client._request.call_args.kwargs["params"] + assert params["plat"] == "0" + assert re.fullmatch(r"\d{13}", params["_time"]) + assert result["balance"] == 2199 def test_exchange_surfaces_daoju_failure_message(self): - self.client._request = Mock(return_value=SimpleNamespace( - text='var buyInfo={"ret":"-1","msg":"点券不足"};', - )) + self.client._request = Mock( + return_value=SimpleNamespace( + text='var buyInfo={"ret":"-1","msg":"点券不足"};', + ) + ) - with self.assertRaisesRegex(DouyuActivityError, '点券不足'): + with pytest.raises(DouyuActivityError, match="点券不足"): self.client.xpd_exchange_goods( - embed_query={'code': 'fresh-code'}, - act_id='46195', - rid='9263298', - commodity_id='goods-1', + embed_query={"code": "fresh-code"}, + act_id="46195", + rid="9263298", + commodity_id="goods-1", price=1200, ) def test_list_goods_requests_and_preserves_inventory(self): - self.client._request = Mock(return_value=SimpleNamespace( - text=( - 'var recommend={"data":{"client_data":{"itemsdetail":[{' - '"iGoodsId":"goods-1","sGoodsName":"测试商品","iPrice":"6800",' - '"iJb2Price":"0","iGoodsLeft":"0"}]}}};' - ), - )) - - result = self.client.xpd_list_goods( - embed_query={}, act_id='46195', openid='openid', roleid='roleid', + self.client._request = Mock( + return_value=SimpleNamespace( + text=( + 'var recommend={"data":{"client_data":{"itemsdetail":[{' + '"iGoodsId":"goods-1","sGoodsName":"测试商品","iPrice":"6800",' + '"iJb2Price":"0","iGoodsLeft":"0"}]}}};' + ), + ) ) - self.assertEqual(result['goods'][0]['price'], 6800) - self.assertEqual(result['goods'][0]['goods_left'], 0) - params = self.client._request.call_args.kwargs['params'] - self.assertEqual(params['excludeFields'], 'iCategoryId_76') - self.assertNotIn('category', params) + result = self.client.xpd_list_goods( + embed_query={}, + act_id="46195", + openid="openid", + roleid="roleid", + ) + + assert result["goods"][0]["price"] == 6800 + assert result["goods"][0]["goods_left"] == 0 + params = self.client._request.call_args.kwargs["params"] + assert params["excludeFields"] == "iCategoryId_76" + assert "category" not in params def test_list_goods_fetches_every_page_and_deduplicates_goods(self): first_page = [ - {'iGoodsId': f'goods-{index}', 'sGoodsName': f'商品 {index}'} + {"iGoodsId": f"goods-{index}", "sGoodsName": f"商品 {index}"} for index in range(10) ] second_page = [ - {'iGoodsId': 'goods-9', 'sGoodsName': '重复商品'}, - {'iGoodsId': 'goods-10', 'sGoodsName': '商品 10'}, + {"iGoodsId": "goods-9", "sGoodsName": "重复商品"}, + {"iGoodsId": "goods-10", "sGoodsName": "商品 10"}, ] - self.client._request = Mock(side_effect=[ - SimpleNamespace(text=f'var recommend={{"errcode":0,"data":{{"client_data":{{"itemsdetail":{json.dumps(first_page)}}}}}}};'), - SimpleNamespace(text=f'var recommend={{"errcode":0,"data":{{"client_data":{{"itemsdetail":{json.dumps(second_page)}}}}}}};'), - ]) - - result = self.client.xpd_list_goods( - embed_query={}, act_id='46195', openid='openid', roleid='roleid', + self.client._request = Mock( + side_effect=[ + SimpleNamespace( + text=f'var recommend={{"errcode":0,"data":{{"client_data":{{"itemsdetail":{json.dumps(first_page)}}}}}}};' + ), + SimpleNamespace( + text=f'var recommend={{"errcode":0,"data":{{"client_data":{{"itemsdetail":{json.dumps(second_page)}}}}}}};' + ), + ] ) - self.assertEqual(len(result['goods']), 11) - self.assertEqual([item['commodity_id'] for item in result['goods']][-1], 'goods-10') - self.assertEqual(self.client._request.call_count, 2) - first_params = self.client._request.call_args_list[0].kwargs['params'] - second_params = self.client._request.call_args_list[1].kwargs['params'] - self.assertEqual((first_params['page_begin'], first_params['page_num']), ('0', '1')) - self.assertEqual((second_params['page_begin'], second_params['page_num']), ('10', '2')) + result = self.client.xpd_list_goods( + embed_query={}, + act_id="46195", + openid="openid", + roleid="roleid", + ) + + assert len(result["goods"]) == 11 + assert [item["commodity_id"] for item in result["goods"]][-1] == "goods-10" + assert self.client._request.call_count == 2 + first_params = self.client._request.call_args_list[0].kwargs["params"] + second_params = self.client._request.call_args_list[1].kwargs["params"] + assert (first_params["page_begin"], first_params["page_num"]) == ("0", "1") + assert (second_params["page_begin"], second_params["page_num"]) == ("10", "2") def test_purchase_records_parses_order_list_and_paginates(self): first_record = { - 'sSerialNum': 'order-1', - 'dtBuyTime': '2026-08-08 01:00:42', - 'dtPayTime': '2026-08-08 01:00:55', - 'iStatus': '3', - 'iPrice': '800', - 'iPayAmount': '800', - 'sRoleName': '测试角色', - 'sGoodsInfo': json.dumps({ - 'list': [{'sGoodsName': '改名卡', 'sGoodsPic': 'https://cdn.example/item.jpg'}], - }), + "sSerialNum": "order-1", + "dtBuyTime": "2026-08-08 01:00:42", + "dtPayTime": "2026-08-08 01:00:55", + "iStatus": "3", + "iPrice": "800", + "iPayAmount": "800", + "sRoleName": "测试角色", + "sGoodsInfo": json.dumps( + { + "list": [ + { + "sGoodsName": "改名卡", + "sGoodsPic": "https://cdn.example/item.jpg", + } + ], + } + ), } - self.client._request = Mock(side_effect=[ - SimpleNamespace(text=f'var info={{"ret":"0","msg":"ok","data":{{"cnt":11,"list":{json.dumps([first_record])}}}}};'), - SimpleNamespace(text='var info={"ret":"0","msg":"ok","data":{"cnt":11,"list":[]}};'), - ]) + self.client._request = Mock( + side_effect=[ + SimpleNamespace( + text=f'var info={{"ret":"0","msg":"ok","data":{{"cnt":11,"list":{json.dumps([first_record])}}}}};' + ), + SimpleNamespace( + text='var info={"ret":"0","msg":"ok","data":{"cnt":11,"list":[]}};' + ), + ] + ) result = self.client.xpd_purchase_records( embed_query={ - 'gameId': 'cjm', 'actId': '18882', 'appId': 'bp_cf', - 'livePlatId': 'douyu', 'code': 'fresh-code', - 'timestamp': '123', 'v': '2.0', 'sig': 'fresh-signature', + "gameId": "cjm", + "actId": "18882", + "appId": "bp_cf", + "livePlatId": "douyu", + "code": "fresh-code", + "timestamp": "123", + "v": "2.0", + "sig": "fresh-signature", }, - act_id='46195', + act_id="46195", page_size=1, ) - self.assertEqual(result['total'], 11) - self.assertEqual(result['records'][0]['goods_name'], '改名卡') - self.assertEqual(result['records'][0]['serial'], 'order-1') - self.assertEqual(self.client._request.call_count, 2) - params = self.client._request.call_args_list[0].kwargs['params'] - self.assertEqual(params['_service'], 'order.list') - self.assertEqual(params['pn'], '1') - self.assertEqual(params['code'], 'fresh-code') - + assert result["total"] == 11 + assert result["records"][0]["goods_name"] == "改名卡" + assert result["records"][0]["serial"] == "order-1" + assert self.client._request.call_count == 2 + params = self.client._request.call_args_list[0].kwargs["params"] + assert params["_service"] == "order.list" + assert params["pn"] == "1" + assert params["code"] == "fresh-code" def test_exchange_keeps_raw_price_from_goods_list(self): - self.client._request = Mock(side_effect=[ - SimpleNamespace( - text='var recommend={"errcode":0,"data":{"client_data":{"itemsdetail":[{' - '"iGoodsId":"goods-1","sGoodsName":"测试商品","iPrice":"6800",' - '"iJb2Price":"0","iGoodsLeft":"-1"}]}}};', - ), - SimpleNamespace(text='var buyInfo={"ret":"0","msg":"ok","serial":"order-1","newBalance":88};'), - ]) - - goods = self.client.xpd_list_goods( - embed_query={}, act_id='46195', openid='openid', roleid='roleid', - )['goods'] - result = self.client.xpd_exchange_goods( - embed_query={'code': 'fresh-code'}, - act_id='46195', - rid='9263298', - commodity_id=goods[0]['commodity_id'], - price=goods[0]['price'], - picture='', - pay_type=1, - action_id='46195', + self.client._request = Mock( + side_effect=[ + SimpleNamespace( + text='var recommend={"errcode":0,"data":{"client_data":{"itemsdetail":[{' + '"iGoodsId":"goods-1","sGoodsName":"测试商品","iPrice":"6800",' + '"iJb2Price":"0","iGoodsLeft":"-1"}]}}};', + ), + SimpleNamespace( + text='var buyInfo={"ret":"0","msg":"ok","serial":"order-1","newBalance":88};' + ), + ] ) - self.assertEqual(goods[0]['price'], 6800) - self.assertEqual(result['serial'], 'order-1') - params = self.client._request.call_args.kwargs['params'] - self.assertEqual(json.loads(params['appext'])['user_price'], 6800) + goods = self.client.xpd_list_goods( + embed_query={}, + act_id="46195", + openid="openid", + roleid="roleid", + )["goods"] + result = self.client.xpd_exchange_goods( + embed_query={"code": "fresh-code"}, + act_id="46195", + rid="9263298", + commodity_id=goods[0]["commodity_id"], + price=goods[0]["price"], + picture="", + pay_type=1, + action_id="46195", + ) - -if __name__ == '__main__': - unittest.main() + assert goods[0]["price"] == 6800 + assert result["serial"] == "order-1" + params = self.client._request.call_args.kwargs["params"] + assert json.loads(params["appext"])["user_price"] == 6800 diff --git a/tests/test_fish_fin_recharge_api.py b/tests/test_fish_fin_recharge_api.py index 68b82c2..659338d 100644 --- a/tests/test_fish_fin_recharge_api.py +++ b/tests/test_fish_fin_recharge_api.py @@ -1,6 +1,7 @@ """鱼翅直充供应商 API 客户端测试。""" -import unittest +import pytest + from unittest.mock import Mock from core.douyu.recharge_api import ( @@ -10,8 +11,8 @@ from core.douyu.recharge_api import ( ) -class FishFinRechargeClientTests(unittest.TestCase): - def setUp(self): +class TestFishFinRechargeClient: + def setup_method(self): self.config = FishFinRechargeConfig( base_url="https://supplier.example/", app_id="15945681", @@ -21,18 +22,21 @@ class FishFinRechargeClientTests(unittest.TestCase): self.client = FishFinRechargeClient(self.config, session=self.session) def test_sign_matches_document_example(self): - sign = self.client.sign({ - "app_id": "15945681 ", - "timestamp": 1740916504, - "charge_account": "18888888888", - "buy_num": 2, - "customer_price": 1.21, - "customer_order_no": "DD202507010010208888", - "product_id": "66668888", - "empty": " \n", - "sign": "ignored", - }, "POST") - self.assertEqual(sign, "4087a959f3488ecb13efe6ef58e3bc67") + sign = self.client.sign( + { + "app_id": "15945681 ", + "timestamp": 1740916504, + "charge_account": "18888888888", + "buy_num": 2, + "customer_price": 1.21, + "customer_order_no": "DD202507010010208888", + "product_id": "66668888", + "empty": " \n", + "sign": "ignored", + }, + "POST", + ) + assert sign == "4087a959f3488ecb13efe6ef58e3bc67" def test_create_order_posts_new_document_payload(self): response = Mock() @@ -48,40 +52,52 @@ class FishFinRechargeClientTests(unittest.TestCase): ext_arg={"skuid": 12}, ) - self.assertEqual(result["code"], 200) + assert result["code"] == 200 kwargs = self.session.post.call_args.kwargs - self.assertEqual(self.session.post.call_args.args[0], "https://supplier.example/adapter-apiaccess/open/api/createOrderV2") - self.assertEqual(kwargs["json"]["pay_amount"], 1.2) - self.assertEqual(kwargs["json"]["out_order_id"], "OUT202503010010208888") - self.assertEqual(kwargs["json"]["order_type"], 0) - self.assertEqual(kwargs["json"]["recharge_arg"], '[{"templateName":"斗鱼账号","templateVal":"10001"}]') - self.assertEqual(kwargs["json"]["ext_arg"], '{"skuid":12}') - self.assertNotIn("charge_account", kwargs["json"]) - self.assertNotIn("customer_price", kwargs["json"]) - self.assertNotIn("customer_order_no", kwargs["json"]) - self.assertEqual(kwargs["json"]["sign"], self.client.sign(kwargs["json"], "POST")) - self.assertEqual(kwargs["timeout"], (8, 20)) + assert ( + self.session.post.call_args.args[0] + == "https://supplier.example/adapter-apiaccess/open/api/createOrderV2" + ) + assert kwargs["json"]["pay_amount"] == 1.2 + assert kwargs["json"]["out_order_id"] == "OUT202503010010208888" + assert kwargs["json"]["order_type"] == 0 + assert ( + kwargs["json"]["recharge_arg"] + == '[{"templateName":"斗鱼账号","templateVal":"10001"}]' + ) + assert kwargs["json"]["ext_arg"] == '{"skuid":12}' + assert "charge_account" not in kwargs["json"] + assert "customer_price" not in kwargs["json"] + assert "customer_order_no" not in kwargs["json"] + assert kwargs["json"]["sign"] == self.client.sign(kwargs["json"], "POST") + assert kwargs["timeout"] == (8, 20) - def test_create_order_omits_empty_optional_fields_and_serializes_integer_price(self): + def test_create_order_omits_empty_optional_fields_and_serializes_integer_price( + self, + ): response = Mock() response.json.return_value = {"code": 200} self.session.post.return_value = response self.client.create_order( - buy_num=1, pay_amount="1.0", - out_order_id="merchant-001", product_id="111570", + buy_num=1, + pay_amount="1.0", + out_order_id="merchant-001", + product_id="111570", recharge_arg=[{"templateName": "斗鱼UID", "templateVal": "10001"}], ) body = self.session.post.call_args.kwargs["json"] - self.assertEqual(body["pay_amount"], 1) - self.assertNotIn("notify_url", body) - self.assertEqual(body["recharge_arg"], '[{"templateName":"斗鱼UID","templateVal":"10001"}]') - self.assertNotIn("ext_arg", body) - self.assertEqual(body["sign"], self.client.sign(body, "POST")) + assert body["pay_amount"] == 1 + assert "notify_url" not in body + assert ( + body["recharge_arg"] == '[{"templateName":"斗鱼UID","templateVal":"10001"}]' + ) + assert "ext_arg" not in body + assert body["sign"] == self.client.sign(body, "POST") def test_query_order_requires_identifier_and_uses_get_params(self): - with self.assertRaisesRegex(ValueError, "out_order_id"): + with pytest.raises(ValueError, match="out_order_id"): self.client.query_order("") response = Mock() @@ -90,13 +106,18 @@ class FishFinRechargeClientTests(unittest.TestCase): self.client.query_order("OUT202503010010208888") kwargs = self.session.get.call_args.kwargs - self.assertEqual(self.session.get.call_args.args[0], "https://supplier.example/adapter-apiaccess/open/api/queryOrderV2") - self.assertEqual(kwargs["params"]["out_order_id"], "OUT202503010010208888") - self.assertEqual(kwargs["params"]["sign"], self.client.sign(kwargs["params"], "GET")) + assert ( + self.session.get.call_args.args[0] + == "https://supplier.example/adapter-apiaccess/open/api/queryOrderV2" + ) + assert kwargs["params"]["out_order_id"] == "OUT202503010010208888" + assert kwargs["params"]["sign"] == self.client.sign(kwargs["params"], "GET") def test_missing_configuration_is_rejected_before_request(self): - with self.assertRaises(FishFinRechargeConfigError): - FishFinRechargeClient(FishFinRechargeConfig(base_url="", app_id="", app_secret="")) + with pytest.raises(FishFinRechargeConfigError): + FishFinRechargeClient( + FishFinRechargeConfig(base_url="", app_id="", app_secret="") + ) def test_trace_excludes_replayable_signature_and_nested_account_data(self): response = Mock() @@ -104,47 +125,62 @@ class FishFinRechargeClientTests(unittest.TestCase): response.json.return_value = {"code": 200, "result": {"order_status": 0}} self.session.post.return_value = response events = [] - client = FishFinRechargeClient(self.config, session=self.session, trace=events.append) + client = FishFinRechargeClient( + self.config, session=self.session, trace=events.append + ) client.create_order( - buy_num=1, pay_amount="1", - out_order_id="merchant-001", product_id="111570", + buy_num=1, + pay_amount="1", + out_order_id="merchant-001", + product_id="111570", recharge_arg=[{"templateName": "斗鱼UID", "templateVal": "10001"}], ) - self.assertEqual([event["stage"] for event in events], ["request", "response"]) - self.assertNotIn("sign", events[0]["params"]) - self.assertNotIn("recharge_arg", events[0]["params"]) - self.assertEqual(events[0]["params"]["pay_amount"], 1) + assert [event["stage"] for event in events] == ["request", "response"] + assert "sign" not in events[0]["params"] + assert "recharge_arg" not in events[0]["params"] + assert events[0]["params"]["pay_amount"] == 1 def test_debug_trace_includes_complete_request_and_response_bodies(self): response = Mock() response.status_code = 200 - response.headers = {"Content-Type": "application/json", "X-Request-Id": "request-1"} - response.text = '{"code":1000,"message":"未传递支付金额","sign":"response-sign"}' - response.json.return_value = {"code": 1000, "message": "未传递支付金额", "sign": "response-sign"} + response.headers = { + "Content-Type": "application/json", + "X-Request-Id": "request-1", + } + response.text = ( + '{"code":1000,"message":"未传递支付金额","sign":"response-sign"}' + ) + response.json.return_value = { + "code": 1000, + "message": "未传递支付金额", + "sign": "response-sign", + } self.session.post.return_value = response events = [] config = FishFinRechargeConfig( - base_url=self.config.base_url, app_id=self.config.app_id, - app_secret=self.config.app_secret, debug=True, + base_url=self.config.base_url, + app_id=self.config.app_id, + app_secret=self.config.app_secret, + debug=True, + ) + client = FishFinRechargeClient( + config, session=self.session, trace=events.append ) - client = FishFinRechargeClient(config, session=self.session, trace=events.append) client.create_order( - buy_num=1, pay_amount="1", - out_order_id="merchant-001", product_id="111570", + buy_num=1, + pay_amount="1", + out_order_id="merchant-001", + product_id="111570", recharge_arg=[{"templateName": "斗鱼UID", "templateVal": "10001"}], ) - self.assertEqual(events[0]["json_body"]["sign"], "") - self.assertEqual(events[0]["json_body"]["pay_amount"], 1) - self.assertNotIn("sign", events[0]["sign_params"]) - self.assertNotIn(self.config.app_secret, str(events[0])) - self.assertIn("sign_source_digest", events[0]) - self.assertEqual(events[1]["response_body"]["sign"], "") - self.assertNotIn(self.config.app_secret, str(events[1])) - - -if __name__ == "__main__": - unittest.main() + assert events[0]["json_body"]["sign"] == "" + assert events[0]["json_body"]["pay_amount"] == 1 + assert "sign" not in events[0]["sign_params"] + assert self.config.app_secret not in str(events[0]) + assert "sign_source_digest" in events[0] + assert events[1]["response_body"]["sign"] == "" + assert self.config.app_secret not in str(events[1]) diff --git a/tests/test_huya_app_login.py b/tests/test_huya_app_login.py index eef817c..4aed4fb 100644 --- a/tests/test_huya_app_login.py +++ b/tests/test_huya_app_login.py @@ -1,9 +1,10 @@ """虎牙 App 密码登录及相关组件测试。""" +import pytest + import base64 import os import struct -import unittest from unittest.mock import patch, MagicMock from sqlalchemy import create_engine @@ -32,16 +33,18 @@ from core.huya.wup_encoder import build_password_login_wup from web.backend.database import Base from web.backend.models import User, HuyaAccount -from web.backend.schemas import HuyaAppPasswordLoginRequest, HuyaPasswordLoginSelectedRequest +from web.backend.schemas import ( + HuyaAppPasswordLoginRequest, + HuyaPasswordLoginSelectedRequest, +) from web.backend.routers.huya import ( app_password_login_account, app_password_login_selected_accounts, ) -class TestHuyaAppLogin(unittest.TestCase): - - def setUp(self): +class TestHuyaAppLogin: + def setup_method(self): self.engine = create_engine("sqlite://") Base.metadata.create_all(self.engine) self.session = sessionmaker(bind=self.engine)() @@ -49,7 +52,7 @@ class TestHuyaAppLogin(unittest.TestCase): self.session.add(self.admin) self.session.commit() - def tearDown(self): + def teardown_method(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() @@ -59,54 +62,54 @@ class TestHuyaAppLogin(unittest.TestCase): plain = b"Hello, Huya App Login Protocol!" enc = udb_encrypt(key, plain) dec = udb_decrypt(key, enc) - self.assertEqual(dec, plain) + assert dec == plain def test_cert_and_nonce_forge(self): uid = 1199666914671 rnd = gen_nonce(uid, K1_DEFAULT) - self.assertEqual(len(rnd), 20) + assert len(rnd) == 20 fp = b"02df398797432eadefcc12767119ad5e80999389" cred = b"\x0a" + os.urandom(113) p1 = build_p1(b"5008", fp, cred, rnd=rnd) - self.assertEqual(len(p1), 187) + assert len(p1) == 187 cert = forge_cert(p1) dec = decrypt_cert(cert) parsed = parse_p1(dec) - self.assertEqual(parsed["rnd"], rnd) - self.assertEqual(parsed["fingerprint"], fp) - self.assertEqual(parsed["cred"], cred) + assert parsed["rnd"] == rnd + assert parsed["fingerprint"] == fp + assert parsed["cred"] == cred def test_envelope_patching(self): env = Envelope.load() - self.assertGreater(env.uid, 0) - self.assertEqual(len(env.cert_b64), 260) + assert env.uid > 0 + assert len(env.cert_b64) == 260 new_uid = 1199666911746 env.patch_uid(new_uid) - self.assertEqual(env.uid, new_uid) + assert env.uid == new_uid fake_cert = bytes([0x0C, 0x20]) + os.urandom(192) env.patch_cert(fake_cert) - self.assertEqual(len(env.cert_b64), 260) + assert len(env.cert_b64) == 260 wup_b64 = env.wup_b64() - self.assertGreater(len(wup_b64), 0) + assert len(wup_b64) > 0 def test_device_profile_generation(self): p1 = generate_profile() - self.assertEqual(p1["os"], "android") - self.assertEqual(len(p1["fingerprint"]), 40) - self.assertEqual(len(p1["device_id"]), 40) - self.assertEqual(p1["hdid"], "ed0db8334cadd236c00cadf7e11ab5a5") + assert p1["os"] == "android" + assert len(p1["fingerprint"]) == 40 + assert len(p1["device_id"]) == 40 + assert p1["hdid"] == "ed0db8334cadd236c00cadf7e11ab5a5" # 画像不再承载 safedeviceid：该令牌由 dfp_register 注册链每次登录前签发 - self.assertNotIn("safedeviceid", p1) - self.assertNotIn("safedeviceid", DEFAULT_GOLDEN_DEV) + assert "safedeviceid" not in p1 + assert "safedeviceid" not in DEFAULT_GOLDEN_DEV p2 = get_profile("test_user_account_123") p3 = get_profile("test_user_account_123") - self.assertEqual(p2["fingerprint"], p3["fingerprint"]) + assert p2["fingerprint"] == p3["fingerprint"] def test_wup_encoder_output(self): dev = generate_profile() @@ -121,18 +124,22 @@ class TestHuyaAppLogin(unittest.TestCase): user_action_json="{}", device_info=dev, ) - self.assertGreater(len(pkt), 500) + assert len(pkt) > 500 total_len = struct.unpack(">I", pkt[:4])[0] - self.assertEqual(total_len, len(pkt)) + assert total_len == len(pkt) # ---- 新设备注册链 (core/huya/dfp_register) 生产接入测试 ---- def test_wup_login_skips_registration_when_safedeviceid_given(self): """显式传入 safedeviceid 时不再触发注册链（风控重试复用同一注册结果）。""" - with patch("core.huya.app_login.register_device") as m_reg, \ - patch("core.huya.app_login.build_password_login_wup", return_value=b"pkt"), \ - patch("core.huya.app_login.requests.post", - return_value=MagicMock(status_code=200, content=b"")): + with ( + patch("core.huya.app_login.register_device") as m_reg, + patch("core.huya.app_login.build_password_login_wup", return_value=b"pkt"), + patch( + "core.huya.app_login.requests.post", + return_value=MagicMock(status_code=200, content=b""), + ), + ): wup_password_login_raw("300023887", "pw", safedeviceid="A" * 180) m_reg.assert_not_called() @@ -146,37 +153,48 @@ class TestHuyaAppLogin(unittest.TestCase): captured["args"] = args return b"pkt" - with patch("core.huya.app_login.register_device", - return_value=("a" * 32, new_action, new_device_id)) as m_reg, \ - patch("core.huya.app_login.build_password_login_wup", side_effect=fake_build), \ - patch("core.huya.app_login.requests.post", - return_value=MagicMock(status_code=200, content=b"")): + with ( + patch( + "core.huya.app_login.register_device", + return_value=("a" * 32, new_action, new_device_id), + ) as m_reg, + patch( + "core.huya.app_login.build_password_login_wup", side_effect=fake_build + ), + patch( + "core.huya.app_login.requests.post", + return_value=MagicMock(status_code=200, content=b""), + ), + ): wup_password_login_raw("300023887", "pw") m_reg.assert_called_once() # args: uid_str, sha1, safedeviceid, hdid, session, traceId, ua, dev - self.assertEqual(captured["args"][2], new_action) - self.assertEqual(captured["args"][7]["device_id"], new_device_id) + assert captured["args"][2] == new_action + assert captured["args"][7]["device_id"] == new_device_id # 画像默认值里的旧 device_id 被注册结果覆盖，而非沿用 - self.assertNotEqual( - captured["args"][7]["device_id"], - DEFAULT_GOLDEN_DEV["device_id"], - ) + assert captured["args"][7]["device_id"] != DEFAULT_GOLDEN_DEV["device_id"] def test_wup_login_registration_failure_is_explicit(self): """注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。""" - with patch("core.huya.app_login.register_device", - side_effect=DfpRegistrationError("注册链超时")) as m_reg, \ - patch("core.huya.app_login.requests.post") as m_post: - with self.assertRaises(HuyaAppLoginError): + with ( + patch( + "core.huya.app_login.register_device", + side_effect=DfpRegistrationError("注册链超时"), + ) as m_reg, + patch("core.huya.app_login.requests.post") as m_post, + ): + with pytest.raises(HuyaAppLoginError): wup_password_login_raw("300023887", "pw") m_reg.assert_called_once() m_post.assert_not_called() def test_login_cred_flow_registration_failure_is_explicit(self): """login_cred_with_flow 注册失败同样包装为 HuyaAppLoginError 显式失败。""" - with patch("core.huya.app_login.register_device", - side_effect=DfpRegistrationError("注册链 HTTP 500")): - with self.assertRaisesRegex(HuyaAppLoginError, "注册失败"): + with patch( + "core.huya.app_login.register_device", + side_effect=DfpRegistrationError("注册链 HTTP 500"), + ): + with pytest.raises(HuyaAppLoginError, match="注册失败"): login_cred_with_flow("300023887", "pw") def test_router_functions(self): @@ -187,26 +205,32 @@ class TestHuyaAppLogin(unittest.TestCase): sdid="mock_sdid_123", ) - with patch("web.backend.routers.huya.login_huya_app_password", return_value=mock_res): + with patch( + "web.backend.routers.huya.login_huya_app_password", return_value=mock_res + ): req = HuyaAppPasswordLoginRequest( username="mock_test_huya_user", password="mock_password_123", tag="test_tag", ) - resp = app_password_login_account(req=req, db=self.session, current=self.admin) - self.assertTrue(resp["success"]) - self.assertIn("App", resp["message"]) - self.assertEqual(resp["account"].username, "mock_test_huya_user") + resp = app_password_login_account( + req=req, db=self.session, current=self.admin + ) + assert resp["success"] + assert "App" in resp["message"] + assert resp["account"].username == "mock_test_huya_user" # Test batch app login router function - acc = self.session.query(HuyaAccount).filter(HuyaAccount.username == "mock_test_huya_user").first() + acc = ( + self.session.query(HuyaAccount) + .filter(HuyaAccount.username == "mock_test_huya_user") + .first() + ) batch_req = HuyaPasswordLoginSelectedRequest(account_ids=[acc.id]) - batch_resp = app_password_login_selected_accounts(req=batch_req, db=self.session, current=self.admin) - self.assertTrue(batch_resp["success"]) - self.assertEqual(batch_resp["count"], 1) - self.assertEqual(len(batch_resp["results"]), 1) - self.assertTrue(batch_resp["results"][0]["success"]) - - -if __name__ == "__main__": - unittest.main() + batch_resp = app_password_login_selected_accounts( + req=batch_req, db=self.session, current=self.admin + ) + assert batch_resp["success"] + assert batch_resp["count"] == 1 + assert len(batch_resp["results"]) == 1 + assert batch_resp["results"][0]["success"] diff --git a/tests/test_huya_dfp_register.py b/tests/test_huya_dfp_register.py index 88cd92c..5b4c674 100644 --- a/tests/test_huya_dfp_register.py +++ b/tests/test_huya_dfp_register.py @@ -7,9 +7,10 @@ 注册结果三元组返回; * 失败语义: 模板缺失/HTTP 异常 -> DfpRegistrationError (不静默、不发登录帧)。 """ + +import pytest import base64 import json -import unittest from pathlib import Path from unittest.mock import patch @@ -29,9 +30,7 @@ TAIL = bytes.fromhex("3600400c0b8c980ca80c") def _fake_resp() -> bytes: """合成 dfpReport 响应: t1(32hex) + t2(180B base64 action) + t5(40hex)。""" - return (b"\x16\x20" + b"a" * 32 - + b"\x26\xb4" + b"A" * 180 - + b"\x56\x28" + b"b" * 40) + return b"\x16\x20" + b"a" * 32 + b"\x26\xb4" + b"A" * 180 + b"\x56\x28" + b"b" * 40 _GOLDEN_FP = b"02df398797432eadefcc12767119ad5e80999389" @@ -46,28 +45,26 @@ def _dummy_chain() -> dict: } -class TestRandomDfpBody(unittest.TestCase): - +class TestRandomDfpBody: def test_body_shape(self): body = _build_random_dfp_body() - self.assertEqual(len(body), 4226) + assert len(body) == 4226 cw = body[-CW_LEN:] - self.assertEqual(len(cw), CW_LEN) # 4146 - self.assertTrue(cw.endswith(TAIL), "cw 尾部应为固定 10B") + assert len(cw) == CW_LEN # 4146 + assert cw.endswith(TAIL) def test_body_is_randomized(self): b1 = _build_random_dfp_body() b2 = _build_random_dfp_body() - self.assertNotEqual(b1, b2, "每次注册应生成不同的随机 cw") + assert b1 != b2, "每次注册应生成不同的随机 cw" -class TestParseResponse(unittest.TestCase): - +class TestParseResponse: def test_parse_ok(self): t1, t2, t5 = _parse_response(_fake_resp()) - self.assertEqual(t1, "a" * 32) - self.assertEqual(t2, "A" * 180) - self.assertEqual(t5, "b" * 40) + assert t1 == "a" * 32 + assert t2 == "A" * 180 + assert t5 == "b" * 40 def test_parse_golden_evidence(self): """真实抓包模板响应必须可解析 (证据: evidence/dfp_chain_golden.json)。""" @@ -76,77 +73,87 @@ class TestParseResponse(unittest.TestCase): data = json.loads(CHAIN_FILE.read_text(encoding="utf-8")) resp = base64.b64decode(data["dfpReport"]["resp_b64"]) t1, t2, t5 = _parse_response(resp) - self.assertEqual(len(t1), 32) - self.assertEqual(len(t2), 180) - self.assertEqual(len(t5), 40) + assert len(t1) == 32 + assert len(t2) == 180 + assert len(t5) == 40 def test_parse_missing_fields_raises(self): - with self.assertRaises(DfpRegistrationError): + with pytest.raises(DfpRegistrationError): _parse_response(b"\x16\x20" + b"a" * 32) # 缺 t2/t5 - with self.assertRaises(DfpRegistrationError): + with pytest.raises(DfpRegistrationError): _parse_response(b"\x26\xb4" + b"A" * 180) # 缺 t1/t5 - with self.assertRaises(DfpRegistrationError): + with pytest.raises(DfpRegistrationError): _parse_response(b"\x56\x28" + b"b" * 40) # 缺 t1/t2 -class TestRegisterDevice(unittest.TestCase): - +class TestRegisterDevice: def test_flow_three_steps_and_returns_triple(self): calls = [] - def spy_post(body, content_type="application/octet-stream", - timeout=20, proxies=None): + def spy_post( + body, content_type="application/octet-stream", timeout=20, proxies=None + ): calls.append(content_type) return _fake_resp() - with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \ - patch("core.huya.dfp_register._post", side_effect=spy_post): + with ( + patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), + patch("core.huya.dfp_register._post", side_effect=spy_post), + ): t1, t2, t5 = register_device(fingerprint=None) - self.assertEqual((t1, t2, t5), ("a" * 32, "A" * 180, "b" * 40)) + assert (t1, t2, t5) == ("a" * 32, "A" * 180, "b" * 40) # getDfpConfig -> selectOperator -> dfpReport 的 content-type 序列 - self.assertEqual( - calls, - ["application/octet-stream", "application/x-wup", "application/octet-stream"], - ) + assert calls == [ + "application/octet-stream", + "application/x-wup", + "application/octet-stream", + ] def test_select_operator_injects_fingerprint(self): captured = [] - def spy_post(body, content_type="application/octet-stream", - timeout=20, proxies=None): + def spy_post( + body, content_type="application/octet-stream", timeout=20, proxies=None + ): captured.append((body, content_type)) return _fake_resp() new_fp = b"0" * 20 + b"f" * 20 - with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \ - patch("core.huya.dfp_register._post", side_effect=spy_post): + with ( + patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), + patch("core.huya.dfp_register._post", side_effect=spy_post), + ): register_device(fingerprint=new_fp.decode("ascii")) sel_body, sel_ct = captured[1] - self.assertEqual(sel_ct, "application/x-wup") - self.assertIn(new_fp, sel_body, "selectOperator 应注入当前账号画像 fingerprint") - self.assertNotIn(_GOLDEN_FP, sel_body) + assert sel_ct == "application/x-wup" + assert new_fp in sel_body, "selectOperator 应注入当前账号画像 fingerprint" + assert _GOLDEN_FP not in sel_body # dfpReport 请求体为随机 cw 结构 (零设备生成) dfp_body, dfp_ct = captured[2] - self.assertEqual(len(dfp_body), 4226) - self.assertEqual(dfp_ct, "application/octet-stream") + assert len(dfp_body) == 4226 + assert dfp_ct == "application/octet-stream" def test_chain_template_missing_raises(self): """模板缺失 -> 注册链直接报错, 不应发起任何网络请求。""" - with patch("core.huya.dfp_register.CHAIN_FILE", - Path("/tmp/definitely_missing_chain.json")), \ - patch("core.huya.dfp_register._post") as m_post: - with self.assertRaises(DfpRegistrationError): + with ( + patch( + "core.huya.dfp_register.CHAIN_FILE", + Path("/tmp/definitely_missing_chain.json"), + ), + patch("core.huya.dfp_register._post") as m_post, + ): + with pytest.raises(DfpRegistrationError): register_device() m_post.assert_not_called() def test_http_failure_raises(self): """HTTP 层异常由 _post 转换为 DfpRegistrationError 后向上传播 (不静默)。""" - with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \ - patch("core.huya.dfp_register.requests.post", - side_effect=requests.RequestException("connection reset")): - with self.assertRaises(DfpRegistrationError): + with ( + patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), + patch( + "core.huya.dfp_register.requests.post", + side_effect=requests.RequestException("connection reset"), + ), + ): + with pytest.raises(DfpRegistrationError): register_device() - - -if __name__ == "__main__": - unittest.main() \ No newline at end of file diff --git a/tests/test_logger.py b/tests/test_logger.py index 8bcb598..9d41ed3 100644 --- a/tests/test_logger.py +++ b/tests/test_logger.py @@ -3,23 +3,28 @@ import gzip import logging import tempfile -import unittest from datetime import datetime, timedelta from pathlib import Path -from utils.logger import _SensitiveDataFilter, _SizeAndDayRotatingFileHandler, _parse_size +from utils.logger import ( + _SensitiveDataFilter, + _SizeAndDayRotatingFileHandler, + _parse_size, +) -class LoggerTests(unittest.TestCase): +class TestLogger: def test_size_parser_accepts_human_readable_values(self): - self.assertEqual(_parse_size("2K"), 2 * 1024) - self.assertEqual(_parse_size("3MiB"), 3 * 1024 * 1024) - self.assertEqual(_parse_size("invalid", default=123), 123) + assert _parse_size("2K") == 2 * 1024 + assert _parse_size("3MiB") == 3 * 1024 * 1024 + assert _parse_size("invalid", default=123) == 123 def test_file_log_redacts_credentials_and_compresses_rotation(self): with tempfile.TemporaryDirectory() as tmpdir: log_path = Path(tmpdir) / "app-2026-08-28.log" - handler = _SizeAndDayRotatingFileHandler(log_path, max_bytes=1, retention_days=14) + handler = _SizeAndDayRotatingFileHandler( + log_path, max_bytes=1, retention_days=14 + ) handler.setFormatter(logging.Formatter("%(message)s")) handler.addFilter(_SensitiveDataFilter()) test_logger = logging.getLogger("tests.logger.redaction") @@ -34,20 +39,24 @@ class LoggerTests(unittest.TestCase): test_logger.handlers = [] archives = list(Path(tmpdir).glob("app-2026-08-28.log.*.gz")) - self.assertEqual(len(archives), 1) + assert len(archives) == 1 archived_content = gzip.open(archives[0], "rt", encoding="utf-8").read() current_content = log_path.read_text(encoding="utf-8") combined = archived_content + current_content - self.assertNotIn("super-secret", combined) - self.assertNotIn("top-secret-token", combined) - self.assertIn("Cookie: [REDACTED]", combined) - self.assertIn("Authorization: [REDACTED]", combined) + assert "super-secret" not in combined + assert "top-secret-token" not in combined + assert "Cookie: [REDACTED]" in combined + assert "Authorization: [REDACTED]" in combined def test_daily_log_switches_to_a_new_dated_file(self): with tempfile.TemporaryDirectory() as tmpdir: today = datetime.now().date() - old_path = Path(tmpdir) / f"app-{(today - timedelta(days=1)).isoformat()}.log" - handler = _SizeAndDayRotatingFileHandler(old_path, max_bytes=1024, retention_days=14) + old_path = ( + Path(tmpdir) / f"app-{(today - timedelta(days=1)).isoformat()}.log" + ) + handler = _SizeAndDayRotatingFileHandler( + old_path, max_bytes=1024, retention_days=14 + ) handler.setFormatter(logging.Formatter("%(message)s")) handler._active_day = today - timedelta(days=1) test_logger = logging.getLogger("tests.logger.daily") @@ -61,9 +70,5 @@ class LoggerTests(unittest.TestCase): test_logger.handlers = [] current_path = Path(tmpdir) / f"app-{today.isoformat()}.log" - self.assertTrue(current_path.exists()) - self.assertEqual(current_path.read_text(encoding="utf-8").strip(), "today") - - -if __name__ == "__main__": - unittest.main() + assert current_path.exists() + assert current_path.read_text(encoding="utf-8").strip() == "today" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index c31c850..48cee58 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -11,7 +11,6 @@ """ import tempfile -import unittest from pathlib import Path from alembic import command @@ -42,28 +41,24 @@ def _migration_config() -> Config: return config -class MigrationSmokeTests(unittest.TestCase): +class TestMigrationSmoke: def test_chain_is_linear_and_head_matches_latest_file(self): directory = ScriptDirectory.from_config(_migration_config()) heads = directory.get_heads() - self.assertEqual(len(heads), 1, f"迁移链出现多个头版本（存在分叉）: {heads}") - self.assertEqual(heads[0], HEAD_REVISION, f"头版本 {heads[0]} 与最新迁移不一致") + assert len(heads) == 1, f"迁移链出现多个头版本（存在分叉）: {heads}" + assert heads[0] == HEAD_REVISION, f"头版本 {heads[0]} 与最新迁移不一致" version_files = sorted(path.name for path in VERSIONS_DIR.glob("*.py")) revisions = list(directory.walk_revisions()) revision_ids = {revision.revision for revision in revisions} - self.assertEqual( - len(revision_ids), - len(version_files), - "迁移文件数量与已加载修订数量不一致（存在无法加载的脚本）", + assert len(revision_ids) == len(version_files), ( + "迁移文件数量与已加载修订数量不一致（存在无法加载的脚本）" ) for revision in revisions: if revision.down_revision is not None: - self.assertIn( - revision.down_revision, - revision_ids, - f"{revision.revision} 引用了不存在的父版本 {revision.down_revision}", + assert revision.down_revision in revision_ids, ( + f"{revision.revision} 引用了不存在的父版本 {revision.down_revision}" ) def test_all_migration_scripts_compile(self): @@ -86,18 +81,20 @@ class MigrationSmokeTests(unittest.TestCase): inspector = inspect(create_engine(f"sqlite:///{db_path}")) db_tables = set(inspector.get_table_names()) - self.assertTrue(db_tables, "迁移链未创建任何表") + assert db_tables model_tables = Base.metadata.tables missing_tables = set(model_tables) - db_tables - self.assertFalse(missing_tables, f"模型要求但迁移未创建的表: {missing_tables}") + assert not missing_tables all_db_indexes = set() for name, table in model_tables.items(): - db_columns = {column["name"] for column in inspector.get_columns(name)} + db_columns = { + column["name"] for column in inspector.get_columns(name) + } model_columns = set(table.columns.keys()) missing_columns = model_columns - db_columns - self.assertFalse(missing_columns, f"{name} 迁移后缺少列: {missing_columns}") + assert not missing_columns db_indexes = { index["name"] @@ -105,16 +102,16 @@ class MigrationSmokeTests(unittest.TestCase): if not index["name"].startswith("sqlite_autoindex") } all_db_indexes |= db_indexes - model_indexes = {index.name for index in table.indexes if index.name} + model_indexes = { + index.name for index in table.indexes if index.name + } missing_indexes = model_indexes - db_indexes - self.assertFalse( - missing_indexes, f"{name} 迁移后缺少索引: {missing_indexes}" - ) + assert not missing_indexes # 迁移新增的复合查询索引（不入模型，但应存在）。 for index_name in EXTRA_INDEXES: - self.assertIn( - index_name, all_db_indexes, f"迁移链未创建索引 {index_name}" + assert index_name in all_db_indexes, ( + f"迁移链未创建索引 {index_name}" ) finally: - database_module.DATABASE_URL = original_url \ No newline at end of file + database_module.DATABASE_URL = original_url diff --git a/tests/test_user_deletion.py b/tests/test_user_deletion.py index e207d1c..a086d0a 100644 --- a/tests/test_user_deletion.py +++ b/tests/test_user_deletion.py @@ -1,4 +1,4 @@ -import unittest +import pytest from types import SimpleNamespace from sqlalchemy import create_engine, event @@ -10,8 +10,8 @@ from web.backend.routers.users import delete_user, list_users, rename_user from web.backend.schemas import UserRename -class UserDeletionTests(unittest.TestCase): - def setUp(self): +class TestUserDeletion: + def setup_method(self): self.engine = create_engine("sqlite://") @event.listens_for(self.engine, "connect") @@ -24,23 +24,27 @@ class UserDeletionTests(unittest.TestCase): self.user = User(username="support", password_hash="hash", role="support") self.session.add_all([self.admin, self.user]) self.session.commit() - self.session.add(Account( - username="douyu-user", - password="password", - email="mail@example.com", - email_password="mail-password", - assigned_to=self.user.id, - )) + self.session.add( + Account( + username="douyu-user", + password="password", + email="mail@example.com", + email_password="mail-password", + assigned_to=self.user.id, + ) + ) self.session.commit() account = self.session.query(Account).one() - self.session.add(LoginTask( - batch_id="batch-1", - account_id=account.id, - created_by=self.user.id, - )) + self.session.add( + LoginTask( + batch_id="batch-1", + account_id=account.id, + created_by=self.user.id, + ) + ) self.session.commit() - def tearDown(self): + def teardown_method(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() @@ -54,19 +58,19 @@ class UserDeletionTests(unittest.TestCase): deleted = self.session.get(User, self.user.id) account = self.session.query(Account).one() - self.assertTrue(result["success"]) - self.assertEqual(result["released_accounts"], 1) - self.assertFalse(deleted.is_active) - self.assertIsNotNone(deleted.deleted_at) - self.assertEqual(deleted.deleted_username, "support") - self.assertTrue(deleted.username.startswith(f"deleted-{deleted.id}-")) - self.assertIsNone(account.assigned_to) - self.assertEqual(self.session.query(AuditLog).one().target, "support") - self.assertEqual(self.session.query(LoginTask).one().created_by, deleted.id) - self.assertEqual([item.id for item in list_users(db=self.session)], [self.admin.id]) + assert result["success"] + assert result["released_accounts"] == 1 + assert not deleted.is_active + assert deleted.deleted_at is not None + assert deleted.deleted_username == "support" + assert deleted.username.startswith(f"deleted-{deleted.id}-") + assert account.assigned_to is None + assert self.session.query(AuditLog).one().target == "support" + assert self.session.query(LoginTask).one().created_by == deleted.id + assert [item.id for item in list_users(db=self.session)] == [self.admin.id] def test_cannot_delete_current_user(self): - with self.assertRaisesRegex(Exception, "不能删除当前登录用户"): + with pytest.raises(Exception, match="不能删除当前登录用户"): delete_user( self.user.id, db=self.session, @@ -84,12 +88,8 @@ class UserDeletionTests(unittest.TestCase): renamed = self.session.get(User, self.user.id) account = self.session.query(Account).one() task = self.session.query(LoginTask).one() - self.assertEqual(result.username, "renamed-support") - self.assertEqual(renamed.id, self.user.id) - self.assertEqual(account.assigned_to, self.user.id) - self.assertEqual(task.created_by, self.user.id) - self.assertEqual(self.session.query(AuditLog).one().target, "support -> renamed-support") - - -if __name__ == "__main__": - unittest.main() + assert result.username == "renamed-support" + assert renamed.id == self.user.id + assert account.assigned_to == self.user.id + assert task.created_by == self.user.id + assert self.session.query(AuditLog).one().target == "support -> renamed-support" diff --git a/uv.lock b/uv.lock index 84fa56b..9feb71e 100644 --- a/uv.lock +++ b/uv.lock @@ -255,6 +255,7 @@ dependencies = [ [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "ruff" }, ] [package.metadata] @@ -283,7 +284,10 @@ requires-dist = [ ] [package.metadata.requires-dev] -dev = [{ name = "pytest", specifier = ">=9.1.1" }] +dev = [ + { name = "pytest", specifier = ">=9.1.1" }, + { name = "ruff", specifier = ">=0.12.0" }, +] [[package]] name = "ecdsa" @@ -751,6 +755,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/64/8d/0133e4eb4beed9e425d9a98ed6e081a55d195481b7632472be1af08d2f6b/rsa-4.9.1-py3-none-any.whl", hash = "sha256:68635866661c6836b8d39430f97a996acbd61bfa49406748ea243539fe239762", size = 34696, upload-time = "2025-04-16T09:51:17.142Z" }, ] +[[package]] +name = "ruff" +version = "0.16.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f3/85/c8e12473c93018f92d19dd988a294202e1c27426c47ec4de53ffb847b8d8/ruff-0.16.5.tar.gz", hash = "sha256:1b88500f9ffbcab3dedb0082c9f9492e91ec3d618aac1236a3e0189938f7040b", size = 4912003, upload-time = "2026-08-27T16:34:18.258Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c6/b6/77c90a970fe2dae17a723acbd011043ea97c98d7deacccefdc4ba74ec512/ruff-0.16.5-py3-none-linux_armv6l.whl", hash = "sha256:12e5f673e774c35fbb62f288809c7653b73445f8ecec6b6063fd6ea3521aa14b", size = 10011941, upload-time = "2026-08-27T16:33:41.287Z" }, + { url = "https://files.pythonhosted.org/packages/4b/46/6cf67cf6411885a1d6f7f6d801682f155536a85176d10b605e2ceffed8bd/ruff-0.16.5-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:eda58a5802de40e7ed5b32b64e0b32539338cc6fcd2c78f61e3ad6a0d79f51c3", size = 10204049, upload-time = "2026-08-27T16:33:44.056Z" }, + { url = "https://files.pythonhosted.org/packages/46/fd/c8720ca7a090abf0c2fef4abe8a5ef6e5127ed15196d8886ff75a2b370e2/ruff-0.16.5-py3-none-macosx_11_0_arm64.whl", hash = "sha256:c5ae9a7b9a8875131f40f8fe967cc86abf899779efd663cb7ce3d572d01da7eb", size = 9809037, upload-time = "2026-08-27T16:33:46.257Z" }, + { url = "https://files.pythonhosted.org/packages/43/45/a684caacdedaca180f52bacccc40bf0789d2c5a7c75f25324853e9eaedb5/ruff-0.16.5-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7b719b0a1f4d59710d283ab2965f621684a108a9e41da622e3b23f0326cd0025", size = 9964129, upload-time = "2026-08-27T16:33:48.352Z" }, + { url = "https://files.pythonhosted.org/packages/9e/f2/5d2bcdaca6b5b93d1b4dfc166cd2aebf7680143a1b38a28759df13a94d31/ruff-0.16.5-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:2298f2780ed1be0c5cb1361e32ab7b1467f3cce7dabe101d2210a314f2fe42e9", size = 9821518, upload-time = "2026-08-27T16:33:50.57Z" }, + { url = "https://files.pythonhosted.org/packages/aa/ff/011cce29accf9257d5974145b733fc653a37985ed6825413a3987cefbfe0/ruff-0.16.5-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:258f29035a2dd021e7861e631b227a5b3f14e50c1184c9a6a122c5f4576154d7", size = 10534835, upload-time = "2026-08-27T16:33:52.522Z" }, + { url = "https://files.pythonhosted.org/packages/d7/5a/f0cf109bada9bba0e96c90c21c9f9251803f57225c32d293327a03c710d6/ruff-0.16.5-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:b9a4f0432966834019c74d1b7e5c51224305d7713f3d7faf3e7451f1a3be3cde", size = 11252550, upload-time = "2026-08-27T16:33:54.521Z" }, + { url = "https://files.pythonhosted.org/packages/63/4d/1d481aaea2046c6a7ed7c291f9004c669cce3c087b6b376ed5b08271e3fe/ruff-0.16.5-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b5eb3a8c3d0ade9cea42b591fd530368e8798380e30e0a308b85a5cf718f09ea", size = 10777949, upload-time = "2026-08-27T16:33:56.88Z" }, + { url = "https://files.pythonhosted.org/packages/ee/34/ee245ca55f64443233034b3d02b03236b19242004281247c079390b7facd/ruff-0.16.5-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ef0f69e191a13a3c9816f63163c88790cb12cd157bbbb384e9c44745702ab105", size = 10311656, upload-time = "2026-08-27T16:33:59.12Z" }, + { url = "https://files.pythonhosted.org/packages/a7/4d/c33a333e341c0a2b96c715b52d89a606f5a34cd4ac493cd9b8d0187186b8/ruff-0.16.5-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:0eeab41fbea2c42f98dfb9822cdccda9d24ba38d49f6dc945b5c236d48f0ef29", size = 10532125, upload-time = "2026-08-27T16:34:01.166Z" }, + { url = "https://files.pythonhosted.org/packages/30/e1/a64cef78b40192497bb98a27a8aa8f2c98ee9ee15bc97f7712d94ef32937/ruff-0.16.5-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:f0768e9df4300713fff30733c87575f68b6f1d8de41184e505b7fdd9c0c95eaf", size = 10097648, upload-time = "2026-08-27T16:34:03.16Z" }, + { url = "https://files.pythonhosted.org/packages/cc/4e/4cdc9ed3c3e109d2f71e62572a37457298d7bc7501ec3138babb7ed32bbd/ruff-0.16.5-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:95cc70cdc7aa80c338de356279d2adbeb2de0f520b9ecd8aba75b94e95e02f91", size = 9829344, upload-time = "2026-08-27T16:34:05.134Z" }, + { url = "https://files.pythonhosted.org/packages/39/4a/31ed35ce31729955fc583ee0d176d6e784c1290cb0b0a75cb2134c1ab72a/ruff-0.16.5-py3-none-musllinux_1_2_i686.whl", hash = "sha256:d185c8398ded1bfd91c0c2cb258346307571eccc473a8490af8c3977399c384a", size = 10277117, upload-time = "2026-08-27T16:34:07.425Z" }, + { url = "https://files.pythonhosted.org/packages/a8/a0/60356d86687b4b666d593df213f4dc3041750d024cb7bf2cfa81cfd65c2e/ruff-0.16.5-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:fb8e3a3c4c6a784150a7ced53b015f4b253fc2bf97a610886419ead64b4756ef", size = 10711653, upload-time = "2026-08-27T16:34:09.712Z" }, + { url = "https://files.pythonhosted.org/packages/ed/20/656d67f5b25ca9bda4e02b1de25867b2954e1d19e03648060f167ad0f4cc/ruff-0.16.5-py3-none-win32.whl", hash = "sha256:288b0a5f080492fe5635db849f9e2e84aa3cce7b7f0e955997d416c507c76a26", size = 10034250, upload-time = "2026-08-27T16:34:11.8Z" }, + { url = "https://files.pythonhosted.org/packages/5b/42/ee8e68a207b9127fcde6c3d7e197def432f346cb1af159e1fa14ca0d1cdc/ruff-0.16.5-py3-none-win_amd64.whl", hash = "sha256:ddc6385fb2137f616357ca03d6c74f4be987f80fed4008566b754f6032b8546f", size = 10516714, upload-time = "2026-08-27T16:34:13.963Z" }, + { url = "https://files.pythonhosted.org/packages/73/e3/7df5a396e445b9ba49ce9a9437439a4d80042c61c0ade199abf8d16de1ac/ruff-0.16.5-py3-none-win_arm64.whl", hash = "sha256:a64abe90968719b851bb7cedffaa8753fbdbdadab483089682db623f3edc587e", size = 10391564, upload-time = "2026-08-27T16:34:16.064Z" }, +] + [[package]] name = "scipy" version = "1.18.0" diff --git a/web/backend/models.py b/web/backend/models.py index 12dd9a4..cab9fb3 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -2,7 +2,15 @@ from datetime import datetime, timezone from sqlalchemy import ( - Column, Integer, String, Boolean, Text, DateTime, ForeignKey, JSON, UniqueConstraint, + Column, + Integer, + String, + Boolean, + Text, + DateTime, + ForeignKey, + JSON, + UniqueConstraint, ) from sqlalchemy.orm import relationship from .database import Base @@ -23,27 +31,39 @@ def _utcnow(): class User(Base): """系统用户""" + __tablename__ = "users" id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(64), unique=True, nullable=False, index=True) password_hash = Column(String(256), nullable=False) - role = Column(String(32), nullable=False, default="support") # super_admin / operation / support + role = Column( + String(32), nullable=False, default="support" + ) # super_admin / operation / support is_active = Column(Boolean, default=True) remark = Column(String(256), default="") - custom_permissions = Column(JSON, nullable=True, comment="自定义权限列表，null表示使用角色默认权限") + custom_permissions = Column( + JSON, nullable=True, comment="自定义权限列表，null表示使用角色默认权限" + ) created_at = Column(DateTime, default=_utcnow) updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow) deleted_at = Column(DateTime, nullable=True, index=True) deleted_username = Column(String(64), default="") # 客服被分配的账号 - assigned_accounts = relationship("Account", back_populates="assigned_user", foreign_keys="Account.assigned_to") - huya_accounts = relationship("HuyaAccount", back_populates="assigned_user", foreign_keys="HuyaAccount.assigned_to") + assigned_accounts = relationship( + "Account", back_populates="assigned_user", foreign_keys="Account.assigned_to" + ) + huya_accounts = relationship( + "HuyaAccount", + back_populates="assigned_user", + foreign_keys="HuyaAccount.assigned_to", + ) class Account(Base): """斗鱼账号""" + __tablename__ = "accounts" id = Column(Integer, primary_key=True, autoincrement=True) @@ -83,19 +103,24 @@ class Account(Base): created_at = Column(DateTime, default=_utcnow) updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow) - assigned_user = relationship("User", back_populates="assigned_accounts", foreign_keys=[assigned_to]) + assigned_user = relationship( + "User", back_populates="assigned_accounts", foreign_keys=[assigned_to] + ) login_tasks = relationship("LoginTask", back_populates="account") douyu_tasks = relationship("DouyuTask", back_populates="account") class LoginTask(Base): """单次登录任务""" + __tablename__ = "login_tasks" id = Column(Integer, primary_key=True, autoincrement=True) batch_id = Column(String(64), nullable=False, index=True) # 批次ID account_id = Column(Integer, ForeignKey("accounts.id"), nullable=False) - status = Column(String(32), default="pending") # pending / running / success / failed / error + status = Column( + String(32), default="pending" + ) # pending / running / success / failed / error cookie = Column(EncryptedText(), default="") message = Column(String(512), default="") created_by = Column(Integer, ForeignKey("users.id"), nullable=False) @@ -103,7 +128,9 @@ class LoginTask(Base): finished_at = Column(DateTime, nullable=True) # Cookie 有效性检测结果（持久化，刷新不丢失） ck_check_status = Column(String(16), default="") # valid / invalid / 空=未检测 - ck_check_result = Column(JSON, nullable=True) # {fish_ball, nickname, level, message} + ck_check_result = Column( + JSON, nullable=True + ) # {fish_ball, nickname, level, message} ck_checked_at = Column(DateTime, nullable=True) account = relationship("Account", back_populates="login_tasks") @@ -111,6 +138,7 @@ class LoginTask(Base): class DouyuTask(Base): """斗鱼业务任务""" + __tablename__ = "douyu_tasks" id = Column(Integer, primary_key=True, autoincrement=True) @@ -133,6 +161,7 @@ class DouyuTask(Base): class DouyuWorkbenchAccount(Base): """用户在指定斗鱼工作台中启用的账号，跨浏览器同步。""" + __tablename__ = "douyu_workbench_accounts" id = Column(Integer, primary_key=True, autoincrement=True) @@ -142,12 +171,15 @@ class DouyuWorkbenchAccount(Base): created_at = Column(DateTime, default=_utcnow) __table_args__ = ( - UniqueConstraint("user_id", "handbook_scope", "account_id", name="uq_douyu_workbench_account"), + UniqueConstraint( + "user_id", "handbook_scope", "account_id", name="uq_douyu_workbench_account" + ), ) class DouyuWorkbench(Base): """工作台配置哨兵，令空账号集合也能跨浏览器同步。""" + __tablename__ = "douyu_workbenches" id = Column(Integer, primary_key=True, autoincrement=True) @@ -165,6 +197,7 @@ class YybRechargeTask(Base): YYB 登录身份与斗鱼账号无关，因此任务只关联创建者，不复用 Account。 """ + __tablename__ = "yyb_recharge_tasks" id = Column(Integer, primary_key=True, autoincrement=True) @@ -197,6 +230,7 @@ class YybRechargeTask(Base): class DouyuConfig(Base): """斗鱼业务配置""" + __tablename__ = "douyu_config" id = Column(Integer, primary_key=True, autoincrement=True) @@ -228,6 +262,7 @@ class DouyuConfig(Base): class DouyuGoodsSnapshot(Base): """斗鱼兑换商品快照""" + __tablename__ = "douyu_goods_snapshot" id = Column(Integer, primary_key=True, autoincrement=True) @@ -241,6 +276,7 @@ class DouyuGoodsSnapshot(Base): class DouyuXpdGoodsSnapshot(Base): """斗鱼和平小店商品快照""" + __tablename__ = "douyu_xpd_goods_snapshot" id = Column(Integer, primary_key=True, autoincrement=True) @@ -256,6 +292,7 @@ class DouyuXpdGoodsSnapshot(Base): class DouyuEsportsGoodsSnapshot(Base): """斗鱼电竞手册皮肤商城商品快照""" + __tablename__ = "douyu_esports_goods_snapshot" id = Column(Integer, primary_key=True, autoincrement=True) @@ -269,6 +306,7 @@ class DouyuEsportsGoodsSnapshot(Base): class HuyaAccount(Base): """虎牙账号""" + __tablename__ = "huya_accounts" id = Column(Integer, primary_key=True, autoincrement=True) @@ -292,12 +330,15 @@ class HuyaAccount(Base): created_at = Column(DateTime, default=_utcnow) updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow) - assigned_user = relationship("User", back_populates="huya_accounts", foreign_keys=[assigned_to]) + assigned_user = relationship( + "User", back_populates="huya_accounts", foreign_keys=[assigned_to] + ) tasks = relationship("HuyaTask", back_populates="account") class HuyaRegisterBatch(Base): """虎牙自动注册批次（持久化元数据）""" + __tablename__ = "huya_register_batches" id = Column(Integer, primary_key=True, autoincrement=True) @@ -320,15 +361,20 @@ class HuyaRegisterBatch(Base): started_at = Column(DateTime, nullable=True) finished_at = Column(DateTime, nullable=True) - items = relationship("HuyaRegisterItem", back_populates="batch", cascade="all, delete-orphan") + items = relationship( + "HuyaRegisterItem", back_populates="batch", cascade="all, delete-orphan" + ) class HuyaRegisterItem(Base): """虎牙自动注册批次内单条手机号状态""" + __tablename__ = "huya_register_items" id = Column(Integer, primary_key=True, autoincrement=True) - batch_db_id = Column(Integer, ForeignKey("huya_register_batches.id"), nullable=False, index=True) + batch_db_id = Column( + Integer, ForeignKey("huya_register_batches.id"), nullable=False, index=True + ) batch_id = Column(String(64), nullable=False, index=True) line = Column(Integer, nullable=False) phone = Column(String(64), default="", index=True) @@ -354,6 +400,7 @@ class HuyaRegisterItem(Base): class HuyaRegisterSuccessLog(Base): """虎牙自动注册成功流水（只追加，供随时导出）""" + __tablename__ = "huya_register_success_logs" id = Column(Integer, primary_key=True, autoincrement=True) @@ -373,6 +420,7 @@ class HuyaRegisterSuccessLog(Base): class HuyaTask(Base): """虎牙业务任务""" + __tablename__ = "huya_tasks" id = Column(Integer, primary_key=True, autoincrement=True) @@ -391,6 +439,7 @@ class HuyaTask(Base): class HuyaConfig(Base): """虎牙业务配置""" + __tablename__ = "huya_config" id = Column(Integer, primary_key=True, autoincrement=True) @@ -404,6 +453,7 @@ class HuyaConfig(Base): class HuyaGoodsSnapshot(Base): """虎牙兑换商品快照""" + __tablename__ = "huya_goods_snapshot" id = Column(Integer, primary_key=True, autoincrement=True) @@ -417,6 +467,7 @@ class HuyaGoodsSnapshot(Base): class HuyaRechargeGoodsSnapshot(Base): """虎牙充值商品快照""" + __tablename__ = "huya_recharge_goods_snapshot" id = Column(Integer, primary_key=True, autoincrement=True) @@ -436,6 +487,7 @@ class HuyaRechargeGoodsSnapshot(Base): class ProxyConfig(Base): """代理配置（全局单条记录）""" + __tablename__ = "proxy_config" id = Column(Integer, primary_key=True, autoincrement=True) @@ -454,6 +506,7 @@ class ProxyConfig(Base): class AuditLog(Base): """操作审计日志""" + __tablename__ = "audit_logs" id = Column(Integer, primary_key=True, autoincrement=True)