"""AWS Organizations cross-account fan-out (2026-07-21). The management/payer account runs ~0 EC2 — all compute lives in member accounts. When ``org_fanout`` is on, ``list_servers`` / ``list_server_metrics`` STS-assume ``org_member_role`` into every active member account and stamp each server's ``raw`` with ``_aws_account_id`` so metrics re-assume the right account. Off by default: a single-account integration is unchanged. Gated OFF at ship time — this suite is the contract for when the AWS org grants ``sts:AssumeRole`` and the flag flips on. """ from __future__ import annotations from unittest.mock import MagicMock, patch import pytest from app.domains.cmdb.providers.aws import AWSProvider from app.domains.cmdb.providers.base import ProviderCredentials, ServerMetricTarget def _provider(**extra: str) -> AWSProvider: fields = { "access_key_id": "AKIA" + "X" * 16, "secret_access_key": "secret" + "X" * 32, } fields.update(extra) return AWSProvider(ProviderCredentials( provider_id="aws-test", provider_type="aws", account_alias="test", fields=fields, )) def _server(instance_id: str, raw: dict | None = None): from app.domains.cmdb.providers.base import CloudServer return CloudServer( instance_id=instance_id, name=instance_id, status="running", region="us-east-1", zone=None, vcpu=2, memory_mb=None, public_ips=[], private_ips=[], image_id=None, os_type="linux", tags={}, raw=raw if raw is not None else {"id": instance_id}, ) def _client_error(code: str, message: str = ""): from botocore.exceptions import ClientError return ClientError( error_response={"Error": {"Code": code, "Message": message or code}}, operation_name="AssumeRole", ) # ── config parsing ──────────────────────────────────────────────────── def test_org_fanout_defaults_off_and_role_default() -> None: p = _provider() assert p._org_fanout is False assert p._org_member_role == "OrganizationAccountAccessRole" @pytest.mark.parametrize("val", ["true", "1", "yes", "on", "TRUE", "On"]) def test_org_fanout_truthy_values(val: str) -> None: assert _provider(org_fanout=val)._org_fanout is True @pytest.mark.parametrize("val", ["false", "0", "no", "", "off"]) def test_org_fanout_falsy_values(val: str) -> None: assert _provider(org_fanout=val)._org_fanout is False def test_org_member_role_override() -> None: p = _provider(org_fanout="true", org_member_role="HelixCloudReadOnly") assert p._org_member_role == "HelixCloudReadOnly" # ── region denylist ─────────────────────────────────────────────────── def test_region_denylist_default_empty() -> None: assert _provider()._region_denylist == set() @pytest.mark.parametrize("val,expect", [ ("me-south-1", {"me-south-1"}), ("me-south-1, af-south-1", {"me-south-1", "af-south-1"}), ("me-south-1 af-south-1", {"me-south-1", "af-south-1"}), ("ME-South-1", {"me-south-1"}), # case-insensitive ]) def test_region_denylist_parsing(val: str, expect: set) -> None: assert _provider(region_denylist=val)._region_denylist == expect def test_ec2_regions_sync_excludes_denylisted() -> None: """A denylisted opt-in region (empty + unreachable, e.g. me-south-1) is dropped from discovery so it never produces a connect-timeout leg.""" p = _provider(region_denylist="me-south-1") def _client_factory(service: str, region_name: str): client = MagicMock() client.describe_regions.return_value = { "Regions": [ {"RegionName": "us-east-1"}, {"RegionName": "me-south-1"}, {"RegionName": "ap-northeast-1"}, ] } return client with patch.object(p._session, "client", side_effect=_client_factory): regions = p._ec2_regions_sync() assert "me-south-1" not in regions assert set(regions) == {"us-east-1", "ap-northeast-1"} # ── list_servers ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_list_servers_fanout_off_uses_single_account() -> None: """Flag off → plain single-account path, no member enumeration.""" p = _provider() with patch.object(p, "_list_servers_sync", return_value=[_server("i-1")]) as sync, \ patch.object(p, "_list_member_accounts_sync") as members, \ patch.object(p, "_assume_session_sync") as assume: servers = await p.list_servers() assert [s.instance_id for s in servers] == ["i-1"] sync.assert_called_once_with() members.assert_not_called() assume.assert_not_called() @pytest.mark.asyncio async def test_list_servers_fanout_stamps_account_id() -> None: """Flag on → each member assumed; servers stamped with their account.""" p = _provider(org_fanout="true") def _sync(session=None): if session is None: # management account — 0 EC2 return [] # ``session`` is the account_id our fake assume returns. return [_server(f"i-{session}", raw={"id": session})] with patch.object(p, "_list_member_accounts_sync", return_value=[("111", "acctA"), ("222", "acctB")]), \ patch.object(p, "_assume_session_sync", side_effect=lambda acct: acct), \ patch.object(p, "_list_servers_sync", side_effect=_sync): servers = await p.list_servers() by_acct = {s.raw["_aws_account_id"]: s for s in servers} assert set(by_acct) == {"111", "222"} assert by_acct["111"].raw["_aws_account_name"] == "acctA" assert by_acct["222"].instance_id == "i-222" assert p.partial_failures == [] @pytest.mark.asyncio async def test_list_servers_fanout_assume_denied_is_skipped_not_paged() -> None: """A member whose assume_role is DENIED (invited account with no OrganizationAccountAccessRole) → skipped_legs (silent, admin-visible), NOT partial_failures — otherwise every sync tick pages the failure- notifier once per structurally-unreachable account. Others survive.""" p = _provider(org_fanout="true") def _assume(acct: str): if acct == "222": raise _client_error("AccessDenied", "not authorized to sts:AssumeRole") return acct def _sync(session=None): if session is None: return [] return [_server(f"i-{session}", raw={"id": session})] with patch.object(p, "_list_member_accounts_sync", return_value=[("111", "acctA"), ("222", "acctB")]), \ patch.object(p, "_assume_session_sync", side_effect=_assume), \ patch.object(p, "_list_servers_sync", side_effect=_sync): servers = await p.list_servers() assert [s.raw["_aws_account_id"] for s in servers] == ["111"] # Denied account is a quiet skip, never a page. assert p.partial_failures == [] assert len(p.skipped_legs) == 1 leg = p.skipped_legs[0] assert leg["leg"] == "account:222" assert leg["reason"] == "member_role_unavailable" @pytest.mark.asyncio async def test_list_servers_fanout_assume_transport_error_is_partial_failure() -> None: """A transient transport error assuming the role stays a partial_failure (retryable — safe-sweep pauses, failure-notifier surfaces it).""" from botocore.exceptions import EndpointConnectionError p = _provider(org_fanout="true") def _assume(acct: str): if acct == "222": raise EndpointConnectionError(endpoint_url="https://sts.amazonaws.com") return acct def _sync(session=None): return [] if session is None else [_server(f"i-{session}", raw={"id": session})] with patch.object(p, "_list_member_accounts_sync", return_value=[("111", "acctA"), ("222", "acctB")]), \ patch.object(p, "_assume_session_sync", side_effect=_assume), \ patch.object(p, "_list_servers_sync", side_effect=_sync): servers = await p.list_servers() assert [s.raw["_aws_account_id"] for s in servers] == ["111"] assert p.skipped_legs == [] assert len(p.partial_failures) == 1 assert p.partial_failures[0]["reason"] == "assume_role_transport_error" @pytest.mark.asyncio async def test_list_servers_fanout_listaccounts_denied_raises() -> None: """If the management credential can't even list org accounts, that's a structural auth failure → ProviderAuthError (not a silent empty sync).""" from app.domains.cmdb.providers.exceptions import ProviderAuthError p = _provider(org_fanout="true") with patch.object(p, "_list_member_accounts_sync", side_effect=_client_error("AccessDenied", "organizations:ListAccounts")): with pytest.raises(ProviderAuthError): await p.list_servers() # ── list_server_metrics ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_list_server_metrics_fanout_groups_by_account() -> None: """Targets are grouped by stamped ``_aws_account_id``; each account is assumed once and its CloudWatch session passed to the sync worker.""" p = _provider(org_fanout="true") targets = [ ServerMetricTarget(instance_id="i-a1", region="us-east-1", raw={"_aws_account_id": "111"}), ServerMetricTarget(instance_id="i-a2", region="us-east-1", raw={"_aws_account_id": "111"}), ServerMetricTarget(instance_id="i-b1", region="us-west-2", raw={"_aws_account_id": "222"}), ] seen_sessions: list = [] def _metrics(tgts, lookback_days, period_seconds, session=None): seen_sessions.append(session) return [] # values irrelevant; we assert the assume routing with patch.object(p, "_assume_session_sync", side_effect=lambda acct: f"sess-{acct}") as assume, \ patch.object(p, "_list_metrics_sync", side_effect=_metrics): await p.list_server_metrics(targets) assumed = {c.args[0] for c in assume.call_args_list} assert assumed == {"111", "222"} assert set(seen_sessions) == {"sess-111", "sess-222"} @pytest.mark.asyncio async def test_list_server_metrics_assume_denied_is_skipped() -> None: """Metrics leg mirrors the servers leg: an AssumeRole denial for an account is a quiet skip, not a page.""" p = _provider(org_fanout="true") targets = [ ServerMetricTarget(instance_id="i-b1", region="us-west-2", raw={"_aws_account_id": "222"}), ] with patch.object(p, "_assume_session_sync", side_effect=_client_error("AccessDenied", "no role")), \ patch.object(p, "_list_metrics_sync", return_value=[]): pts = await p.list_server_metrics(targets) assert pts == [] assert p.partial_failures == [] assert len(p.skipped_legs) == 1 assert p.skipped_legs[0]["reason"] == "member_role_unavailable" assert p.skipped_legs[0]["capability"] == "metrics" @pytest.mark.asyncio async def test_list_server_metrics_fanout_off_single_call() -> None: """Flag off → one plain metrics call on the base session, no assume.""" p = _provider() targets = [ServerMetricTarget(instance_id="i-1", region="us-east-1")] with patch.object(p, "_list_metrics_sync", return_value=[]) as sync, \ patch.object(p, "_assume_session_sync") as assume: await p.list_server_metrics(targets) sync.assert_called_once() assume.assert_not_called() @pytest.mark.asyncio async def test_list_server_metrics_unstamped_target_uses_base_session() -> None: """A target with no ``_aws_account_id`` (e.g. management account server) queries on the base session — no assume attempted for that group.""" p = _provider(org_fanout="true") targets = [ServerMetricTarget(instance_id="i-mgmt", region="us-east-1", raw={})] sessions: list = [] with patch.object(p, "_assume_session_sync") as assume, \ patch.object(p, "_list_metrics_sync", side_effect=lambda t, lb, ps, session=None: sessions.append(session) or []): await p.list_server_metrics(targets) assert sessions == [None] assume.assert_not_called()