Files
live-hub-py/tests/test_huya_sms_url.py
yml2213 65a4590291 refactor(time): 时间列统一走 UTCDateTime,并把 MySQL 会话时区对齐成 UTC
问题不是「UTC 还是本地时间」,而是这套口径有两个半吊子的地方:

1. 库里存 naive UTC、读出来也是裸时间,于是每一处比较/输出都得自己补时区。
   项目里因此散了三份 _as_utc 和一堆 replace(tzinfo=UTC),P1a 排下一轮时间
   时还炸过一次 TypeError: can't compare offset-naive and offset-aware。
2. MySQL 会话时区跟随系统 +08:00,而库里存的是 UTC —— SQL 里的 NOW()/
   CURDATE() 比库里的时间快 8 小时,查询会静默返回空(P1a 排查时就遇到过,
   看着像「没有数据」)。

改动:

- 新增 database.UTCDateTime:写时 aware→转 UTC 去 tzinfo、naive 视为 UTC;
  读时一律补 UTC 时区。底层 DDL 仍是 DATETIME,**不需要迁移**。models.py 的
  28 个时间列全部改用它,模型上的时间字段从此不再是裸值;
- docker-compose 给 mysql 加 --default-time-zone=+00:00,让 SQL 时间函数与存储
  口径一致。安全性已核实:全库 0 个 TIMESTAMP 列、0 个 CURRENT_TIMESTAMP
  默认值,改这里不会改变任何已有数据的语义;
- 新增 tests/test_utc_datetime.py:读出必须 aware、能直接和 datetime.now(UTC)
  比较、带 +08:00 的写入要落成 UTC、naive 写入视为 UTC,并有一条元测试扫描
  Base.metadata,防止以后新增时间列忘了用 UTCDateTime。

保留既有的 _as_utc 补丁(对 aware 值是幂等的,属防御性代码);新代码不再需要。
「存 UTC、算 UTC、显示本地」三条线至此各自明确:存储与比较统一 UTC,日志按
容器 TZ 显示本地时间,前端 formatTime 负责转本地。
2026-10-02 20:27:07 +08:00

160 lines
5.3 KiB
Python

"""虎牙四段式账号 (账号----密码----手机号----接码链接) 导入与短信自动求解测试。"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from core.huya.verification.solver import (
HuyaSmsAuthRequiredError,
HuyaVerificationSolver,
)
from web.backend.services.huya_service import (
import_huya_password_accounts,
split_huya_password_line,
)
def test_parse_password_line_with_sms_url():
"""测试四段式账号行解析 (账号----密码----手机号----接码链接)。"""
line = (
"hy_299112483----aa778899----+14504409332----"
"http://194.163.45.144/api/get_sms?key=c16fe33219924363fa362ed69dc6fd4f"
)
parsed = split_huya_password_line(line)
assert parsed is not None
assert parsed.username == "hy_299112483"
assert parsed.password == "aa778899"
assert parsed.phone == "+14504409332"
assert (
parsed.sms_url
== "http://194.163.45.144/api/get_sms?key=c16fe33219924363fa362ed69dc6fd4f"
)
def test_parse_password_line_backward_compatible():
"""测试老格式向后兼容性。"""
# 2段
p2 = split_huya_password_line("user123----pass456")
assert p2 is not None
assert p2.username == "user123"
assert p2.password == "pass456"
assert p2.phone == ""
assert p2.sms_url == ""
# 3段 (带手机号或cookie)
p3 = split_huya_password_line("user123----pass456----13800138000")
assert p3 is not None
assert p3.username == "user123"
assert p3.password == "pass456"
assert p3.cookie == "13800138000"
assert p3.sms_url == ""
def test_import_password_accounts_persists_sms_url():
"""测试导入账号密码服务会正确写入 sms_url 和 game_phone。"""
mock_db = MagicMock()
# 模拟无现有账号
mock_db.query().filter().first.return_value = None
text = (
"hy_299112483----aa778899----+14504409332----"
"http://194.163.45.144/api/get_sms?key=c16fe33219924363fa362ed69dc6fd4f"
)
created, skipped = import_huya_password_accounts(mock_db, text, tag="测试接码")
assert created == 1
assert skipped == 0
added_account = mock_db.add.call_args[0][0]
assert added_account.username == "hy_299112483"
assert added_account.game_phone == "+14504409332"
assert (
added_account.sms_url
== "http://194.163.45.144/api/get_sms?key=c16fe33219924363fa362ed69dc6fd4f"
)
assert added_account.tag == "测试接码"
def test_import_password_accounts_strictly_four_segments():
"""测试账号密码导入默认严格限制仅允许四段格式。"""
mock_db = MagicMock()
mock_db.query().filter().first.return_value = None
# 两段、三段会被跳过;仅四段有效行会被导入
text = (
"user1----pass1\n"
"user2----pass2----13800138000\n"
"user3----pass3----+14504409332----http://194.163.45.144/api/get_sms?key=xxx\n"
"user4----pass4----not_a_url\n"
)
created, skipped = import_huya_password_accounts(mock_db, text, tag="四段测试")
assert created == 1
assert skipped == 3
def test_solver_strategy_with_and_without_sms_url():
"""测试在遇到 sms_auth2 时,有接码链接不抛异常,无接码链接抛出短信异常。"""
url = "https://aq.huya.com/p/safe_auth/sms_auth2.html?param=xyz&mobile=14504409332"
# 无 sms_url,应触发 HuyaSmsAuthRequiredError
solver_no_sms = HuyaVerificationSolver(page_url=url)
with pytest.raises(HuyaSmsAuthRequiredError):
solver_no_sms._parse_strategy(url)
# 有 sms_url,应正常解析为 sms_auth
solver_with_sms = HuyaVerificationSolver(
page_url=url,
sms_url="http://194.163.45.144/api/get_sms?key=test",
)
solver_with_sms._parse_strategy(url)
assert solver_with_sms.auth_type == "sms_auth"
def test_solver_sms_flow():
"""测试 solver._solve_sms_strategy 自动发码、接码与提码闭环。"""
from core.sms_provider import SmsPollResult
url = "https://aq.huya.com/p/safe_auth/sms_auth2.html?param=mock_param&mobile=14504409332&appId=5002"
solver = HuyaVerificationSolver(
page_url=url,
sms_url="http://194.163.45.144/api/get_sms?key=test_key",
account="hy_test",
)
solver._parse_strategy(url)
mock_send_res = {"returnCode": 0, "message": "发送成功"}
mock_verify_res = {
"returnCode": 0,
"data": {"authId": "auth_token_success_123"},
}
with (
patch.object(
solver,
"_request_json",
return_value=mock_send_res,
) as mock_req_json,
patch.object(
solver,
"_submit_verify",
return_value=mock_verify_res["data"],
) as mock_submit,
patch(
"core.sms_provider.SmsProviderClient.poll",
return_value=SmsPollResult(status="code", code="654321"),
) as mock_poll,
):
data = solver._solve_sms_strategy()
assert data["authId"] == "auth_token_success_123"
mock_req_json.assert_called_once()
assert "send-sms-code" in mock_req_json.call_args[0][1]
mock_poll.assert_called_once()
mock_submit.assert_called_once()
verify_url, verify_payload = mock_submit.call_args[0]
assert "verify-sms-code" in verify_url
assert verify_payload["data"]["smsCode"] == b"654321".hex()