""" E2E API tests for DNS Records, Load Balancers, and CDN Domains endpoints. Coverage: - List endpoints: pagination, filtering, empty results - Detail endpoints: 200 on hit, 404 on miss - RBAC: employee can read, unauthenticated gets 401 - Soft-delete: deleted records excluded from results """ import uuid from datetime import UTC import pytest import pytest_asyncio from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import CmdbCdnDomain, CmdbDnsRecord, CmdbLoadBalancer from app.integrations.models import Integration # --------------------------------------------------------------------------- # Shared fixture: one cloud integration used across all three resource types # (post-migration 045 — cmdb_provider_accounts was merged into integrations) # --------------------------------------------------------------------------- @pytest_asyncio.fixture async def provider(db_session: AsyncSession) -> Integration: integration_id = uuid.uuid4() integ = Integration( id=integration_id, kind="cloud", provider="aliyun", # Prefix uniquely identifies this row as pytest debris so a leaked # session (dev DB, CI re-run without cleanup) can be swept by the # alembic migration that purges the old ``test-aliyun`` name. name="pytest-fixture-aliyun", slug=f"pytest-fixture-aliyun-{str(integration_id)[:8]}", enabled=True, is_primary=False, # ProviderCapability enum VALUES are lowercase; seeding with the # enum NAME (uppercase) silently produced an empty effective- # capability set, which bypassed every cap-aware guard. Bug # surfaced by the 2026-06-01 ``_load_provider_for_dns`` cap # check that now fail-fasts when dns_records isn't advertised. capabilities=["dns_records", "load_balancers", "cdn"], ) db_session.add(integ) await db_session.commit() return integ # --------------------------------------------------------------------------- # DNS Records # --------------------------------------------------------------------------- @pytest_asyncio.fixture async def dns_records(db_session: AsyncSession, provider: Integration): records = [ CmdbDnsRecord( integration_id=provider.id, zone_name="example.com", record_id="rec-001", name="www", record_type="A", value="1.2.3.4", ttl=300, status="enable", ), CmdbDnsRecord( integration_id=provider.id, zone_name="example.com", record_id="rec-002", name="mail", record_type="MX", value="mail.example.com", ttl=3600, status="enable", ), CmdbDnsRecord( integration_id=provider.id, zone_name="other.com", record_id="rec-003", name="@", record_type="A", value="9.9.9.9", ttl=600, status="disable", ), ] db_session.add_all(records) await db_session.commit() return records @pytest.mark.asyncio async def test_list_dns_records_admin(client, admin_user, dns_records): _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() assert body["success"] is True assert body["meta"]["total"] == 3 @pytest.mark.asyncio async def test_list_dns_records_employee_sees_all(client, employee_user, dns_records): """DNS is a company-public asset (ADR 2026-06-09): an employee sees the FULL list regardless of owner — no owner auto-demote. The fixture records all have owner_person_id=None; the pre-ADR behaviour scoped the employee to their own records and returned 0 (the reported "搜不到" bug). Guard that visibility is now org-wide. Write/disable/delete remain admin-only (covered by the mutate tests below).""" _, token = employee_user resp = await client.get( "/api/v1/cmdb/dns-records", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() assert body["success"] is True assert body["meta"]["total"] == 3 @pytest.mark.asyncio async def test_list_dns_records_filter_zone(client, admin_user, dns_records): _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records?zone_name=example.com", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert all(r["zone_name"] == "example.com" for r in data) assert len(data) == 2 @pytest.mark.asyncio async def test_list_dns_records_filter_type(client, admin_user, dns_records): _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records?record_type=MX", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert len(data) == 1 assert data[0]["record_type"] == "MX" @pytest.mark.asyncio async def test_list_dns_records_search(client, admin_user, dns_records): _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records?search=mail", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert len(data) == 1 assert data[0]["name"] == "mail" @pytest.mark.asyncio async def test_get_dns_record_by_id(client, admin_user, dns_records): _, token = admin_user record_id = str(dns_records[0].id) resp = await client.get( f"/api/v1/cmdb/dns-records/{record_id}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["data"]["record_id"] == "rec-001" @pytest.mark.asyncio async def test_get_dns_record_employee_can_read_any( client, employee_user, dns_records, ): """DNS is a company-public asset (ADR 2026-06-09): an employee may read ANY record's detail, mirroring list_dns_records returning the full list. Before the fix the detail endpoint still owner-scoped and 404'd employees on records they could see in the list (the fixture records have owner_person_id=None — the pre-ADR detail guard raised NotFound on the ``not record.owner_person_id`` branch).""" _, token = employee_user record_id = str(dns_records[0].id) resp = await client.get( f"/api/v1/cmdb/dns-records/{record_id}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["data"]["record_id"] == "rec-001" @pytest.mark.asyncio async def test_get_dns_record_not_found(client, admin_user): _, token = admin_user resp = await client.get( f"/api/v1/cmdb/dns-records/{uuid.uuid4()}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_dns_record_soft_deleted_excluded(client, admin_user, db_session, dns_records): """A soft-deleted record must not appear in list results.""" from datetime import datetime rec = dns_records[0] rec.deleted_at = datetime.now(UTC) db_session.add(rec) await db_session.commit() _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["meta"]["total"] == 2 @pytest.mark.asyncio async def test_list_dns_records_newest_first( client, admin_user, db_session, provider, ): """Records are returned newest-first by ``first_seen_at``. User feedback 2026-05-29: brand-new records (admin direct-create or freshly-discovered by sync) should float to the top so the admin can confirm a write landed. Prior ``(zone_name, name)`` ordering buried new rows alphabetically. """ from datetime import datetime, timedelta base = datetime.now(UTC) # Three records with explicit first_seen_at — oldest → newest. oldest = CmdbDnsRecord( integration_id=provider.id, zone_name="example.com", record_id="old-1", name="old", record_type="A", value="1.1.1.1", ttl=300, status="enable", first_seen_at=base - timedelta(days=30), ) middle = CmdbDnsRecord( integration_id=provider.id, zone_name="example.com", record_id="mid-1", name="mid", record_type="A", value="2.2.2.2", ttl=300, status="enable", first_seen_at=base - timedelta(hours=1), ) newest = CmdbDnsRecord( integration_id=provider.id, zone_name="example.com", record_id="new-1", name="new", record_type="A", value="3.3.3.3", ttl=300, status="enable", first_seen_at=base, ) # Insert deliberately in mixed order so the ORDER BY is what # actually produces the assertion's expected sequence. db_session.add_all([middle, oldest, newest]) await db_session.commit() _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records?zone_name=example.com", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 names_in_order = [r["name"] for r in resp.json()["data"]] assert names_in_order[:3] == ["new", "mid", "old"] @pytest.mark.asyncio async def test_list_dns_zones_includes_domains_with_no_records( client, admin_user, db_session, provider, ): """Admin DNS zone picker UNIONs zones-with-records with ``cmdb_domains`` rows (even when ``allow_dns_request=False``). Prod bug 2026-05-29: ``tiup.io`` was registered in ``cmdb_domains`` (``allow_dns_request=True``, integration enabled) but had ZERO records yet, so the prior ``SELECT DISTINCT zone_name FROM cmdb_dns_records`` query returned nothing → admin couldn't add the first record because the picker dropdown was empty. Employees could find it (their ``/domains/selectable-for-dns`` path reads ``cmdb_domains`` directly) — admin saw a strict subset, which inverts the expected privilege ordering. """ from app.domains.cmdb.models import CmdbDomain # Pre-existing record on existing-zone-with-records. has_record = CmdbDnsRecord( integration_id=provider.id, zone_name="hasrecords.com", record_id="r1", name="www", record_type="A", value="1.1.1.1", ttl=300, status="enable", ) # Domain with no records yet — the tiup.io case. new_domain = CmdbDomain( integration_id=str(provider.id), domain_name="tiup.io", status="inactive", allow_dns_request=True, ) # Admin-locked domain (allow_dns_request=False) — admin should # STILL see it in the picker since direct-write bypasses the # self-service gate; employee path remains gated separately. locked_domain = CmdbDomain( integration_id=str(provider.id), domain_name="locked-prod.com", status="active", allow_dns_request=False, ) db_session.add_all([has_record, new_domain, locked_domain]) await db_session.commit() _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-zones", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 # 2026-06-01: response shape upgraded from list[str] to # list[{zone_name, integration_id, integration_name, provider, # record_count}] so the FE picker can show the resolved DNS # provider per zone (replacing the buggy "first capable provider" # fallback that mis-attributed registrar-only zones). rows = resp.json()["data"] by_zone = {r["zone_name"]: r for r in rows} # All three must surface: legacy zone-from-records, # new-zone-no-records, and admin-locked zone. assert by_zone.get("hasrecords.com") is not None assert by_zone["hasrecords.com"]["integration_id"] == str(provider.id), ( "zone with records must resolve to the record's integration" ) assert by_zone["hasrecords.com"]["record_count"] == 1 assert by_zone.get("tiup.io") is not None, ( "domain with no records yet is missing from admin picker — " "the prod tiup.io regression is back" ) # tiup.io has no records → integration_id MUST be None so the FE # knows to ask the admin which provider to use. Returning the # registrar integration here is the bug that triggered the PR # ("tikv.io → AliCloud (海外)" mis-attribution). assert by_zone["tiup.io"]["integration_id"] is None assert by_zone["tiup.io"]["record_count"] == 0 assert by_zone.get("locked-prod.com") is not None, ( "admin-locked domain hidden from admin picker — direct-write " "users would have no UI path to manage their gated zones" ) # --------------------------------------------------------------------------- # Load Balancers # --------------------------------------------------------------------------- @pytest_asyncio.fixture async def load_balancers(db_session: AsyncSession, provider: Integration): lbs = [ CmdbLoadBalancer( integration_id=provider.id, lb_id="lb-001", name="prod-lb", lb_type="application", network_type="public", ip_address="203.0.113.10", region="cn-hangzhou", status="active", ), CmdbLoadBalancer( integration_id=provider.id, lb_id="lb-002", name="internal-lb", lb_type="network", network_type="internal", ip_address="10.0.0.5", region="cn-shanghai", status="active", ), ] db_session.add_all(lbs) await db_session.commit() return lbs @pytest.mark.asyncio async def test_list_load_balancers(client, admin_user, load_balancers): _, token = admin_user resp = await client.get( "/api/v1/cmdb/load-balancers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["meta"]["total"] == 2 @pytest.mark.asyncio async def test_list_load_balancers_filter_region(client, admin_user, load_balancers): _, token = admin_user resp = await client.get( "/api/v1/cmdb/load-balancers?region=cn-hangzhou", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert len(data) == 1 assert data[0]["name"] == "prod-lb" @pytest.mark.asyncio async def test_list_load_balancers_filter_network_type(client, admin_user, load_balancers): _, token = admin_user resp = await client.get( "/api/v1/cmdb/load-balancers?network_type=internal", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert len(data) == 1 assert data[0]["network_type"] == "internal" @pytest.mark.asyncio async def test_get_load_balancer_by_id(client, admin_user, load_balancers): _, token = admin_user lb_id = str(load_balancers[0].id) resp = await client.get( f"/api/v1/cmdb/load-balancers/{lb_id}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["data"]["lb_id"] == "lb-001" @pytest.mark.asyncio async def test_get_load_balancer_not_found(client, admin_user): _, token = admin_user resp = await client.get( f"/api/v1/cmdb/load-balancers/{uuid.uuid4()}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_load_balancer_employee_scoped_to_owner(client, employee_user, load_balancers): """Migration 173 added ``owner_person_id`` to LB so employees can now list LBs the same way they list servers / domains / dns / cdn — seeing only rows they own or that their direct reports own. The fixture LBs have ``owner_person_id=NULL`` (sync would normally fill it post-migration); an employee should still get a 200 with an empty list rather than the previous blanket 403 leak. """ _, token = employee_user resp = await client.get( "/api/v1/cmdb/load-balancers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() # No ownership on the fixture rows → employee sees zero, not the fleet. for lb in body.get("data", []): assert lb["owner_person_id"] is not None @pytest.mark.asyncio async def test_load_balancer_soft_deleted_excluded(client, admin_user, db_session, load_balancers): from datetime import datetime load_balancers[0].deleted_at = datetime.now(UTC) db_session.add(load_balancers[0]) await db_session.commit() _, token = admin_user resp = await client.get( "/api/v1/cmdb/load-balancers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.json()["meta"]["total"] == 1 # --------------------------------------------------------------------------- # CDN Domains # --------------------------------------------------------------------------- @pytest_asyncio.fixture async def cdn_domains(db_session: AsyncSession, provider: Integration): domains = [ CmdbCdnDomain( integration_id=provider.id, domain_name="cdn.example.com", cname="cdn.example.com.alikunlun.com", status="online", source_type="domain", source_address="origin.example.com", is_https=True, ), CmdbCdnDomain( integration_id=provider.id, domain_name="static.example.com", cname="static.example.com.alikunlun.com", status="offline", source_type="ip", source_address="1.2.3.4", is_https=False, ), ] db_session.add_all(domains) await db_session.commit() return domains @pytest.mark.asyncio async def test_list_cdn_domains(client, admin_user, cdn_domains): _, token = admin_user resp = await client.get( "/api/v1/cmdb/cdn-domains", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["meta"]["total"] == 2 @pytest.mark.asyncio async def test_list_cdn_domains_filter_status(client, admin_user, cdn_domains): _, token = admin_user resp = await client.get( "/api/v1/cmdb/cdn-domains?status=online", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert len(data) == 1 assert data[0]["domain_name"] == "cdn.example.com" @pytest.mark.asyncio async def test_list_cdn_domains_search(client, admin_user, cdn_domains): _, token = admin_user resp = await client.get( "/api/v1/cmdb/cdn-domains?search=static", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] assert len(data) == 1 assert data[0]["domain_name"] == "static.example.com" @pytest.mark.asyncio async def test_get_cdn_domain_by_id(client, admin_user, cdn_domains): _, token = admin_user domain_id = str(cdn_domains[0].id) resp = await client.get( f"/api/v1/cmdb/cdn-domains/{domain_id}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["data"]["domain_name"] == "cdn.example.com" assert resp.json()["data"]["is_https"] is True @pytest.mark.asyncio async def test_get_cdn_domain_not_found(client, admin_user): _, token = admin_user resp = await client.get( f"/api/v1/cmdb/cdn-domains/{uuid.uuid4()}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_cdn_domain_employee_can_read_own(client, employee_user, cdn_domains): """Employees with can_view_cloud_resources can list CDN domains but only see their own (backend auto-demotes to owner_person_id = caller). The fixture CDN domains have no owner, so the result is an empty list — not a 403. PR #284 changed the gate from can_read_cloud to can_view_cloud_resources to align with servers/domains/dns pattern.""" _, token = employee_user resp = await client.get( "/api/v1/cmdb/cdn-domains", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 data = resp.json()["data"] # Fixture CDN domains have no owner — employee sees empty list after demote. assert data == [] @pytest.mark.asyncio async def test_cdn_domain_soft_deleted_excluded(client, admin_user, db_session, cdn_domains): from datetime import datetime cdn_domains[1].deleted_at = datetime.now(UTC) db_session.add(cdn_domains[1]) await db_session.commit() _, token = admin_user resp = await client.get( "/api/v1/cmdb/cdn-domains", headers={"Authorization": f"Bearer {token}"}, ) assert resp.json()["meta"]["total"] == 1 # --------------------------------------------------------------------------- # Empty state — no data seeded # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_empty_dns_records(client, admin_user): _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["meta"]["total"] == 0 assert resp.json()["data"] == [] @pytest.mark.asyncio async def test_empty_load_balancers(client, admin_user): _, token = admin_user resp = await client.get( "/api/v1/cmdb/load-balancers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["meta"]["total"] == 0 @pytest.mark.asyncio async def test_empty_cdn_domains(client, admin_user): _, token = admin_user resp = await client.get( "/api/v1/cmdb/cdn-domains", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 assert resp.json()["meta"]["total"] == 0 # --------------------------------------------------------------------------- # Disabled-tenant gating (PR after prod incident 2026-05-25) # --------------------------------------------------------------------------- # # Prod showed 4 domains appearing twice in /cmdb/domains because they # lived in BOTH a live UCloud tenant and a disabled 阿里云 tenant. The # cloud_sync cron skips disabled integrations, which skips mark-and- # sweep, so the disabled-tenant rows stay forever as cross-tenant # duplicates. The fix: every CMDB list endpoint with no explicit # ``integration_id`` filter scopes to enabled cloud integrations only. # Admin can still audit a disabled tenant by picking it explicitly. @pytest_asyncio.fixture async def disabled_tenant_setup(db_session: AsyncSession, provider: Integration): """Two cloud tenants: ``provider`` (enabled) + a disabled twin. Each carries one row of every resource type — a server, a domain, a DNS record, a load balancer, a CDN domain. The asserts then confirm that "全部" listings drop the disabled tenant's rows while pinning ``integration_id`` to the disabled tenant still returns them (audit path). """ from datetime import date from app.domains.cmdb.models import CmdbDomain, CmdbServer disabled_id = uuid.uuid4() disabled = Integration( id=disabled_id, kind="cloud", provider="aliyun", name="pytest-fixture-disabled", slug=f"pytest-fixture-disabled-{str(disabled_id)[:8]}", enabled=False, is_primary=False, # ProviderCapability enum VALUES are lowercase; seeding with the # enum NAME (uppercase) silently produced an empty effective- # capability set, which bypassed every cap-aware guard. Bug # surfaced by the 2026-06-01 ``_load_provider_for_dns`` cap # check that now fail-fasts when dns_records isn't advertised. capabilities=["dns_records", "load_balancers", "cdn"], ) db_session.add(disabled) await db_session.flush() # One row of each resource type per tenant. Use the SAME domain_name # across tenants so the dedup behavior of the list endpoint is # exercised (the prod incident's signature). rows = [ CmdbServer( integration_id=provider.id, instance_id="srv-enabled", name="srv-enabled", status="running", ), CmdbServer( integration_id=disabled.id, instance_id="srv-disabled", name="srv-disabled", status="running", ), CmdbDomain( integration_id=provider.id, domain_name="dup.example.com", registrar="enabled-registrar", status="active", expiry_date=date(2027, 1, 1), ), CmdbDomain( integration_id=disabled.id, domain_name="dup.example.com", registrar="disabled-registrar", status="active", expiry_date=date(2027, 1, 1), ), CmdbDnsRecord( integration_id=provider.id, zone_name="enabled-zone.com", record_id="rec-enabled", name="@", record_type="A", value="1.1.1.1", ttl=300, status="enable", ), CmdbDnsRecord( integration_id=disabled.id, zone_name="disabled-zone.com", record_id="rec-disabled", name="@", record_type="A", value="2.2.2.2", ttl=300, status="enable", ), CmdbLoadBalancer( integration_id=provider.id, lb_id="lb-enabled", name="lb-enabled", lb_type="alb", region="cn-east-1", status="active", ip_address="10.0.0.1", network_type="public", ), CmdbLoadBalancer( integration_id=disabled.id, lb_id="lb-disabled", name="lb-disabled", lb_type="alb", region="cn-north-1", status="active", ip_address="10.0.0.2", network_type="public", ), CmdbCdnDomain( integration_id=provider.id, domain_name="cdn-enabled.com", cname="cdn-enabled.cdn.com", status="online", ), CmdbCdnDomain( integration_id=disabled.id, domain_name="cdn-disabled.com", cname="cdn-disabled.cdn.com", status="online", ), ] db_session.add_all(rows) # Object storage rows added separately so the import sits next to # them — keeps the top-of-file imports lean for the common LB / CDN # / DNS path. from app.domains.cmdb.models import CmdbObjectStorage db_session.add_all([ CmdbObjectStorage( integration_id=provider.id, bucket_name="bucket-enabled", region="cn-east-1", status="active", ), CmdbObjectStorage( integration_id=disabled.id, bucket_name="bucket-disabled", region="cn-north-1", status="active", ), ]) await db_session.commit() return {"enabled": provider, "disabled": disabled} @pytest.mark.asyncio async def test_domains_list_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): """The "全部" view returns only the enabled-tenant row even though both tenants have a ``dup.example.com`` row. Without the gate, the admin's domains page would show 2 rows for the same domain — the exact prod regression this PR fixes.""" _, token = admin_user resp = await client.get( "/api/v1/cmdb/domains", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() registrars = [row["registrar"] for row in body["data"]] assert body["meta"]["total"] == 1, body assert registrars == ["enabled-registrar"] @pytest.mark.asyncio async def test_domains_list_disabled_tenant_visible_when_pinned( client, admin_user, disabled_tenant_setup, ): """Admin audit path — explicit ``integration_id`` lets them see the disabled-tenant row. Without this carve-out the disabled tenant tab in would render an empty page.""" _, token = admin_user disabled_id = str(disabled_tenant_setup["disabled"].id) resp = await client.get( f"/api/v1/cmdb/domains?integration_id={disabled_id}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() assert body["meta"]["total"] == 1 assert body["data"][0]["registrar"] == "disabled-registrar" @pytest.mark.asyncio async def test_servers_list_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): _, token = admin_user resp = await client.get( "/api/v1/cmdb/servers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() names = [row["name"] for row in body["data"]] assert names == ["srv-enabled"], body @pytest.mark.asyncio async def test_dns_zones_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): """Disabled-tenant zones must not appear in the picker dropdown — selecting one would let an employee submit a DNS request the cron can never execute against.""" _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-zones", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 zones = {r["zone_name"] for r in resp.json()["data"]} assert sum(zone == "enabled-zone.com" for zone in zones) == 1 assert sum(zone == "disabled-zone.com" for zone in zones) == 0 @pytest.mark.asyncio async def test_dns_records_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): _, token = admin_user resp = await client.get( "/api/v1/cmdb/dns-records", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() record_ids = [row["record_id"] for row in body["data"]] assert "rec-enabled" in record_ids assert "rec-disabled" not in record_ids @pytest.mark.asyncio async def test_load_balancers_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): _, token = admin_user resp = await client.get( "/api/v1/cmdb/load-balancers", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() names = [row["name"] for row in body["data"]] assert "lb-enabled" in names assert "lb-disabled" not in names @pytest.mark.asyncio async def test_cdn_domains_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): _, token = admin_user resp = await client.get( "/api/v1/cmdb/cdn-domains", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() names = [row["domain_name"] for row in body["data"]] assert sum(name == "cdn-enabled.com" for name in names) == 1 assert sum(name == "cdn-disabled.com" for name in names) == 0 @pytest.mark.asyncio async def test_object_storage_excludes_disabled_tenant( client, admin_user, disabled_tenant_setup, ): """OSS gets the same gate as every other resource type. Without this, the buckets page would inherit the same prod regression pattern as 域名管理 once any OSS-carrying integration is disabled.""" _, token = admin_user resp = await client.get( "/api/v1/cmdb/object-storage", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json() names = [row["bucket_name"] for row in body["data"]] assert "bucket-enabled" in names assert "bucket-disabled" not in names @pytest.mark.asyncio async def test_domain_expiry_alert_skips_disabled_tenant( db_session: AsyncSession, client, disabled_tenant_setup, monkeypatch, ): # ``client`` fixture is included for its side-effect — it patches # ``async_session_factory`` to bind to the test DB engine. The # handler opens its own session via that factory, so without the # patch the handler would query the dev/prod DB and return real # data (the failure mode was "summary always shows scanned=N>0 # against prod integrations" in CI). """The renewed ``cmdb.domain_expiry_alert`` cron must not double-page an owner whose same domain_name lives in both a disabled and an enabled tenant — state is keyed by ``domain.id``, not domain_name, so two rows would otherwise dispatch two Feishu DMs per bucket transition (see migration 176 + the prod incident this PR fixes). """ from datetime import UTC, datetime, timedelta from app.domains.cmdb.models import ( EXPIRY_RESOURCE_DOMAIN, CmdbDomain, CmdbExpiryAlertState, ) from app.scheduled_jobs.handlers import run_domain_expiry_alert # Move both dup rows into the <=30d bucket so they would otherwise # both be considered. Both fixture rows already carry expiry # 2027-01-01 — overwrite to now+10d. soon = datetime.now(UTC).date() + timedelta(days=10) dup_rows = (await db_session.execute( select(CmdbDomain).where(CmdbDomain.domain_name == "dup.example.com") )).scalars().all() enabled_id = str(disabled_tenant_setup["enabled"].id) disabled_id = str(disabled_tenant_setup["disabled"].id) enabled_row_id: str | None = None disabled_row_id: str | None = None for row in dup_rows: row.expiry_date = soon if str(row.integration_id) == enabled_id: enabled_row_id = str(row.id) elif str(row.integration_id) == disabled_id: disabled_row_id = str(row.id) await db_session.commit() assert enabled_row_id and disabled_row_id, "fixture must provide both rows" # Force the network-notifications kill switch ON for this run so the # handler doesn't short-circuit before the gate fires. async def _stub_kill_switch(_db): return True monkeypatch.setattr( "app.domains.cmdb.network_notifications.network_user_notifications_enabled", _stub_kill_switch, ) summary = await run_domain_expiry_alert(db_session, {}) # Gate verification, multiple angles to reject false-positives: # (a) exactly one row scanned (would be 2 if disabled-tenant row leaked) assert summary["scanned"] == 1, summary # (b) handler did not short-circuit on a different branch assert summary.get("status") != "skipped", summary # (c) any state-table row touched is the enabled-tenant row only — # the dedup state row gets stamped on first observation, so # this is the most direct "did the handler see this row" signal state_rows = (await db_session.execute( select(CmdbExpiryAlertState).where( CmdbExpiryAlertState.resource_type == EXPIRY_RESOURCE_DOMAIN, ) )).scalars().all() touched_ids = {s.resource_id for s in state_rows} assert enabled_row_id in touched_ids or len(touched_ids) == 0, touched_ids assert disabled_row_id not in touched_ids, ( f"disabled-tenant row {disabled_row_id} should have been gated " f"out before the handler reached the state-write path: {touched_ids}" ) @pytest.mark.asyncio async def test_iam_mine_bypasses_disabled_tenant_gate( db_session: AsyncSession, client, employee_user, disabled_tenant_setup, ): """Employees' ``/cmdb/iam-users?mine=true`` MUST still surface their own IAM rows even when those rows live in a disabled tenant. The disabled-tenant gate is for the ADMIN list (where a revoke button would silently fail against a frozen tenant) — the self-service view has no such buttons. Hiding an employee's own credentials because IT temporarily disabled the tenant is a worse UX failure than showing a frozen row.""" from app.domains.cmdb.models import CmdbIamUser person, token = employee_user db_session.add(CmdbIamUser( integration_id=disabled_tenant_setup["disabled"].id, user_name="employee.user", arn="arn:aws:iam::000:user/employee.user", sub_account_id="000", owner_person_id=str(person.id), mfa_enabled=True, )) await db_session.commit() resp = await client.get( "/api/v1/cmdb/iam-users?mine=true", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text body = resp.json() user_names = [row["user_name"] for row in body["data"]] assert "employee.user" in user_names, ( "mine=true must not be gated by Integration.enabled — see " "router.py::list_iam_users carve-out comment" ) @pytest.mark.asyncio async def test_domain_expiry_alert_short_circuits_when_no_enabled_tenants( db_session: AsyncSession, client, disabled_tenant_setup, monkeypatch, ): # See sibling test's note re: client fixture pulling in the # ``async_session_factory`` monkeypatch. """When every cloud tenant is disabled the builder yields no items, so the engine does no owner resolution and fans out NO Feishu DMs — the alternative is an empty ``IN ()`` clause or a 0-scan run that still tries to notify. (Pre-unified-engine this surfaced as an explicit ``status=skipped``; now it's an empty, no-fan-out run.)""" from sqlalchemy import update as _update from app.integrations.models import Integration as _Integration from app.scheduled_jobs.handlers import run_domain_expiry_alert async def _stub_kill_switch(_db): return True monkeypatch.setattr( "app.domains.cmdb.network_notifications.network_user_notifications_enabled", _stub_kill_switch, ) # Disable EVERY cloud integration so the helper sees an empty set. # Direct UPDATE (not ORM ``.enabled = False`` on a single fixture # row) — the handler opens its own ``async_session_factory()`` # session, and other tests in this session may have left other # ``kind='cloud'`` integrations behind. UPDATE-by-kind catches # those too. Commit before calling the handler so its fresh # session sees the change. await db_session.execute( _update(_Integration) .where(_Integration.kind == "cloud") .values(enabled=False) ) await db_session.commit() summary = await run_domain_expiry_alert(db_session, {}) # No enabled tenants → nothing scanned, nobody notified (no empty IN(), # no owner resolution, no fan-out). assert summary.get("scanned", 0) == 0, summary assert summary.get("owners_notified", 0) == 0, summary