"""Regression coverage for ``_cloud_shared._CloudDriverBase.test_connection``. Background — incident 2026-05-16: both AWS integrations and the GCP integration showed ``last_test_status='ok'`` while every scheduled sync failed (``UnauthorizedOperation`` on every region for AWS, project-level 403 for GCP). Root cause: per-region / per-project fan-out helpers absorb leg failures into ``provider.partial_failures`` and return the surviving subset. When 100% of legs failed, the underlying call returned ``[]`` and the connection-test interpreted that as "OK · enumerated 0 servers" — green for the credentials that demonstrably could not observe a single instance. The fix in ``_cloud_shared.test_connection`` now: * Flips ``ok=False`` when ``len(servers) == 0`` AND any leg failed. * Surfaces a grouped, length-bounded leg breakdown in the message so the admin can see WHICH legs failed and the underlying reason. * Runs raw exception strings through ``redact_provider_error`` before they land in ``last_test_message`` (same rule the sync path follows). These tests pin all five branches: happy path, legitimate empty cloud, partial-leg-failure with data, all-leg-failure with no data, and the exception-redaction path. """ from __future__ import annotations from typing import Any import pytest from app.domains.cmdb.providers.base import ( CloudProvider, CloudServer, ProviderCapability, ProviderCredentials, ) from app.integrations.driver_base import CloudDriver from app.integrations.drivers._cloud_shared import ( _LEG_DETAIL_MAX, _LEG_SUMMARY_MAX, _CloudDriverBase, _redact_leg, _summarize_leg_failures, ) # ── Fake provider + driver scaffolding ──────────────────────────── class _FakeProvider(CloudProvider): """Minimal stub honoring the partial_failures contract. Instantiated by ``_CloudDriverBase._provider()`` — needs the ``ProviderCredentials`` ctor signature inherited from the base. """ # Tests mutate these before invoking the driver. canned_servers: list[CloudServer] = [] canned_partial: list[dict[str, Any]] = [] canned_exc: Exception | None = None close_called: bool = False @classmethod def provider_type(cls) -> str: return "fake" @classmethod def supported_capabilities(cls) -> frozenset[ProviderCapability]: return frozenset({ProviderCapability.SERVERS}) async def list_servers(self) -> list[CloudServer]: if type(self).canned_exc is not None: raise type(self).canned_exc # Drivers append to self.partial_failures during the sweep — # mirror that contract here so the test path is identical. self.partial_failures.extend(type(self).canned_partial) return list(type(self).canned_servers) async def close(self) -> None: type(self).close_called = True class _FakeDriver(_CloudDriverBase): provider_cls = _FakeProvider name = "fake" provider = "fake" def _driver() -> _FakeDriver: return _FakeDriver(integration_id="fake-test", config={}) def _reset_fake() -> None: _FakeProvider.canned_servers = [] _FakeProvider.canned_partial = [] _FakeProvider.canned_exc = None _FakeProvider.close_called = False def _server(instance_id: str) -> CloudServer: return CloudServer( instance_id=instance_id, name=instance_id, status="running", region="us-east-1", zone=None, vcpu=None, memory_mb=None, public_ips=[], private_ips=[], image_id=None, os_type="linux", tags={}, raw={}, ) # ── _summarize_leg_failures ────────────────────────────────────── def test_summarize_groups_by_reason_and_caps_examples() -> None: legs = [ {"leg": f"region:r{i}", "reason": "UnauthorizedOperation"} for i in range(_LEG_SUMMARY_MAX + 3) ] + [{"leg": "region:rate", "reason": "Throttling"}] summary = _summarize_leg_failures(legs) assert "UnauthorizedOperation × " in summary assert f"(+{3} more)" in summary # 8 - _LEG_SUMMARY_MAX(5) = 3 assert "Throttling × 1" in summary # No raw access-key fragments leak through (covered by redactor # tests, but assert the call path goes through it). assert "AKIA" not in summary def test_summarize_handles_missing_fields() -> None: legs = [{"leg": None, "reason": None}] summary = _summarize_leg_failures(legs) assert "unknown" in summary assert "?" in summary def test_summarize_empty_returns_empty_string() -> None: assert _summarize_leg_failures([]) == "" def test_summarize_caps_distinct_reason_groups() -> None: """A pathological 10k-unique-reason payload would otherwise produce a multi-hundred-KB message string. Pin the outer-group cap + "+N more reason(s)" tail.""" legs = [{"leg": f"r{i}", "reason": f"reason_{i}"} for i in range(100)] summary = _summarize_leg_failures(legs) # Only _LEG_SUMMARY_MAX distinct reasons rendered + 1 tail entry. assert summary.count("reason_") <= _LEG_SUMMARY_MAX assert f"+{100 - _LEG_SUMMARY_MAX} more reason(s)" in summary def test_summarize_2000_char_hard_ceiling() -> None: """Final defense-in-depth: even with maximally-long reasons (each individually pre-truncated to 300 chars by ``redact_provider_error``) across the full _LEG_SUMMARY_MAX of distinct reason groups, the joined string stays bounded. The 2000-char outer guard rarely fires in practice given the inner caps, but pin the contract.""" legs = [ {"leg": f"r{i}", "reason": "x" * 5000} for i in range(_LEG_SUMMARY_MAX) ] summary = _summarize_leg_failures(legs) assert len(summary) <= 2000 # ── test_connection branches ───────────────────────────────────── @pytest.mark.asyncio async def test_happy_path_returns_ok_with_count() -> None: """Per-cap probe (2026-05-17): message now reports per-cap shape (``servers=ok(N) · ...``) rather than the legacy ``enumerated N servers``. The legacy ``sample_server_count`` / ``leg_failures`` keys stay populated for callers that haven't migrated to ``capability_probes``.""" _reset_fake() _FakeProvider.canned_servers = [_server("i-1"), _server("i-2")] result = await _driver().test_connection() assert result.ok is True assert "servers=ok(2)" in result.message assert result.detail["sample_server_count"] == 2 assert result.detail["leg_failures"] == [] # New structured payload — admin UI table reads from this. probes = result.detail["capability_probes"] assert probes["servers"]["status"] == "ok" assert probes["servers"]["sample_count"] == 2 assert _FakeProvider.close_called is True @pytest.mark.asyncio async def test_empty_cloud_no_failures_is_ok() -> None: """Legitimate empty cloud (no instances, no leg failures) stays ok.""" _reset_fake() result = await _driver().test_connection() assert result.ok is True assert "servers=ok(0)" in result.message assert result.detail["leg_failures"] == [] assert result.detail["capability_probes"]["servers"]["status"] == "ok" @pytest.mark.asyncio async def test_zero_servers_with_leg_failures_flips_to_not_ok() -> None: """The prod incident: 5/5 AWS regions UnauthorizedOperation → ok=False. Per-cap probe (2026-05-17): the leg breakdown moved from ``message`` to ``detail.capability_probes['servers'].reason`` — the message field stays one-line for the integrations table. """ _reset_fake() _FakeProvider.canned_partial = [ {"leg": f"region:r{i}", "reason": "UnauthorizedOperation"} for i in range(5) ] result = await _driver().test_connection() assert result.ok is False assert "servers=failed" in result.message servers_probe = result.detail["capability_probes"]["servers"] assert servers_probe["status"] == "failed" assert servers_probe["sample_count"] == 0 assert "0 rows enumerated" in servers_probe["reason"] assert "5 leg(s) failed" in servers_probe["reason"] assert "UnauthorizedOperation × 5" in servers_probe["reason"] assert result.detail["sample_server_count"] == 0 assert len(result.detail["leg_failures"]) == 5 @pytest.mark.asyncio async def test_partial_coverage_with_data_stays_ok_with_warning() -> None: """3 servers visible but 57 GCP projects 403 — still ok, with warning. Partial-coverage now reads from the per-cap probe — message carries the compact ``partial(N;M legs failed)`` token; the leg summary itself lives on ``detail.capability_probes['servers'].reason`` (admin UI renders the table). The top-level ``ok`` stays True because the credential demonstrably worked on at least one leg. """ _reset_fake() _FakeProvider.canned_servers = [_server("i-a"), _server("i-b"), _server("i-c")] _FakeProvider.canned_partial = [ {"leg": f"project:p{i}", "reason": "http_403"} for i in range(57) ] result = await _driver().test_connection() assert result.ok is True assert "servers=partial(3;57 legs failed)" in result.message servers_probe = result.detail["capability_probes"]["servers"] assert servers_probe["status"] == "partial_failure" assert servers_probe["sample_count"] == 3 assert servers_probe["leg_count"] == 57 assert "http_403 × 57" in servers_probe["reason"] @pytest.mark.asyncio async def test_exception_path_is_redacted_and_returns_not_ok() -> None: """Exceptions get redacted before landing in last_test_message. A raise inside the provider's ``list_servers`` is caught by the per-cap probe's generic ``except`` arm and lands in ``capability_probes['servers'].reason`` as ``f"{type(exc).__name__}: {redact_provider_error(exc)}"`` — the type prefix is REQUIRED so admins reading ``last_test_message`` can tell driver-side ``RuntimeError`` apart from cloud-side ``HttpError`` / ``ClientError``. The aggregated message embeds the redacted reason (with type) inline so the failure- notifier dedup key bites on a structurally distinct signature. """ _reset_fake() # boto3-shape error string carrying a credential fragment. _FakeProvider.canned_exc = RuntimeError( "Signature=ABCDEF0123456789abcdef0123456789 not valid" ) result = await _driver().test_connection() assert result.ok is False # The signature payload must be redacted, not echoed verbatim. assert "ABCDEF0123456789" not in result.message assert "" in result.message # Type name surfaces both in the message AND the per-cap reason — # admins distinguish driver-side from cloud-side errors at a glance. assert "RuntimeError" in result.message servers_probe = result.detail["capability_probes"]["servers"] assert servers_probe["status"] == "failed" assert servers_probe["reason"].startswith("RuntimeError: ") assert "" in servers_probe["reason"] assert _FakeProvider.close_called is True @pytest.mark.asyncio async def test_close_runs_even_when_test_fails() -> None: _reset_fake() _FakeProvider.canned_exc = ValueError("boom") await _driver().test_connection() assert _FakeProvider.close_called is True @pytest.mark.asyncio async def test_leg_summary_length_bounded() -> None: """200-leg payload must not flood last_test_message OR detail. The 200-leg summary now lives on ``detail.capability_probes['servers'].reason`` (per-cap probe); the top-level ``leg_failures`` array is the union across caps, capped + sentinel symmetrically with the old contract. """ _reset_fake() _FakeProvider.canned_partial = [ {"leg": f"region:r{i}", "reason": "UnauthorizedOperation"} for i in range(200) ] result = await _driver().test_connection() assert result.ok is False servers_reason = result.detail["capability_probes"]["servers"]["reason"] # All 200 legs counted in the per-cap reason summary; only # ``_LEG_SUMMARY_MAX`` rendered per group + a ``(+N more)`` suffix. assert "UnauthorizedOperation × 200" in servers_reason expected_extra = 200 - _LEG_SUMMARY_MAX assert f"(+{expected_extra} more)" in servers_reason # Hard cap on message size stays — the message is the one-line # summary, not the detail payload. assert len(result.message) < 2048 # Top-level union ``leg_failures`` capped at ``_LEG_DETAIL_MAX + # 1 sentinel`` so the wire payload stays bounded. assert len(result.detail["leg_failures"]) == _LEG_DETAIL_MAX + 1 sentinel = result.detail["leg_failures"][-1] assert sentinel["leg"] == "_truncated_" assert str(200 - _LEG_DETAIL_MAX) in sentinel["reason"] @pytest.mark.asyncio async def test_detail_cap_not_applied_below_threshold() -> None: """A small number of legs (e.g. 3) must NOT emit a truncation sentinel.""" _reset_fake() _FakeProvider.canned_partial = [ {"leg": f"region:r{i}", "reason": "Throttling"} for i in range(3) ] result = await _driver().test_connection() assert len(result.detail["leg_failures"]) == 3 # No sentinel — every entry is a real leg. assert all(e["leg"] != "_truncated_" for e in result.detail["leg_failures"]) servers_reason = result.detail["capability_probes"]["servers"]["reason"] assert "Throttling × 3" in servers_reason @pytest.mark.asyncio async def test_summary_reports_true_count_when_detail_truncated() -> None: """Summary must read 'X × N' where N is the FULL count, not the capped detail-list count — admin needs accurate scale of the failure, not the wire-payload size.""" _reset_fake() # Push beyond _LEG_DETAIL_MAX with two distinct reasons so the # summary's grouped breakdown is verifiable. _FakeProvider.canned_partial = ( [{"leg": f"r{i}", "reason": "http_403"} for i in range(80)] + [{"leg": f"r{i}", "reason": "Throttling"} for i in range(20)] ) result = await _driver().test_connection() servers_reason = result.detail["capability_probes"]["servers"]["reason"] # True counts surface in the per-cap reason summary. assert "http_403 × 80" in servers_reason assert "Throttling × 20" in servers_reason # Top-level union detail is capped (50 entries + 1 sentinel = 51). assert len(result.detail["leg_failures"]) == _LEG_DETAIL_MAX + 1 # ── Detail-payload redaction (Round-1 review finding) ─────────── def test_redact_leg_scrubs_reason_field() -> None: """A leg entry whose ``reason`` accidentally carries a credential fragment must be scrubbed before reaching ``detail.leg_failures``. No current provider does this, but the redactor wrap is what protects the on-wire surface against future drivers that build leg labels from raw SDK error payloads. """ entry = { "leg": "region:us-east-1", "reason": "X-Amz-Signature=ABCDEF0123456789abcdef0123456789 expired", } out = _redact_leg(entry) assert "ABCDEF0123456789" not in out["reason"] assert "" in out["reason"] # Leg label untouched (no credential pattern). assert out["leg"] == "region:us-east-1" def test_redact_leg_preserves_other_keys() -> None: """Non-leg/reason keys (e.g. ``capability``) pass through unchanged.""" out = _redact_leg({ "leg": "project:foo", "reason": "http_403", "capability": "servers", }) assert out["capability"] == "servers" assert out["leg"] == "project:foo" assert out["reason"] == "http_403" @pytest.mark.asyncio async def test_test_connection_detail_legs_are_redacted() -> None: """End-to-end: leg payload reaching API response must not echo creds. Verifies the same finding that triggered the redaction wrap — ``IntegrationTestResult.model_dump()`` is returned verbatim by POST /integrations/{slug}/test, so ``detail.leg_failures[*].reason`` is on the wire. """ _reset_fake() _FakeProvider.canned_partial = [ { "leg": "region:us-east-1", "reason": ( "AccessKeyId=AKIAEXAMPLE0123456789 was denied by IAM" ), }, ] result = await _driver().test_connection() leg = result.detail["leg_failures"][0] assert "AKIAEXAMPLE0123456789" not in leg["reason"] assert "" in leg["reason"] # And the message-level summary is also clean (already covered by # the helper test, this is the wire-level pin). assert "AKIAEXAMPLE0123456789" not in result.message # ── Defensive shape guards (Round-1 review finding) ───────────── @pytest.mark.asyncio async def test_test_connection_handles_missing_partial_failures_attr() -> None: """A driver that forgets ``super().__init__()`` would not have the ``partial_failures`` attribute. The base must not crash.""" _reset_fake() _FakeProvider.canned_servers = [_server("i-1")] # Wrap _provider so the returned object lacks ``partial_failures``. class _Headless: async def list_servers(self) -> list[CloudServer]: return [_server("i-1")] async def close(self) -> None: pass driver = _driver() driver._provider = lambda: _Headless() # type: ignore[method-assign] result = await driver.test_connection() assert result.ok is True assert result.detail["leg_failures"] == [] @pytest.mark.asyncio async def test_test_connection_handles_non_list_partial_failures() -> None: """A driver that shadows the attr with a dict / None must not crash.""" _reset_fake() _FakeProvider.canned_servers = [_server("i-1")] class _MisTyped: partial_failures = {"not": "a list"} async def list_servers(self) -> list[CloudServer]: return [_server("i-1")] async def close(self) -> None: pass driver = _driver() driver._provider = lambda: _MisTyped() # type: ignore[method-assign] result = await driver.test_connection() assert result.ok is True assert result.detail["leg_failures"] == [] # ── CloudDriver contract sanity ───────────────────────────────── def test_fake_driver_is_a_clouddriver() -> None: """Sanity: the test scaffold actually exercises the real base class.""" assert isinstance(_driver(), CloudDriver) assert _FakeDriver.__mro__[1] is _CloudDriverBase def test_fake_provider_inherits_partial_failures_from_base() -> None: """If the base ever drops the partial_failures attr we want to know.""" creds = ProviderCredentials( provider_id="x", provider_type="fake", account_alias="x", fields={}, ) p = _FakeProvider(creds) assert p.partial_failures == []