"""Regression tests for round-21 pre-launch audit fixes (PR — 2026-05-17). Covers seven independent production-incident fixes bundled into one PR: 1. ``feishu.approval_pending_reaper`` — naive vs aware ``datetime`` comparison crash (incident 2026-05-16: ``TypeError: can't compare offset-naive and offset-aware datetimes`` aborted the daily reaper the moment any pending row was found). 2. GCP ``cmdb.billing_daily_sync`` reconciliation pass — ``sorted()`` ``TypeError`` when a billing-export row has ``project.id IS NULL`` (org-level credit / refund). 3. GCP ``sys-*`` system projects (Apigee / AppEngine / Firebase) — filtered from ``list_scopes`` AND ``_list_active_projects_sync`` so their 403s stop poisoning ``cmdb.grants_sync`` / ``cmdb.resource_sync object_storage``. 4. Tencent CAM ``ListAttachedUserAllPolicies`` — ``Page`` / ``Rp`` are REQUIRED (the prior code stripped them based on a mistaken transfer of the ``ListUsers`` rejection shape). Production logged ``MissingParameter Page`` continuously. 5. ``cmdb.sync`` — 100% leg ``auth_error`` + zero upserts now degrades the capability to ``skipped`` (with structured reason + leg breakdown), not ``partial_failure`` (which spammed the admin chat hourly with the same structurally-impossible reminder). 6. ``_CloudDriverBase.test_connection`` — per-capability dry-run so an integration whose ``object_storage`` capability is 100% auth_error doesn't pass with ``OK · enumerated 692 servers``. 7. ``cmdb.resource_sync`` raise — error_message now embeds the redacted leg-failure summary rather than a generic "status= partial_failure" line, so admins read the breakdown directly instead of SQL-querying ``cmdb_sync_jobs.capability_results``. """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from unittest.mock import MagicMock, patch import pytest # ───────────────────────── 1. Feishu reaper tzinfo ──────────────────── def test_feishu_reaper_uses_aware_utc_helper() -> None: """The reaper imports ``_as_aware_utc`` and applies it on every Python-level comparison site so a naive ``req.created_at`` (TiDB strips tz) doesn't raise ``TypeError`` against aware cutoffs. """ import inspect from app.scheduled_jobs import feishu_handlers src = inspect.getsource(feishu_handlers.run_approval_pending_reaper) # Helper imported into the function body assert "from app.auth.tokens.service import _as_aware_utc" in src # Each of the three Python-level comparison sites wraps the # naive ORM column. assert src.count("_as_aware_utc(req.created_at)") >= 3 def test_as_aware_utc_round_trips_naive_datetime() -> None: """Sanity-check the shared helper itself: naive in, aware out; aware in, identity. The reaper relies on this contract. """ from app.auth.tokens.service import _as_aware_utc naive = datetime(2026, 5, 1, 12, 0, 0) aware = _as_aware_utc(naive) assert aware is not None assert aware.tzinfo is UTC already_aware = datetime(2026, 5, 1, 12, 0, 0, tzinfo=UTC) assert _as_aware_utc(already_aware) is already_aware assert _as_aware_utc(None) is None # ───────────────────────── 2. GCP billing None sort ─────────────────── @pytest.mark.asyncio async def test_gcp_list_billing_daily_tolerates_null_project_id() -> None: """Rows with ``project.id IS NULL`` (org-level credits / refunds in GCP's BQ export) must not raise ``TypeError`` during the final ``sorted(by_key.items())``. Pre-fix: the tuple key contained ``None`` and ``None < 'real-project'`` exploded the reconciliation pass with ``'<' not supported between instances of 'NoneType' and 'str'``. The sync helper calls ``self._credentials()`` (which decodes the SA JSON) before reaching the BQ client — we patch that to bypass google-auth in the unit-test surface. """ from datetime import date as _date from app.integrations.drivers import gcp_cloud driver = gcp_cloud.GcpCloudDriver( integration_id="x", config={"service_account_json": "{}", "organization_id": "1"}, ) fake_rows = [ { "day": _date(2026, 5, 10), "project_id": None, "service_name": "Promotional credit", "total_cost": -42.5, "currency": "USD", }, { "day": _date(2026, 5, 10), "project_id": "p-active", "service_name": "Compute Engine", "total_cost": 100.0, "currency": "USD", }, { "day": _date(2026, 5, 11), "project_id": None, "service_name": "Invoice adjustment", "total_cost": -10.0, "currency": "USD", }, ] fake_job = MagicMock() fake_job.result.return_value = iter(fake_rows) fake_client = MagicMock() fake_client.query.return_value = fake_job with ( patch.object(driver, "_credentials", return_value=MagicMock()), patch("google.cloud.bigquery.Client", return_value=fake_client), ): out = driver._list_billing_daily_sync( start=_date(2026, 5, 10), end=_date(2026, 5, 11), scope=None, tables=["host.ds.gcp_billing_export_v1_X"], ) # Three distinct rows: (May10, "", USD), (May10, "p-active", USD), # (May11, "", USD). The fact that we got here at all proves the # ``sorted()`` no longer raises. assert {r.scope.external_id for r in out} == {"", "p-active"} # ───────────────────────── 3. GCP sys-* filter ──────────────────────── def _parent_bounded_crm( projects: list[dict[str, object]], *, organization_id: str = "1", ) -> MagicMock: """Build a CRM v3 mock for recursive projects/folders list traversal.""" crm = MagicMock() root = f"organizations/{organization_id}" folder_names = sorted( { str(project.get("parent")) for project in projects if str(project.get("parent") or "").startswith("folders/") } ) def _projects_list(*, parent, pageToken, showDeleted): # noqa: N803 assert pageToken is None assert showDeleted is False request = MagicMock() request.execute.return_value = { "projects": [p for p in projects if p.get("parent") == parent], } return request def _folders_list(*, parent, pageToken, showDeleted): # noqa: N803 assert pageToken is None assert showDeleted is False request = MagicMock() request.execute.return_value = { "folders": ( [{"name": name, "parent": root} for name in folder_names] if parent == root else [] ), } return request crm.projects.return_value.list.side_effect = _projects_list crm.folders.return_value.list.side_effect = _folders_list return crm def test_is_gcp_system_project_matches_prefix() -> None: from app.domains.cmdb.providers.gcp import is_gcp_system_project assert is_gcp_system_project("sys-98903879594750044794161370") is True assert is_gcp_system_project("sys-") is True # degenerate but matches assert is_gcp_system_project("encoded-net-496415-d6") is False assert is_gcp_system_project("my-systems-prod") is False assert is_gcp_system_project("") is False assert is_gcp_system_project(None) is False @pytest.mark.asyncio async def test_gcp_driver_list_scopes_skips_sys_projects() -> None: """``list_scopes`` MUST drop ``sys-*`` projects so they never become ``cmdb_cloud_scopes`` rows; downstream ``grants_sync`` / ``resource_sync`` then can't 403 on them. """ from app.integrations.drivers import gcp_cloud driver = gcp_cloud.GcpCloudDriver( integration_id="x", config={"service_account_json": "{}", "organization_id": "1"}, ) fake_crm = _parent_bounded_crm( [ {"projectId": "real-prod-app", "displayName": "Real prod", "parent": "organizations/1"}, { "projectId": "sys-12345678901234567890", "displayName": "Apigee internal", "parent": "organizations/1", }, { "projectId": "sys-99887766554433221100", "displayName": "Firebase internal", "parent": "organizations/1", }, { "projectId": "another-real-one", "displayName": "Another real one", "parent": "organizations/1", }, ] ) with patch.object(driver, "_svc", return_value=fake_crm): scopes = await driver.list_scopes() ids = {s.external_id for s in scopes} assert ids == {"real-prod-app", "another-real-one"} def test_gcp_provider_list_active_projects_skips_sys_projects() -> None: """Same filter at the CMDB-provider layer (which ``cmdb.resource_sync`` consumes via ``_fan_out_per_project``).""" from app.domains.cmdb.providers.base import ProviderCredentials from app.domains.cmdb.providers.gcp import GcpProvider with patch( "app.domains.cmdb.providers.gcp.service_account.Credentials.from_service_account_info", return_value=MagicMock(), ): p = GcpProvider( ProviderCredentials( provider_id="x", provider_type="gcp", account_alias="x", fields={"service_account_json": "{}", "organization_id": "1"}, ) ) fake_crm = _parent_bounded_crm( [ {"projectId": "real-prod-app", "parent": "organizations/1"}, {"projectId": "sys-1234", "parent": "organizations/1"}, {"projectId": "real-prod-data", "parent": "organizations/1"}, ] ) with patch.object(p, "_svc", return_value=fake_crm): out = p._list_active_projects_sync() assert out == ["real-prod-app", "real-prod-data"] # ─────── 3b. GCP personal/sandbox filter (incident 2026-05-16) ──────── def test_is_gcp_personal_sandbox_project_matches_known_patterns() -> None: """Personal-project filter — defensive fix for the 2026-05-16 GCP sync incident that flooded ``cmdb_cloud_scopes`` with 1028 rows. Matches the three signals enumerated in the helper's docstring: ``gen-lang-client-*`` (Gemini auto-create), ``My First Project`` (Cloud Console default name), and ``Untitled project*`` (alternate default). Crucially, does NOT match legit business projects even when they happen to be org-root parented — parent alone is too weak a signal. """ from app.domains.cmdb.providers.gcp import is_gcp_personal_sandbox_project # Gemini auto-created assert ( is_gcp_personal_sandbox_project( { "projectId": "gen-lang-client-0923496505", "displayName": "Default Gemini Project", "parent": "organizations/478164292948", } ) is True ) # GCP Console default name assert ( is_gcp_personal_sandbox_project( { "projectId": "abiding-window-424223-t9", "displayName": "My First Project", "parent": "organizations/478164292948", } ) is True ) # Untitled — exact match assert ( is_gcp_personal_sandbox_project( { "projectId": "random-id-1", "displayName": "Untitled project", "parent": "organizations/478164292948", } ) is True ) # Untitled — with trailing variant (e.g. "Untitled project 2") assert ( is_gcp_personal_sandbox_project( { "projectId": "random-id-2", "displayName": "Untitled project 7", "parent": "organizations/478164292948", } ) is True ) # Legit business project — folder-parented assert ( is_gcp_personal_sandbox_project( { "projectId": "tidbcloud-prod-000", "displayName": "tidbcloud-prod-000", "parent": "folders/988345611051", } ) is False ) # Legit business project — org-root parented but non-matching name. # Intentionally NOT filtered (parent alone is too weak a signal — # some customers run org-root infra projects). assert ( is_gcp_personal_sandbox_project( { "projectId": "apac-professional-service", "displayName": "APAC Professional Service", "parent": "organizations/478164292948", } ) is False ) # Defensive: missing / wrong types assert is_gcp_personal_sandbox_project({}) is False assert is_gcp_personal_sandbox_project(None) is False # type: ignore[arg-type] assert is_gcp_personal_sandbox_project({"projectId": None}) is False assert is_gcp_personal_sandbox_project({"projectId": 123}) is False assert is_gcp_personal_sandbox_project({"displayName": None}) is False @pytest.mark.asyncio async def test_gcp_driver_list_scopes_skips_personal_sandbox_projects() -> None: """``list_scopes`` MUST drop personal / sandbox projects so they never become ``cmdb_cloud_scopes`` rows. Root-cause fix for the 2026-05-16 incident. """ from app.integrations.drivers import gcp_cloud driver = gcp_cloud.GcpCloudDriver( integration_id="x", config={ "service_account_json": "{}", "organization_id": "478164292948", }, ) fake_crm = _parent_bounded_crm( [ # legit business projects — kept { "projectId": "tidbcloud-prod-000", "displayName": "tidbcloud-prod-000", "parent": "folders/988345611051", }, { "projectId": "apac-professional-service", "displayName": "APAC Professional Service", "parent": "organizations/478164292948", }, # Gemini auto-create — dropped { "projectId": "gen-lang-client-0923496505", "displayName": "Default Gemini Project", "parent": "organizations/478164292948", }, # Cloud Console default name — dropped { "projectId": "abiding-window-424223-t9", "displayName": "My First Project", "parent": "organizations/478164292948", }, # Untitled variant — dropped { "projectId": "random-id-7", "displayName": "Untitled project 3", "parent": "organizations/478164292948", }, # sys-* (existing filter — still dropped) { "projectId": "sys-12345678901234567890", "displayName": "Apigee internal", "parent": "organizations/478164292948", }, ], organization_id="478164292948", ) with patch.object(driver, "_svc", return_value=fake_crm): scopes = await driver.list_scopes() ids = {s.external_id for s in scopes} assert ids == {"tidbcloud-prod-000", "apac-professional-service"} def test_gcp_provider_list_active_projects_skips_personal_sandbox_projects() -> None: """Same filter at the CMDB-provider layer (``resource_sync`` fans out via this). Two layers must filter independently because they consume different surfaces (driver writes scope rows; provider fans out per-project API calls).""" from app.domains.cmdb.providers.base import ProviderCredentials from app.domains.cmdb.providers.gcp import GcpProvider with patch( "app.domains.cmdb.providers.gcp.service_account.Credentials.from_service_account_info", return_value=MagicMock(), ): p = GcpProvider( ProviderCredentials( provider_id="x", provider_type="gcp", account_alias="x", fields={ "service_account_json": "{}", "organization_id": "478164292948", }, ) ) fake_crm = _parent_bounded_crm( [ { "projectId": "tidbcloud-prod-000", "displayName": "tidbcloud-prod-000", "parent": "folders/988345611051", }, { "projectId": "gen-lang-client-9999", "displayName": "x", "parent": "organizations/478164292948", }, { "projectId": "real-id-1", "displayName": "My First Project", "parent": "organizations/478164292948", }, { "projectId": "real-id-2", "displayName": "Untitled project 12", "parent": "organizations/478164292948", }, {"projectId": "sys-1234", "parent": "organizations/478164292948"}, ], organization_id="478164292948", ) with patch.object(p, "_svc", return_value=fake_crm): out = p._list_active_projects_sync() assert out == ["tidbcloud-prod-000"] def test_gcp_driver_list_grants_returns_empty_for_gen_lang_client_project() -> None: """Defense-in-depth at the grants surface — if an old ``cmdb_cloud_scopes`` row for a ``gen-lang-client-*`` project survived the 2026-05-16 cleanup, the grants call MUST bail without invoking ``getIamPolicy``. The early-return fires BEFORE ``_svc`` so unit tests can exercise it without faking google-auth. """ from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers import gcp_cloud driver = gcp_cloud.GcpCloudDriver( integration_id="x", config={"service_account_json": "{}", "organization_id": "1"}, ) out = driver._list_scope_grants_sync( CloudScopeRef( scope_type="gcp_project", external_id="gen-lang-client-0923496505", display_name="Default Gemini Project", ), ) assert out == [] def test_gcp_driver_list_grants_returns_empty_for_sys_project() -> None: """Defense-in-depth at the grants surface — an existing ``cmdb_cloud_scopes`` row for a ``sys-*`` project (created by an earlier driver version) must NOT call ``getIamPolicy`` and must return ``[]`` so grants_sync doesn't 403. The sys-* check fires BEFORE ``_svc`` so the test does not need to fake google-auth — important: this verifies the optimization that bails early. """ from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers import gcp_cloud driver = gcp_cloud.GcpCloudDriver( integration_id="x", config={"service_account_json": "{}", "organization_id": "1"}, ) # If the early-return guard regressed, this call would attempt to # parse the empty SA JSON and raise MalformedError — fixture # depends on that observable behaviour. out = driver._list_scope_grants_sync( CloudScopeRef( scope_type="gcp_project", external_id="sys-98903879594750044794161370", display_name="Apigee internal", ), ) assert out == [] # ───────────────────────── 4. Tencent Page/Rp ───────────────────────── def test_tencent_list_scope_grants_paginates_attached_policies() -> None: """``ListAttachedUserAllPolicies`` MUST pass ``Page`` and ``Rp`` — the API rejects with ``MissingParameter Page`` otherwise. The driver pages until ``TotalNum`` is exhausted so a >200-policy user doesn't silently truncate.""" from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers import tencent_cloud driver = tencent_cloud.TencentCloudDriver( integration_id="x", config={"secret_id": "sk-id", "secret_key": "sk-key"}, ) # Track every CAM call so we can assert Page/Rp shape. captured: list[Any] = [] class _FakeListUsersResp: Data = [ type( "U", (), { "Name": "alice@example.com", "Email": "alice@example.com", "Uin": 100000001, }, )(), ] class _FakePolicy: def __init__(self, name: str, pid: int) -> None: self.PolicyName = name self.PolicyId = pid self.PolicyType = "User" page_responses = { 1: type( "R", (), { "PolicyList": [_FakePolicy(f"P{i}", i) for i in range(200)], "TotalNum": 350, }, )(), 2: type( "R", (), { "PolicyList": [_FakePolicy(f"P{i}", i) for i in range(200, 350)], "TotalNum": 350, }, )(), } def _fake_call(fn, req, *_args: Any, **_kwargs: Any) -> Any: captured.append((fn.__name__, req)) if fn.__name__ == "ListUsers": return _FakeListUsersResp() if fn.__name__ == "ListAttachedUserAllPolicies": return page_responses[req.Page] raise AssertionError(f"unexpected call {fn.__name__}") fake_client = MagicMock() fake_client.ListUsers.__name__ = "ListUsers" fake_client.ListAttachedUserAllPolicies.__name__ = "ListAttachedUserAllPolicies" with ( patch.object(driver, "_cam_client", return_value=fake_client), patch.object(driver, "_call", side_effect=_fake_call), ): grants = driver._list_scope_grants_sync( CloudScopeRef( scope_type="standalone_iam", external_id="100000001", display_name="acme", ), ) # Two ListAttachedUserAllPolicies calls (200 + 150 → TotalNum=350) policy_calls = [c for c in captured if c[0] == "ListAttachedUserAllPolicies"] assert len(policy_calls) == 2 assert policy_calls[0][1].Page == 1 assert policy_calls[0][1].Rp == 200 assert policy_calls[1][1].Page == 2 assert policy_calls[1][1].Rp == 200 # All 350 policies surfaced — no truncation. assert len(grants) == 350 # ───────────────────────── 5. Structural skip degrade ───────────────── def test_is_structural_all_legs_auth_failure_helper() -> None: from app.domains.cmdb.sync import _is_structural_all_legs_auth_failure # 100% auth_error → structural assert ( _is_structural_all_legs_auth_failure( [ {"leg": "sub:1", "reason": "auth_error"}, {"leg": "sub:2", "reason": "auth_error"}, ] ) is True ) # 100% http_403 → structural (also the GCP shape) assert ( _is_structural_all_legs_auth_failure( [ {"leg": "project:p1", "reason": "buckets.list_http_403"}, {"leg": "project:p2", "reason": "buckets.list_http_403"}, ] ) is True ) # Mixed bag with one transient → NOT structural (keep partial) assert ( _is_structural_all_legs_auth_failure( [ {"leg": "sub:1", "reason": "auth_error"}, {"leg": "sub:2", "reason": "InternalError"}, ] ) is False ) # Empty input → not structural assert _is_structural_all_legs_auth_failure([]) is False def test_structural_auth_regex_covers_other_endpoints() -> None: """Round-21 logic review caught the curated set was object_storage- only. A GCP integration whose SA can't reach Cloud DNS at all returns 100% ``managedZones.list_http_403`` — must also degrade to ``skipped``, not flood admin daily. Regex covers ``_http_(401|403)`` for any provider/endpoint. """ from app.domains.cmdb.sync import _is_structural_all_legs_auth_failure # GCP Cloud DNS — 403 on every project (SA missing dns.managedZones.list) assert ( _is_structural_all_legs_auth_failure( [ {"leg": "project:p1", "reason": "managedZones.list_http_403"}, {"leg": "project:p2", "reason": "managedZones.list_http_403"}, ] ) is True ) # GCP DNS rrsets — 403 assert ( _is_structural_all_legs_auth_failure( [ {"leg": "project:p1/zone:z1", "reason": "rrsets.list_http_403"}, ] ) is True ) # Future driver — IAM users 401 path assert ( _is_structural_all_legs_auth_failure( [ {"leg": "sub:s1", "reason": "iam.users.list_http_401"}, ] ) is True ) # 5xx must NOT degrade — that's a transient cloud-side outage that # admin needs to see (regex is restricted to 4xx). assert ( _is_structural_all_legs_auth_failure( [ {"leg": "project:p1", "reason": "managedZones.list_http_500"}, {"leg": "project:p2", "reason": "managedZones.list_http_500"}, ] ) is False ) # Mixed (one 5xx + one 403) — transient mixed in keeps partial_failure assert ( _is_structural_all_legs_auth_failure( [ {"leg": "project:p1", "reason": "buckets.list_http_403"}, {"leg": "project:p2", "reason": "buckets.list_http_500"}, ] ) is False ) def test_summarize_leg_failures_deterministic_ordering() -> None: """Round-21 logic review: ordering before slicing must be stable across permutations so the failure-notifier dedup signature doesn't flip when per-region fan-out ordering varies between cron ticks. Same {reason: count} multiset → same rendered summary regardless of input leg order. """ from app.integrations.drivers._cloud_shared import _summarize_leg_failures # Same data, different input order. legs_a = [{"leg": f"r{i}", "reason": "alpha"} for i in range(3)] + [ {"leg": f"r{i}", "reason": "beta"} for i in range(7) ] legs_b = list(reversed(legs_a)) # leg order flipped assert _summarize_leg_failures(legs_a) == _summarize_leg_failures(legs_b) # And the higher-count reason renders first (count desc). out = _summarize_leg_failures(legs_a) beta_idx = out.find("beta") alpha_idx = out.find("alpha") assert beta_idx >= 0 and alpha_idx >= 0 assert beta_idx < alpha_idx # ───────────────────────── 6. test_connection per-capability ────────── @pytest.mark.asyncio async def test_test_connection_failed_when_one_capability_all_auth_error() -> None: """Pre-fix: a cloud integration with one healthy capability and one 100%-auth_error capability returned ``ok=True`` with ``OK · enumerated 692 servers``. Post-fix: the failed capability flips the top-level ``ok=False`` and the response detail carries the per-capability probe breakdown. """ from app.domains.cmdb.providers.base import CloudProvider from app.integrations.drivers._cloud_shared import _CloudDriverBase from app.integrations.schemas import IntegrationTestResult class _FakeProvider: provider_id = "x" provider_type = "fake" def __init__(self) -> None: # MUST be an instance attribute — sharing across tests via # a class-level list would cross-contaminate fixtures. self.partial_failures: list[dict[str, Any]] = [] async def list_servers(self) -> list[dict[str, Any]]: return [{"id": "i-1"}, {"id": "i-2"}] async def list_object_storage(self) -> list[dict[str, Any]]: # Simulate the AliCloud-intl prod shape: every leg returns # auth_error, zero rows enumerated. self.partial_failures.extend( [ {"leg": "sub_account:5060778679810487", "reason": "auth_error"}, {"leg": "sub_account:5531977013674801", "reason": "auth_error"}, ] ) return [] async def close(self) -> None: return None def supports(self, _cap: Any) -> bool: return True class _FakeDriver(_CloudDriverBase): provider = "fake" provider_cls = CloudProvider # not used because we patch _provider driver = _FakeDriver( integration_id="i", config={"_alias": "fake"}, enabled_capabilities=["servers", "object_storage"], ) with patch.object(driver, "_provider", return_value=_FakeProvider()): result = await driver.test_connection() assert isinstance(result, IntegrationTestResult) assert result.ok is False, result.message probes = result.detail["capability_probes"] assert probes["servers"]["status"] == "ok" assert probes["servers"]["sample_count"] == 2 assert probes["object_storage"]["status"] == "failed" assert "auth_error" in (probes["object_storage"].get("reason") or "") @pytest.mark.asyncio async def test_test_connection_excludes_billing_and_user_revoke_from_probe() -> None: """Money-cost (billing / cost_facts) and write-side (user_revoke) capabilities MUST stay out of the probe loop — clicking ``测试`` must not burn BigQuery dollars or trigger an offboarding call.""" from app.domains.cmdb.providers.base import CloudProvider from app.integrations.drivers._cloud_shared import _CloudDriverBase list_called: list[str] = [] class _FakeProvider: def __init__(self) -> None: self.partial_failures: list[dict[str, Any]] = [] async def list_servers(self) -> list[dict[str, Any]]: list_called.append("servers") return [] async def list_object_storage(self) -> list[dict[str, Any]]: list_called.append("object_storage") return [] # If the probe ever reaches these we'd crash because they don't # exist; the test asserts they were never called. def supports(self, _cap: Any) -> bool: return True async def close(self) -> None: return None class _FakeDriver(_CloudDriverBase): provider = "fake" provider_cls = CloudProvider driver = _FakeDriver( integration_id="i", config={"_alias": "fake"}, enabled_capabilities=[ "servers", "object_storage", "user_revoke", "billing", "cost_facts", ], ) with patch.object(driver, "_provider", return_value=_FakeProvider()): result = await driver.test_connection() assert "user_revoke" not in result.detail["capability_probes"] assert "billing" not in result.detail["capability_probes"] assert "cost_facts" not in result.detail["capability_probes"] assert list_called == ["servers", "object_storage"] # ───────────────────────── 7. resource_sync error_message summary ───── def test_summarize_leg_failures_groups_by_reason() -> None: """The summary helper groups by reason — same shape the resource_sync raise now uses to embed actionable detail in the error_message (admins no longer SQL-query ``cmdb_sync_jobs.capability_results``).""" from app.integrations.drivers._cloud_shared import _summarize_leg_failures out = _summarize_leg_failures( [ {"leg": "project:sys-AAA", "reason": "buckets.list_http_403"}, {"leg": "project:sys-BBB", "reason": "buckets.list_http_403"}, {"leg": "project:real-prod", "reason": "buckets.list_http_403"}, {"leg": "project:other-prod", "reason": "InternalError"}, ] ) # Reason grouping intact; counts accurate. assert "buckets.list_http_403 × 3" in out assert "InternalError × 1" in out def test_resource_sync_raise_text_shape() -> None: """The handler must format the raise message as ``cmdb.resource_sync /: · leg(s): `` so the failure-notifier dedup signature key bites on the structural shape (e.g. ``buckets.list_http_403``) not on a generic ``status=partial_failure`` line. Verifying the literal format string is brittle — instead we exercise the helper the raise composes from and assert it produces the expected tokens. """ import inspect from app.integrations.drivers._cloud_shared import _summarize_leg_failures from app.scheduled_jobs import handlers legs = [ {"leg": "project:sys-AAA", "reason": "buckets.list_http_403"}, {"leg": "project:sys-BBB", "reason": "buckets.list_http_403"}, ] summary = _summarize_leg_failures(legs) assert "buckets.list_http_403 × 2" in summary # Belt-and-braces: the handler's source must reference the helper # so a future refactor doesn't silently drop the summary call. src = inspect.getsource(handlers.run_resource_sync) assert "_summarize_leg_failures" in src assert "partial_failures" in src assert "leg(s)" in src