"""Unit tests for the AWS cloud driver — Cost Explorer billing paths.

Cost Explorer's ``GetCostAndUsage`` is NOT in boto3's paginator
registry, so calling ``ce.get_paginator("get_cost_and_usage")``
raises ``OperationNotPageableError``. The driver drives the
``NextPageToken`` cursor manually through ``_ce_paged``. These tests
cover the pagination + per-service breakdown + error fallback paths.

Everything else the AWS driver does (scopes, grants, IAM revoke,
route53 sync) is covered by the broader integration suite.
"""
from __future__ import annotations

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

import pytest

from app.integrations.driver_base import CloudAccessContext, CloudScopeRef
from app.integrations.drivers import aws_cloud
from app.integrations.drivers.aws_cloud import AwsCloudDriver, _ce_paged

_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=_CFG)


# ── _ce_paged helper ────────────────────────────────────────────


def test_ce_paged_stops_when_no_next_page_token() -> None:
    ce = MagicMock()
    ce.get_cost_and_usage.return_value = {"ResultsByTime": [{"x": 1}]}
    pages = list(_ce_paged(ce, TimePeriod={"Start": "a", "End": "b"}))
    assert pages == [{"ResultsByTime": [{"x": 1}]}]
    assert ce.get_cost_and_usage.call_count == 1


def test_ce_paged_follows_next_page_token() -> None:
    ce = MagicMock()
    ce.get_cost_and_usage.side_effect = [
        {"ResultsByTime": [{"page": 1}], "NextPageToken": "tok-2"},
        {"ResultsByTime": [{"page": 2}], "NextPageToken": "tok-3"},
        {"ResultsByTime": [{"page": 3}]},  # no more cursor
    ]
    pages = list(_ce_paged(ce, Granularity="DAILY"))
    assert [p["ResultsByTime"][0]["page"] for p in pages] == [1, 2, 3]
    # The cursor is passed back in subsequent calls.
    calls = ce.get_cost_and_usage.call_args_list
    assert "NextPageToken" not in calls[0].kwargs
    assert calls[1].kwargs["NextPageToken"] == "tok-2"
    assert calls[2].kwargs["NextPageToken"] == "tok-3"


def test_ce_paged_rejects_repeated_token() -> None:
    ce = MagicMock()
    ce.get_cost_and_usage.return_value = {
        "ResultsByTime": [{"page": 1}],
        "NextPageToken": "stuck",
    }

    with pytest.raises(RuntimeError, match="repeated NextPageToken"):
        list(_ce_paged(ce, Granularity="DAILY"))

    assert ce.get_cost_and_usage.call_count == 2


def test_ce_paged_fails_loudly_at_page_cap(monkeypatch) -> None:
    monkeypatch.setattr(aws_cloud, "_MAX_CE_PAGES", 2)
    ce = MagicMock()
    ce.get_cost_and_usage.side_effect = [
        {"ResultsByTime": [{"page": 1}], "NextPageToken": "tok-2"},
        {"ResultsByTime": [{"page": 2}], "NextPageToken": "tok-3"},
    ]

    with pytest.raises(RuntimeError, match="exceeded MAX_CE_PAGES=2"):
        list(_ce_paged(ce, Granularity="DAILY"))

    assert ce.get_cost_and_usage.call_count == 2


# ── list_billing_daily: end-to-end with pagination ──────────────


def _mock_ce_client(pages: list[dict]) -> MagicMock:
    ce = MagicMock()
    ce.get_cost_and_usage.side_effect = list(pages)
    return ce


@pytest.mark.asyncio
async def test_list_billing_daily_merges_across_pages() -> None:
    """First pass (linked-account group) spans two pages; second pass
    (per-service breakdown for one scope) spans two pages. Both must
    merge into the final result without dropping rows."""
    acct = "123456789012"
    # Pass 1: LINKED_ACCOUNT group, 2 pages, 1 bucket per page for 2 dates
    pass1_p1 = {
        "ResultsByTime": [{
            "TimePeriod": {"Start": "2026-04-01", "End": "2026-04-02"},
            "Groups": [{
                "Keys": [acct],
                "Metrics": {"UnblendedCost": {"Amount": "100.5", "Unit": "USD"}},
            }],
        }],
        "NextPageToken": "p1-2",
    }
    pass1_p2 = {
        "ResultsByTime": [{
            "TimePeriod": {"Start": "2026-04-02", "End": "2026-04-03"},
            "Groups": [{
                "Keys": [acct],
                "Metrics": {"UnblendedCost": {"Amount": "75.25", "Unit": "USD"}},
            }],
        }],
    }
    # Pass 2: SERVICE group filtered to the scope, also 2 pages.
    pass2_p1 = {
        "ResultsByTime": [{
            "TimePeriod": {"Start": "2026-04-01", "End": "2026-04-02"},
            "Groups": [{
                "Keys": ["EC2"],
                "Metrics": {"UnblendedCost": {"Amount": "60.5", "Unit": "USD"}},
            }],
        }],
        "NextPageToken": "p2-2",
    }
    pass2_p2 = {
        "ResultsByTime": [{
            "TimePeriod": {"Start": "2026-04-01", "End": "2026-04-02"},
            "Groups": [{
                "Keys": ["S3"],
                "Metrics": {"UnblendedCost": {"Amount": "40.0", "Unit": "USD"}},
            }],
        }],
    }
    # 双口径白名单遍(RECORD_TYPE=Usage)插在 totals 之后、breakdown
    # 之前——单页,给出两天的纯用量金额(4/1 混入了 -$5 credit)。
    wl_page = {
        "ResultsByTime": [
            {
                "TimePeriod": {"Start": "2026-04-01", "End": "2026-04-02"},
                "Groups": [{
                    "Keys": [acct],
                    "Metrics": {"UnblendedCost": {"Amount": "105.5", "Unit": "USD"}},
                }],
            },
            {
                "TimePeriod": {"Start": "2026-04-02", "End": "2026-04-03"},
                "Groups": [{
                    "Keys": [acct],
                    "Metrics": {"UnblendedCost": {"Amount": "75.25", "Unit": "USD"}},
                }],
            },
        ],
    }
    ce = _mock_ce_client([pass1_p1, pass1_p2, wl_page, pass2_p1, pass2_p2])
    scope = CloudScopeRef(scope_type="aws_account", external_id=acct)

    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        rows = await _driver().list_billing_daily(
            date(2026, 4, 1), date(2026, 4, 2), scope=scope,
        )
    assert ce.get_cost_and_usage.call_count == 5
    by_date = {r.date: r for r in rows}
    assert by_date[date(2026, 4, 1)].total_cost_usd == pytest.approx(100.5)
    assert by_date[date(2026, 4, 2)].total_cost_usd == pytest.approx(75.25)
    # 双口径:list = 白名单遍金额(发票口径里的 credit 被剔除)。
    assert by_date[date(2026, 4, 1)].total_cost_usd_list == pytest.approx(105.5)
    assert by_date[date(2026, 4, 2)].total_cost_usd_list == pytest.approx(75.25)
    # CE 无公开牌价度量 → public 恒 None。
    assert by_date[date(2026, 4, 1)].total_cost_usd_public is None
    # Per-service breakdown must carry rows from BOTH pages of pass 2.
    april1_services = by_date[date(2026, 4, 1)].cost_by_service
    assert april1_services["EC2"] == pytest.approx(60.5)
    assert april1_services["S3"] == pytest.approx(40.0)


@pytest.mark.asyncio
async def test_list_billing_daily_raises_on_first_pass_error() -> None:
    """A ClientError or BotoCoreError on the primary call must
    propagate as a RuntimeError — the lifecycle service catches it
    per-scope and records the failure against ``scheduled_job_runs.
    errors`` so the admin sees it. The previous "swallow + return []"
    behavior masked outages and let spike-detection run on stale
    7-day averages without anyone noticing."""
    from botocore.exceptions import ClientError

    ce = MagicMock()
    ce.get_cost_and_usage.side_effect = ClientError(
        {"Error": {"Code": "AccessDenied", "Message": "nope"}},
        "GetCostAndUsage",
    )
    scope = CloudScopeRef(scope_type="aws_account", external_id="111")
    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        with pytest.raises(RuntimeError, match="Cost Explorer daily fetch failed"):
            await _driver().list_billing_daily(
                date(2026, 4, 1), date(2026, 4, 2), scope=scope,
            )


@pytest.mark.asyncio
async def test_list_billing_daily_raises_on_paginator_error() -> None:
    """OperationNotPageableError inherits from BotoCoreError, not
    ClientError — the driver must propagate both as RuntimeError so
    the failure surfaces in scheduled_job_runs (no silent return [])."""
    from botocore.exceptions import OperationNotPageableError

    ce = MagicMock()
    ce.get_cost_and_usage.side_effect = OperationNotPageableError(
        operation_name="GetCostAndUsage",
    )
    scope = CloudScopeRef(scope_type="aws_account", external_id="111")
    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        with pytest.raises(RuntimeError, match="Cost Explorer daily fetch failed"):
            await _driver().list_billing_daily(
                date(2026, 4, 1), date(2026, 4, 2), scope=scope,
            )


@pytest.mark.asyncio
async def test_list_cost_facts_non_usd_unit_zeros_amount_usd() -> None:
    """AWS Cost Explorer in standard partitions always returns
    ``Unit=USD``. AWS China (Sinnet / NWCD partitions) bills in CNY
    and CE there returns ``Unit=CNY``. Pre-fix the driver dropped the
    raw amount into ``amount_usd`` regardless of Unit, silently
    inflating USD totals on any AWS-China tenant. Post-fix non-USD
    rows persist ``amount_native`` for native-currency analytics and
    set ``amount_usd=0`` so the row stays out of USD aggregates until
    an operator intervenes (mirrors the GCP fallback when
    ``cost_in_usd`` is NULL on a non-USD billing account)."""
    # Use a date within Cost Explorer's 14-day resource-level window
    # (relative to today), since the driver short-circuits older
    # windows. Anchor on "yesterday" for both start and end.
    yesterday = (datetime.now(UTC).date() - timedelta(days=1))
    page = {
        "ResultsByTime": [{
            "TimePeriod": {
                "Start": yesterday.isoformat(),
                "End": (yesterday + timedelta(days=1)).isoformat(),
            },
            "Groups": [
                {
                    "Keys": ["AmazonEC2", "i-aaa"],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "100.0", "Unit": "USD"},
                    },
                },
                {
                    "Keys": ["AmazonRDS", "db-bbb"],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "700.0", "Unit": "CNY"},
                    },
                },
            ],
        }],
    }
    ce = MagicMock()
    ce.get_cost_and_usage_with_resources.return_value = page
    scope = CloudScopeRef(scope_type="aws_account", external_id="333000000000")
    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        facts = await _driver().list_cost_facts(
            yesterday, yesterday, scope=scope,
        )
    by_resource = {f.resource_id: f for f in facts}
    assert by_resource["i-aaa"].amount_usd == pytest.approx(100.0)
    assert by_resource["i-aaa"].currency == "USD"
    # Non-USD: native amount preserved, USD column zeroed.
    assert by_resource["db-bbb"].amount_native == pytest.approx(700.0)
    assert by_resource["db-bbb"].amount_usd == 0.0
    assert by_resource["db-bbb"].currency == "CNY"


@pytest.mark.asyncio
async def test_list_billing_daily_keeps_totals_if_breakdown_fails() -> None:
    """If the per-service pass fails (e.g. filter rejected), the
    per-scope daily totals from the first pass should still be
    returned. Cost breakdown is a nice-to-have, not a hard dep."""
    from botocore.exceptions import ClientError

    acct = "222"
    pass1 = {
        "ResultsByTime": [{
            "TimePeriod": {"Start": "2026-04-01", "End": "2026-04-02"},
            "Groups": [{
                "Keys": [acct],
                "Metrics": {"UnblendedCost": {"Amount": "50", "Unit": "USD"}},
            }],
        }],
    }
    pass2_err = ClientError(
        {"Error": {"Code": "ValidationException", "Message": "bad filter"}},
        "GetCostAndUsage",
    )
    ce = MagicMock()
    # 调用序:totals → 白名单遍(RECORD_TYPE=Usage)→ breakdown。
    # 白名单遍与 breakdown 都失败:totals 仍须存活,list 降级 None。
    ce.get_cost_and_usage.side_effect = [pass1, pass2_err, pass2_err]
    scope = CloudScopeRef(scope_type="aws_account", external_id=acct)
    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        rows = await _driver().list_billing_daily(
            date(2026, 4, 1), date(2026, 4, 1), scope=scope,
        )
    assert len(rows) == 1
    assert rows[0].total_cost_usd == pytest.approx(50.0)
    # 白名单遍失败 → list/public 落 None(口径未接入),不伪造 0。
    assert rows[0].total_cost_usd_list is None
    assert rows[0].total_cost_usd_public is None
    # Breakdown defaults to whatever CloudDailyCost's default is (empty
    # dict) — the important invariant is "totals survive".
    assert rows[0].cost_by_service == {} or rows[0].cost_by_service is None


# ── list_cost_facts: IAM principal-tag mode (AWS SG / Bedrock) ──


@pytest.mark.asyncio
async def test_list_cost_facts_principal_tag_emits_per_owner_facts() -> None:
    """When ``principal_tag_keys=["Owner"]`` is passed, the driver
    queries Cost Explorer's ``GetCostAndUsage`` with
    ``GroupBy=[SERVICE, TAG:iamPrincipal/Owner]`` and emits one
    ``CostFact`` per (date × service × owner) cell.

    This is what the AWS SG / Bedrock account uses — Bedrock on-demand
    has no taggable billable resource, so per-resource granularity
    can't carry the ``Owner`` tag. The IAM-principal-tag path sidesteps
    that limit by grouping at the dimension Cost Explorer surfaces under
    **``iamPrincipal/Owner``** once activated in Billing Console → Cost
    allocation tags. (``aws:PrincipalTag/<Key>`` is the IAM-policy
    evaluator dimension and is NOT a valid CE GroupBy key — passing it
    silently collapses every row to ``Owner=``, which the audit on
    2026-05-12 traced as the root cause of all production Bedrock cost
    landing in the ``(unallocated)`` bucket.)
    """
    acct = "555000000000"
    page = {
        "ResultsByTime": [{
            "TimePeriod": {"Start": "2026-04-15", "End": "2026-04-16"},
            "Groups": [
                {
                    # CE returns tag values as ``"key$value"``.
                    "Keys": ["Amazon Bedrock", "Owner$alice@pingcap.com"],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "12.50", "Unit": "USD"},
                        "UsageQuantity": {"Amount": "100", "Unit": "Tokens"},
                    },
                },
                {
                    "Keys": ["Amazon Bedrock", "Owner$bob@pingcap.com"],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "8.00", "Unit": "USD"},
                        "UsageQuantity": {"Amount": "50", "Unit": "Tokens"},
                    },
                },
                {
                    # Untagged spend — CE returns an empty tag value
                    # (no ``key$value`` separator). Driver folds onto a
                    # ``(unallocated)`` synthetic id so the chargeback
                    # ladder leaves it on the unallocated bucket.
                    "Keys": ["Amazon Bedrock", ""],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "3.00", "Unit": "USD"},
                    },
                },
                {
                    # Zero-amount rows are silently dropped — CE
                    # occasionally returns them for principals that
                    # had access but no billable activity that day.
                    "Keys": ["Amazon Bedrock", "Owner$ghost@pingcap.com"],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "0", "Unit": "USD"},
                    },
                },
            ],
        }],
    }
    ce = MagicMock()
    ce.get_cost_and_usage.return_value = page
    scope = CloudScopeRef(scope_type="aws_account", external_id=acct)

    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        facts = await _driver().list_cost_facts(
            date(2026, 4, 15), date(2026, 4, 15), scope=scope,
            principal_tag_keys=["Owner"],
        )

    # Three non-zero cells -> three facts; zero-amount row dropped.
    assert len(facts) == 3
    by_owner = {f.tags.get("Owner", "(unallocated)"): f for f in facts}
    assert set(by_owner) == {
        "alice@pingcap.com", "bob@pingcap.com", "(unallocated)",
    }
    alice = by_owner["alice@pingcap.com"]
    assert alice.amount_usd == pytest.approx(12.5)
    assert alice.service == "Amazon Bedrock"
    # Synthetic resource_id keeps fact_key unique per (key, owner,
    # service) so re-ingestion converges instead of duplicating rows.
    assert alice.resource_id.startswith("aws-principal:Owner=alice@")
    assert "service=Amazon Bedrock" in alice.resource_id
    assert alice.usage_quantity == pytest.approx(100.0)
    assert alice.usage_unit == "Tokens"

    # Untagged spend lands on the (unallocated) bucket — empty
    # tags-dict so the allocation rule ladder skips per-Owner matches
    # and falls through to any catch-all default.
    unalloc = by_owner["(unallocated)"]
    assert unalloc.tags == {}
    assert "Owner=(unallocated)" in unalloc.resource_id

    # Cost Explorer was called with the correct group + filter shape.
    call = ce.get_cost_and_usage.call_args
    assert call.kwargs["Granularity"] == "DAILY"
    assert {"Type": "DIMENSION", "Key": "SERVICE"} in call.kwargs["GroupBy"]
    assert {
        "Type": "TAG", "Key": "iamPrincipal/Owner",
    } in call.kwargs["GroupBy"]
    assert call.kwargs["Filter"]["Dimensions"]["Values"] == [acct]


@pytest.mark.asyncio
async def test_list_cost_facts_principal_tag_requires_account_scope() -> None:
    """Like the per-resource path, principal-tag mode refuses tenant-
    wide queries — Cost Explorer needs the LINKED_ACCOUNT pin to
    avoid quietly substituting the caller's own account on Org-master
    credentials. ``NotImplementedError`` (not RuntimeError) so the
    cron handler treats it as 'skip this scope, not a fatal' —
    consistent with the existing behavior."""
    with patch.object(AwsCloudDriver, "_boto_client", return_value=MagicMock()):
        with pytest.raises(NotImplementedError, match="explicit scope"):
            await _driver().list_cost_facts(
                date(2026, 4, 15), date(2026, 4, 15),
                scope=None,
                principal_tag_keys=["Owner"],
            )


@pytest.mark.asyncio
async def test_list_cost_facts_principal_tag_falls_back_to_per_resource_when_empty() -> None:
    """Empty / None ``principal_tag_keys`` keeps the existing
    per-resource path — the new mode is opt-in via platform_setting,
    no behaviour change for accounts that haven't activated IAM
    principal tags."""
    yesterday = (datetime.now(UTC).date() - timedelta(days=1))
    page = {
        "ResultsByTime": [{
            "TimePeriod": {
                "Start": yesterday.isoformat(),
                "End": (yesterday + timedelta(days=1)).isoformat(),
            },
            "Groups": [{
                "Keys": ["AmazonEC2", "i-zzz"],
                "Metrics": {
                    "UnblendedCost": {"Amount": "1.0", "Unit": "USD"},
                },
            }],
        }],
    }
    ce = MagicMock()
    ce.get_cost_and_usage_with_resources.return_value = page
    # The principal-tag method must NOT be reachable when keys is empty.
    ce.get_cost_and_usage.side_effect = AssertionError(
        "principal-tag path called with empty keys",
    )
    scope = CloudScopeRef(scope_type="aws_account", external_id="666000000000")
    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        facts = await _driver().list_cost_facts(
            yesterday, yesterday, scope=scope,
            principal_tag_keys=None,
        )
    assert len(facts) == 1
    assert facts[0].resource_id == "i-zzz"


# ── CUR tag-extraction helpers (PR 2026-05-22) ──────────────────


class _FakeRow:
    """Stand-in for a google-cloud-bigquery ``Row`` — supports both
    attribute (``row.tag_kv``) and subscript (``row['tag_kv']``)
    access on the fields the helpers consult."""

    def __init__(self, **fields):
        self.__dict__.update(fields)

    def __getitem__(self, k):
        try:
            return self.__dict__[k]
        except KeyError as exc:
            raise KeyError(k) from exc


def test_cur1_extract_tags_skips_null_and_empty_string_values() -> None:
    """CUR 1.0 path — empty/None tag columns drop out; non-empty ones
    land under their alias name (not the upstream tag key)."""
    row = _FakeRow(
        tag_owner="alice",
        tag_application=None,
        tag_environment="",
        tag_project="helix",
        tag_department=None,
    )
    aliases = [
        ("owner", "Owner"),
        ("application", "Application"),
        ("environment", "Environment"),
        ("project", "Project"),
        ("department", "Department"),
    ]
    tags = AwsCloudDriver._cur1_extract_tags_from_row(row, aliases)
    assert tags == {"owner": "alice", "project": "helix"}


def test_cur1_missing_alias_raises_loudly() -> None:
    """A missing ``tag_<alias>`` attribute signals SQL-vs-Python
    drift (per-pick SELECT didn't project the column the helper
    expects). The helper must raise rather than silently emit
    ``tags={}`` — quietly losing data is a worse outcome than a
    visible ingest failure."""
    row = _FakeRow()  # no tag_* attrs at all
    with pytest.raises(AttributeError):
        AwsCloudDriver._cur1_extract_tags_from_row(
            row, [("owner", "Owner")],
        )


def test_cur2_strips_user_prefix_and_keeps_only_customer_tags() -> None:
    """CUR 2.0 path — ``user_X`` keys are surfaced as ``X``;
    ``aws_*`` keys (CloudFormation stack name, etc) are AWS-managed
    metadata, NOT user attribution, and must be skipped."""
    row = _FakeRow(tag_kv=[
        _FakeRow(key="user_Owner", value="alice"),
        _FakeRow(key="user_Environment", value="prod"),
        # AWS-managed entry — must NOT land in our tags dict.
        _FakeRow(key="aws_cloudformation_stack_name", value="helix-stack"),
    ])
    tags, malformed = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert tags == {"Owner": "alice", "Environment": "prod"}
    assert malformed == 0


def test_cur2_accepts_dict_struct_elements() -> None:
    """Some google-cloud-bigquery client versions hand back nested
    STRUCT elements as plain dicts instead of Row objects. The
    helper must work on both shapes."""
    row = _FakeRow(tag_kv=[
        {"key": "user_Project", "value": "helix"},
    ])
    tags, malformed = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert tags == {"Project": "helix"}
    assert malformed == 0


def test_cur2_drops_empty_or_invalid_entries_silently() -> None:
    """Empty / None / prefix-only / null-value entries get dropped
    without inflating the malformed counter (they're parseable, just
    not useful)."""
    row = _FakeRow(tag_kv=[
        _FakeRow(key="user_", value="x"),       # empty after prefix strip
        _FakeRow(key="", value="x"),            # empty key
        _FakeRow(key=None, value="x"),          # null key
        _FakeRow(key="user_X", value=None),     # null value
        _FakeRow(key="user_X", value=""),       # empty value
        _FakeRow(key="not_user_prefixed", value="x"),  # missing user_
    ])
    tags, malformed = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert tags == {}
    assert malformed == 0


def test_cur2_counts_malformed_structs_for_aggregated_warn() -> None:
    """STRUCT items missing ``key`` or ``value`` fields can't be
    parsed — increment the malformed counter so the caller can emit
    one aggregated WARN per fact row (per-item logging would drown
    the log on a degenerate CUR snapshot)."""
    row = _FakeRow(tag_kv=[
        _FakeRow(key="user_Owner", value="alice"),  # ok
        {"foo": "bar"},                              # missing key + value
        _FakeRow(),                                  # missing both
    ])
    tags, malformed = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert tags == {"Owner": "alice"}
    assert malformed == 2


def test_cur2_caps_key_at_128_chars_drops_longer() -> None:
    """A 128-char key is the AWS limit (boundary kept); 129+ signals
    corrupt CUR data and is dropped to prevent JSON-row bloat."""
    row = _FakeRow(tag_kv=[
        _FakeRow(key="user_" + "a" * 128, value="kept"),
        _FakeRow(key="user_" + "b" * 129, value="dropped"),
    ])
    tags, _ = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert ("a" * 128) in tags
    assert ("b" * 129) not in tags


def test_cur2_truncates_value_at_256_chars() -> None:
    """Long values (URLs in tags are common) are truncated, NOT
    dropped — the prefix is still useful for attribution."""
    row = _FakeRow(tag_kv=[
        _FakeRow(key="user_Big", value="x" * 500),
    ])
    tags, _ = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert tags == {"Big": "x" * 256}


def test_cur2_caps_at_50_tags_per_fact() -> None:
    """AWS spec is 50 tags/resource; mirror that on the read side so
    a future cap-bump on the AWS side can't unbounded-grow our JSON
    row payload."""
    row = _FakeRow(tag_kv=[
        _FakeRow(key=f"user_T{i:03d}", value="v") for i in range(60)
    ])
    tags, _ = AwsCloudDriver._cur2_extract_tags_from_row(row)
    assert len(tags) == 50


def test_cur2_handles_none_and_empty_tag_kv() -> None:
    """A fact row with no resource tags at all — ``tag_kv`` either
    None (BQ returns this when the ARRAY is empty/NULL across every
    contributing pick) or an explicit empty list. Both must yield
    ``({}, 0)``, NOT raise."""
    assert AwsCloudDriver._cur2_extract_tags_from_row(
        _FakeRow(tag_kv=None),
    ) == ({}, 0)
    assert AwsCloudDriver._cur2_extract_tags_from_row(
        _FakeRow(tag_kv=[]),
    ) == ({}, 0)
    assert AwsCloudDriver._cur2_extract_tags_from_row(
        _FakeRow(),  # tag_kv attr missing entirely
    ) == ({}, 0)


# ── SSO create_account_assignment status (#1650) ─────────────────

_SSO_CFG = {
    **_CFG,
    "account_mode": "sso",
    "sso_instance_arn": "arn:aws:sso:::instance/ssoins-abc",
    "sso_region": "us-east-1",
}


def _sso_ctx() -> CloudAccessContext:
    return CloudAccessContext(
        principal_email="alice@example.com",
        principal_display_name="Alice",
        scope=CloudScopeRef(scope_type="aws_account", external_id="123456789012"),
        role_key="arn:aws:sso:::permissionSet/ssoins-abc/ps-1",
    )


def _sso_driver() -> AwsCloudDriver:
    return AwsCloudDriver(integration_id="aws-sso-test", config=_SSO_CFG)


@pytest.mark.asyncio
async def test_grant_sso_failed_status_is_not_ok() -> None:
    """#1650: ``create_account_assignment`` is async server-side and
    returns 200 with a Status of IN_PROGRESS | SUCCEEDED | FAILED. A
    synchronous FAILED (permission-set not provisioned to the account /
    SCIM principal mismatch) previously landed ok=True — the grant row
    went active while AWS granted nothing. Post-fix FAILED yields
    ok=False so the caller surfaces + retries."""
    sso = MagicMock()
    sso.create_account_assignment.return_value = {
        "AccountAssignmentCreationStatus": {
            "Status": "FAILED",
            "RequestId": "req-1",
            "FailureReason": "PermissionSet not provisioned to account",
        },
    }
    with patch.object(
        AwsCloudDriver, "_boto_client", return_value=sso,
    ), patch.object(
        AwsCloudDriver, "_resolve_identity_store_user_id",
        return_value="user-uid-1",
    ):
        result = await _sso_driver().grant_access(_sso_ctx())
    assert result.ok is False
    assert "FAILED" in (result.error_message or "")
    assert result.detail["status"] == "FAILED"


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["SUCCEEDED", "IN_PROGRESS"])
async def test_grant_sso_succeeded_or_in_progress_is_ok(status: str) -> None:
    """SUCCEEDED (synchronously applied) and IN_PROGRESS (AWS will apply
    it shortly) both remain ok=True — the fix only tightens FAILED, it
    must not regress the two non-failure states."""
    sso = MagicMock()
    sso.create_account_assignment.return_value = {
        "AccountAssignmentCreationStatus": {
            "Status": status,
            "RequestId": "req-1",
        },
    }
    with patch.object(
        AwsCloudDriver, "_boto_client", return_value=sso,
    ), patch.object(
        AwsCloudDriver, "_resolve_identity_store_user_id",
        return_value="user-uid-1",
    ):
        result = await _sso_driver().grant_access(_sso_ctx())
    assert result.ok is True
    assert result.external_ref == "123456789012::" + _sso_ctx().role_key + "::user-uid-1"
    assert result.detail["status"] == status


@pytest.mark.asyncio
async def test_revoke_sso_failed_status_is_not_ok() -> None:
    """Post-review sibling of #1650: ``delete_account_assignment`` is ALSO
    async server-side (returns AccountAssignmentDeletionStatus). A FAILED
    revoke must not be reported ok=True — that would close an offboarding
    blocker while the leaver keeps the permission set."""
    sso = MagicMock()
    sso.delete_account_assignment.return_value = {
        "AccountAssignmentDeletionStatus": {
            "Status": "FAILED",
            "RequestId": "req-del-1",
            "FailureReason": "throttled",
        },
    }
    with patch.object(
        AwsCloudDriver, "_boto_client", return_value=sso,
    ), patch.object(
        AwsCloudDriver, "_resolve_identity_store_user_id",
        return_value="user-uid-1",
    ):
        result = await _sso_driver().revoke_access(_sso_ctx())
    assert result.ok is False
    assert "FAILED" in (result.error_message or "")
    assert result.detail["status"] == "FAILED"


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["SUCCEEDED", "IN_PROGRESS"])
async def test_revoke_sso_succeeded_or_in_progress_is_ok(status: str) -> None:
    sso = MagicMock()
    sso.delete_account_assignment.return_value = {
        "AccountAssignmentDeletionStatus": {"Status": status, "RequestId": "r"},
    }
    with patch.object(
        AwsCloudDriver, "_boto_client", return_value=sso,
    ), patch.object(
        AwsCloudDriver, "_resolve_identity_store_user_id",
        return_value="user-uid-1",
    ):
        result = await _sso_driver().revoke_access(_sso_ctx())
    assert result.ok is True


@pytest.mark.asyncio
async def test_principal_tag_whitelist_pass_drives_list_and_bucket() -> None:
    """双口径（成本口径改造 PR3）：principal-tag 模式先跑一遍
    RECORD_TYPE=Usage 白名单查询 → 每单元的纯用量金额落 list；
    发票口径单元里混入的 Credit 只影响 amount，不进 list；白名单里
    完全不存在的单元（纯 credit）按符号归 credit 桶、list None。"""
    acct = "555000000000"
    day = {"Start": "2026-06-15", "End": "2026-06-16"}
    wl_page = {
        "ResultsByTime": [{
            "TimePeriod": day,
            "Groups": [
                {
                    "Keys": ["Amazon Bedrock", "Owner$alice@pingcap.com"],
                    "Metrics": {"UnblendedCost": {"Amount": "12.50", "Unit": "USD"}},
                },
            ],
        }],
    }
    main_page = {
        "ResultsByTime": [{
            "TimePeriod": day,
            "Groups": [
                {
                    # 发票口径:alice 单元混入 -0.5 credit → 12.0。
                    "Keys": ["Amazon Bedrock", "Owner$alice@pingcap.com"],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "12.00", "Unit": "USD"},
                        "UsageQuantity": {"Amount": "100", "Unit": "Tokens"},
                    },
                },
                {
                    # 纯 credit 单元:白名单遍里不存在。
                    "Keys": ["AWS Credits", ""],
                    "Metrics": {
                        "UnblendedCost": {"Amount": "-3.00", "Unit": "USD"},
                    },
                },
            ],
        }],
    }
    ce = MagicMock()
    ce.get_cost_and_usage.side_effect = [wl_page, main_page]
    scope = CloudScopeRef(scope_type="aws_account", external_id=acct)

    with patch.object(AwsCloudDriver, "_boto_client", return_value=ce):
        facts = await _driver().list_cost_facts(
            date(2026, 6, 15), date(2026, 6, 15), scope=scope,
            principal_tag_keys=["Owner"],
        )

    assert ce.get_cost_and_usage.call_count == 2
    # 第一个调用带 RECORD_TYPE=Usage 白名单过滤。
    first_kwargs = ce.get_cost_and_usage.call_args_list[0].kwargs
    assert "And" in first_kwargs["Filter"]
    assert any(
        f.get("Dimensions", {}).get("Key") == "RECORD_TYPE"
        for f in first_kwargs["Filter"]["And"]
    )
    by_service = {f.service: f for f in facts}
    alice = by_service["Amazon Bedrock"]
    assert alice.amount_usd == pytest.approx(12.0)   # 发票口径(含 credit)
    assert alice.amount_usd_list == pytest.approx(12.5)  # 纯用量
    assert alice.cost_bucket == "usage"
    credit = by_service["AWS Credits"]
    assert credit.amount_usd == pytest.approx(-3.0)
    assert credit.amount_usd_list is None
    assert credit.cost_bucket == "credit"