""" PR #559 — server lifecycle endpoints (开机/关机/重启/删除). Covers the 4 ``POST /cmdb/servers/{id}/{action}`` routes that replicate the legacy CMDB's 开机/关机/重启/删除 action menu: * RBAC: ``can_manage_cloud`` required (it_admin / cloud_admin); employees and read-only auditors get 403. * 404 on unknown / soft-deleted server id. * Provider lookup via ``_load_provider_for_dns`` (helper name is historical — it loads any cloud provider, mocked here so we exercise the router branches without a real cloud API call). * ``CapabilityNotSupported`` raised by the driver surfaces 501 with code ``CLOUD_LIFECYCLE_NOT_SUPPORTED`` (matches the legacy reinstall/ticket paths that aren't implemented for every provider). * Successful power-on/off/restart bumps the row's ``status`` to the optimistic transitional value and stamps the modifier fields; the next sync cron reconciles to the cloud-authoritative state. * Successful destroy ALSO soft-deletes the row (``deleted_at`` set) so it drops from the active list immediately and lands in the recycle bin. * Audit log row written on every call (success AND failure paths) — governance + auditor evidence trail. """ import uuid from unittest.mock import AsyncMock, patch import pytest import pytest_asyncio from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.models import AuditLog from app.domains.cmdb.models import CmdbServer from app.domains.cmdb.providers.exceptions import CapabilityNotSupported from app.integrations.models import Integration @pytest_asyncio.fixture async def integ(db_session: AsyncSession) -> Integration: integration_id = uuid.uuid4() integ = Integration( id=integration_id, kind="cloud", provider="ucloud", name="pytest-fixture-ucloud-lifecycle", slug=f"pytest-fixture-ucloud-lifecycle-{str(integration_id)[:8]}", enabled=True, is_primary=False, capabilities=["SERVERS"], ) db_session.add(integ) await db_session.commit() return integ @pytest_asyncio.fixture async def server_row(db_session: AsyncSession, integ: Integration) -> CmdbServer: s = CmdbServer( integration_id=integ.id, instance_id="uhost-test-559", name="lifecycle-host", status="running", region="cn-bj2", project_id="org-test559", public_ips=["1.2.3.4"], private_ips=["10.9.0.1"], ) db_session.add(s) await db_session.commit() return s def _mock_provider(method_name: str, *, side_effect=None): """Build an AsyncMock CloudProvider with one stubbed lifecycle method. The other 3 lifecycle methods are also stubbed so any test that routes to the wrong method gets a clear AttributeError instead of a confusing AsyncMock auto-attribute return value. """ provider = AsyncMock() provider.power_on = AsyncMock() provider.power_off = AsyncMock() provider.restart_server = AsyncMock() provider.destroy_server = AsyncMock() if side_effect is not None: getattr(provider, method_name).side_effect = side_effect provider.close = AsyncMock(return_value=None) return provider # ───────────────────────────────────────────────────────────────────── # Success paths — each of 4 actions # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.parametrize( "action,method,transitional_status", [ ("power-on", "power_on", "starting"), ("power-off", "power_off", "stopping"), ("restart", "restart_server", "rebooting"), ("destroy", "destroy_server", "terminated"), ], ) async def test_lifecycle_success_updates_status_and_stamps_modifier( client, admin_user, server_row, db_session, action, method, transitional_status, ): person, token = admin_user provider = _mock_provider(method) with patch( "app.domains.cmdb.router._load_provider_for_dns", new_callable=AsyncMock, ) as mock_loader: mock_loader.return_value = provider resp = await client.post( f"/api/v1/cmdb/servers/{server_row.id}/{action}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text # Driver received instance_id + region + project_id straight off the row. getattr(provider, method).assert_awaited_once() call_kwargs = getattr(provider, method).await_args.kwargs assert call_kwargs["instance_id"] == "uhost-test-559" assert call_kwargs["region"] == "cn-bj2" assert call_kwargs["project_id"] == "org-test559" # Status flipped to the transitional value + modifier stamped. await db_session.refresh(server_row) assert server_row.status == transitional_status assert server_row.last_modified_by_person_id == str(person.id) assert server_row.last_modified_at is not None # Destroy ALSO soft-deletes the row (recycle bin restoration). if action == "destroy": assert server_row.deleted_at is not None else: assert server_row.deleted_at is None # ───────────────────────────────────────────────────────────────────── # RBAC: only can_manage_cloud admins # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.parametrize("action", ["power-on", "power-off", "restart", "destroy"]) async def test_lifecycle_employee_forbidden( client, employee_user, server_row, action, ): """Employees can't power-cycle infrastructure even if they own the row. Cloud lifecycle is an admin-only operation in the legacy CMDB too; Helix preserves the policy.""" _, token = employee_user resp = await client.post( f"/api/v1/cmdb/servers/{server_row.id}/{action}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403 # ───────────────────────────────────────────────────────────────────── # 404 — unknown id # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_lifecycle_unknown_server_404(client, admin_user): _, token = admin_user resp = await client.post( f"/api/v1/cmdb/servers/{uuid.uuid4()}/power-on", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 404 # ───────────────────────────────────────────────────────────────────── # 501 — provider doesn't implement the action # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_lifecycle_capability_not_supported_returns_501( client, admin_user, server_row, db_session, ): """When the integration's provider has no lifecycle impl (Azure / GCP / Tencent / KSyun today), surface 501 with a precise code so the UI can present an actionable message rather than a 500.""" _, token = admin_user provider = _mock_provider( "power_on", side_effect=CapabilityNotSupported( "tencent does not support server power-on", ), ) with patch( "app.domains.cmdb.router._load_provider_for_dns", new_callable=AsyncMock, ) as mock_loader: mock_loader.return_value = provider resp = await client.post( f"/api/v1/cmdb/servers/{server_row.id}/power-on", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 501 assert "CLOUD_LIFECYCLE_NOT_SUPPORTED" in resp.text # Status NOT bumped on failure. await db_session.refresh(server_row) assert server_row.status == "running" # ───────────────────────────────────────────────────────────────────── # Audit trail — success + failure both write # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_lifecycle_writes_audit_log_on_success( client, admin_user, server_row, db_session, ): person, token = admin_user provider = _mock_provider("restart_server") with patch( "app.domains.cmdb.router._load_provider_for_dns", new_callable=AsyncMock, ) as mock_loader: mock_loader.return_value = provider resp = await client.post( f"/api/v1/cmdb/servers/{server_row.id}/restart", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 rows = (await db_session.execute( select(AuditLog) .where(AuditLog.resource_type == "cmdb_server") .where(AuditLog.resource_id == str(server_row.id)) .where(AuditLog.action == "cmdb.server.restart") )).scalars().all() assert len(rows) >= 1 success_row = next((r for r in rows if r.status == "success"), None) assert success_row is not None assert success_row.actor_user_id == person.id assert success_row.detail["instance_id"] == "uhost-test-559" assert success_row.detail["new_status"] == "rebooting" @pytest.mark.asyncio async def test_lifecycle_writes_audit_log_on_capability_failure( client, admin_user, server_row, db_session, ): _, token = admin_user provider = _mock_provider( "destroy_server", side_effect=CapabilityNotSupported("ksyun does not support termination"), ) with patch( "app.domains.cmdb.router._load_provider_for_dns", new_callable=AsyncMock, ) as mock_loader: mock_loader.return_value = provider resp = await client.post( f"/api/v1/cmdb/servers/{server_row.id}/destroy", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 501 rows = (await db_session.execute( select(AuditLog) .where(AuditLog.resource_type == "cmdb_server") .where(AuditLog.resource_id == str(server_row.id)) .where(AuditLog.action == "cmdb.server.destroy") .where(AuditLog.status == "failure") )).scalars().all() assert len(rows) == 1 assert rows[0].detail["reason"] == "capability_not_supported" # ───────────────────────────────────────────────────────────────────── # Region missing → 422 rather than 500 # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_lifecycle_missing_region_returns_422( client, admin_user, integ: Integration, db_session, ): """A row imported before region/project was reliably captured by the sync engine could legitimately have NULL region. The provider layer raises ValueError; the router turns that into a 422 so the admin sees the missing field rather than a generic 500.""" s = CmdbServer( integration_id=integ.id, instance_id="uhost-missing-region", name="no-region-host", status="running", region=None, # the bug case ) db_session.add(s) await db_session.commit() _, token = admin_user # Mock the driver to raise ValueError the way the real drivers do # when region is missing — keeps the test independent of which # driver is wired. provider = _mock_provider( "power_on", side_effect=ValueError( "UCloud StartUHostInstance requires region — server row has no region", ), ) with patch( "app.domains.cmdb.router._load_provider_for_dns", new_callable=AsyncMock, ) as mock_loader: mock_loader.return_value = provider resp = await client.post( f"/api/v1/cmdb/servers/{s.id}/power-on", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 422 # ───────────────────────────────────────────────────────────────────── # Provider response payload — integration_name + provider plumbed # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_server_list_carries_provider_meta( client, admin_user, server_row, ): """The frontend gates its "更多 ↓" action menu per-provider; make sure the list response actually carries the joined provider field so the per-row gating works.""" _, token = admin_user resp = await client.get( "/api/v1/cmdb/servers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 rows = resp.json()["data"] row = next(r for r in rows if r["instance_id"] == "uhost-test-559") assert row["provider"] == "ucloud" assert row["integration_name"] == "pytest-fixture-ucloud-lifecycle" @pytest.mark.asyncio async def test_server_get_detail_carries_provider_meta( client, admin_user, server_row, ): _, token = admin_user resp = await client.get( f"/api/v1/cmdb/servers/{server_row.id}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 row = resp.json()["data"] assert row["provider"] == "ucloud" assert row["integration_name"] == "pytest-fixture-ucloud-lifecycle" # ───────────────────────────────────────────────────────────────────── # Sync resurrection guard (Round-1 review fix) # # Pre-fix the upsert had ``deleted_at=None`` in its SET list, which # silently un-soft-deleted any matching row on every sync tick. After # an admin clicked 删除 the destroy router sets status='terminated' + # deleted_at=now(), but the cloud-side terminate is async — UCloud / # Aliyun / AWS keep returning the instance in 'stopping' / # 'shutting-down' state for up to a few minutes. The next sync tick # would reset ``deleted_at=NULL`` and the row would silently reappear # in the admin's list. # # Fix (sync.py): drop ``deleted_at`` from the upsert SET list entirely. # Recycle bin is the canonical un-delete path now — admins click # Restore explicitly rather than rely on sync to auto-resurrect. (Tried # a CASE-WHEN guard first; MySQL evaluates SET clauses sequentially # and the column ref reads the post-update value, so the guard always # saw status='stopping' regardless of the pre-update value. The # straight removal is the production-grade fix.) # # **Schema note**: the unique constraint ``uq_cmdb_servers`` on # ``(integration_id, instance_id)`` was created by migration 006 and # is the key the upsert's ON DUPLICATE KEY UPDATE clause matches # against. It's not declared on the SQLAlchemy model (historical), so # the test fixture (which builds the DB via ``Base.metadata.create_all`` # instead of alembic) doesn't get it for free — these tests inject the # constraint inline before exercising the upsert. Production DBs run # the migration chain so they DO have the constraint. # ───────────────────────────────────────────────────────────────────── async def _ensure_cmdb_servers_unique_key(db_session: AsyncSession) -> None: """Test fixture helper — add the production unique constraint to ``cmdb_servers`` if it's missing. Idempotent.""" from sqlalchemy import text as _text exists = (await db_session.execute(_text( "SELECT COUNT(*) FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = 'cmdb_servers' " "AND INDEX_NAME = 'uq_cmdb_servers'" ))).scalar() if exists == 0: # Clean up any duplicates from prior test runs before adding # the constraint — otherwise the ALTER TABLE will fail. await db_session.execute(_text( "DELETE s1 FROM cmdb_servers s1 " "INNER JOIN cmdb_servers s2 " "WHERE s1.id > s2.id " "AND s1.integration_id = s2.integration_id " "AND s1.instance_id = s2.instance_id" )) await db_session.execute(_text( "ALTER TABLE cmdb_servers " "ADD UNIQUE KEY uq_cmdb_servers (integration_id, instance_id)" )) await db_session.commit() @pytest.mark.asyncio async def test_sync_preserves_destroy_soft_delete( db_session: AsyncSession, integ: Integration, ): """An instance the admin destroyed must NOT be resurrected by sync while the cloud side is still propagating the terminate. Drives the real production code path via ``_sync_servers`` against a mocked provider that returns the still-terminating instance — no manual upsert duplication of the SET-clause ordering. """ from datetime import UTC, datetime from app.domains.cmdb.providers.base import CloudServer as ProvServer from app.domains.cmdb.sync import _sync_servers await _ensure_cmdb_servers_unique_key(db_session) destroyed_at = datetime.now(UTC) s = CmdbServer( integration_id=integ.id, instance_id="uhost-destroyed-by-admin", name="destroyed-host", status="terminated", region="cn-bj2", deleted_at=destroyed_at, last_seen_sync_job_id=None, ) db_session.add(s) await db_session.commit() s_id = s.id # Mocked provider returns the same instance the admin just # destroyed — cloud-side terminate is still propagating. mock_provider = AsyncMock() mock_provider.list_servers = AsyncMock(return_value=[ ProvServer( instance_id="uhost-destroyed-by-admin", name="destroyed-host", status="stopping", # cloud reports terminating region="cn-bj2", zone=None, vcpu=None, memory_mb=None, public_ips=[], private_ips=[], image_id=None, os_type=None, tags={}, raw={}, ), ]) mock_provider.consume_partial_failures = lambda: [] mock_provider.consume_skipped_legs = lambda: [] await _sync_servers( db_session, integ, job_id=str(uuid.uuid4()), provider=mock_provider, ) from sqlalchemy import text raw = (await db_session.execute(text( "SELECT deleted_at FROM cmdb_servers WHERE id = :id" ), {"id": str(s_id)})).first() assert raw is not None assert raw[0] is not None, ( "sync silently resurrected an admin-destroyed server" ) @pytest.mark.asyncio async def test_sync_does_not_auto_resurrect_soft_deleted_row( db_session: AsyncSession, integ: Integration, ): """After Round-1 fix: sync NEVER auto-resurrects a soft-deleted row. Recycle bin Restore is the explicit un-delete path. A row that was swept by mistake and reappears in the next list_servers stays soft-deleted until the admin clicks Restore.""" from datetime import UTC, datetime from app.domains.cmdb.providers.base import CloudServer as ProvServer from app.domains.cmdb.sync import _sync_servers await _ensure_cmdb_servers_unique_key(db_session) prior_job_id = str(uuid.uuid4()) s = CmdbServer( integration_id=integ.id, instance_id="uhost-was-swept", name="swept-then-reappeared", status="running", region="cn-bj2", deleted_at=datetime.now(UTC), last_seen_sync_job_id=prior_job_id, ) db_session.add(s) await db_session.commit() s_id = s.id mock_provider = AsyncMock() mock_provider.list_servers = AsyncMock(return_value=[ ProvServer( instance_id="uhost-was-swept", name="swept-then-reappeared", status="running", # cloud reports healthy region="cn-bj2", zone=None, vcpu=None, memory_mb=None, public_ips=[], private_ips=[], image_id=None, os_type=None, tags={}, raw={}, ), ]) mock_provider.consume_partial_failures = lambda: [] mock_provider.consume_skipped_legs = lambda: [] await _sync_servers( db_session, integ, job_id=str(uuid.uuid4()), provider=mock_provider, ) from sqlalchemy import text raw = (await db_session.execute(text( "SELECT deleted_at, status FROM cmdb_servers WHERE id = :id" ), {"id": str(s_id)})).first() assert raw is not None # Post-Round-1: deleted_at is no longer cleared by sync. Row stays # soft-deleted until admin clicks Restore in the recycle bin. assert raw[0] is not None, ( "sync unexpectedly auto-resurrected a soft-deleted row " "(should require explicit admin Restore via recycle bin)" ) # The other columns still get updated — only deleted_at is preserved. assert raw[1] == "running" # ───────────────────────────────────────────────────────────────────── # Audit log persistence — Round-1 fix verified end-to-end # ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_audit_row_persisted_after_capability_failure( client, admin_user, server_row, db_session, ): """The earlier shape committed state then logged audit then returned — leaving the audit row uncommitted on failure paths. Verify the audit row is observable from a fresh query after the request completes (i.e. it was committed).""" _, token = admin_user provider = _mock_provider( "power_on", side_effect=CapabilityNotSupported("azure does not support power-on"), ) with patch( "app.domains.cmdb.router._load_provider_for_dns", new_callable=AsyncMock, ) as mock_loader: mock_loader.return_value = provider resp = await client.post( f"/api/v1/cmdb/servers/{server_row.id}/power-on", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 501 # Open a fresh-isolated read to confirm the row really committed — # not just sitting in the test session's identity map. await db_session.commit() # release the request's session state rows = (await db_session.execute( select(AuditLog).where( AuditLog.action == "cmdb.server.power_on", AuditLog.status == "failure", AuditLog.resource_id == str(server_row.id), ) )).scalars().all() assert len(rows) == 1, ( "capability-failure audit row was not committed to DB" ) assert rows[0].detail["reason"] == "capability_not_supported"