Files
live-hub-py/services/yyb-worker/runtime/tests/test_jsdom_pay.py
T

156 lines
7.6 KiB
Python

"""YYB 付款表单的离线回归测试。"""
from __future__ import annotations
import importlib.util
import sys
import unittest
from pathlib import Path
from urllib.parse import parse_qs
_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "jsdom-pay.py"
_SPEC = importlib.util.spec_from_file_location("yyb_jsdom_pay", _SCRIPT)
assert _SPEC and _SPEC.loader
_MODULE = importlib.util.module_from_spec(_SPEC)
sys.modules[_SPEC.name] = _MODULE
_SPEC.loader.exec_module(_MODULE)
from pyvm.payment_errors import describe_payment_failure
from pyvm.algorithm import derive_key1_from_key16, generate_encrypt_msg_offline
from pyvm.protocol import goods_material_diagnostics, validate_goods_materials
class TestJsdomPay(unittest.TestCase):
def test_build_save_body_has_current_payment_context(self):
body = _MODULE.build_save_body(
{"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"},
{"openid": "openid", "accesstoken": "access-token", "logintype": "WX"},
"web-token", "anti-token", "cipher", "2",
"mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap", 600,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["type"], ["bg"])
self.assertEqual(fields["pf"], ["mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap"])
self.assertEqual(fields["zoneid"], ["2"])
self.assertEqual(fields["pay_method"], ["wechat"])
self.assertEqual(fields["wcp"], ["type=CNY&amt=600"])
self.assertEqual(fields["encrypt_way"], ["web_new_encrypt"])
self.assertEqual(fields["webversion"], ["minipayv2"])
def test_qq_payment_uses_qq_oauth_session_fields(self):
body = _MODULE.build_save_body(
{"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"},
{"openid": "openid", "accesstoken": "access-token", "logintype": "QC", "appid": "102033112"},
"web-token", "anti-token", "cipher", "1", "pf", 100,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["session_id"], ["openid"])
self.assertEqual(fields["session_type"], ["kp_accesstoken"])
self.assertEqual(fields["wx_appid"], [""])
self.assertEqual(fields["offerid_for_qq_appid"], ["102033112"])
self.assertNotIn("qq_appid", fields)
def test_wechat_payment_does_not_include_qq_offer_field(self):
body = _MODULE.build_save_body(
{"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"},
{"openid": "openid", "accesstoken": "access-token", "logintype": "WX"},
"web-token", "anti-token", "cipher", "1", "pf", 100,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertNotIn("offerid_for_qq_appid", fields)
def test_qq_page_info_matches_the_web_flow(self):
body = _MODULE.build_page_info_body(
{"token_id": "token", "transaction_id": "transaction", "out_trade_no": "trade"},
{"openid": "openid", "accesstoken": "access-token", "logintype": "QC", "appid": "102033112"},
"anti-token", "1", "pf",
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["isusempaymode"], ["1"])
self.assertEqual(fields["offerid_for_qq_appid"], ["102033112"])
self.assertNotIn("pay_method", fields)
def test_qq_risk_rejection_is_not_reported_as_qr_error(self):
message = describe_payment_failure("QC", "payment", "ret=1099 err_code=1099-1007-0")
self.assertIn("风控/限流", message)
self.assertIn("未生成微信付款码", message)
self.assertIn("停止连续重试", message)
def test_wechat_rejection_is_not_labeled_as_risk_control(self):
message = describe_payment_failure("WX", "payment", "ret=1099 err_code=1099-1007-0")
self.assertIn("服务端拒绝", message)
self.assertNotIn("风控", message)
def test_game_role_error_is_classified_separately(self):
message = describe_payment_failure("QC", "order", "inner=10014 gamerole err")
self.assertIn("角色或区服校验失败", message)
self.assertIn("重新选择角色和区服", message)
def test_qq_encrypt_rand_aligns_plaintext_to_aes_block(self):
params = {
"token_id": "t", "openid": "o", "openkey": "k", "session_id": "openid",
"session_type": "kp_accesstoken", "zoneid": "1", "pay_method": "wechat",
"buy_quantity": "1", "from_h5": "1", "webversion": "minipayv2",
}
rand_value = _MODULE.make_encrypt_rand(params, "tdrc_session%3Dpay-test", "1775990000", True)
plaintext = _MODULE.build_plaintext(params, "tdrc_session%3Dpay-test", "1775990000", rand_value)
self.assertEqual(len(plaintext.encode("latin-1")) % 16, 0)
self.assertGreaterEqual(len(rand_value), 8)
def test_wechat_encrypt_rand_preserves_historical_control_byte(self):
rand_value = _MODULE.make_encrypt_rand({}, "tdrc_session%3Dpay-test", "1775990000", False)
self.assertEqual(len(rand_value), 9)
self.assertEqual(rand_value[-1], "\x01")
def test_goods_protocol_template_is_compatible(self):
template = _MODULE.load_template_args()
validate_goods_materials(template, [0] * 59640)
diagnostics = goods_material_diagnostics(_MODULE.ROOT, template, [0] * 59640)
self.assertEqual(diagnostics["goods_xmidas_ops_length"], 59640)
self.assertEqual(diagnostics["args_template_slots"], 18)
self.assertEqual(len(diagnostics["bytecode_sha256_prefix"]), 12)
def test_private_evidence_writer_restricts_permissions(self):
import tempfile
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "evidence.json"
_MODULE.write_private_text(path, "sensitive")
self.assertEqual(path.read_text(encoding="utf-8"), "sensitive")
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
def test_goods_vm_generates_expected_full_length_ciphertext(self):
"""528B 明文应走完 33 个块,输出 1056 个 hex 字符。"""
import json
template = _MODULE.load_template_args()
xmidas = json.loads((_MODULE.ROOT / "replay" / "xmidasops.json").read_text(encoding="utf-8"))
params = {
"token_id": "token", "openid": "openid", "openkey": "key", "session_id": "openid",
"session_type": "kp_accesstoken", "zoneid": "1", "pay_method": "wechat",
"buy_quantity": "1", "mb_pwd": "", "pay_id": "", "auth_key": "", "card_value": "",
"accounttype": "", "provide_uin": "", "extend": "", "from_h5": "1",
"webversion": "minipayv2",
}
timestamp = "1775990000"
fk_extend = "tdrc_session%3Dpay-test"
for length in range(512):
params["extend"] = "x" * length
rand_value = _MODULE.make_encrypt_rand(params, fk_extend, timestamp, True)
plaintext = _MODULE.build_plaintext(params, fk_extend, timestamp, rand_value)
if len(plaintext.encode("latin-1")) == 528:
break
else:
self.fail("无法构造 528B goods 测试明文")
key16 = list(range(16))
key1 = derive_key1_from_key16(key16, [template[i][0] for i in (1, 2, 3, 4)], template[5][0])
ciphertext = generate_encrypt_msg_offline(
params, fk_extend, timestamp, rand_value, key16=key16, key1=key1,
args_template=template, xmidas=xmidas, xmidas_token="A" * 96,
)
self.assertEqual(len(ciphertext), 1056)
if __name__ == "__main__":
unittest.main()