import asyncio
import json
import sys
from pathlib import Path
from time import perf_counter
from uuid import UUID, uuid4

import httpx
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401
from test_fulfillment import approve, connected, operation_for  # noqa: F401

from zhifu.modules.identity.auth import now
from zhifu.modules.integrations.fulfillment import execute_due, prepare_event
from zhifu.modules.integrations.models import ExternalOperation, Reservation
from zhifu.oms.models import Exchange


async def test_terminate_after_oms_commit_and_recover(connected):  # noqa: F811
    sessions = connected["app"].state.sessions
    ticket, event = await approve(connected)
    await prepare_event(sessions, event, 5)
    operation = await operation_for(connected, ticket)
    name = "process-recovery-" + uuid4().hex
    process = None
    started = perf_counter()
    async with httpx.AsyncClient(base_url="http://127.0.0.1:18474", timeout=10) as proxy:
        response = await proxy.post(
            "/proxies",
            json={
                "name": name,
                "listen": "0.0.0.0:18100",
                "upstream": "host.docker.internal:8100",
            },
        )
        assert response.status_code == 201, response.text
        try:
            async with asyncio.timeout(180):
                response = await proxy.post(
                    f"/proxies/{name}/toxics",
                    json={
                        "name": "hold-response",
                        "type": "latency",
                        "stream": "downstream",
                        "attributes": {"latency": 60000, "jitter": 0},
                    },
                )
                assert response.status_code == 200, response.text
                async with sessions.begin() as session:
                    current = await session.get(ExternalOperation, operation.id)
                    current.base_url = "http://127.0.0.1:18100"
                process = await asyncio.create_subprocess_exec(
                    sys.executable,
                    "scripts/recovery_probe.py",
                    str(UUID(bytes=operation.id)),
                    "external",
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                # 独立查询真实OMS事务，只有确认受理后才终止正在等待HTTP响应的工作进程。
                for _ in range(250):
                    async with connected["oms_sessions"]() as session:
                        exchange = await session.get(Exchange, operation.id)
                    if exchange is not None:
                        break
                    assert process.returncode is None
                    await asyncio.sleep(0.1)
                assert exchange is not None and exchange.status == "ACCEPTED"
                process.kill()
                await process.wait()
                response = await proxy.delete(f"/proxies/{name}/toxics/hold-response")
                assert response.status_code == 204
                async with sessions() as session:
                    current = await session.get(ExternalOperation, operation.id)
                    assert current.status == "PREPARED"
                    assert (await session.get(Reservation, operation.id)).status == "HELD"
                    expiry, token = current.lease_until, current.lease_token
                assert await execute_due(sessions, 5, operation.id) is False
                await asyncio.sleep(max(0, (expiry - now()).total_seconds()) + 0.05)
                assert await execute_due(sessions, 5, operation.id)
                async with sessions() as session:
                    current = await session.get(ExternalOperation, operation.id)
                    assert current.status == "ACCEPTED" and current.lease_token > token
                    assert (await session.get(Reservation, operation.id)).status == "CONSUMED"
                async with connected["oms_sessions"]() as session:
                    count = await session.scalar(
                        select(func.count())
                        .select_from(Exchange)
                        .where(Exchange.tenant_id == connected["tenant"].bytes)
                    )
                    assert count == 1
                report = {
                    "child_exit": process.returncode,
                    "seconds": perf_counter() - started,
                    "natural_lease_expiry": True,
                    "external_applications": count,
                    "recovered_status": current.status,
                    "generation_attempts": 0,
                }
                await asyncio.to_thread(
                    Path(".work/external-process-recovery.json").write_text,
                    json.dumps(report, ensure_ascii=False, indent=2) + "\n",
                    encoding="utf-8",
                )
        finally:
            if process is not None and process.returncode is None:
                process.kill()
                await process.wait()
            response = await proxy.delete(f"/proxies/{name}")
            assert response.status_code == 204