"""Bailian HTTP client signing + idempotency.

These tests don't hit Aliyun. We patch ``httpx.AsyncClient.request`` to
capture the headers + body the signer produces, and verify:

  1. ``Authorization`` header is well-formed (ACS3-HMAC-SHA256, has all
     required signed-headers, signature is hex).
  2. ``CreateUser`` then ``CreateUser`` again (same user_name) produces
     EntityAlreadyExists handling — service falls back to GetUser and
     marks ``created=False``.
  3. ``attach_policy`` swallows EntityAlreadyExists and reports
     already_attached=True instead of raising.
"""
from __future__ import annotations

from unittest.mock import AsyncMock, patch

import httpx
import pytest

from app.core.exceptions import AppException
from app.core.pagination import PaginationGuardError
from app.domains.bailian.client import BailianClient

_CREDS = {
    "access_key_id": "LTAItest",
    "access_key_secret": "secret-test",
    # Bailian only runs on the 海外 tenant — the client defaults to
    # alicloud_intl + ap-southeast-1.
    "_provider": "alicloud_intl",
}


@pytest.mark.asyncio
async def test_workspace_pagination_preserves_order(monkeypatch):
    client = BailianClient(_CREDS)
    pages = iter([
        {"workspaces": [{"workspaceId": "1"}], "nextToken": "A"},
        {"workspaces": [{"workspaceId": "2"}], "nextToken": None},
    ])
    monkeypatch.setattr(client, "_acs3_call", AsyncMock(side_effect=pages))
    rows = await client.list_workspaces()
    assert [row["workspace_id"] for row in rows] == ["1", "2"]


@pytest.mark.asyncio
async def test_workspace_pagination_rejects_repeated_token(monkeypatch):
    client = BailianClient(_CREDS)
    call = AsyncMock(return_value={
        "workspaces": [{"workspaceId": "1"}],
        "nextToken": "workspace-secret",
    })
    monkeypatch.setattr(client, "_acs3_call", call)
    with pytest.raises(PaginationGuardError) as exc_info:
        await client.list_workspaces()
    assert call.await_count == 2
    assert "workspace-secret" not in str(exc_info.value)


@pytest.mark.asyncio
async def test_workspace_pagination_rejects_oversized_page(monkeypatch):
    client = BailianClient(_CREDS)
    monkeypatch.setattr(
        "app.domains.bailian.client._MAX_WORKSPACE_ITEMS", 1,
    )
    call = AsyncMock(return_value={
        "workspaces": [{"workspaceId": "1"}, {"workspaceId": "2"}],
    })
    monkeypatch.setattr(client, "_acs3_call", call)
    with pytest.raises(PaginationGuardError, match="max_items_exceeded"):
        await client.list_workspaces()
    assert call.await_count == 1


@pytest.mark.asyncio
async def test_workspace_pagination_rejects_non_string_token(monkeypatch):
    client = BailianClient(_CREDS)
    monkeypatch.setattr(client, "_acs3_call", AsyncMock(return_value={
        "workspaces": [], "nextToken": {"not": "opaque"},
    }))
    with pytest.raises(PaginationGuardError, match="malformed_response"):
        await client.list_workspaces()


@pytest.mark.asyncio
async def test_api_key_pagination_rejects_alternating_token(monkeypatch):
    client = BailianClient(_CREDS)
    pages = iter([
        {"apiKeys": [{"apiKeyId": "1"}], "nextToken": "secret-A"},
        {"apiKeys": [{"apiKeyId": "2"}], "nextToken": "secret-B"},
        {"apiKeys": [{"apiKeyId": "3"}], "nextToken": "secret-A"},
    ])
    call = AsyncMock(side_effect=pages)
    monkeypatch.setattr(client, "_acs3_call", call)
    with pytest.raises(PaginationGuardError) as exc_info:
        await client.list_api_keys(workspace_id="ws")
    assert call.await_count == 3
    assert "secret-A" not in str(exc_info.value)


@pytest.mark.asyncio
async def test_api_key_pagination_rejects_oversized_page(monkeypatch):
    client = BailianClient(_CREDS)
    monkeypatch.setattr(
        "app.domains.bailian.client._MAX_API_KEY_ITEMS", 1,
    )
    call = AsyncMock(return_value={
        "apiKeys": [{"apiKeyId": "1"}, {"apiKeyId": "2"}],
    })
    monkeypatch.setattr(client, "_acs3_call", call)
    with pytest.raises(PaginationGuardError, match="max_items_exceeded"):
        await client.list_api_keys(workspace_id="ws")
    assert call.await_count == 1


@pytest.mark.asyncio
async def test_api_key_pagination_rejects_non_list_items(monkeypatch):
    client = BailianClient(_CREDS)
    monkeypatch.setattr(client, "_acs3_call", AsyncMock(return_value={
        "apiKeys": {"apiKeyId": "not-a-list"},
    }))
    with pytest.raises(PaginationGuardError, match="malformed_response"):
        await client.list_api_keys(workspace_id="ws")


@pytest.mark.asyncio
async def test_instance_bill_pagination_rejects_repeated_token(monkeypatch):
    client = BailianClient(_CREDS)
    call = AsyncMock(return_value={
        "Data": {"Items": {"Item": []}, "NextToken": "billing-secret"},
    })
    monkeypatch.setattr(client, "_acs3_call", call)
    with pytest.raises(PaginationGuardError) as exc_info:
        await client.query_instance_bill_daily(
            start_date="2026-06-01", end_date="2026-06-01",
            cny_to_usd_rate=0.14,
        )
    assert call.await_count == 2
    assert "billing-secret" not in str(exc_info.value)


@pytest.mark.asyncio
async def test_instance_bill_pagination_rejects_oversized_page(monkeypatch):
    client = BailianClient(_CREDS)
    monkeypatch.setattr(
        "app.domains.bailian.client._MAX_BILLING_ITEMS_PER_DAY", 1,
    )
    call = AsyncMock(return_value={
        "Data": {"Items": {"Item": [{}, {}]}},
    })
    monkeypatch.setattr(client, "_acs3_call", call)
    with pytest.raises(PaginationGuardError, match="max_items_exceeded"):
        await client.query_instance_bill_daily(
            start_date="2026-06-01", end_date="2026-06-01",
            cny_to_usd_rate=0.14,
        )
    assert call.await_count == 1


@pytest.mark.asyncio
async def test_instance_bill_pagination_rejects_non_list_items(monkeypatch):
    client = BailianClient(_CREDS)
    monkeypatch.setattr(client, "_acs3_call", AsyncMock(return_value={
        "Data": {"Items": {"Item": {"not": "a-list"}}},
    }))
    with pytest.raises(PaginationGuardError, match="malformed_response"):
        await client.query_instance_bill_daily(
            start_date="2026-06-01", end_date="2026-06-01",
            cny_to_usd_rate=0.14,
        )


def _mock_response(status_code: int, json_body):
    resp = httpx.Response(
        status_code=status_code,
        json=json_body,
        request=httpx.Request("POST", "https://example.com/"),
    )
    return resp


@pytest.mark.asyncio
async def test_create_access_key_returns_pair():
    client = BailianClient(_CREDS)

    async def fake_request(method, url, **kwargs):
        return _mock_response(200, {"AccessKey": {
            "AccessKeyId": "AK-new", "AccessKeySecret": "AKS-new",
        }})

    with patch.object(httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request)):
        result = await client.create_access_key(user_name="bailian-tester")
    assert result["access_key_id"] == "AK-new"
    assert result["access_key_secret"] == "AKS-new"


@pytest.mark.asyncio
async def test_create_access_key_empty_response_raises():
    client = BailianClient(_CREDS)

    async def fake_request(method, url, **kwargs):
        return _mock_response(200, {"AccessKey": {}})

    with patch.object(httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request)):
        with pytest.raises(AppException) as exc_info:
            await client.create_access_key(user_name="bailian-tester")
    assert exc_info.value.code == "RAM_AK_MINT_FAILED"


@pytest.mark.asyncio
async def test_attach_policy_already_attached_idempotent():
    client = BailianClient(_CREDS)

    async def fake_request(method, url, **kwargs):
        return _mock_response(
            400,
            {"Code": "EntityAlreadyExists.User.Policy", "Message": "already attached"},
        )

    with patch.object(httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request)):
        result = await client.attach_policy(
            user_name="bailian-tester", policy_name="AliyunBailianFullAccess",
        )
    assert result == {"already_attached": True}


@pytest.mark.asyncio
async def test_no_credentials_raises_at_construction():
    with pytest.raises(AppException) as exc_info:
        BailianClient({})
    assert exc_info.value.code == "BAILIAN_NO_CREDENTIALS"


@pytest.mark.asyncio
async def test_create_api_key_returns_sk_value():
    """``create_api_key`` calls ModelStudio CreateApiKey (POST
    /modelstudio/apikeys) and exposes the plaintext sk-... that's only
    revealed once. Captured URL + body so we don't accidentally call
    the legacy CN bailian RPC product.
    """
    client = BailianClient(_CREDS)
    captured: list[dict] = []

    async def fake_request(method, url, **kwargs):
        captured.append({
            "method": method,
            "url": url,
            "headers": kwargs.get("headers"),
            "content": kwargs.get("content"),
        })
        return _mock_response(200, {
            "apiKey": {
                "apiKeyId": "ak-real-1234",
                "apiKeyValue": "sk-real0987654321deadbeef",
                "workspaceId": "ws-x",
                "createdBy": "subuser-x",
                "disabled": False,
                "gmtCreate": 1747500000000,
                "description": "test",
            },
        })

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        result = await client.create_api_key(
            workspace_id="ws-x", description="test",
        )

    assert result["api_key_id"] == "ak-real-1234"
    assert result["api_key_value"] == "sk-real0987654321deadbeef"
    assert result["workspace_id"] == "ws-x"
    assert captured[0]["method"] == "POST"
    assert "/modelstudio/apikeys" in captured[0]["url"]
    assert captured[0]["headers"]["x-acs-action"] == "CreateApiKey"
    assert captured[0]["headers"]["x-acs-version"] == "2026-02-10"
    assert captured[0]["headers"]["host"] == "modelstudio.ap-southeast-1.aliyuncs.com"


@pytest.mark.asyncio
async def test_create_api_key_empty_response_raises():
    """Aliyun has been known to return 200 with an empty body on
    quota-exceeded / region-mismatch edge cases. The wrapper must
    refuse rather than silently writing an empty sk- to the bundle.
    """
    client = BailianClient(_CREDS)

    async def fake_request(method, url, **kwargs):
        return _mock_response(200, {"apiKey": {}})

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        with pytest.raises(AppException) as exc_info:
            await client.create_api_key(workspace_id="ws-x")
    assert exc_info.value.code == "BAILIAN_CREATE_APIKEY_FAILED"


@pytest.mark.asyncio
async def test_delete_api_key_idempotent_on_404():
    """The revoke retry path must not bomb on a key already removed
    manually in the Bailian console. NotFound / NotExist → success
    with ``already_absent=True``.
    """
    client = BailianClient(_CREDS)

    async def fake_request(method, url, **kwargs):
        return _mock_response(
            404,
            {"Code": "ApiKey.NotFound", "Message": "API key not found"},
        )

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        result = await client.delete_api_key(
            api_key_id="ak-missing", workspace_id="ws-x",
        )
    assert result["deleted"] is True
    assert result["already_absent"] is True


@pytest.mark.asyncio
async def test_delete_api_key_unknown_error_propagates():
    """A genuine Aliyun error (5xx / signature mismatch / quota) must
    propagate so the admin sees a real failure on the revoke audit row,
    not a phantom success.
    """
    client = BailianClient(_CREDS)

    async def fake_request(method, url, **kwargs):
        return _mock_response(
            500,
            {"Code": "InternalError", "Message": "bailian backend down"},
        )

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        with pytest.raises(AppException) as exc_info:
            await client.delete_api_key(api_key_id="ak-x")
    assert exc_info.value.code == "ALIYUN_API_ERROR"


@pytest.mark.asyncio
async def test_alicloud_intl_routes_to_singapore_hosts():
    """list_workspaces uses the ROA-style intl ModelStudio API
    (modelstudio.ap-southeast-1, GET /modelstudio/workspaces, version
    2026-02-10) — NOT the CN bailian RPC product. BSS billing stays
    on business.ap-southeast-1.aliyuncs.com (the intl BSS host).
    Calling the CN BSS host from a 海外 AK returns ``NotApplicable``,
    so this routing is the difference between billing data showing up
    and the page going silently empty.
    """
    creds = {
        "access_key_id": "LTAIintl",
        "access_key_secret": "secret-intl",
        "_provider": "alicloud_intl",
    }
    client = BailianClient(creds)
    captured: list[dict] = []
    responses = [
        {"workspaces": [], "nextToken": None},
        {"Data": {"Items": {"Item": []}, "NextToken": None}},
    ]

    async def fake_request(method, url, **kwargs):
        captured.append({"url": url, "headers": kwargs.get("headers")})
        return _mock_response(200, responses[len(captured) - 1])

    with patch.object(httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request)):
        await client.list_workspaces()
        await client.query_instance_bill_daily(
            start_date="2026-04-28",
            end_date="2026-04-28",
            cny_to_usd_rate=0.14,
        )

    assert captured[0]["headers"]["host"] == "modelstudio.ap-southeast-1.aliyuncs.com"
    assert "/modelstudio/workspaces" in captured[0]["url"]
    assert captured[0]["headers"]["x-acs-version"] == "2026-02-10"
    assert captured[1]["headers"]["host"] == "business.ap-southeast-1.aliyuncs.com"


@pytest.mark.asyncio
async def test_instance_bill_per_api_key_and_model():
    """query_instance_bill_daily reads BSS ``DescribeInstanceBill`` (DAILY),
    keeps ONLY Bailian (ProductCode='sfm') consumption rows, parses the
    ``InstanceID`` composite ``apiKeyId;workspaceId;model;tokenType;channel``
    into per-(day × account × API key × model) rows, folds token-types of the
    same model together, and excludes adjustment / non-Bailian rows.
    """
    client = BailianClient(_CREDS)
    # DescribeInstanceBill wraps Items as ``{"Item": [...]}``. Two token-type
    # rows for the same (key, model) must fold into ONE output row.
    items = [
        {"ProductCode": "sfm", "ProductName": "Alibaba Cloud Model Studio",
         "Currency": "USD", "OwnerID": "5464160084398809",
         "BillAccountName": "sa@pingcap.com",
         "InstanceID": "132245;ws-8a6e;qwen3.6-plus;context_0-256k_input_token;;0",
         "PretaxGrossAmount": "10.00", "InvoiceDiscount": "7.00",
         "PretaxAmount": "3.00"},
        {"ProductCode": "sfm", "ProductName": "Alibaba Cloud Model Studio",
         "Currency": "USD", "OwnerID": "5464160084398809",
         "BillAccountName": "sa@pingcap.com",
         "InstanceID": "132245;ws-8a6e;qwen3.6-plus;context_0-256k_output_token;;0",
         "PretaxGrossAmount": "6.00", "InvoiceDiscount": "5.00",
         "PretaxAmount": "1.00"},
        {"ProductCode": "sfm", "ProductName": "Alibaba Cloud Model Studio",
         "Currency": "USD", "OwnerID": "5962436752113096",
         "BillAccountName": "sa@pingcap.com",
         "InstanceID": "164964;llm-tbe;qwen-max;output_token;bailian-claude-code-proxy;0",
         "PretaxGrossAmount": "2.00", "InvoiceDiscount": "1.50",
         "PretaxAmount": "0.50"},
        # Adjustment (调账): gross<=0 → excluded.
        {"ProductCode": "sfm", "ProductName": "Alibaba Cloud Model Studio",
         "Currency": "USD", "InstanceID": "x;y;z;t;;0",
         "PretaxGrossAmount": "0", "PretaxAmount": "-9.00"},
        # Non-Bailian commodity → excluded.
        {"ProductCode": "ecs", "ProductName": "Elastic Compute Service",
         "Currency": "USD", "PretaxGrossAmount": "100.00",
         "PretaxAmount": "80.00"},
    ]
    resp = {"Data": {"Items": {"Item": items}, "NextToken": None}}

    async def fake_request(method, url, **kwargs):
        return _mock_response(200, resp)

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        rows = await client.query_instance_bill_daily(
            start_date="2026-06-16",
            end_date="2026-06-16",  # one day → one DescribeInstanceBill call
            cny_to_usd_rate=0.14,
        )

    by_key = {(r["bailian_api_key_id"], r["model_name"]): r for r in rows}
    assert len(rows) == 2  # two (key, model) groups; adjustment + ecs excluded
    # qwen3.6-plus on key 132245: two token-types folded → gross 16, net 4.
    plus = by_key[("132245", "qwen3.6-plus")]
    assert plus["usage_date"] == "2026-06-16"
    assert plus["aliyun_account_id"] == "5464160084398809"
    assert plus["workspace_id"] == "ws-8a6e"
    assert plus["cost_usd_gross"] == pytest.approx(16.00)
    assert plus["cost_usd_after_discount"] == pytest.approx(4.00)  # 16 − 12
    assert plus["cost_usd"] == pytest.approx(4.00)                 # net 3 + 1
    assert plus["cost_cny"] == pytest.approx(4.00 / 0.14, abs=0.5)  # USD/rate
    # qwen-max on key 164964 carries the invocation channel.
    mx = by_key[("164964", "qwen-max")]
    assert mx["aliyun_account_id"] == "5962436752113096"
    assert mx["channels"] == ["bailian-claude-code-proxy"]
    assert mx["cost_usd"] == pytest.approx(0.50)


@pytest.mark.asyncio
async def test_instance_bill_currency_aware():
    """CNY-billed Bailian rows convert cost_usd = native × rate."""
    client = BailianClient(_CREDS)
    items = [
        {"ProductCode": "sfm", "ProductName": "百炼", "Currency": "CNY",
         "OwnerID": "5962436752113096",
         "InstanceID": "999;ws-cn;qwen-plus;input_token;;0",
         "PretaxGrossAmount": "100.00", "InvoiceDiscount": "30.00",
         "PretaxAmount": "70.00"},
    ]
    resp = {"Data": {"Items": {"Item": items}, "NextToken": None}}

    async def fake_request(method, url, **kwargs):
        return _mock_response(200, resp)

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        rows = await client.query_instance_bill_daily(
            start_date="2026-05-01", end_date="2026-05-01", cny_to_usd_rate=0.14,
        )

    assert len(rows) == 1
    r = rows[0]
    assert r["cost_cny_gross"] == pytest.approx(100.00)
    assert r["cost_cny_after_discount"] == pytest.approx(70.00)  # 100 − 30
    assert r["cost_cny"] == pytest.approx(70.00)
    assert r["cost_usd"] == pytest.approx(70.00 * 0.14)   # CNY → USD ×rate
    assert r["cost_cny_gross"] == pytest.approx(100.00)
    assert r["cost_usd_gross"] == pytest.approx(100.00 * 0.14)


@pytest.mark.asyncio
async def test_instance_bill_mixed_currency_in_one_slot_raises():
    """Two line items that fall into the SAME (date, account, key, ws, model)
    slot but carry different currencies must fail loudly — native amounts are
    summed before a single conversion, so silently keeping the first-seen
    currency would misstate spend (CNY↔USD is a ~7x error)."""
    client = BailianClient(_CREDS)
    items = [
        {"ProductCode": "sfm", "ProductName": "百炼", "Currency": "USD",
         "OwnerID": "5962436752113096",
         "InstanceID": "132245;ws-8a6e;qwen-plus;input_token;;0",
         "PretaxGrossAmount": "10.00", "InvoiceDiscount": "0", "PretaxAmount": "10.00"},
        {"ProductCode": "sfm", "ProductName": "百炼", "Currency": "CNY",
         "OwnerID": "5962436752113096",
         "InstanceID": "132245;ws-8a6e;qwen-plus;output_token;;0",
         "PretaxGrossAmount": "20.00", "InvoiceDiscount": "0", "PretaxAmount": "20.00"},
    ]
    resp = {"Data": {"Items": {"Item": items}, "NextToken": None}}

    async def fake_request(method, url, **kwargs):
        return _mock_response(200, resp)

    with patch.object(
        httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request),
    ):
        with pytest.raises(RuntimeError, match="spans currencies"):
            await client.query_instance_bill_daily(
                start_date="2026-05-01", end_date="2026-05-01",
                cny_to_usd_rate=0.14,
            )


@pytest.mark.asyncio
async def test_acs3_signature_headers_present():
    """The signer produces an Authorization header with ACS3-HMAC-SHA256
    plus the standard signed-headers list. We capture the kwargs and
    inspect.
    """
    client = BailianClient(_CREDS)
    captured: dict = {}

    async def fake_request(method, url, **kwargs):
        captured["method"] = method
        captured["url"] = url
        captured["headers"] = kwargs.get("headers")
        return _mock_response(200, {"workspaces": [], "nextToken": None})

    with patch.object(httpx.AsyncClient, "request", new=AsyncMock(side_effect=fake_request)):
        await client.list_workspaces()

    headers = captured["headers"]
    assert headers["host"] == "modelstudio.ap-southeast-1.aliyuncs.com"
    auth = headers["authorization"]
    assert auth.startswith("ACS3-HMAC-SHA256 Credential=LTAItest")
    assert "SignedHeaders=" in auth
    assert "Signature=" in auth
    assert headers["x-acs-action"] == "ListWorkspaces"
    assert headers["x-acs-version"] == "2026-02-10"