from __future__ import annotations

import json

import httpx
import pytest

from investibook.clients.run9 import (
    RUN9_STDIN_CHUNK_SIZE,
    Run9ApiError,
    Run9Client,
    decode_run9_output_frames,
)


@pytest.mark.asyncio
async def test_access_key_and_secret_key_use_basic_auth():
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(200, json={"box_id": "box-1"}, request=request)

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        access_key="access",
        secret_key="secret",
        project_cid="project",
        client=http_client,
    )

    await client.get_box("box-1")

    assert requests[0].headers["Authorization"] == "Basic YWNjZXNzOnNlY3JldA=="
    await http_client.aclose()


def test_partial_basic_auth_credentials_are_rejected():
    with pytest.raises(ValueError, match="both access_key and secret_key"):
        Run9Client(
            base_url="https://run9.test",
            access_key="access",
            project_cid="project",
        )


def test_decode_run9_output_frames_extracts_output_and_ignores_control_frames():
    def frame(kind: int, sequence: int, payload: bytes = b"") -> bytes:
        return (
            bytes([kind])
            + sequence.to_bytes(8, "big")
            + len(payload).to_bytes(4, "big")
            + payload
        )

    data = (
        frame(1, 1)
        + frame(2, 2, b"hello")
        + frame(6, 3, b"\x00\x00\x00\x00")
        + frame(7, 4, b"started")
    )

    assert decode_run9_output_frames(data) == b"hello"
    assert decode_run9_output_frames(b'{"jsonrpc":"2.0"}\n') == b'{"jsonrpc":"2.0"}\n'


def test_decode_run9_output_frames_rejects_truncated_payload():
    data = b"\x02" + (1).to_bytes(8, "big") + (5).to_bytes(4, "big") + b"bad"

    with pytest.raises(RuntimeError, match="payload is truncated"):
        decode_run9_output_frames(data)


@pytest.mark.asyncio
async def test_write_stdin_splits_payload_and_closes_only_final_chunk():
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(200, request=request)

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )
    payload = b"x" * (RUN9_STDIN_CHUNK_SIZE * 2 + 7)

    await client.write_stdin(exec_id="exec/1", data=payload, close_stdin=True)

    assert [len(request.content) for request in requests] == [
        RUN9_STDIN_CHUNK_SIZE,
        RUN9_STDIN_CHUNK_SIZE,
        7,
    ]
    assert [request.headers["X-Run9-Close-Stdin"] for request in requests] == [
        "false",
        "false",
        "true",
    ]
    assert requests[0].url.raw_path.endswith(b"/execs/exec%2F1/write-stdin")
    await http_client.aclose()


@pytest.mark.asyncio
async def test_pull_output_preserves_raw_bytes_and_status_headers():
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            content=b"part-1\xe4\xbd",
            headers={
                "X-Run9-Next-Cursor": "cursor-2",
                "X-Run9-Exec-State": "running",
                "X-Run9-Exit-Code": "",
            },
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    chunk = await client.pull_output(exec_id="exec-1", cursor="cursor-1")

    assert chunk.data == b"part-1\xe4\xbd"
    assert chunk.cursor == "cursor-2"
    assert chunk.state == "running"
    assert chunk.exit_code is None
    await http_client.aclose()


@pytest.mark.asyncio
async def test_ensure_box_rejects_owner_label_mismatch():
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "box_id": "box-1",
                "labels": {"workspace_id": "ws-1", "owner_user_id": "other"},
            },
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    with pytest.raises(RuntimeError, match="ownership labels do not match"):
        await client.ensure_box(
            box_id="box-1",
            description="workspace",
            labels={"workspace_id": "ws-1", "owner_user_id": "user-1"},
        )
    await http_client.aclose()


@pytest.mark.asyncio
async def test_api_error_does_not_treat_non_404_as_missing_box():
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(401, text="unauthorized", request=request)

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    with pytest.raises(Run9ApiError) as exc_info:
        await client.ensure_box(box_id="box-1", description="workspace", labels={})
    assert exc_info.value.status_code == 401
    await http_client.aclose()


@pytest.mark.asyncio
async def test_ensure_box_accepts_run9_legacy_400_missing_response():
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        if request.method == "GET":
            return httpx.Response(400, json={"error": "box not found"}, request=request)
        return httpx.Response(
            200,
            json={"box_id": "box-1", "labels": {"workspace_id": "ws-1"}},
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    box = await client.ensure_box(
        box_id="box-1",
        description="workspace",
        labels={"workspace_id": "ws-1"},
    )

    assert box["box_id"] == "box-1"
    assert [request.method for request in requests] == ["GET", "POST"]
    create_payload = json.loads(requests[1].content)
    assert create_payload["network_mode"] == "managed"
    assert "security_mode" not in create_payload
    await http_client.aclose()


@pytest.mark.asyncio
async def test_ensure_box_keeps_existing_network_mode():
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        if request.method == "GET":
            return httpx.Response(
                200,
                json={"box_id": "box-1", "labels": {}, "network_mode": "normal"},
                request=request,
            )
        raise AssertionError("existing box must not be patched")

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    box = await client.ensure_box(
        box_id="box-1", description="workspace", labels={}, network_mode="managed"
    )

    assert box["network_mode"] == "normal"
    assert [request.method for request in requests] == ["GET"]
    await http_client.aclose()


@pytest.mark.asyncio
async def test_ensure_box_uses_extra_labels_only_when_creating():
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        if request.method == "GET":
            return httpx.Response(404, json={"error": "not found"}, request=request)
        payload = json.loads(request.content)
        return httpx.Response(
            200,
            json={"box_id": "box-1", "labels": payload["labels"]},
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    box = await client.ensure_box(
        box_id="box-1",
        description="workspace",
        labels={"workspace_id": "ws-1"},
        create_labels={
            "workspace_id": "ws-1",
            "investibook_secret_delivery": "box-secrets-v1",
        },
    )

    assert box["labels"]["investibook_secret_delivery"] == "box-secrets-v1"
    assert json.loads(requests[1].content)["labels"] == box["labels"]
    await http_client.aclose()


@pytest.mark.asyncio
async def test_upsert_box_secret_updates_value_and_returns_placeholder():
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(
            200,
            json={
                "secret_id": "secret-1",
                "name": "clinic",
                "placeholder": "run9secret_abc123",
            },
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    view = await client.upsert_box_secret(
        box_id="box-1",
        name="clinic",
        value="real-value",
        allowed_hosts=("clinic.pingcap.com",),
        inject_header_name="Authorization",
        existing=({"secret_id": "secret-1", "name": "clinic"},),
    )

    assert view["placeholder"] == "run9secret_abc123"
    assert requests[0].method == "PATCH"
    assert requests[0].url.raw_path.endswith(b"/boxes/box-1/secrets/secret-1")
    assert json.loads(requests[0].content) == {
        "name": "clinic",
        "value": "real-value",
        "allowed_hosts": ["clinic.pingcap.com"],
        "inject_header_name": "Authorization",
    }
    await http_client.aclose()


@pytest.mark.asyncio
async def test_upsert_box_secret_generates_placeholder_when_creating(monkeypatch):
    requests: list[httpx.Request] = []
    monkeypatch.setattr(
        "investibook.clients.run9.secrets.token_hex",
        lambda size: "abc123" if size == 16 else "unexpected",
    )

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        payload = json.loads(request.content)
        return httpx.Response(
            201,
            json={
                "secret_id": "secret-1",
                "name": payload["name"],
                "placeholder": payload["placeholder"],
            },
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    view = await client.upsert_box_secret(
        box_id="box-1",
        name="clinic",
        value="real-value",
        allowed_hosts=("clinic.pingcap.com",),
        inject_header_name="Authorization",
        existing=(),
    )

    payload = json.loads(requests[0].content)
    assert requests[0].method == "POST"
    assert payload["placeholder"] == "run9secret_abc123"
    assert view["placeholder"] == payload["placeholder"]
    await http_client.aclose()


@pytest.mark.asyncio
async def test_kill_exec_accepts_empty_success_response():
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, content=b"", request=request)

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    assert await client.kill_exec("exec-1") is None
    await http_client.aclose()


@pytest.mark.asyncio
async def test_run_command_collects_output_until_success():
    pulls = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal pulls
        if request.url.path.endswith("/background-execs"):
            return httpx.Response(200, json={"exec_id": "exec-1"}, request=request)
        pulls += 1
        if pulls == 1:
            return httpx.Response(
                200,
                content=b"first-",
                headers={
                    "X-Run9-Next-Cursor": "1",
                    "X-Run9-Exec-State": "running",
                },
                request=request,
            )
        return httpx.Response(
            200,
            content=b"second",
            headers={
                "X-Run9-Next-Cursor": "2",
                "X-Run9-Exec-State": "succeeded",
                "X-Run9-Exit-Code": "0",
            },
            request=request,
        )

    http_client = httpx.AsyncClient(
        base_url="https://run9.test", transport=httpx.MockTransport(handler)
    )
    client = Run9Client(
        base_url="https://run9.test",
        authorization="test-token",
        project_cid="project",
        client=http_client,
    )

    output = await client.run_command(
        box_id="box-1", command=["python3", "-c", "print('ok')"]
    )

    assert output == b"first-second"
    await http_client.aclose()