"""Tests for the resource-sync safety-floor admin override. User report 2026-05-14: ``cmdb.resource_sync`` for AWS-SG IAM users failed with ``provider returned 9 rows vs 20 existing (>5% drop) — refusing to sweep``; clicking 「标记为已确认」 hid the alert but the next cron tick refailed identically and the failure_notifier re-paged on every tick. The fix is a separate admin action that re-runs sync with ``force_sweep=True`` so the missing rows are finally soft-deleted; ack remains "I saw this", force-sweep is "and I accept the new data". Coverage: * ``_safe_sweep`` unit tests - 5%-drop guard raises by default but is bypassed when force_sweep=True - 0-rows guard raises by default but is bypassed when force_sweep=True - forced_meta is populated with which guard tripped and the existing count - Healthy sync (no guard trip) does NOT set forced_meta even with force=True * ``POST /api/v1/cmdb/sync/{id}/force-sweep`` endpoint tests - 403 for non-admin - 404 for unknown / non-cloud integration - 422 for empty resource_types (Pydantic min_length=1) - 422 for capability not supported by the integration's provider - 200 happy path with audit row written + job row visible """ from __future__ import annotations import uuid from datetime import UTC, datetime from typing import Any from unittest.mock import AsyncMock import pytest from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.audit.models import AuditLog from app.domains.cmdb.models import CmdbIamUser, CmdbSyncJob from app.domains.cmdb.providers.exceptions import ProviderTransientError from app.domains.cmdb.sync import _safe_sweep from app.integrations.models import Integration # ── Helpers ───────────────────────────────────────────────────── async def _make_integration( db: AsyncSession, *, enabled: bool = True, capabilities: list[str] | None = None, provider: str = "aws_cloud", ) -> Integration: iid = uuid.uuid4() integ = Integration( id=iid, kind="cloud", provider=provider, name=f"force-sweep-{str(iid)[:6]}", slug=f"fs-{str(iid)[:6]}", enabled=enabled, is_primary=False, capabilities=capabilities, # secrets_* left null — we patch ``run_sync`` so the provider # is never loaded; the row only needs to satisfy # ``_load_cloud_integration`` (kind='cloud') and the enabled # check. ) db.add(integ) await db.commit() return integ async def _make_iam_user( db: AsyncSession, integration: Integration, *, user_name: str, last_seen_sync_job_id: str | None = None, ) -> CmdbIamUser: row = CmdbIamUser( id=uuid.uuid4(), integration_id=str(integration.id), user_name=user_name, sub_account_id="", arn=f"arn:aws:iam::123:user/{user_name}", user_id_external=f"AIDA{uuid.uuid4().hex.upper()[:16]}", attached_policies=[], groups=[], mfa_enabled=False, tags={}, raw_data={}, last_seen_sync_job_id=last_seen_sync_job_id, last_synced_at=datetime.now(UTC), ) db.add(row) await db.commit() return row # ── Unit tests for _safe_sweep ────────────────────────────────── @pytest.mark.asyncio async def test_safe_sweep_drop_ratio_raises_by_default( db_session: AsyncSession, ): """20 existing rows, 1 row 'upserted' this run → 95% drop, beyond any sane tolerance value → raise (default 20% tolerance → floor=16, 1<16).""" integ = await _make_integration(db_session) # Stamp every existing row with a *prior* job id so the sweep # update considers them eligible for soft-delete in the next run. prior_job_id = str(uuid.uuid4()) for i in range(20): await _make_iam_user( db_session, integ, user_name=f"alice-{i}", last_seen_sync_job_id=prior_job_id, ) new_job_id = str(uuid.uuid4()) with pytest.raises(ProviderTransientError) as exc: await _safe_sweep( db_session, model=CmdbIamUser, integration_id=str(integ.id), job_id=new_job_id, upserted=1, # 1 < int(20 * 0.80) = 16 → drop-tolerance guard trips capability_label="iam_users", ) assert "refusing to sweep" in str(exc.value) # No rows were actually soft-deleted. refreshed = (await db_session.execute( select(CmdbIamUser).where( CmdbIamUser.integration_id == str(integ.id), CmdbIamUser.deleted_at.is_(None), ) )).scalars().all() assert len(refreshed) == 20 @pytest.mark.asyncio async def test_safe_sweep_drop_ratio_bypassed_with_force_sweep( db_session: AsyncSession, ): """Same setup as above, but force_sweep=True bypasses the guard and the 19 stale rows get soft-deleted (1 was 'upserted' this run).""" integ = await _make_integration(db_session) prior_job_id = str(uuid.uuid4()) new_job_id = str(uuid.uuid4()) # 1 row stamped with the NEW job id (i.e. "upserted this run"), # 19 rows stamped with the PRIOR job id (i.e. missing from this run). await _make_iam_user( db_session, integ, user_name="alice-0", last_seen_sync_job_id=new_job_id, ) for i in range(1, 20): await _make_iam_user( db_session, integ, user_name=f"alice-{i}", last_seen_sync_job_id=prior_job_id, ) forced_meta: dict[str, Any] = {} deleted = await _safe_sweep( db_session, model=CmdbIamUser, integration_id=str(integ.id), job_id=new_job_id, upserted=1, capability_label="iam_users", force_sweep=True, forced_meta=forced_meta, ) await db_session.commit() assert deleted == 19, f"expected 19 soft-deletes, got {deleted}" # Meta surface: which guard tripped + baseline source + pre-sweep # counts + the tolerance that was active at trip time (so a future # audit can reconstruct what threshold the admin chose to bypass). # No ``capability_key`` was passed → falls back to the ``existing`` # count comparison (still 20 here). assert forced_meta == { "forced_sweep_guard": "drop_ratio", "forced_sweep_baseline": 20, "forced_sweep_baseline_source": "existing", "forced_sweep_existing": 20, "forced_sweep_upserted": 1, "forced_sweep_observed": 1, # Resolver default (no platform_settings row in the test DB + # settings default of 20). Pinning this asserts the audit # surface stays self-describing — a future tolerance change # would surface immediately in this test. "forced_sweep_tolerance_percent": 20, } # The single row stamped with the new job_id survived. survivors = (await db_session.execute( select(CmdbIamUser).where( CmdbIamUser.integration_id == str(integ.id), CmdbIamUser.deleted_at.is_(None), ) )).scalars().all() assert len(survivors) == 1 assert survivors[0].user_name == "alice-0" @pytest.mark.asyncio async def test_safe_sweep_zero_rows_guard_raises_by_default( db_session: AsyncSession, ): """``upserted=0`` with existing rows is the partial-driver-failure case — almost always a swallowed exception, not real deletion.""" integ = await _make_integration(db_session) for i in range(5): await _make_iam_user( db_session, integ, user_name=f"bob-{i}", last_seen_sync_job_id=str(uuid.uuid4()), ) with pytest.raises(ProviderTransientError) as exc: await _safe_sweep( db_session, model=CmdbIamUser, integration_id=str(integ.id), job_id=str(uuid.uuid4()), upserted=0, capability_label="iam_users", ) assert "provider returned 0 rows" in str(exc.value) @pytest.mark.asyncio async def test_safe_sweep_zero_rows_guard_bypassed_with_force_sweep( db_session: AsyncSession, ): integ = await _make_integration(db_session) prior_job_id = str(uuid.uuid4()) for i in range(5): await _make_iam_user( db_session, integ, user_name=f"bob-{i}", last_seen_sync_job_id=prior_job_id, ) new_job_id = str(uuid.uuid4()) forced_meta: dict[str, Any] = {} deleted = await _safe_sweep( db_session, model=CmdbIamUser, integration_id=str(integ.id), job_id=new_job_id, upserted=0, capability_label="iam_users", force_sweep=True, forced_meta=forced_meta, ) await db_session.commit() assert deleted == 5 assert forced_meta == { "forced_sweep_guard": "zero_rows", "forced_sweep_baseline": 5, "forced_sweep_baseline_source": "existing", "forced_sweep_existing": 5, } @pytest.mark.asyncio async def test_safe_sweep_no_guard_trip_leaves_forced_meta_empty( db_session: AsyncSession, ): """Healthy sync (no drop): force_sweep=True is a no-op signal — forced_meta stays empty so the UI doesn't flag "forced" on a row that wasn't actually overridden.""" integ = await _make_integration(db_session) prior_job_id = str(uuid.uuid4()) new_job_id = str(uuid.uuid4()) # 10 rows, all stamped with the new job_id (this sync saw all of # them) → ``upserted=10``, ``existing=10``, 0% drop. for i in range(10): await _make_iam_user( db_session, integ, user_name=f"carol-{i}", last_seen_sync_job_id=new_job_id, ) # Plus one stale row that genuinely should be swept regardless of # the floor — keeps ``upserted < existing`` strict (11 existing, 10 # upserted; 90.9% of existing, well above the 95% floor → no trip). # Actually: 10 / 11 = 90.9% < 95% → would trip. To avoid the trip, # pad upserted to 11 (everyone present) and have NO stale row. _ = prior_job_id # unused — kept for symmetry with the drop tests forced_meta: dict[str, Any] = {} deleted = await _safe_sweep( db_session, model=CmdbIamUser, integration_id=str(integ.id), job_id=new_job_id, upserted=10, # equals existing — 0% drop, no guard capability_label="iam_users", force_sweep=True, forced_meta=forced_meta, ) await db_session.commit() assert deleted == 0 assert forced_meta == {}, ( "force_sweep=True must not flag the row when no guard tripped" ) # ── HTTP endpoint tests ───────────────────────────────────────── @pytest.fixture def _patched_session_factory(db_engine, monkeypatch): """Point ``async_session_factory`` at the test engine so the endpoint's fresh session writes see the fixture rows. Patches the source module (``app.core.database``) AND every downstream module that does ``from app.core.database import async_session_factory`` at import time — those modules captured the original factory object at module load, so rebinding the db_module attribute alone leaves their local names stale. """ factory = async_sessionmaker(db_engine, expire_on_commit=False) from app.core import database as db_module monkeypatch.setattr(db_module, "async_session_factory", factory) # Modules that did ``from app.core.database import async_session_factory`` # at import time hold their own name binding to the original factory. # Patch each one we exercise so the real ``run_sync_with_lock`` / # ``run_sync`` cross-session writes land in the test DB. from app.domains.cmdb import sync as cmdb_sync_module monkeypatch.setattr( cmdb_sync_module, "async_session_factory", factory, ) return factory @pytest.mark.asyncio async def test_force_sweep_endpoint_rejects_employee( client: AsyncClient, db_session: AsyncSession, employee_headers, ): integ = await _make_integration(db_session, capabilities=["iam"]) resp = await client.post( f"/api/v1/cmdb/sync/{integ.id}/force-sweep", headers=employee_headers, json={"resource_types": ["iam"]}, ) assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_force_sweep_endpoint_404_on_unknown_integration( client: AsyncClient, admin_headers, ): resp = await client.post( f"/api/v1/cmdb/sync/{uuid.uuid4()}/force-sweep", headers=admin_headers, json={"resource_types": ["iam"]}, ) assert resp.status_code == 404, resp.text @pytest.mark.asyncio async def test_force_sweep_endpoint_requires_non_empty_resource_types( client: AsyncClient, db_session: AsyncSession, admin_headers, ): integ = await _make_integration(db_session, capabilities=["iam"]) resp = await client.post( f"/api/v1/cmdb/sync/{integ.id}/force-sweep", headers=admin_headers, json={"resource_types": []}, ) # Pydantic min_length=1 → 422 from FastAPI validation layer. assert resp.status_code == 422, resp.text @pytest.mark.asyncio async def test_force_sweep_endpoint_rejects_unsupported_capability( client: AsyncClient, db_session: AsyncSession, admin_headers, _patched_session_factory, ): # Provider supports only servers (capabilities=["servers"]); admin # asks to force-sweep iam → 422 (project ``ValidationError`` → # status_code=422) with the standard error envelope. integ = await _make_integration(db_session, capabilities=["servers"]) resp = await client.post( f"/api/v1/cmdb/sync/{integ.id}/force-sweep", headers=admin_headers, json={"resource_types": ["iam"]}, ) assert resp.status_code == 422, resp.text body = resp.json() # Project's AppException envelope, not FastAPI's auto-validation one. assert body["success"] is False err = body["errors"][0] assert err["code"] == "VALIDATION_ERROR" assert "not supported" in err["message"].lower() assert "iam" in err["message"] @pytest.mark.asyncio async def test_force_sweep_endpoint_happy_path_writes_audit_and_queues_job( client: AsyncClient, db_session: AsyncSession, admin_headers, monkeypatch, _patched_session_factory, ): """202 + initial queued job row + audit committed at request time. The endpoint went async on 2026-05-18: it now returns the freshly- created job row (``status='queued'``) immediately and queues the real sync as a FastAPI ``BackgroundTask``. The frontend polls ``/cmdb/sync-jobs/{id}`` to observe terminal status. The BG task runs inside the ASGI cycle that ``httpx.AsyncClient`` drives, so its side-effects (job-row update via the patched session factory) are observable by the time ``await client.post(...)`` returns. """ integ = await _make_integration(db_session, capabilities=["iam"]) # Stub ``run_sync_with_lock`` so neither the lock nor the provider # is touched — the real wrapper is covered by the unit test below. # This test exercises routing, RBAC, audit, response shape, and # that the BG task is actually scheduled and runs with the right # args. captured: dict[str, Any] = {} async def _fake_run_with_lock( integration_id, job_id, capabilities, triggered_by_user_id, *, force_sweep=False, lock_ttl=1200, ): captured["integration_id"] = integration_id captured["job_id"] = job_id captured["capabilities"] = capabilities captured["force_sweep"] = force_sweep captured["lock_ttl"] = lock_ttl # Update the job row to look like a successful sync. The # frontend polls ``/sync-jobs/{id}`` to observe this — we # assert the same state by re-reading from the DB below. async with _patched_session_factory() as worker_db: from sqlalchemy import update as _update from app.domains.cmdb.models import CmdbSyncJob as _Job await worker_db.execute( _update(_Job) .where(_Job.id == uuid.UUID(job_id)) .values( status="success", total_upserted=9, total_deleted=11, finished_at=datetime.now(UTC), capability_results={ "iam": { "status": "success", "upserted": 9, "deleted": 11, "forced_sweep": True, "forced_sweep_guard": "drop_ratio", "forced_sweep_existing": 20, "forced_sweep_upserted": 9, } }, ) ) await worker_db.commit() monkeypatch.setattr( "app.domains.cmdb.router.run_sync_with_lock", AsyncMock(side_effect=_fake_run_with_lock), ) resp = await client.post( f"/api/v1/cmdb/sync/{integ.id}/force-sweep", headers=admin_headers, json={"resource_types": ["iam"], "reason": "admin closed 11 users"}, ) # 202 Accepted — the BG task has been queued. assert resp.status_code == 202, resp.text body = resp.json() assert body["success"] is True job = body["data"] # Response carries the INITIAL job row (status='queued'). The BG # task fires after the response is sent. The committed-state of # the row is asserted below via the DB. assert job["triggered_by"] == "force_sweep" assert job["resource_types"] == ["iam"] initial_job_id = job["id"] # BG task was scheduled and ran with force_sweep=True assert captured["force_sweep"] is True assert captured["capabilities"] == ["iam"] assert captured["job_id"] == initial_job_id assert captured["lock_ttl"] == 1200 # Audit row written + tied to the requesting admin (audit happens # at request time, not at BG-task time — the decision is made now). audits = ( await db_session.execute( select(AuditLog).where( AuditLog.action == "cmdb.sync.force_sweep_override", AuditLog.resource_id == str(integ.id), ) ) ).scalars().all() assert len(audits) == 1 detail = audits[0].detail assert detail["integration_id"] == str(integ.id) assert detail["resource_types"] == ["iam"] assert detail["reason"] == "admin closed 11 users" assert detail["job_id"] == initial_job_id # Sync job row persisted; the BG task's commit (via the patched # session factory) updated it to status='success'. End the test # session's read snapshot before re-querying so we observe the # cross-session commit under MySQL REPEATABLE READ. await db_session.commit() persisted = (await db_session.execute( select(CmdbSyncJob).where( CmdbSyncJob.integration_id == str(integ.id), ) )).scalars().all() assert len(persisted) == 1 final = persisted[0] assert final.triggered_by == "force_sweep" assert final.status == "success" assert final.total_upserted == 9 assert final.total_deleted == 11 assert final.capability_results["iam"]["forced_sweep"] is True assert final.capability_results["iam"]["forced_sweep_guard"] == "drop_ratio" @pytest.mark.asyncio async def test_force_sweep_endpoint_404_on_disabled_integration( client: AsyncClient, db_session: AsyncSession, admin_headers, ): integ = await _make_integration( db_session, enabled=False, capabilities=["iam"], ) resp = await client.post( f"/api/v1/cmdb/sync/{integ.id}/force-sweep", headers=admin_headers, json={"resource_types": ["iam"]}, ) # ``_load_cloud_integration`` allows disabled rows; the explicit # ``not enabled`` check fires here. assert resp.status_code == 404, resp.text @pytest.mark.asyncio async def test_force_sweep_endpoint_lock_busy_marks_job_failed( client: AsyncClient, db_session: AsyncSession, admin_headers, monkeypatch, _patched_session_factory, ): """Lock contention is surfaced via the job row, not via a 409. Pre-2026-05-18 the endpoint ran sync synchronously inside the request and could raise ``CLOUD_SYNC_ALREADY_RUNNING`` 409 if the ``cloud_sync:`` lock was held by a concurrent scope / grants path. After the async refactor the request always returns 202 (the BG task hasn't tried the lock yet at response time), so contention is surfaced as ``status='failed'`` on the job row with a clear retry message — the polling UI shows it and the failure notifier dedup keeps the alert from re-paging. """ integ = await _make_integration(db_session, capabilities=["iam"]) # Stub ``run_sync_with_lock`` to mimic its LockBusy code path: # mark the job row failed with the canonical retry message. The # real wrapper's LockBusy branch is covered by the unit test # below. async def _fake_lock_busy( integration_id, job_id, capabilities, triggered_by_user_id, *, force_sweep=False, lock_ttl=600, ): from sqlalchemy import update as _update from app.domains.cmdb.models import CmdbSyncJob as _Job async with _patched_session_factory() as worker_db: await worker_db.execute( _update(_Job) .where(_Job.id == uuid.UUID(job_id)) .values( status="failed", finished_at=datetime.now(UTC), error_message=( "Another cloud-sync operation is already running " "for this integration; wait for it to complete " "and retry the force-sweep." ), ) ) await worker_db.commit() monkeypatch.setattr( "app.domains.cmdb.router.run_sync_with_lock", AsyncMock(side_effect=_fake_lock_busy), ) resp = await client.post( f"/api/v1/cmdb/sync/{integ.id}/force-sweep", headers=admin_headers, json={"resource_types": ["iam"]}, ) # Request always returns 202; outcome is on the job row. assert resp.status_code == 202, resp.text await db_session.commit() # drop snapshot before cross-session read persisted = (await db_session.execute( select(CmdbSyncJob).where( CmdbSyncJob.integration_id == str(integ.id), ) )).scalars().one() assert persisted.status == "failed" assert persisted.error_message is not None assert "already running" in persisted.error_message assert "retry" in persisted.error_message.lower() @pytest.mark.asyncio async def test_run_sync_with_lock_marks_job_failed_on_lock_busy( db_session: AsyncSession, monkeypatch, _patched_session_factory, ): """Unit test for ``run_sync_with_lock``'s LockBusy handling. The endpoint test above patches ``run_sync_with_lock`` wholesale; this test covers the wrapper itself: simulate a busy lock by patching ``app.core.distributed_lock.distributed_lock`` and assert the job row lands in the canonical ``status='failed'`` shape. """ integ = await _make_integration(db_session, capabilities=["iam"]) # Create a queued job row the wrapper will update. from app.domains.cmdb.service import CmdbService job = await CmdbService.create_sync_job( db_session, integration_id=str(integ.id), resource_types=["iam"], triggered_by_user_id=None, triggered_by="force_sweep", ) await db_session.commit() job_id = str(job.id) from contextlib import asynccontextmanager from app.core.distributed_lock import LockBusy @asynccontextmanager async def _busy_lock(_key: str, *, ttl: int = 1200): raise LockBusy(_key) yield # unreachable # The wrapper imports ``distributed_lock`` lazily inside the # function body, so we patch the source module — every late import # resolves through it. monkeypatch.setattr( "app.core.distributed_lock.distributed_lock", _busy_lock, ) # ``run_sync`` must NOT be reached when the lock is busy. from app.domains.cmdb import sync as cmdb_sync monkeypatch.setattr( cmdb_sync, "run_sync", AsyncMock(side_effect=AssertionError( "run_sync called despite busy lock", )), ) await cmdb_sync.run_sync_with_lock( str(integ.id), job_id, ["iam"], None, force_sweep=True, ) # ``CmdbService.create_sync_job`` inserted + refreshed the job in # ``db_session``'s identity map, so the cached ORM instance still # carries ``status='queued'``. The wrapper's UPDATE landed in a # different session (its own ``async_session_factory()`` opened # under the lock-busy branch); a plain ``select()`` over the same # PK reuses the cached object and does NOT refresh attributes from # the freshly-committed row. ``populate_existing`` forces the # attributes to be repopulated from the DB row, which is what we # actually need to observe here. await db_session.commit() # drop snapshot before cross-session read refreshed = (await db_session.execute( select(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .execution_options(populate_existing=True) )).scalars().one() assert refreshed.status == "failed" assert "already running" in (refreshed.error_message or "") assert refreshed.finished_at is not None # Parity with ``_fail_job`` — see ``_mark_force_sweep_job_failed``. assert refreshed.total_errors == 1 @pytest.mark.asyncio async def test_mark_force_sweep_job_failed_does_not_clobber_terminal_row( db_session: AsyncSession, _patched_session_factory, ): """``_mark_force_sweep_job_failed``'s ``WHERE status IN (queued, running)`` guard must no-op when the row already landed a terminal status — never overwrite a real ``success`` / ``partial_failure`` with a misleading ``failed`` (review feedback 2026-05-18; the practical race is impossible because job_ids are UUID4-unique per request, but the predicate is the contract). """ integ = await _make_integration(db_session, capabilities=["iam"]) from app.domains.cmdb.service import CmdbService job = await CmdbService.create_sync_job( db_session, integration_id=str(integ.id), resource_types=["iam"], triggered_by_user_id=None, triggered_by="force_sweep", ) # Land the row as ``success`` to simulate "wrapper hit the # last-resort branch AFTER run_sync already finalized". job.status = "success" job.total_upserted = 7 job.total_deleted = 0 job.finished_at = datetime.now(UTC) await db_session.commit() job_id = str(job.id) from app.domains.cmdb import sync as cmdb_sync await cmdb_sync._mark_force_sweep_job_failed( job_id, error_message="should be ignored — row already terminal", ) await db_session.commit() final = (await db_session.execute( select(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .execution_options(populate_existing=True) )).scalars().one() # Terminal row preserved verbatim — no clobber. assert final.status == "success" assert final.total_upserted == 7 assert final.error_message is None assert final.total_errors == 0 @pytest.mark.asyncio async def test_run_sync_with_lock_marks_job_failed_on_unexpected_exception( db_session: AsyncSession, monkeypatch, _patched_session_factory, ): """Unit test for ``run_sync_with_lock``'s last-resort exception branch. Review feedback 2026-05-18: without a bare ``except Exception`` around the lock acquisition + ``run_sync`` call, a Redis-down ``ConnectionRefusedError`` (or any other non-LockBusy failure at lock-acquire time) would kill the BG task silently — job row stranded as ``status='queued'`` and the failure notifier blind. Verify the last-resort branch materializes the failure on the job row. """ integ = await _make_integration(db_session, capabilities=["iam"]) from app.domains.cmdb.service import CmdbService job = await CmdbService.create_sync_job( db_session, integration_id=str(integ.id), resource_types=["iam"], triggered_by_user_id=None, triggered_by="force_sweep", ) await db_session.commit() job_id = str(job.id) from contextlib import asynccontextmanager @asynccontextmanager async def _broken_lock(_key: str, *, ttl: int = 1200): # Simulate any non-LockBusy failure (Redis down, programming # error). Critically NOT a ``LockBusy`` — must hit the # last-resort branch, not the LockBusy branch. raise ConnectionRefusedError("redis unreachable") yield # unreachable monkeypatch.setattr( "app.core.distributed_lock.distributed_lock", _broken_lock, ) from app.domains.cmdb import sync as cmdb_sync monkeypatch.setattr( cmdb_sync, "run_sync", AsyncMock(side_effect=AssertionError( "run_sync called despite broken lock", )), ) # Must NOT raise — the wrapper's last-resort except absorbs it. await cmdb_sync.run_sync_with_lock( str(integ.id), job_id, ["iam"], None, force_sweep=True, ) await db_session.commit() refreshed = (await db_session.execute( select(CmdbSyncJob) .where(CmdbSyncJob.id == uuid.UUID(job_id)) .execution_options(populate_existing=True) )).scalars().one() assert refreshed.status == "failed" assert refreshed.error_message is not None assert "BG task failed before sync ran" in refreshed.error_message # The visible error_message uses only the exception class name — # ``str(exc)`` (which can carry redis URIs with embedded passwords) # stays in worker logs, not in the admin-visible job row. assert "ConnectionRefusedError" in refreshed.error_message assert "redis unreachable" not in refreshed.error_message assert refreshed.finished_at is not None # Parity with ``_fail_job`` — the row must count as 1 error so the # sync-history UI doesn't show "failed but 0 errors". assert refreshed.total_errors == 1