"""Azure Cost Management custom-dimension billing query.

Azure's request is a POST body (not a paginator like AWS), and the
response carries dimension columns under their PascalCase native
names — but with one twist: tag groupings come back under just the
tag key, not the literal string ``"Tag"``. Tests lock in:

  * Helix-canonical alias → Cost Management ``Dimension``/``Tag``
    grouping spec.
  * Filter-clause shape (And of leaf clauses, not the AWS-specific
    ``Dimensions``/``Tags`` keys).
  * Column-index resolution that handles the tag-key-as-column
    quirk so we don't drop tag dimensions from the response.
"""
from __future__ import annotations

from datetime import date
from typing import Any
from unittest.mock import AsyncMock, patch

import httpx
import pytest

from app.integrations.driver_base import CloudScopeRef
from app.integrations.drivers.azure_cloud import AzureCloudDriver

_CFG = {
    "client_id": "test-client",
    "client_secret": "test-secret",
    "tenant_id": "test-tenant",
}


def _driver() -> AzureCloudDriver:
    return AzureCloudDriver(integration_id="azure-test", config=dict(_CFG))


def test_supported_dimensions() -> None:
    dims = AzureCloudDriver.supported_billing_dimensions()
    # Subscription is the user's specifically-mentioned use case.
    assert "subscription" in dims
    assert "service" in dims
    assert "resource_group" in dims
    assert "tag" in dims  # marker for tag:KEY grouping
    # Convenience alias: ``region`` → ``ResourceLocation`` (CE-style).
    assert "region" in dims


def test_grouping_spec_dimension() -> None:
    d = _driver()
    assert d._cm_grouping_spec("subscription") == {
        "type": "Dimension", "name": "SubscriptionId",
    }
    assert d._cm_grouping_spec("service") == {
        "type": "Dimension", "name": "ServiceName",
    }
    assert d._cm_grouping_spec("region") == {
        "type": "Dimension", "name": "ResourceLocation",
    }


def test_grouping_spec_tag() -> None:
    d = _driver()
    assert d._cm_grouping_spec("tag:Project") == {
        "type": "Tag", "name": "Project",
    }


def test_grouping_spec_unknown_dimension_raises() -> None:
    d = _driver()
    with pytest.raises(ValueError, match="does not support dimension 'foobar'"):
        d._cm_grouping_spec("foobar")


def test_grouping_spec_empty_tag_key_raises() -> None:
    d = _driver()
    with pytest.raises(ValueError, match="tag dimension requires a key"):
        d._cm_grouping_spec("tag:")


def test_filter_expr_dimension_clause() -> None:
    d = _driver()
    expr = d._cm_filter_expr({"service": ["AKS"]})
    # Cost Management uses ``In`` operator, not AWS's ``Values`` list.
    assert expr == {
        "dimensions": {
            "name": "ServiceName", "operator": "In", "values": ["AKS"],
        },
    }


def test_filter_expr_tag_clause() -> None:
    d = _driver()
    expr = d._cm_filter_expr({"tag:Project": ["tidb"]})
    assert expr == {
        "tags": {"name": "Project", "operator": "In", "values": ["tidb"]},
    }


def test_filter_expr_combines_with_and() -> None:
    d = _driver()
    expr = d._cm_filter_expr({
        "service": ["AKS"],
        "tag:Project": ["tidb"],
    })
    # Order is dict-iteration order which is insertion-stable in
    # Python 3.7+; we accept whatever order shows up but require both.
    assert "and" in expr
    assert len(expr["and"]) == 2
    types = {next(iter(c)) for c in expr["and"]}
    assert types == {"dimensions", "tags"}


def test_filter_expr_empty_returns_none() -> None:
    d = _driver()
    assert d._cm_filter_expr({}) is None
    assert d._cm_filter_expr({"service": []}) is None


# ── End-to-end with mocked Cost Management response ─────────────


def _cm_response(rows: list[list[Any]], cols: list[str]) -> dict:
    return {
        "properties": {
            "columns": [{"name": n, "type": "Number"} for n in cols],
            "rows": rows,
        },
    }


class _FakeAsyncClient:
    """Mimics httpx.AsyncClient enough for the driver's POST loop."""

    def __init__(self, payloads: list[dict]) -> None:
        self._payloads = list(payloads)
        self.posts: list[tuple[str, dict]] = []

    async def __aenter__(self):
        return self

    async def __aexit__(self, *_):
        return None

    async def post(self, url: str, json: dict, headers: dict) -> Any:
        self.posts.append((url, json))
        body = self._payloads.pop(0) if self._payloads else {"properties": {}}
        return httpx.Response(200, json=body)


@pytest.mark.asyncio
async def test_subscription_grouped_query() -> None:
    """The user's specific ask: Azure cost grouped by subscription.
    Response columns come back as ``[Cost, UsageDate, SubscriptionId,
    Currency]``; the driver maps SubscriptionId back to the
    ``subscription`` alias."""
    d = _driver()
    fake = _FakeAsyncClient([_cm_response(
        rows=[
            [12.34, "20260422", "11111111-1111-1111-1111-111111111111", "USD"],
            [56.78, "20260422", "22222222-2222-2222-2222-222222222222", "USD"],
        ],
        cols=["PreTaxCost", "UsageDate", "SubscriptionId", "Currency"],
    )])
    with patch.object(d, "_token", AsyncMock(return_value="t0k3n")), \
         patch.object(d, "_http", return_value=fake):
        rows = await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["subscription"],
            scope=CloudScopeRef(
                scope_type="azure_subscription",
                external_id="33333333-3333-3333-3333-333333333333",
                display_name="prod",
            ),
        )
    assert len(rows) == 2
    assert rows[0].date == date(2026, 4, 22)
    assert rows[0].dimensions == {
        "subscription": "11111111-1111-1111-1111-111111111111",
    }
    assert rows[0].cost == 12.34
    # The POST went to the scope's subscription, not some default.
    sent_url = fake.posts[0][0]
    assert "/subscriptions/33333333-3333-3333-3333-333333333333/" in sent_url
    # Body should carry the right grouping spec.
    sent_body = fake.posts[0][1]
    assert sent_body["dataset"]["grouping"] == [
        {"type": "Dimension", "name": "SubscriptionId"},
    ]


@pytest.mark.asyncio
async def test_tag_dimension_with_filter() -> None:
    """Group by tag:Project, filter to one project value. Locks in
    that the filter clause shape (``tags.operator=In``) reaches the
    POST body intact."""
    d = _driver()
    fake = _FakeAsyncClient([_cm_response(
        # When grouping by Tag, the column header is the tag KEY
        # itself (``Project``), not the literal string ``Tag``.
        rows=[
            [12.34, "20260422", "tidb", "USD"],
        ],
        cols=["PreTaxCost", "UsageDate", "Project", "Currency"],
    )])
    with patch.object(d, "_token", AsyncMock(return_value="t0k3n")), \
         patch.object(d, "_http", return_value=fake):
        rows = await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["tag:Project"],
            scope=CloudScopeRef(
                scope_type="azure_subscription",
                external_id="11111111-1111-1111-1111-111111111111",
                display_name="prod",
            ),
            filters={"tag:Project": ["tidb"]},
        )
    assert len(rows) == 1
    assert rows[0].dimensions == {"tag:Project": "tidb"}
    sent_body = fake.posts[0][1]
    # Grouping body uses the tag key directly (``Project``).
    assert sent_body["dataset"]["grouping"] == [
        {"type": "Tag", "name": "Project"},
    ]
    # Filter body must be present, In-operator.
    assert sent_body["dataset"]["filter"] == {
        "tags": {"name": "Project", "operator": "In", "values": ["tidb"]},
    }


@pytest.mark.asyncio
async def test_no_scope_returns_empty() -> None:
    """Cost Management queries are subscription-scoped — calling
    without a scope is a no-op (the caller iterates per scope).
    Don't blow up."""
    d = _driver()
    rows = await d.list_billing_by_dimension(
        date(2026, 4, 22), date(2026, 4, 22),
        group_by=["service"],
        scope=None,
    )
    assert rows == []


@pytest.mark.asyncio
async def test_empty_group_by_rejected() -> None:
    d = _driver()
    with pytest.raises(ValueError, match="at least one dimension"):
        await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=[],
            scope=CloudScopeRef(
                scope_type="azure_subscription",
                external_id="x", display_name=None,
            ),
        )


class _FakeAsyncClientNon200:
    """Mimics httpx.AsyncClient that returns a non-200 response."""

    def __init__(self, status_code: int, body: str) -> None:
        self._status_code = status_code
        self._body = body

    async def __aenter__(self):
        return self

    async def __aexit__(self, *_):
        return None

    async def post(self, url: str, json: dict, headers: dict) -> Any:
        return httpx.Response(self._status_code, text=self._body)


@pytest.mark.asyncio
async def test_non_200_raises_for_symmetric_with_daily_and_cost_facts() -> None:
    """Round-19 fix: ``list_billing_by_dimension`` was the holdout in
    Azure's three billing paths. ``list_billing_daily`` (PR #234) and
    ``list_cost_facts`` already raise on non-200; this path silently
    returned ``[]`` on 401/403/429/5xx, masking a real Azure outage as
    "no spend on this dimension". Pin that all three paths now raise."""
    d = _driver()
    # Body intentionally includes an Authorization header echoed by a
    # misconfigured proxy WAF — verify it gets sanitised before
    # landing in the RuntimeError message (otherwise it propagates
    # into ``scheduled_job_runs.errors`` and the failure-notifier).
    fake = _FakeAsyncClientNon200(
        403,
        '{"error":"Forbidden"}\nAuthorization: Bearer secret-token-123',
    )
    with patch.object(d, "_token", AsyncMock(return_value="t0k3n")), \
         patch.object(d, "_http", return_value=fake):
        with pytest.raises(RuntimeError, match="status=403") as exc_info:
            await d.list_billing_by_dimension(
                date(2026, 4, 22), date(2026, 4, 22),
                group_by=["service"],
                scope=CloudScopeRef(
                    scope_type="azure_subscription",
                    external_id="11111111-1111-1111-1111-111111111111",
                    display_name="prod",
                ),
            )
    # Body in the error message has no newlines (control chars
    # sanitised) and stays under the 200-char cap.
    msg = str(exc_info.value)
    assert "\n" not in msg
    assert "\r" not in msg
    assert "\t" not in msg


@pytest.mark.asyncio
async def test_missing_requested_dimension_column_raises() -> None:
    """Round-19 fix: pre-fix when a requested grouping column wasn't in
    the response (mis-typed tag, removed dimension, or Cost Management
    not yet exposing it on this subscription) the driver silently
    dropped the alias from per-row output schema. The MCP tool / dashboard
    treated absence as "row has no Project tag" leading to silent
    allocation drift. Pin that the driver now raises so the caller
    knows the request shape can't be honoured."""
    d = _driver()
    # Response is missing the requested ``Project`` tag column.
    fake = _FakeAsyncClient([_cm_response(
        rows=[[12.34, "20260422", "USD"]],
        cols=["PreTaxCost", "UsageDate", "Currency"],
    )])
    with patch.object(d, "_token", AsyncMock(return_value="t0k3n")), \
         patch.object(d, "_http", return_value=fake):
        with pytest.raises(RuntimeError, match="missing requested columns"):
            await d.list_billing_by_dimension(
                date(2026, 4, 22), date(2026, 4, 22),
                group_by=["tag:Project"],
                scope=CloudScopeRef(
                    scope_type="azure_subscription",
                    external_id="11111111-1111-1111-1111-111111111111",
                    display_name="prod",
                ),
            )