"""AWS Cost Explorer custom-dimension billing query.

Production audit (2026-04-25) needed two things from this code path:

1. Pull bills sliced by ``tag:Project`` (the AWS-SG case the PR
   triple-down request specifically called out). This locks in the
   tag-key encoding that goes onto the wire and the tag-value
   stripping that comes back (``"Project$tidb"`` → ``"tidb"``).
2. Honor the ``LINKED_ACCOUNT`` filter pin only when the scope's
   ``external_id`` is a real 12-digit AWS account id — the
   org-not-in-use synthetic scope used to carry the integration
   UUID, which silently dropped every CE row.

Plus: alias mapping, 2-dimension cap, ``ValueError`` on bad
dimension, and the ``filters`` shape that the router validates.
"""
from __future__ import annotations

from datetime import date
from unittest.mock import MagicMock, patch

import pytest

from app.integrations.driver_base import (
    CloudScopeRef,
)
from app.integrations.drivers.aws_cloud import AwsCloudDriver

_CFG = {
    "access_key_id": "AKIA-test",
    "secret_access_key": "test-sk",
    "default_region": "ap-southeast-1",
}


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


# ── _ce_group_spec: alias → CE GroupBy translation ──────────────


def test_group_spec_dimension_alias() -> None:
    d = _driver()
    assert d._ce_group_spec("service") == {"Type": "DIMENSION", "Key": "SERVICE"}
    # ``account`` is a UI-friendly alias for ``LINKED_ACCOUNT``.
    assert d._ce_group_spec("account") == {"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"}
    assert d._ce_group_spec("linked_account") == {"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"}


def test_group_spec_tag_key_round_trips() -> None:
    """``tag:Project`` is the canonical wire form. The CE GroupBy
    payload requires the literal tag key — case preserved — under
    a separate ``Type=TAG`` shape."""
    d = _driver()
    assert d._ce_group_spec("tag:Project") == {"Type": "TAG", "Key": "Project"}
    # Spaces around the colon should not survive — admins paste
    # casually and the CE API rejects whitespace.
    assert d._ce_group_spec("tag: CostCenter") == {"Type": "TAG", "Key": "CostCenter"}


def test_group_spec_unknown_dimension_raises() -> None:
    """A typo / nonexistent dimension must be a hard error so the
    router can return 400 instead of silently grouping by garbage."""
    d = _driver()
    with pytest.raises(ValueError, match="does not support dimension 'foobar'"):
        d._ce_group_spec("foobar")


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


def test_group_spec_aws_principal_tag_namespace_rejected_with_hint() -> None:
    """Catch the silent-fail trap admins easily walk into — AWS docs
    use ``aws:PrincipalTag/<Key>`` heavily in IAM policy ``Condition``
    examples, but CE GroupBy[Type=TAG] does NOT support that namespace.
    AWS does not error; CE returns an empty string for every row's
    tag value, collapsing all spend into a single bucket. The admin
    sees a misleading 0-spend chart with no indication anything is
    wrong. Helix rejects up-front with a 400 and a hint that points
    at the correct ``iamPrincipal/<Key>`` namespace (the literal key
    Billing Console surfaces once the cost-allocation tag is activated).

    Mirrors the 2026-05-12 Bedrock production audit where 100% of
    Bedrock cost landed in ``(unallocated)`` because the driver had
    used the wrong namespace internally. This guard catches the
    parallel admin-typed mistake at the API surface.
    """
    d = _driver()
    with pytest.raises(ValueError, match="iamPrincipal/Owner") as exc:
        d._ce_group_spec("tag:aws:PrincipalTag/Owner")
    # Error message must explicitly mention the correct alternative
    # so the admin doesn't have to read the source code to figure
    # out what to type.
    assert "iamPrincipal/Owner" in str(exc.value)
    assert "aws:PrincipalTag/Owner" in str(exc.value)


def test_group_spec_iamprincipal_namespace_accepted() -> None:  # noqa: N802
    """The correct cost-allocation namespace for IAM principal tags
    is ``iamPrincipal/<Key>`` (no ``aws:`` prefix). It passes through
    cleanly so the admin can drill into per-IAM-user cost when the
    activated cost-allocation tag covers that dimension."""
    d = _driver()
    assert d._ce_group_spec("tag:iamPrincipal/Owner") == {
        "Type": "TAG", "Key": "iamPrincipal/Owner",
    }


def test_supported_dimensions_includes_tag_marker() -> None:
    """The router checks ``"tag" in supported_dimensions()`` to know
    whether ``tag:KEY`` requests should be allowed for this provider."""
    dims = AwsCloudDriver.supported_billing_dimensions()
    assert "tag" in dims
    assert "service" in dims
    assert "linked_account" in dims


# ── _ce_filter_expr: scope pin + filter map ─────────────────────


def test_filter_pins_real_aws_account_id() -> None:
    """Real 12-digit account id → CE LINKED_ACCOUNT pin. This is the
    happy path that started working the moment scope sync stopped
    using the integration UUID as external_id."""
    d = _driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
        display_name="prod",
    )
    expr = d._ce_filter_expr(scope, {})
    assert expr == {"Dimensions": {"Key": "LINKED_ACCOUNT", "Values": ["123456789012"]}}


def test_filter_skips_pin_when_external_id_isnt_real_account() -> None:
    """Defensive: if a legacy scope still carries a UUID-shaped
    external_id, don't add a guaranteed-empty filter clause to CE.
    (The scope-sync fix backfills these but a stale row could
    still flow through during the deploy window.)"""
    d = _driver()
    scope = CloudScopeRef(
        scope_type="aws_account",
        external_id="a99dd95d-9734-4a7b-942a-a2836e9448a9",  # UUID
        display_name="legacy",
    )
    expr = d._ce_filter_expr(scope, {})
    assert expr is None


def test_filter_combines_scope_with_dimension_filter() -> None:
    """Scope (LINKED_ACCOUNT pin) AND-ed with explicit dimension
    filter. AWS CE caller sees one ``And`` clause whose contents
    flow back into the GET, not separate top-level filters."""
    d = _driver()
    scope = CloudScopeRef(
        scope_type="aws_account", external_id="123456789012",
        display_name="prod",
    )
    expr = d._ce_filter_expr(scope, {"service": ["EC2 - Other", "S3"]})
    assert expr == {"And": [
        {"Dimensions": {"Key": "LINKED_ACCOUNT", "Values": ["123456789012"]}},
        {"Dimensions": {"Key": "SERVICE", "Values": ["EC2 - Other", "S3"]}},
    ]}


def test_filter_tag_clause() -> None:
    d = _driver()
    expr = d._ce_filter_expr(None, {"tag:Project": ["tidb", "tikv"]})
    assert expr == {"Tags": {"Key": "Project", "Values": ["tidb", "tikv"]}}


def test_filter_empty_returns_none() -> None:
    """No scope + no filters = no Filter on the CE request."""
    d = _driver()
    assert d._ce_filter_expr(None, {}) is None
    # Empty values for a dimension should also drop the clause —
    # ``filter_dim=[]`` from the router shouldn't accidentally pin
    # to "no values match".
    assert d._ce_filter_expr(None, {"service": []}) is None


# ── list_billing_by_dimension: end-to-end shape ─────────────────


def _ce_response(date_str: str, groups: list[tuple[list[str], str]]) -> dict:
    return {
        "ResultsByTime": [{
            "TimePeriod": {"Start": date_str, "End": date_str},
            "Groups": [
                {
                    "Keys": list(keys),
                    "Metrics": {"UnblendedCost": {"Amount": amt, "Unit": "USD"}},
                }
                for keys, amt in groups
            ],
        }],
    }


@pytest.mark.asyncio
async def test_two_dim_query_with_tag() -> None:
    """The exact shape the user asked for: AWS-SG bill grouped by
    service AND tag:Project. CE returns ``"Project$tidb"`` for tag
    values; the driver strips the prefix so the API payload carries
    just ``"tidb"`` (UI doesn't need to know CE's wire encoding)."""
    d = _driver()
    ce = MagicMock()
    # Real driver uses ``_ce_paged`` which calls
    # ``ce.get_cost_and_usage(**kwargs)`` directly (CE has no boto3
    # paginator — see _ce_paged docstring). Stub that, not paginator.
    ce.get_cost_and_usage.return_value = _ce_response("2026-04-22", [
        (["EC2 - Other", "Project$tidb"], "12.50"),
        (["EC2 - Other", "Project$tikv"], "8.00"),
        (["S3", "Project$tidb"], "0.34"),
    ])
    with patch.object(d, "_boto_client", return_value=ce):
        rows = await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["service", "tag:Project"],
            scope=CloudScopeRef(
                scope_type="aws_account",
                external_id="123456789012",
                display_name="prod",
            ),
        )
    assert len(rows) == 3
    assert rows[0].dimensions == {"service": "EC2 - Other", "tag:Project": "tidb"}
    assert rows[0].cost == 12.50
    assert rows[0].currency == "USD"
    assert rows[1].dimensions["tag:Project"] == "tikv"
    assert rows[2].dimensions["service"] == "S3"
    # The kwargs CE was called with should carry the LINKED_ACCOUNT
    # filter (proving the scope_id round-trips into the request).
    called_kwargs = ce.get_cost_and_usage.call_args.kwargs
    assert called_kwargs["GroupBy"] == [
        {"Type": "DIMENSION", "Key": "SERVICE"},
        {"Type": "TAG", "Key": "Project"},
    ]
    assert called_kwargs["Filter"] == {
        "Dimensions": {"Key": "LINKED_ACCOUNT", "Values": ["123456789012"]},
    }


@pytest.mark.asyncio
async def test_three_dim_query_rejected_at_two_dim_cap() -> None:
    """AWS Cost Explorer caps GROUP BY at 2. We refuse the request
    rather than silently dropping the third dimension."""
    d = _driver()
    with pytest.raises(ValueError, match="at most 2"):
        await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["service", "region", "usage_type"],
        )


@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=[],
        )


@pytest.mark.asyncio
async def test_filter_dimension_key_aliased_lowercase() -> None:
    """``filters={"service":["EC2 - Other"]}`` with lowercase alias
    must be translated to CE's PascalCase ``SERVICE`` before the
    request goes out — UI doesn't speak AWS dialect."""
    d = _driver()
    ce = MagicMock()
    ce.get_cost_and_usage.return_value = _ce_response("2026-04-22", [
        (["EC2 - Other"], "12.50"),
    ])
    with patch.object(d, "_boto_client", return_value=ce):
        await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["service"],
            filters={"service": ["EC2 - Other"]},
        )
    called_kwargs = ce.get_cost_and_usage.call_args.kwargs
    assert called_kwargs["Filter"] == {
        "Dimensions": {"Key": "SERVICE", "Values": ["EC2 - Other"]},
    }


@pytest.mark.asyncio
async def test_dimension_response_without_tag_prefix() -> None:
    """Non-tag dimensions return raw key as-is — no ``$`` prefix to
    strip. Locks in that the prefix-strip logic doesn't accidentally
    eat ``$`` characters that legitimately appear in service names."""
    d = _driver()
    ce = MagicMock()
    ce.get_cost_and_usage.return_value = _ce_response("2026-04-22", [
        (["AWS - Other"], "1.23"),
    ])
    with patch.object(d, "_boto_client", return_value=ce):
        rows = await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["service"],
        )
    assert rows[0].dimensions == {"service": "AWS - Other"}


@pytest.mark.asyncio
async def test_missing_boto3_raises_for_symmetric_with_daily_and_cost_facts() -> None:
    """Round-19 fix: ``list_billing_by_dimension`` was the holdout in
    AWS's three billing paths. ``list_billing_daily`` and
    ``list_cost_facts`` already raise on missing boto3 (deployment
    defect); this path silently returned ``[]`` masking the missing
    dependency. Pin that all three paths now raise with a clear
    deployment-defect message."""
    d = _driver()
    with patch(
        "app.integrations.drivers.aws_cloud._BOTO3_AVAILABLE", False,
    ), pytest.raises(RuntimeError, match="boto3 not importable"):
        await d.list_billing_by_dimension(
            date(2026, 4, 22), date(2026, 4, 22),
            group_by=["service"],
        )