zhifu-platform/tests/deployment_acceptance.py

import asyncio
import importlib.util
import json
from pathlib import Path
from uuid import uuid4

import httpx
import pytest
from test_after_sales import scenario  # noqa: F401
from test_fulfillment import connected  # noqa: F401


async def test_container_business_and_database_recovery(connected):  # noqa: F811
    async with httpx.AsyncClient(
        base_url="http://127.0.0.1:28000",
        timeout=15,
        trust_env=False,
        headers=connected["headers"]["customer"],
    ) as client:
        orders = await client.get(f"/openapi/v1/orders?store_id={connected['store']}")
        assert orders.status_code == 200, orders.text
        assert str(connected["order_id"]) in orders.text
        created = await client.post(
            "/openapi/v1/after-sales/proposals",
            json=connected["proposal"],
            headers={"Idempotency-Key": uuid4().hex},
        )
        assert created.status_code == 201, created.text
        proposal = await client.get(f"/openapi/v1/tickets/{created.json()['ticket_id']}")
        assert proposal.status_code == 200, proposal.text
    spec = importlib.util.spec_from_file_location("mysql_recovery", "scripts/mysql_recovery.py")
    recovery = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(recovery)
    directory = await asyncio.to_thread(recovery.backup)
    report = await asyncio.to_thread(recovery.restore, directory)
    assert report["databases"]["zhifu"]["orders"] >= 1
    assert report["databases"]["zhifu"]["tickets"] >= 1
    assert report["databases"]["oms"]["orders"] >= 1
    report["container_order_read"] = orders.status_code
    report["container_ticket_read"] = proposal.status_code
    report["container_proposal_create"] = created.status_code
    path = directory / "zhifu.sql"
    original = await asyncio.to_thread(path.read_bytes)
    try:
        await asyncio.to_thread(path.write_bytes, original + b"\n")
        with pytest.raises(ValueError, match="SHA256"):
            await asyncio.to_thread(recovery.restore, directory)
    finally:
        await asyncio.to_thread(path.write_bytes, original)
    report["corrupted_backup_rejected"] = True
    await asyncio.to_thread(
        Path("docs/verification/deployment-recovery.json").write_text,
        json.dumps(report, indent=2),
        encoding="utf-8",
    )