zhifu-platform/tests/test_platform_management.py

import asyncio
import hashlib
import hmac
import ipaddress
import json
import os
import ssl
import threading
from datetime import UTC, datetime, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from uuid import UUID, uuid4

import pytest
import pytest_asyncio
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from sqlalchemy import delete, func, select
from test_after_sales import make_proposal, scenario  # noqa: F401

from zhifu.modules.agent_runtime.models import TokenAccount, TokenPolicy
from zhifu.modules.identity.auth import now, password_hasher
from zhifu.modules.identity.models import Tenant, TenantQuota
from zhifu.modules.merchant.models import ChannelApplication
from zhifu.modules.platform.delivery import enqueue, execute_one
from zhifu.modules.platform.models import (
    ChannelDelivery,
    PlatformAccount,
    PlatformAudit,
    PlatformCommand,
    PlatformSession,
    TenantRegistration,
)
from zhifu.modules.platform.schemas import PlatformState

pytestmark = pytest.mark.integration
BASE = "/openapi/v1/platform"


@pytest_asyncio.fixture
async def operator(scenario):  # noqa: F811
    sessions = scenario["app"].state.sessions
    actor_id = uuid4()
    username = "platform-" + actor_id.hex
    password = "integration-" + uuid4().hex
    created = []
    async with sessions.begin() as session:
        session.add(PlatformAccount(
            id=actor_id.bytes, username=username,
            password_hash=password_hasher.hash(password), status="ACTIVE",
            permissions=["platform:manage"], created_at=now(),
        ))
    client = scenario["client"]
    response = await client.post(BASE + "/auth/sessions", json={
        "username": username, "password": password,
    })
    assert response.status_code == 200, response.text
    token = response.json()["access_token"]

    async def call(method, path, body=None, key=None):
        return await client.request(method, BASE + path, json=body, headers={
            "Authorization": "Bearer " + token, "Idempotency-Key": key or str(uuid4()),
        })

    try:
        yield {
            "call": call, "actor_id": actor_id, "created": created,
            "username": username, "password": password,
        }
    finally:
        async with sessions.begin() as session:
            await session.execute(delete(ChannelDelivery).where(
                ChannelDelivery.tenant_id == scenario["tenant"].bytes,
            ))
            for model in (PlatformCommand, PlatformAudit):
                await session.execute(delete(model).where(model.actor_id == actor_id.bytes))
            await session.execute(delete(PlatformSession).where(
                PlatformSession.account_id == actor_id.bytes,
            ))
            await session.execute(delete(PlatformAccount).where(
                PlatformAccount.id == actor_id.bytes,
            ))
            if created:
                await session.execute(delete(TenantRegistration).where(
                    TenantRegistration.tenant_id.in_(created),
                ))
                await session.execute(delete(Tenant).where(Tenant.id.in_(created)))


async def test_platform_identity_quota_and_audit(scenario, operator):  # noqa: F811
    call = operator["call"]
    client = scenario["client"]
    sessions = scenario["app"].state.sessions
    assert (await client.get(BASE + "/state")).status_code == 401
    foreign = await client.get(BASE + "/state", headers=scenario["headers"]["agent"])
    assert foreign.status_code == 401
    wrong = await client.post(BASE + "/auth/sessions", json={
        "username": operator["username"], "password": "wrong-password",
    })
    assert wrong.status_code == 401
    identity = await call("GET", "/auth/me")
    assert identity.json()["kind"] == "PLATFORM"
    initial = await call("GET", "/state")
    assert initial.status_code == 200, initial.text
    PlatformState.model_validate(initial.json())
    assert len(initial.json()["runtime"]["trend"]) == 7
    assert initial.json()["runtime"]["metrics"]["tenants"] >= 2

    name = "平台接口验收-" + uuid4().hex
    body = {"name": name, "contact": "接口验收负责人"}
    key = str(uuid4())
    first, second = await asyncio.gather(
        call("POST", "/tenants", body, key), call("POST", "/tenants", body, key),
    )
    assert first.status_code == second.status_code == 201
    assert first.json() == second.json()
    tenant_id = UUID(first.json()["id"])
    operator["created"].append(tenant_id.bytes)
    assert first.json()["status"] == "PENDING"
    assert (await call("POST", "/tenants", body)).status_code == 409
    changed = await call("POST", "/tenants", {**body, "contact": "另一位负责人"}, key)
    assert changed.status_code == 409
    pending_status = await call("PUT", f"/tenants/{tenant_id}/status", {
        "expected_version": 1, "status": "ENABLED",
    })
    assert pending_status.json()["code"] == "TENANT_NOT_READY"

    original_id = scenario["tenant"]
    status_path = f"/tenants/{original_id}/status"
    disabled = await call("PUT", status_path, {"expected_version": 1, "status": "DISABLED"})
    assert disabled.status_code == 200, disabled.text
    assert (await scenario["call"]("GET", "/openapi/v1/auth/me")).status_code == 401
    stale = await call("PUT", status_path, {"expected_version": 1, "status": "ENABLED"})
    assert stale.status_code == 409
    enabled = await call("PUT", status_path, {"expected_version": 2, "status": "ENABLED"})
    assert enabled.status_code == 200
    assert (await scenario["call"]("GET", "/openapi/v1/auth/me")).status_code == 200

    capacity_path = f"/tenants/{tenant_id}/capacity"
    change = {
        "expected_version": 1, "expected_token_version": 1, "interactive": 3,
        "reserved": 0, "background": 1, "tokens": 900000, "reason": "真实额度验证",
    }
    saved = await call("PUT", capacity_path, change)
    assert saved.status_code == 200, saved.text
    assert saved.json()["tokens"] == 900000
    assert (await call("PUT", capacity_path, change)).status_code == 409
    excess = await call("PUT", capacity_path, {
        **change, "expected_version": 2, "expected_token_version": 2,
        "interactive": 8, "reserved": 8,
    })
    assert excess.json()["code"] == "CAPACITY_EXCEEDED"
    async with sessions.begin() as session:
        assert (await session.get(TenantQuota, tenant_id.bytes)).interactive_limit == 3
        assert (await session.get(TokenPolicy, tenant_id.bytes)).token_limit == 900000
        session.add(TokenAccount(
            tenant_id=tenant_id.bytes, period=now().date(), token_limit=900000,
            reserved=25, consumed=50,
        ))
    smaller = await call("PUT", capacity_path, {
        **change, "expected_version": 2, "expected_token_version": 2, "tokens": 0,
    })
    assert smaller.status_code == 200
    async with sessions() as session:
        account = await session.get(TokenAccount, (tenant_id.bytes, now().date()))
        assert (account.token_limit, account.reserved, account.consumed) == (0, 25, 50)
        count = await session.scalar(select(func.count()).select_from(PlatformAudit).where(
            PlatformAudit.actor_id == operator["actor_id"].bytes,
            PlatformAudit.action == "登记商家",
        ))
        assert count == 1
    audits = (await call("GET", "/state")).json()["audit"]
    assert any("真实额度验证" in row["action"] for row in audits)
    async with sessions.begin() as session:
        actor = await session.get(PlatformAccount, operator["actor_id"].bytes)
        actor.status = "SUSPENDED"
    assert (await call("GET", "/auth/me")).status_code == 401
    disabled_login = await client.post(BASE + "/auth/sessions", json={
        "username": operator["username"], "password": operator["password"],
    })
    assert disabled_login.status_code == 401
    async with sessions.begin() as session:
        actor = await session.get(PlatformAccount, operator["actor_id"].bytes)
        actor.status = "ACTIVE"
        actor.permissions = []
    assert (await call("GET", "/state")).status_code == 403
    unauthorized_login = await client.post(BASE + "/auth/sessions", json={
        "username": operator["username"], "password": operator["password"],
    })
    assert unauthorized_login.status_code == 401
    async with sessions.begin() as session:
        actor = await session.get(PlatformAccount, operator["actor_id"].bytes)
        actor.permissions = ["platform:manage"]
    assert (await call("DELETE", "/auth/sessions/current")).status_code == 204
    assert (await call("GET", "/auth/me")).status_code == 401
    assert (await scenario["call"]("GET", "/openapi/v1/auth/me")).status_code == 200


@pytest.fixture
def receiver():
    folder = Path(".work/platform-receiver") / uuid4().hex
    folder.mkdir(parents=True)
    private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "platform-local-receiver")])
    clock = datetime.now(UTC)
    certificate = (
        x509.CertificateBuilder().subject_name(subject).issuer_name(subject)
        .public_key(private.public_key()).serial_number(x509.random_serial_number())
        .not_valid_before(clock - timedelta(minutes=1))
        .not_valid_after(clock + timedelta(days=1))
        .add_extension(x509.SubjectAlternativeName([
            x509.IPAddress(ipaddress.ip_address("127.0.0.1")),
        ]), critical=False).sign(private, hashes.SHA256())
    )
    cert = folder / "certificate.pem"
    key = folder / "private.pem"
    cert.write_bytes(certificate.public_bytes(serialization.Encoding.PEM))
    key.write_bytes(private.private_bytes(
        serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption(),
    ))
    secret = uuid4().hex + uuid4().hex
    received = []
    accepted = threading.Event()

    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            content = self.rfile.read(int(self.headers["Content-Length"]))
            timestamp = self.headers["X-Zhifu-Timestamp"]
            signature = hmac.new(
                secret.encode(), timestamp.encode() + b"." + content, hashlib.sha256,
            ).hexdigest()
            assert hmac.compare_digest(signature, self.headers["X-Zhifu-Signature"])
            received.append(json.loads(content))
            # 接收器维护期间真实返回 HTTP 503，解除维护后才接受并确认通知。
            self.send_response(204 if accepted.is_set() else 503)
            self.send_header("Content-Length", "0")
            self.end_headers()

        def log_message(self, *_):
            pass

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(cert, key)
    server.socket = context.wrap_socket(server.socket, server_side=True)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    prior_ca = os.environ.get("SSL_CERT_FILE")
    credential_env = "PLATFORM_TEST_SIGNING_" + uuid4().hex
    os.environ["SSL_CERT_FILE"] = str(cert.resolve())
    os.environ[credential_env] = secret
    try:
        yield {
            "url": f"https://127.0.0.1:{server.server_port}/events",
            "credential_env": credential_env, "received": received, "accepted": accepted,
        }
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)
        if prior_ca is None:
            os.environ.pop("SSL_CERT_FILE", None)
        else:
            os.environ["SSL_CERT_FILE"] = prior_ca
        os.environ.pop(credential_env, None)


async def test_real_tls_delivery_failure_retry_and_receipt(scenario, operator, receiver):  # noqa: F811
    await make_proposal(scenario)
    sessions = scenario["app"].state.sessions
    async with sessions.begin() as session:
        session.add(ChannelApplication(
            tenant_id=scenario["tenant"].bytes, store_id=scenario["store"].bytes,
            name="本地 TLS 验收接收器", callback_url=receiver["url"],
            credential_env=receiver["credential_env"], enabled=True, version=1,
            scopes=["tickets:results"], updated_at=now(),
        ))
    assert await enqueue(sessions, tenant_id=scenario["tenant"].bytes) == 1
    assert await enqueue(sessions, tenant_id=scenario["tenant"].bytes) == 0
    async with sessions() as session:
        row = await session.scalar(select(ChannelDelivery).where(
            ChannelDelivery.tenant_id == scenario["tenant"].bytes,
        ))
        delivery_id = row.id
    # 实际使用 30 秒重试间隔，整个测试最多等待 75 秒，生成模型调用次数为 0。
    async with asyncio.timeout(75):
        while True:
            await execute_one(sessions, delivery_id)
            async with sessions() as session:
                row = await session.get(ChannelDelivery, delivery_id)
                if row.status == "FAILED":
                    break
            await asyncio.sleep(1)
    assert row.attempts == 3 and row.last_error == "HTTP_503"
    assert len(receiver["received"]) == 3
    event_id = receiver["received"][0]["event_id"]
    path = f"/deliveries/{UUID(bytes=delivery_id)}/retry"
    call = operator["call"]
    key = str(uuid4())
    body = {"expected_version": row.version}
    first = await call("POST", path, body, key)
    repeated = await call("POST", path, body, key)
    assert first.status_code == repeated.status_code == 200
    assert first.json() == repeated.json()
    assert first.json()["status"] == "QUEUED"
    assert (await call("POST", path, body)).status_code == 409
    receiver["accepted"].set()
    await execute_one(sessions, delivery_id)
    async with sessions() as session:
        row = await session.get(ChannelDelivery, delivery_id)
        assert row.status == "SENT" and row.response_status == 204 and row.attempts == 4
    assert {entry["event_id"] for entry in receiver["received"]} == {event_id}
    assert {entry["delivery_id"] for entry in receiver["received"]} == {
        str(UUID(bytes=delivery_id)),
    }
    assert await execute_one(sessions, delivery_id) is False