"""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()