import os
import time
import uuid
from pathlib import Path

TEST_DB = Path("/tmp/flashops-e2e-{0}.db".format(os.getpid()))
TEST_OBJECTS = Path("/tmp/flashops-e2e-objects-{0}".format(os.getpid()))
os.environ["FLASHOPS_DATABASE_URL"] = "sqlite+aiosqlite:///" + str(TEST_DB)
os.environ["FLASHOPS_OBJECT_STORE_URL"] = str(TEST_OBJECTS)
os.environ["FLASHOPS_TIME_SCALE"] = "0.01"

from fastapi.testclient import TestClient

from flashops_control.config import reset_settings
from flashops_control.config import get_settings
from flashops_control.main import app


def setup_module():
    reset_settings()
    TEST_DB.unlink(missing_ok=True)


def teardown_module():
    TEST_DB.unlink(missing_ok=True)


def test_demo_run_recovers_and_builds_evidence():
    with TestClient(app) as client:
        health = client.get("/api/v1/health")
        assert health.status_code == 200

        preflight = client.post("/api/v1/preflight", json={})
        assert preflight.status_code == 200
        assert preflight.json()["passed"] is True
        assert len(preflight.json()["checks"]) == 6

        created = client.post(
            "/api/v1/runs",
            json={"loops": 4, "inject_failure": True, "created_by": "pytest"},
        )
        assert created.status_code == 201
        run_id = created.json()["id"]

        run = None
        for _ in range(100):
            response = client.get("/api/v1/runs/" + run_id)
            assert response.status_code == 200
            run = response.json()
            if run["state"] == "COMPLETED":
                break
            time.sleep(0.02)

        assert run is not None
        assert run["state"] == "COMPLETED"
        assert run["verdict"] == "PASS"
        assert run["loop_done"] == 4
        assert run["recovery_count"] == 3
        assert run["evidence"]["completeness"] == 1.0
        assert Path(run["evidence"]["uri"], "manifest.json").exists()

        events = client.get("/api/v1/runs/{0}/events".format(run_id)).json()
        kinds = [event["kind"] for event in events]
        assert [event["seq"] for event in events] == list(range(1, len(events) + 1))
        assert "HEARTBEAT_LOST" in kinds
        assert kinds.count("RECOVERY_STARTED") == 3
        assert "RECOVERY_SUCCEEDED" in kinds
        assert kinds[-1] == "RUN_COMPLETED"


def test_mismatched_serial_is_rejected_before_run_creation():
    with TestClient(app) as client:
        response = client.post(
            "/api/v1/preflight", json={"discovered_serial": "SYSTEM-DISK"}
        )
        assert response.status_code == 200
        payload = response.json()
        assert payload["passed"] is False
        serial = next(check for check in payload["checks"] if check["key"] == "serial")
        assert serial["passed"] is False


def test_pause_resume_and_emergency_stop_are_audited():
    with TestClient(app) as client:
        created = client.post(
            "/api/v1/runs",
            json={"loops": 50, "inject_failure": False, "created_by": "pytest-actions"},
        )
        assert created.status_code == 201
        run_id = created.json()["id"]

        run = created.json()
        for _ in range(50):
            run = client.get("/api/v1/runs/" + run_id).json()
            if run["state"] == "RUNNING":
                break
            time.sleep(0.01)
        assert run["state"] == "RUNNING"

        paused = client.post(
            "/api/v1/runs/{0}/pause".format(run_id),
            json={"actor": "pytest", "reason": "检查暂停"},
        )
        assert paused.status_code == 200
        assert paused.json()["state"] == "PAUSED"

        resumed = client.post(
            "/api/v1/runs/{0}/resume".format(run_id),
            json={"actor": "pytest", "reason": "检查继续"},
        )
        assert resumed.status_code == 200
        assert resumed.json()["state"] == "RUNNING"

        stopped = client.post(
            "/api/v1/runs/{0}/emergency-stop".format(run_id),
            json={"actor": "pytest", "reason": "检查急停"},
        )
        assert stopped.status_code == 200
        assert stopped.json()["state"] == "ABORTED"
        assert stopped.json()["human_touches"] == 3
        events = client.get("/api/v1/runs/{0}/events".format(run_id)).json()
        kinds = [event["kind"] for event in events]
        assert "RUN_PAUSED" in kinds
    assert "RUN_RESUMED" in kinds
    assert kinds[-1] == "RUN_ABORTED"


def test_agent_registration_heartbeat_and_token_rotation():
    with TestClient(app) as client:
        registered = client.post(
            "/api/v1/agent/register",
            json={
                "host_name": "WS-05",
                "agent_version": "0.1.0-test",
                "os_family": "linux",
                "os_version": "pytest",
                "kernel": "test-kernel",
                "ip_address": "192.0.2.50",
                "toolchain": {"fio": "fio-3.36"},
                "labels": {"system_device_paths": ["/dev/nvme0n1"]},
            },
        )
        assert registered.status_code == 201
        credentials = registered.json()
        assert credentials["agent_id"].startswith("agent_")
        assert credentials["host_id"]
        assert len(credentials["token"]) >= 32

        missing_token = client.post(
            "/api/v1/agent/{0}/heartbeat".format(credentials["agent_id"]),
            json={"healthy": True},
        )
        assert missing_token.status_code == 401

        heartbeat = client.post(
            "/api/v1/agent/{0}/heartbeat".format(credentials["agent_id"]),
            headers={"Authorization": "Bearer " + credentials["token"]},
            json={
                "healthy": True,
                "agent_version": "0.1.1-test",
                "device_probe": {"nvme": ["nvme1n1"]},
            },
        )
        assert heartbeat.status_code == 200
        assert heartbeat.json()["ok"] is True
        assert heartbeat.json()["commands"] == []

        rotated = client.post(
            "/api/v1/agent/register",
            json={
                "host_id": credentials["host_id"],
                "agent_version": "0.1.1-test",
            },
        )
        assert rotated.status_code == 201
        rotated_credentials = rotated.json()
        assert rotated_credentials["agent_id"] == credentials["agent_id"]
        assert rotated_credentials["token"] != credentials["token"]

        old_token = client.post(
            "/api/v1/agent/{0}/heartbeat".format(credentials["agent_id"]),
            headers={"Authorization": "Bearer " + credentials["token"]},
            json={"healthy": True},
        )
        assert old_token.status_code == 401

        new_token = client.post(
            "/api/v1/agent/{0}/heartbeat".format(credentials["agent_id"]),
            headers={"Authorization": "Bearer " + rotated_credentials["token"]},
            json={"healthy": False},
        )
        assert new_token.status_code == 200

        hosts = client.get("/api/v1/agent/hosts")
        assert hosts.status_code == 200
        current = next(
            item for item in hosts.json() if item["host_id"] == credentials["host_id"]
        )
        assert current["agent_id"] == credentials["agent_id"]
        assert current["status"] == "DEGRADED"
        assert "token" not in current


def _register_seed_agent(client):
    response = client.post(
        "/api/v1/agent/register",
        json={"host_name": "WS-05", "agent_version": "0.2.0-pytest"},
    )
    assert response.status_code == 201
    return response.json()


def _idem_headers(credentials, key=None):
    return {
        "Authorization": "Bearer " + credentials["token"],
        "Idempotency-Key": key or str(uuid.uuid4()),
    }


def _lease(client, credentials):
    return client.post(
        "/api/v1/agent/{0}/lease".format(credentials["agent_id"]),
        headers={"Authorization": "Bearer " + credentials["token"]},
        json={"wait_timeout_s": 0},
    )


def _ack(client, credentials, lease, key=None):
    return client.post(
        "/api/v1/agent/{0}/steps/{1}/ack".format(
            credentials["agent_id"], lease["step_id"]
        ),
        headers=_idem_headers(credentials, key),
        json={"attempt": lease["attempt"]},
    )


def _complete(client, credentials, lease, key=None):
    return client.post(
        "/api/v1/agent/{0}/steps/{1}/complete".format(
            credentials["agent_id"], lease["step_id"]
        ),
        headers=_idem_headers(credentials, key),
        json={
            "attempt": lease["attempt"],
            "status": "SUCCEEDED",
            "exit_code": 0,
            "result": {"simulated": True},
        },
    )


def test_agent_step_lease_idempotency_and_full_dry_run():
    with TestClient(app) as client:
        credentials = _register_seed_agent(client)
        created = client.post(
            "/api/v1/runs",
            json={
                "loops": 2,
                "inject_failure": False,
                "execution_mode": "agent_dry_run",
                "created_by": "pytest-agent-lease",
            },
        )
        assert created.status_code == 201
        run_id = created.json()["id"]
        assert created.json()["state"] == "RUNNING"
        assert created.json()["execution_mode"] == "agent_dry_run"

        unauthorized = client.post(
            "/api/v1/agent/{0}/lease".format(credentials["agent_id"]),
            json={"wait_timeout_s": 0},
        )
        assert unauthorized.status_code == 401

        first = _lease(client, credentials)
        assert first.status_code == 200
        lease = first.json()
        assert lease["step_key"] == "flash-fw-b"
        assert lease["attempt"] == 1
        assert lease["dry_run"] is True

        ack_key = str(uuid.uuid4())
        acked = _ack(client, credentials, lease, ack_key)
        assert acked.status_code == 200
        assert acked.json()["state"] == "RUNNING"
        replayed_ack = _ack(client, credentials, lease, ack_key)
        assert replayed_ack.status_code == 200
        assert replayed_ack.json()["replayed"] is True

        renew = client.post(
            "/api/v1/agent/{0}/steps/{1}/renew".format(
                credentials["agent_id"], lease["step_id"]
            ),
            headers=_idem_headers(credentials),
            json={"attempt": lease["attempt"]},
        )
        assert renew.status_code == 200

        event_key = str(uuid.uuid4())
        output = client.post(
            "/api/v1/agent/{0}/steps/{1}/events".format(
                credentials["agent_id"], lease["step_id"]
            ),
            headers=_idem_headers(credentials, event_key),
            json={
                "attempt": lease["attempt"],
                "events": [
                    {"message": "dry-run output 1", "payload": {"line": 1}},
                    {"message": "dry-run output 2", "severity": "DEBUG"},
                ],
            },
        )
        assert output.status_code == 200
        assert output.json()["accepted"] == 2
        replayed_output = client.post(
            "/api/v1/agent/{0}/steps/{1}/events".format(
                credentials["agent_id"], lease["step_id"]
            ),
            headers=_idem_headers(credentials, event_key),
            json={
                "attempt": lease["attempt"],
                "events": [
                    {"message": "dry-run output 1", "payload": {"line": 1}},
                    {"message": "dry-run output 2", "severity": "DEBUG"},
                ],
            },
        )
        assert replayed_output.status_code == 200
        assert replayed_output.json()["replayed"] is True

        complete_key = str(uuid.uuid4())
        completed = _complete(client, credentials, lease, complete_key)
        assert completed.status_code == 200
        replayed_complete = _complete(client, credentials, lease, complete_key)
        assert replayed_complete.status_code == 200
        assert replayed_complete.json()["replayed"] is True

        step_count = 1
        while True:
            leased = _lease(client, credentials)
            if leased.status_code == 204:
                break
            assert leased.status_code == 200
            item = leased.json()
            assert _ack(client, credentials, item).status_code == 200
            assert _complete(client, credentials, item).status_code == 200
            step_count += 1

        assert step_count == 7  # flash + (3 loop-body steps * 2 loops)
        run = client.get("/api/v1/runs/" + run_id).json()
        assert run["state"] == "COMPLETED"
        assert run["verdict"] == "PASS"
        assert run["loop_done"] == 2
        assert run["evidence"]["completeness"] == 1.0
        events = client.get("/api/v1/runs/{0}/events".format(run_id)).json()
        assert [event["seq"] for event in events] == list(range(1, len(events) + 1))
        assert sum(event["kind"] == "STEP_DISPATCHED" for event in events) == 7
        assert sum(event["kind"] == "STEP_SUCCEEDED" for event in events) == 7


def test_expired_running_lease_retries_only_idempotent_steps():
    timing = get_settings().timing
    original_ttl = timing.lease_ttl_s
    object.__setattr__(timing, "lease_ttl_s", 0.01)
    try:
        with TestClient(app) as client:
            credentials = _register_seed_agent(client)

            # The first workflow step is a non-idempotent firmware action.  Once
            # ACKed, an expired lease must freeze the run instead of rerunning it.
            created = client.post(
                "/api/v1/runs",
                json={"loops": 1, "execution_mode": "agent_dry_run"},
            )
            assert created.status_code == 201
            run_id = created.json()["id"]
            lease = _lease(client, credentials).json()
            assert lease["idempotent"] is False
            assert _ack(client, credentials, lease).status_code == 200
            time.sleep(0.02)
            assert _lease(client, credentials).status_code == 204
            frozen = client.get("/api/v1/runs/" + run_id).json()
            assert frozen["state"] == "FROZEN"
            assert frozen["verdict"] == "INCONCLUSIVE"

            # On another run, finish the firmware step and let the idempotent fio
            # workload expire.  It is safely reissued once with attempt=2.
            created_retry = client.post(
                "/api/v1/runs",
                json={"loops": 1, "execution_mode": "agent_dry_run"},
            )
            assert created_retry.status_code == 201
            retry_run_id = created_retry.json()["id"]
            firmware = _lease(client, credentials).json()
            assert _ack(client, credentials, firmware).status_code == 200
            assert _complete(client, credentials, firmware).status_code == 200
            workload = _lease(client, credentials).json()
            assert workload["step_key"] == "workload"
            assert workload["idempotent"] is True
            assert _ack(client, credentials, workload).status_code == 200
            time.sleep(0.02)
            retried = _lease(client, credentials)
            assert retried.status_code == 200
            assert retried.json()["step_id"] == workload["step_id"]
            assert retried.json()["attempt"] == 2
            stopped = client.post(
                "/api/v1/runs/{0}/emergency-stop".format(retry_run_id),
                json={"actor": "pytest", "reason": "清理重试场景"},
            )
            assert stopped.status_code == 200
    finally:
        object.__setattr__(timing, "lease_ttl_s", original_ttl)
