"""DNS already knows who owns most of these hostnames. Six of the register's first eight entries had no owner, while Helix was syncing 3,994 DNS records of which a third carry one — kept warm by the `cmdb.dns_owner_recertify` cron. Asking a person to retype what the platform already stores is how the register ends up with blanks again. The line these tests hold is *which* records are safe to believe. A zone-level record's owner administers the ZONE: in production one DNS administrator holds the NS on `tidb-cloud.com` and `labs.pingcap.com` plus the MX on `pingcap.com`. Auto-filling from those invents three owners for three unrelated systems, which is the wrong-routing this domain exists to prevent. """ from __future__ import annotations import uuid import pytest from app.core.coordinators import dns_owner_coordinator as dns_owner from app.domains.system_registry import service as svc from app.identity.person_models import IdentityPerson pytestmark = pytest.mark.asyncio async def _person(db, *, status="active") -> IdentityPerson: p = IdentityPerson( source_system="test", hris_person_id=f"P{uuid.uuid4().hex[:8]}", employee_no=f"E{uuid.uuid4().hex[:8]}", work_email=f"u{uuid.uuid4().hex[:8]}@x.test", display_name=f"Owner {uuid.uuid4().hex[:4]}", employment_status=status, ) db.add(p) await db.flush() return p async def _integration(db): from app.integrations.models import Integration slug = f"pytest-dns-{uuid.uuid4().hex[:6]}" row = Integration( id=uuid.uuid4(), kind="cloud", provider="aws", name=slug, slug=slug, enabled=True, is_primary=False, capabilities=["dns_records"], ) db.add(row) await db.flush() return row async def _dns(db, *, zone, name, rtype, owner, deleted=False): from datetime import UTC, datetime from app.domains.cmdb.models import CmdbDnsRecord integ = await _integration(db) row = CmdbDnsRecord( integration_id=str(integ.id), deleted_at=datetime.now(UTC).replace(tzinfo=None) if deleted else None, zone_name=zone, name=name, record_type=rtype, record_id=f"r-{uuid.uuid4().hex[:10]}", value="192.0.2.1", ttl=600, status="enable", owner_person_id=str(owner.id) if owner else None, ) db.add(row) await db.flush() return row async def test_a_relative_record_matches_by_composed_fqdn(db_session): """`name` holds the RELATIVE label and `zone_name` the zone, so a join on `name` alone returns nothing for every subdomain — indistinguishable from "Helix has no DNS data", and how this whole idea nearly got written off.""" owner = await _person(db_session) await _dns(db_session, zone="example.test", name="docs", rtype="CNAME", owner=owner) found = await dns_owner.suggest_owners_for_hostnames( db_session, ["docs.example.test"], ) assert [c["person_id"] for c in found] == [str(owner.id)] assert found[0]["hostname"] == "docs.example.test" assert found[0]["strength"] == "strong" async def test_an_absolute_record_is_attributed_to_itself_not_the_zone( db_session, ): """Some providers store the FQDN in `name`. Composing blindly gives `docs.example.test.example.test`, and falling back to the zone filed the candidate under the apex — the wrong hostname on the card.""" owner = await _person(db_session) await _dns(db_session, zone="example.test", name="app.example.test", rtype="A", owner=owner) found = await dns_owner.suggest_owners_for_hostnames( db_session, ["app.example.test", "example.test"], ) assert [c["hostname"] for c in found] == ["app.example.test"] async def test_the_zone_apex_matches_an_at_record(db_session): owner = await _person(db_session) await _dns(db_session, zone="apex.test", name="@", rtype="A", owner=owner) found = await dns_owner.suggest_owners_for_hostnames( db_session, ["apex.test"], ) assert [c["hostname"] for c in found] == ["apex.test"] async def test_a_departed_owner_is_never_proposed(db_session): """"Has an owner" has to mean "somebody is accountable". A departed name is worse than a blank — the blank still shows in the coverage gap and gets chased.""" gone = await _person(db_session, status="left") await _dns(db_session, zone="example.test", name="ghost", rtype="A", owner=gone) assert await dns_owner.suggest_owners_for_hostnames( db_session, ["ghost.example.test"], ) == [] async def test_zone_level_records_are_offered_but_never_auto_applied( db_session, ): """The load-bearing distinction. An NS owner administers the zone; in production one person holds the NS on two unrelated systems plus a third zone's MX.""" admin = await _person(db_session) await _dns(db_session, zone="example.test", name="labs", rtype="NS", owner=admin) found = await dns_owner.suggest_owners_for_hostnames( db_session, ["labs.example.test"], ) assert [c["strength"] for c in found] == ["weak"], ( "a zone-level record must be visible to a human deciding" ) assert await dns_owner.resolve_auto_owner( db_session, ["labs.example.test"], ) is None, "and must never be written without one" async def test_two_disagreeing_strong_records_do_not_auto_apply(db_session): """Exactly the case a person should settle.""" a, b = await _person(db_session), await _person(db_session) await _dns(db_session, zone="example.test", name="split", rtype="A", owner=a) await _dns(db_session, zone="example.test", name="split", rtype="CNAME", owner=b) found = await dns_owner.suggest_owners_for_hostnames( db_session, ["split.example.test"], ) assert len(found) == 2, "both are shown" assert await dns_owner.resolve_auto_owner( db_session, ["split.example.test"], ) is None, "but neither is written" async def test_registering_a_system_with_no_owner_adopts_the_dns_owner( db_session, ): """The behaviour asked for: a system added from a hostname that turned up in a researcher's mail should not land ownerless when the platform already knows the answer.""" owner = await _person(db_session) await _dns(db_session, zone="example.test", name="auto", rtype="A", owner=owner) system = await svc.create_system( db_session, name="Auto Filled", slug=f"auto-{uuid.uuid4().hex[:6]}", hostnames=["auto.example.test"], owner_person_id=None, actor=None, ) assert str(system.owner_person_id) == str(owner.id) # And the trail says where the name came from — "DNS said so" is a # weaker claim than "a person decided", and a dispute needs to tell # them apart. events = await svc.list_events(db_session, system.id) assigned = [ e for e in events if e.event_type == "owner_assigned" and (e.meta or {}).get("source") == "dns_record" ] assert len(assigned) == 1, [e.event_type for e in events] assert assigned[0].meta["dns_hostname"] == "auto.example.test" assert assigned[0].meta["dns_record_types"] == ["A"] async def test_an_explicit_owner_is_never_overridden_by_dns(db_session): chosen = await _person(db_session) dns_says = await _person(db_session) await _dns(db_session, zone="example.test", name="explicit", rtype="A", owner=dns_says) system = await svc.create_system( db_session, name="Explicit", slug=f"exp-{uuid.uuid4().hex[:6]}", hostnames=["explicit.example.test"], owner_person_id=chosen.id, actor=None, ) assert str(system.owner_person_id) == str(chosen.id) async def test_a_system_dns_cannot_answer_for_stays_visibly_ownerless( db_session, ): system = await svc.create_system( db_session, name="Unknown", slug=f"unk-{uuid.uuid4().hex[:6]}", hostnames=["nothing-knows-this.example.test"], owner_person_id=None, actor=None, ) assert system.owner_person_id is None, ( "guessing is worse than a blank: the blank is what the coverage " "gap chases" ) async def test_a_soft_deleted_record_does_not_vouch_for_anyone(db_session): """A record that was removed from the zone is not evidence of anything. `cmdb_dns_records` soft-deletes, and the first version of the coordinator did not filter it — so a hostname that had been decommissioned would still have named an owner.""" owner = await _person(db_session) await _dns(db_session, zone="example.test", name="retired", rtype="A", owner=owner, deleted=True) assert await dns_owner.suggest_owners_for_hostnames( db_session, ["retired.example.test"], ) == [] assert await dns_owner.resolve_auto_owner( db_session, ["retired.example.test"], ) is None