wendajia/backend/tests/test_alipay_provider.py

import json

import pytest
from hashlib import md5
from sqlalchemy import create_engine, inspect, text

from app import alipay_provider
from app.main import _migrate_alipay_transfer_attempts


class FakeClient:
    def __init__(self, payload):
        self.payload = payload

    def execute(self, _request):
        return json.dumps(self.payload, ensure_ascii=False)


def test_transfer_parses_official_sdk_json_response(monkeypatch):
    monkeypatch.setattr(alipay_provider, "_client", lambda: FakeClient({
        "code": "10000",
        "msg": "Success",
        "out_biz_no": "RF001",
        "order_id": "ALI001",
        "status": "SUCCESS",
    }))
    result = alipay_provider.transfer(
        out_biz_no="RF001", amount_fen=300, account="buyer@example.com",
        payee_name="测试收款人", order_title="测试返款",
    )
    assert result == {"out_biz_no": "RF001", "trade_no": "ALI001", "status": "SUCCESS"}


def test_transfer_preserves_provider_business_error(monkeypatch):
    monkeypatch.setattr(alipay_provider, "_client", lambda: FakeClient({
        "code": "40004",
        "msg": "Business Failed",
        "sub_code": "PAYEE_NOT_EXIST",
        "sub_msg": "收款账户不存在",
    }))
    with pytest.raises(alipay_provider.AlipayTransferError) as error:
        alipay_provider.transfer(
            out_biz_no="RF002", amount_fen=300, account="missing@example.com",
            payee_name="测试收款人", order_title="测试返款",
        )
    assert error.value.code == "PAYEE_NOT_EXIST"
    assert error.value.ambiguous is False


def test_old_sdk_error_envelope_is_recovered_from_response_exception():
    raw = (
        '[request-id]response shape maybe illegal. '
        '{"alipay_fund_trans_common_query_response":{'
        '"code":"40006","msg":"Insufficient Permissions",'
        '"sub_code":"isv.illegal-client-ip",'
        '"sub_msg":"当前访问IP不在开放平台设置"},'
        '"alipay_cert_sn":"abc","sign":"signature"}'
    )

    error = alipay_provider._provider_error_from_exception(RuntimeError(raw))

    assert error is not None
    assert error.code == "isv.illegal-client-ip"
    assert str(error) == "当前访问IP不在开放平台设置"
    assert error.ambiguous is False


def test_query_parses_processing_status(monkeypatch):
    monkeypatch.setattr(alipay_provider, "_client", lambda: FakeClient({
        "code": "10000",
        "msg": "Success",
        "out_biz_no": "RF003",
        "order_id": "ALI003",
        "status": "DEALING",
    }))
    result = alipay_provider.query(out_biz_no="RF003")
    assert result["status"] == "DEALING"
    assert result["trade_no"] == "ALI003"


def test_legacy_transfer_table_is_migrated_without_losing_rows():
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.execute(text("""
            CREATE TABLE alipay_transfer_attempts (
                id INTEGER PRIMARY KEY,
                target_kind VARCHAR(32) NOT NULL,
                target_id INTEGER NOT NULL,
                amount_fen INTEGER NOT NULL,
                account VARCHAR(128) NOT NULL,
                payee_name VARCHAR(64) NOT NULL,
                created_by_id INTEGER NOT NULL,
                status VARCHAR(32) NOT NULL DEFAULT 'pending',
                created_at DATETIME
            )
        """))
        connection.execute(text("""
            INSERT INTO alipay_transfer_attempts
                (id, target_kind, target_id, amount_fen, account, payee_name, created_by_id)
            VALUES (1, 'answer_submission', 42, 100, 'buyer@example.com', '测试收款人', 7)
        """))

    _migrate_alipay_transfer_attempts(engine)
    columns = {column["name"] for column in inspect(engine).get_columns("alipay_transfer_attempts")}
    assert {"order_no", "shop_name", "payer_account", "operator_name", "error_message", "out_biz_no"} <= columns
    with engine.connect() as connection:
        row = connection.execute(text("SELECT * FROM alipay_transfer_attempts WHERE id = 1")).mappings().one()
    assert row["amount_fen"] == 100
    assert row["out_biz_no"].startswith("LEGACY-1-")
    assert row["status"] == "failed"
    assert "旧版本记录" in row["error_message"]
    engine.dispose()


def test_certificate_serials_use_issuer_and_decimal_serial_number():
    class FakeName:
        def rfc4514_string(self):
            return "CN=Alipay,O=Example,C=CN"

    class FakeCertificate:
        issuer = FakeName()
        serial_number = 0xA12B

        class SignatureAlgorithmOid:
            dotted_string = "1.2.840.113549.1.1.11"

        signature_algorithm_oid = SignatureAlgorithmOid()

    expected = md5(b"CN=Alipay,O=Example,C=CN41259").hexdigest()
    assert alipay_provider._certificate_serial(FakeCertificate()) == expected
    assert alipay_provider._root_certificate_serial([FakeCertificate(), FakeCertificate()]) == f"{expected}_{expected}"


def test_root_certificate_serial_ignores_non_rsa_certificates():
    class FakeName:
        def rfc4514_string(self):
            return "CN=Alipay,O=Example,C=CN"

    class FakeCertificate:
        issuer = FakeName()
        serial_number = 0xA12B

        def __init__(self, oid):
            self.signature_algorithm_oid = type("Oid", (), {"dotted_string": oid})()

    rsa = FakeCertificate("1.2.840.113549.1.1.11")
    sm2 = FakeCertificate("1.2.156.10197.1.501")
    ecc = FakeCertificate("1.2.840.10045.4.3.3")
    expected = md5(b"CN=Alipay,O=Example,C=CN41259").hexdigest()

    assert alipay_provider._root_certificate_serial([sm2, rsa, ecc]) == expected


def test_certificate_bundle_keeps_valid_certificates_when_one_block_is_unsupported(monkeypatch, tmp_path):
    bundle = tmp_path / "alipayRootCert.crt"
    bundle.write_bytes(
        b"-----BEGIN CERTIFICATE-----\nBAD\n-----END CERTIFICATE-----\n"
        b"-----BEGIN CERTIFICATE-----\nGOOD\n-----END CERTIFICATE-----\n"
    )
    valid_certificate = object()

    class FakeX509:
        @staticmethod
        def load_pem_x509_certificate(block):
            if b"BAD" in block:
                raise ValueError("unsupported SM2 certificate")
            return valid_certificate

    monkeypatch.setitem(__import__("sys").modules, "cryptography.x509", FakeX509)

    assert alipay_provider._load_certificate_bundle(str(bundle), "支付宝根证书") == [valid_certificate]