Files
live-hub-py/tests/test_huya_captcha_capture.py
T
2026-09-06 13:02:37 +08:00

210 lines
7.1 KiB
Python

"""虎牙 dx_auth 与验证码捕获落盘功能测试。"""
from __future__ import annotations
import io
import json
from pathlib import Path
from PIL import Image
from core.huya.verification.capture import CaptchaCaptureRecord
from core.huya.verification.solver import (
HuyaQrAuthRequiredError,
HuyaVerificationSolver,
)
def test_dx_auth_strategy_recognized_as_click_not_sms(tmp_path: Path):
"""验证 dx_auth 不再被误杀为短信异常,而是识别为 dx_auth。"""
url = (
"https://aq.huya.com/p/safe_auth/dx_auth.html?"
"param=test_param&mobile=13800000000&sceneType=16&appId=5002"
)
solver = HuyaVerificationSolver(
page_url=url,
account="13800000000",
capture_dir=tmp_path,
enable_capture=True,
)
# 不抛出 HuyaSmsAuthRequiredError
solver._parse_strategy(url)
assert solver.auth_type == "dx_auth"
assert solver.current_capture is not None
assert solver.current_capture.meta["auth_type"] == "dx_auth"
assert solver.current_capture.meta["account"] == "13800000000"
def test_qr_auth_still_raises_qr_error():
"""验证真正的 qr_auth 仍然正确抛出 HuyaQrAuthRequiredError。"""
url = "https://aq.huya.com/p/safe_auth/qr_auth.html?param=test_param&mobile=13800000000"
solver = HuyaVerificationSolver(page_url=url)
try:
solver._parse_strategy(url)
except HuyaQrAuthRequiredError:
pass
else:
raise AssertionError("应该抛出 HuyaQrAuthRequiredError")
def test_click_captcha_capture_saves_files_and_prompts(tmp_path: Path):
"""验证点选验证码捕获时正确落盘图片、提示语与元数据。"""
record = CaptchaCaptureRecord(
auth_type="dx_auth",
account="test_user",
page_url="https://aq.huya.com/p/safe_auth/dx_auth.html?param=123",
scene_type="16",
output_base_dir=tmp_path,
)
img = Image.new("RGB", (100, 100), color="blue")
buf = io.BytesIO()
img.save(buf, format="JPEG")
fake_img_bytes = buf.getvalue()
tip_img = Image.new("RGBA", (80, 40), color="white")
tip_buf = io.BytesIO()
tip_img.save(tip_buf, format="PNG")
fake_tip_bytes = tip_buf.getvalue()
folder = record.capture_click_data(
code="code123456",
dtu_bytes=fake_img_bytes,
hzi_bytes=fake_tip_bytes,
points=[[30, 40], [60, 70]],
raw_data={
"question": "按顺序点击:虎、牙",
"wordSize": 2,
"tuMing": 1,
"description": "安全验证",
},
)
assert folder.exists()
assert (folder / "bg_main.jpg").exists()
assert (folder / "tip_text.png").exists()
assert (folder / "annotated.jpg").exists()
assert (folder / "meta.json").exists()
meta = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
assert meta["auth_type"] == "dx_auth"
assert meta["prompts"]["question"] == "按顺序点击:虎、牙"
assert meta["prompts"]["word_size"] == 2
assert meta["solution"]["points"] == [[30, 40], [60, 70]]
def test_slider_captcha_capture_saves_files_and_prompts(tmp_path: Path):
"""验证滑块验证码捕获时正确落盘背景、拼图与标注图。"""
record = CaptchaCaptureRecord(
auth_type="pt_auth",
account="test_slider_user",
page_url="https://aq.huya.com/p/safe_auth/pt_auth.html?param=123",
scene_type="4",
output_base_dir=tmp_path,
)
img = Image.new("RGB", (260, 180), color="green")
buf = io.BytesIO()
img.save(buf, format="JPEG")
fake_bg_bytes = buf.getvalue()
puzzle = Image.new("RGBA", (50, 50), color="red")
p_buf = io.BytesIO()
puzzle.save(p_buf, format="PNG")
fake_puzzle_bytes = p_buf.getvalue()
folder = record.capture_slider_data(
code="slider_code_888",
dtu_bytes=fake_bg_bytes,
puzzle_bytes=fake_puzzle_bytes,
restored_bg_bytes=fake_bg_bytes,
point_x=120,
raw_data={"location_y": 60, "tip": "拖动滑块完成拼图"},
)
assert folder.exists()
assert (folder / "bg_scrambled.jpg").exists()
assert (folder / "puzzle.png").exists()
assert (folder / "bg_restored.jpg").exists()
assert (folder / "annotated.jpg").exists()
meta = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
assert meta["auth_type"] == "pt_auth"
assert meta["solution"]["distance_x"] == 120
assert meta["prompts"]["location_y"] == 60
def test_submit_result_recorded_in_meta(tmp_path: Path):
"""验证验证码提交结果正确更新进 meta.json。"""
record = CaptchaCaptureRecord(
auth_type="dx_auth",
account="test_user",
output_base_dir=tmp_path,
)
img = Image.new("RGB", (10, 10))
buf = io.BytesIO()
img.save(buf, format="JPEG")
folder = record.capture_click_data(
code="c1", dtu_bytes=buf.getvalue(), hzi_bytes=b""
)
record.record_submit_result({"authId": "auth_token_xyz999", "returnCode": 0})
meta = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
assert meta["submit_result"]["success"] is True
assert meta["submit_result"]["authId"] == "auth_token_xyz999"
def test_sms_auth_capture_saves_meta_and_params(tmp_path: Path):
"""验证遇到短信风控拦截时正确落盘元数据与参数。"""
record = CaptchaCaptureRecord(
auth_type="sms_auth",
account="hy_test_sms",
page_url="https://aq.huya.com/m/safe_auth/sms_auth2.html?param=xyz&mobile=0018888",
scene_type="4",
output_base_dir=tmp_path,
)
folder = record.capture_sms_auth_data(
mobile="0018888",
page_url="https://aq.huya.com/m/safe_auth/sms_auth2.html?param=xyz&mobile=0018888",
scene_type="4",
step=3,
raw_params={"sceneType": "4", "mobile": "0018888", "param": "xyz"},
)
assert folder.exists()
assert (folder / "meta.json").exists()
assert (folder / "params.json").exists()
meta = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
assert meta["auth_type"] == "sms_auth"
assert meta["step"] == 3
assert meta["mobile"] == "0018888"
assert meta["submit_result"]["status"] == "sms_auth_required"
def test_qr_auth_capture_saves_meta_and_params(tmp_path: Path):
"""验证遇到扫码风控拦截时正确落盘元数据与参数。"""
record = CaptchaCaptureRecord(
auth_type="qr_auth",
account="hy_test_qr",
page_url="https://aq.huya.com/m/safe_auth/qr_auth.html?param=abc",
scene_type="4",
output_base_dir=tmp_path,
)
folder = record.capture_qr_auth_data(
mobile="13800138000",
page_url="https://aq.huya.com/m/safe_auth/qr_auth.html?param=abc",
scene_type="4",
step=2,
raw_params={"sceneType": "4", "mobile": "13800138000"},
)
assert folder.exists()
assert (folder / "meta.json").exists()
assert (folder / "params.json").exists()
meta = json.loads((folder / "meta.json").read_text(encoding="utf-8"))
assert meta["auth_type"] == "qr_auth"
assert meta["step"] == 2
assert meta["submit_result"]["status"] == "qr_auth_required"