zhifu-platform/tests/recovery_orchestration_acceptance.py

import asyncio
import json
import subprocess
import sys
from pathlib import Path
from uuid import uuid4

import psutil
from retrieval_evaluation import publish_dataset
from test_after_sales import scenario  # noqa: F401

from zhifu.modules.knowledge.vector import connection


def execute(command):
    with Path(".work/sequential-recovery.log").open("wb") as log:
        subprocess.run(command, check=True, timeout=300, stdout=log, stderr=subprocess.STDOUT)


async def test_sequential_recovery_of_real_sources(scenario):  # noqa: F811
    collection = "zhifu_recovery_sequence_" + uuid4().hex
    output = Path(".work/sequential-recovery.json")
    try:
        await publish_dataset(scenario)
        await asyncio.to_thread(
            execute,
            [sys.executable, "scripts/recovery_exercise.py", "--target-collection", collection,
             "--tenant-id", str(scenario["tenant"]), "--maximum-documents", "12",
             "--maximum-chunks", "64", "--timeout-seconds", "240", "--output", str(output)],
        )
        report = json.loads(await asyncio.to_thread(output.read_text, encoding="utf-8"))
        assert report["status"] == "VERIFIED"
        assert set(report["phases"]) == {"mysql", "redis", "knowledge"}
        assert report["phases"]["knowledge"]["documents_restored"] == 12
        assert report["phases"]["knowledge"]["tenant_id"] == str(scenario["tenant"])
        assert report["temporary_containers_remaining"] == []
        assert report["generation_attempts"] == 0
        assert report["business_targets_switched"] is False
        assert all(row["checksum_equal"] for row in report["phases"]["mysql"]["databases"].values())
        assert report["phases"]["redis"]["sha256_verified"] is True
    finally:
        with connection() as (client, business):
            assert collection != business
            if await asyncio.to_thread(client.has_collection, collection):
                await asyncio.to_thread(client.drop_collection, collection)
            await asyncio.to_thread(
                client.delete, business, filter=f'tenant == "{scenario["tenant"].hex}"'
            )


async def test_timeout_terminates_phase_process_tree_and_containers():
    output = Path(".work/sequential-recovery-timeout.json")
    result = await asyncio.to_thread(
        subprocess.run,
        [sys.executable, "scripts/recovery_exercise.py", "--target-collection",
         "zhifu_recovery_timeout_" + uuid4().hex, "--timeout-seconds", "60",
         "--phase-timeout-seconds", "10", "--output", str(output)],
        capture_output=True, timeout=90,
    )
    await asyncio.to_thread(
        Path(".work/sequential-recovery-timeout.log").write_bytes, result.stdout + result.stderr
    )
    assert result.returncode != 0
    report = json.loads(await asyncio.to_thread(output.read_text, encoding="utf-8"))
    assert report["status"] == "INCOMPLETE"
    assert report["phases"] == {}
    assert report["temporary_containers_cleaned"], "需要在真实恢复容器运行期间验证超时"
    assert report["temporary_containers_remaining"] == []
    identifiers = report["terminated_process_ids"]
    assert identifiers and all(not psutil.pid_exists(identifier) for identifier in identifiers)
    # 阶段进程终止后再次检查，不能出现父工具清理之后新创建的恢复容器。
    await asyncio.sleep(2)
    remaining = await asyncio.to_thread(
        subprocess.run,
        ["docker", "ps", "-aq", "--filter", "label=zhifu.recovery-run=" + report["run_id"]],
        check=True, capture_output=True, timeout=15,
    )
    assert not remaining.stdout.strip()