"""
Migration 173 — modifier tracking + batch annotate + DNS enable/disable.

Covers the new behaviour that ships alongside the legacy-CMDB data
migration. Each test is end-to-end against a real DB; the DNS request
+ provider integration is exercised separately in
``test_dns_request_*`` files (this one focuses on the admin direct
paths and the persistence of ``last_modified_*`` fields).
"""

import uuid
from unittest.mock import AsyncMock, patch

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.models import AuditLog
from app.domains.cmdb.models import (
    CmdbCdnDomain,
    CmdbDnsRecord,
    CmdbDomain,
    CmdbLoadBalancer,
    CmdbObjectStorage,
    CmdbServer,
)
from app.domains.cmdb.providers.exceptions import CapabilityNotSupported
from app.domains.cmdb.schemas import (
    DnsRecordValueProbe,
    _validate_dns_value,
)
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="tencent",
        name="pytest-fixture-tencent-173",
        slug=f"pytest-fixture-tencent-173-{str(integration_id)[:8]}",
        enabled=True,
        is_primary=False,
        capabilities=["DNS_RECORDS", "SERVERS", "LOAD_BALANCERS", "CDN", "DOMAINS"],
    )
    db_session.add(integ)
    await db_session.commit()
    return integ


# ─────────────────────────────────────────────────────────────────────
# DNS value validator (pydantic + helper)
# ─────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "record_type,value",
    [
        ("A", "1.2.3.4"),
        ("AAAA", "::1"),
        ("AAAA", "2606:4700:4700::1111"),
        ("CNAME", "example.com."),
        ("CNAME", "target.example.com"),
        ("NS", "ns1.example.com."),
        ("MX", "10 mail.example.com."),
        ("MX", "65535 mx.example.com."),
        ("SRV", "0 5 5060 sip.example.com."),
        ("CAA", '0 issue "letsencrypt.org"'),
        ("TXT", "v=spf1 -all"),
    ],
)
def test_dns_value_validator_accepts_valid_values(record_type, value):
    assert _validate_dns_value(record_type, value) == value
    DnsRecordValueProbe(record_type=record_type, value=value)


@pytest.mark.parametrize(
    "record_type,value",
    [
        ("A", "not-an-ip"),
        ("A", "256.256.256.256"),  # IPv4 octet overflow caught by ipaddress
        ("AAAA", "1.2.3.4"),       # IPv4 in AAAA slot
        ("CNAME", "not a domain"),  # space breaks FQDN
        ("MX", "10"),               # missing target
        ("MX", "high mail.example.com"),  # non-numeric priority
        ("MX", "70000 mail.example.com"),  # priority overflow
        ("SRV", "0 5 5060"),        # missing target
        ("SRV", "0 5 99999 t.example.com"),  # port overflow
        ("CAA", "0 unknown_tag value"),
    ],
)
def test_dns_value_validator_rejects_invalid_values(record_type, value):
    with pytest.raises(ValueError):
        _validate_dns_value(record_type, value)


def test_dns_value_validator_empty_rejected():
    with pytest.raises(ValueError):
        _validate_dns_value("A", "")
    with pytest.raises(ValueError):
        _validate_dns_value("A", "   ")


# ─────────────────────────────────────────────────────────────────────
# annotate_server: writes last_modified_by + last_modified_at
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def server_row(db_session: AsyncSession, integ: Integration) -> CmdbServer:
    s = CmdbServer(
        integration_id=integ.id,
        instance_id="i-test-173",
        name="m173-host",
        status="running",
        region="cn-beijing",
    )
    db_session.add(s)
    await db_session.commit()
    return s


@pytest.mark.asyncio
async def test_annotate_server_stamps_modifier(
    client, admin_user, server_row, db_session,
):
    person, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/servers/{server_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={"app": "my-app", "remark": "moved from KSY"},
    )
    assert resp.status_code == 200
    # Refresh from DB to see persisted state — the response body would
    # also expose these fields but reading the row is more thorough.
    await db_session.refresh(server_row)
    assert server_row.app == "my-app"
    assert server_row.remark == "moved from KSY"
    assert server_row.last_modified_by_person_id == str(person.id)
    # The exact timestamp depends on whether MySQL session timezone
    # matches the test process — skipping a strict >= check here and
    # only asserting non-null. Wall-clock precision is verified at the
    # service-level unit-test, not this E2E.
    assert server_row.last_modified_at is not None


@pytest.mark.asyncio
async def test_annotate_server_response_carries_modifier_label(
    client, admin_user, server_row,
):
    person, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/servers/{server_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={"app": "another"},
    )
    assert resp.status_code == 200
    body = resp.json()["data"]
    assert body["last_modified_by_person_id"] == str(person.id)
    assert body["last_modified_by_label"]  # populated via _resolve_owner_labels
    assert body["last_modified_at"] is not None
    assert body["app"] == "another"


# ─────────────────────────────────────────────────────────────────────
# Batch annotate — servers
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def server_cohort(db_session: AsyncSession, integ: Integration):
    rows = [
        CmdbServer(
            integration_id=integ.id,
            instance_id=f"i-batch-{i}",
            name=f"batch-host-{i}",
            status="running",
            region="cn-beijing",
        )
        for i in range(3)
    ]
    db_session.add_all(rows)
    await db_session.commit()
    return rows


@pytest.mark.asyncio
async def test_batch_annotate_servers_applies_to_all(
    client, admin_user, server_cohort, db_session,
):
    _, token = admin_user
    resp = await client.post(
        "/api/v1/cmdb/servers/batch-annotate",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "server_ids": [str(s.id) for s in server_cohort],
            "annotation": {"app": "team-alpha", "remark": "post-migration cleanup"},
        },
    )
    assert resp.status_code == 200, resp.text
    summary = resp.json()["data"]
    assert summary["requested"] == 3
    assert summary["updated"] == 3
    assert summary["not_found"] == 0
    for s in server_cohort:
        await db_session.refresh(s)
        assert s.app == "team-alpha"
        assert s.remark == "post-migration cleanup"
        assert s.last_modified_by_person_id is not None


@pytest.mark.asyncio
async def test_batch_annotate_servers_employee_forbidden(
    client, employee_user, server_cohort,
):
    _, token = employee_user
    resp = await client.post(
        "/api/v1/cmdb/servers/batch-annotate",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "server_ids": [str(s.id) for s in server_cohort],
            "annotation": {"app": "evil"},
        },
    )
    assert resp.status_code == 403


# ─────────────────────────────────────────────────────────────────────
# Batch update DNS record owner
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def dns_cohort(db_session: AsyncSession, integ: Integration):
    rows = [
        CmdbDnsRecord(
            integration_id=integ.id,
            zone_name="batch.example.com",
            record_id=f"rec-{i}",
            name=f"sub-{i}",
            record_type="A",
            value=f"10.0.0.{i + 1}",
            ttl=600,
            status="enable",
        )
        for i in range(3)
    ]
    db_session.add_all(rows)
    await db_session.commit()
    return rows


@pytest.mark.asyncio
async def test_batch_dns_owner_update(
    client, admin_user, dns_cohort, db_session,
):
    person, token = admin_user
    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r.id) for r in dns_cohort],
            "owner_person_id": str(person.id),
        },
    )
    assert resp.status_code == 200, resp.text
    summary = resp.json()["data"]
    assert summary["updated"] == 3
    for r in dns_cohort:
        await db_session.refresh(r)
        assert r.owner_person_id == str(person.id)
        assert r.last_modified_by_person_id == str(person.id)


@pytest.mark.asyncio
async def test_batch_dns_owner_clear(
    client, admin_user, dns_cohort, db_session,
):
    person, token = admin_user
    # First assign, then clear.
    await client.post(
        "/api/v1/cmdb/dns-records/batch-owner",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r.id) for r in dns_cohort],
            "owner_person_id": str(person.id),
        },
    )
    resp = await client.post(
        "/api/v1/cmdb/dns-records/batch-owner",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "record_ids": [str(r.id) for r in dns_cohort],
            "owner_person_id": None,
        },
    )
    assert resp.status_code == 200, resp.text
    for r in dns_cohort:
        await db_session.refresh(r)
        assert r.owner_person_id is None


# ─────────────────────────────────────────────────────────────────────
# DNS enable / disable (admin direct path) — mocks the provider call
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def dns_record(db_session: AsyncSession, integ: Integration) -> CmdbDnsRecord:
    r = CmdbDnsRecord(
        integration_id=integ.id,
        zone_name="status.example.com",
        record_id="rec-status",
        name="@",
        record_type="A",
        value="1.1.1.1",
        ttl=600,
        status="enable",
    )
    db_session.add(r)
    await db_session.commit()
    return r


@pytest.mark.asyncio
async def test_admin_disable_dns_record(
    client, admin_user, dns_record, db_session,
):
    _, token = admin_user
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        mock_provider = AsyncMock()
        mock_provider.set_dns_record_status = AsyncMock(return_value=None)
        mock_provider.close = AsyncMock(return_value=None)
        mock_loader.return_value = mock_provider
        resp = await client.post(
            f"/api/v1/cmdb/dns-records/{dns_record.id}/disable",
            headers={"Authorization": f"Bearer {token}"},
        )
    assert resp.status_code == 200, resp.text
    body = resp.json()["data"]
    assert body["status"] == "disable"
    assert body["last_modified_by_person_id"] is not None
    mock_provider.set_dns_record_status.assert_awaited_once()
    args, kwargs = mock_provider.set_dns_record_status.call_args
    # set_dns_record_status(record_id, zone_name, enable_bool)
    assert args[2] is False
    await db_session.refresh(dns_record)
    assert dns_record.status == "disable"


@pytest.mark.asyncio
async def test_admin_enable_dns_record_idempotent(
    client, admin_user, dns_record,
):
    _, token = admin_user
    # Record starts in "enable" — the route short-circuits with no cloud call.
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        resp = await client.post(
            f"/api/v1/cmdb/dns-records/{dns_record.id}/enable",
            headers={"Authorization": f"Bearer {token}"},
        )
    assert resp.status_code == 200
    assert resp.json()["data"]["status"] == "enable"
    mock_loader.assert_not_called()


@pytest.mark.asyncio
async def test_admin_enable_provider_unsupported_returns_501(
    client, admin_user, dns_record, db_session,
):
    """AWS Route 53 has no enable/disable — the provider raises
    CapabilityNotSupported and the router must surface 501 instead of
    a 500."""
    _, token = admin_user
    # Flip to disable first so the route doesn't no-op.
    dns_record.status = "disable"
    await db_session.commit()
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        mock_provider = AsyncMock()
        mock_provider.set_dns_record_status = AsyncMock(
            side_effect=CapabilityNotSupported("aws does not support DNS record enable/disable"),
        )
        mock_provider.close = AsyncMock(return_value=None)
        mock_loader.return_value = mock_provider
        resp = await client.post(
            f"/api/v1/cmdb/dns-records/{dns_record.id}/enable",
            headers={"Authorization": f"Bearer {token}"},
        )
    assert resp.status_code == 501
    assert "DNS_PROVIDER_NO_ENABLE_DISABLE" in resp.text


# ─────────────────────────────────────────────────────────────────────
# DNS PATCH — owner-only edit must NOT touch the cloud provider
# ─────────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_admin_patch_dns_owner_only_skips_cloud_write(
    client, admin_user, dns_record, db_session,
):
    """The edit modal pre-fills name/value/ttl/line from the row, so an
    owner-only edit arrives with all content fields set to their CURRENT
    values. The router must compare against the existing row and skip
    the provider write — otherwise the redundant cloud call surfaces
    either as opaque "更新失败" (provider rejection of no-op) or as
    silent wasted RPCs on every owner reassignment.
    """
    person, token = admin_user
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        resp = await client.patch(
            f"/api/v1/cmdb/dns-records/{dns_record.id}",
            headers={"Authorization": f"Bearer {token}"},
            json={
                # Identical to the existing row — these come from the
                # modal's initialValues, not user edits. ``name`` is
                # absent because host is immutable on update post-
                # 2026-05-28 (schema rejects it with 422).
                "value": dns_record.value,
                "ttl": dns_record.ttl,
                # The only real change.
                "owner_person_id": str(person.id),
                "remark": "transferring ownership",
            },
        )
    assert resp.status_code == 200, resp.text
    mock_loader.assert_not_called()
    await db_session.refresh(dns_record)
    assert dns_record.owner_person_id == str(person.id)
    assert dns_record.remark == "transferring ownership"


@pytest.mark.asyncio
async def test_admin_patch_dns_value_change_still_calls_cloud(
    client, admin_user, dns_record, db_session,
):
    """Sanity counterpart: when the value actually differs from the
    existing row, the cloud update MUST fire — otherwise the DB and
    the cloud drift apart.
    """
    _, token = admin_user
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        mock_provider = AsyncMock()
        mock_provider.update_dns_record = AsyncMock(return_value=None)
        mock_provider.close = AsyncMock(return_value=None)
        mock_loader.return_value = mock_provider
        resp = await client.patch(
            f"/api/v1/cmdb/dns-records/{dns_record.id}",
            headers={"Authorization": f"Bearer {token}"},
            json={
                # host omitted — schema 422s it post-2026-05-28.
                "value": "2.2.2.2",  # actually different
                "ttl": dns_record.ttl,
            },
        )
    assert resp.status_code == 200, resp.text
    mock_provider.update_dns_record.assert_awaited_once()
    await db_session.refresh(dns_record)
    assert dns_record.value == "2.2.2.2"


@pytest_asyncio.fixture
async def dirty_cname_record(
    db_session: AsyncSession, integ: Integration,
) -> CmdbDnsRecord:
    """A CNAME row whose synced ``value`` is a malformed IPv4 — exact
    shape we observed in prod (record_type=CNAME, value=120.92.92.158).
    Could come from a misclassified provider record or a legacy import.
    The admin should still be able to edit owner / remark / ttl on it
    without being blocked by the record_type→value format validator —
    they're not changing the value, after all.
    """
    r = CmdbDnsRecord(
        integration_id=integ.id,
        zone_name="pingcap.example",
        record_id="rec-dirty-cname",
        name="pch",
        record_type="CNAME",
        value="120.92.92.158",  # would fail _validate_dns_value
        ttl=600,
        status="enable",
    )
    db_session.add(r)
    await db_session.commit()
    return r


@pytest.mark.asyncio
async def test_admin_patch_dirty_cname_owner_only_succeeds(
    client, admin_user, dirty_cname_record, db_session,
):
    """Regression: PATCH ``value`` validator only fires when value
    actually changes. A CNAME row with a malformed (IPv4) synced value
    must still accept owner-only edits — otherwise admins can never
    re-assign ownership of any record whose synced data is dirty.
    Same root cause as has_content_change above (frontend re-sends
    initialValues on every PATCH).
    """
    person, token = admin_user
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        resp = await client.patch(
            f"/api/v1/cmdb/dns-records/{dirty_cname_record.id}",
            headers={"Authorization": f"Bearer {token}"},
            json={
                # host omitted — schema 422s it post-2026-05-28.
                "value": dirty_cname_record.value,  # unchanged dirty value
                "ttl": dirty_cname_record.ttl,
                "owner_person_id": str(person.id),
                "remark": "fixing ownership on legacy record",
            },
        )
    assert resp.status_code == 200, resp.text
    mock_loader.assert_not_called()
    await db_session.refresh(dirty_cname_record)
    assert dirty_cname_record.owner_person_id == str(person.id)


@pytest.mark.asyncio
async def test_admin_patch_cname_to_invalid_value_still_rejected(
    client, admin_user, dns_record, db_session,
):
    """Anti-regression for the validator-loosening above: if the admin
    GENUINELY tries to change a value to something the record_type
    rejects, we must still 4xx before any cloud call. ValidationError
    maps to 422 (see app/core/exceptions.py).
    """
    _, token = admin_user
    # Make this row a CNAME so the validator applies.
    dns_record.record_type = "CNAME"
    dns_record.value = "valid.example.com"
    await db_session.commit()
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        resp = await client.patch(
            f"/api/v1/cmdb/dns-records/{dns_record.id}",
            headers={"Authorization": f"Bearer {token}"},
            json={"value": "1.2.3.4"},  # IPv4 on CNAME — invalid
        )
    assert resp.status_code == 422, resp.text
    mock_loader.assert_not_called()


@pytest.mark.asyncio
async def test_admin_patch_dns_with_name_rejected(
    client, admin_user, dns_record, db_session,
):
    """Anti-regression for the 2026-05-28 prod incident: a frontend
    Form-state desync silently rewrote a DNS record's host from
    ``_dnsauth`` to ``feilian`` because ``name`` was editable on update.
    The schema is now ``extra='forbid'`` — any caller that includes
    ``name`` gets a 422 BEFORE the cloud-write branch, no matter what
    the value is. This locks in the contract so a future caller (MCP /
    mobile / curl) cannot re-introduce the footgun.
    """
    _, token = admin_user
    original_name = dns_record.name
    with patch(
        "app.domains.cmdb.router._load_provider_for_dns",
        new_callable=AsyncMock,
    ) as mock_loader:
        resp = await client.patch(
            f"/api/v1/cmdb/dns-records/{dns_record.id}",
            headers={"Authorization": f"Bearer {token}"},
            json={
                "name": "hijacked",  # attempted rename
                "value": dns_record.value,
                "ttl": dns_record.ttl,
            },
        )
    assert resp.status_code == 422, resp.text
    mock_loader.assert_not_called()
    await db_session.refresh(dns_record)
    assert dns_record.name == original_name


# ─────────────────────────────────────────────────────────────────────
# annotate_load_balancer + annotate_cdn — modifier + new editable fields
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def lb_row(db_session: AsyncSession, integ: Integration) -> CmdbLoadBalancer:
    lb = CmdbLoadBalancer(
        integration_id=integ.id,
        lb_id="lb-test-173",
        name="m173-lb",
        lb_type="classic",
        network_type="public",
        status="active",
    )
    db_session.add(lb)
    await db_session.commit()
    return lb


@pytest.mark.asyncio
async def test_annotate_lb_writes_modifier_and_app(
    client, admin_user, lb_row, db_session,
):
    person, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/load-balancers/{lb_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "app": "billing-api",
            "remark": "moved from UCloud",
            "owner_person_id": str(person.id),
        },
    )
    assert resp.status_code == 200, resp.text
    await db_session.refresh(lb_row)
    assert lb_row.app == "billing-api"
    assert lb_row.remark == "moved from UCloud"
    assert lb_row.owner_person_id == str(person.id)
    assert lb_row.last_modified_by_person_id == str(person.id)


@pytest_asyncio.fixture
async def cdn_row(db_session: AsyncSession, integ: Integration) -> CmdbCdnDomain:
    cdn = CmdbCdnDomain(
        integration_id=integ.id,
        domain_name="img.example.com",
        cname="img.example.com.cdn.example",
        status="online",
        source_type="domain",
        source_address="origin.example.com",
        is_https=False,
    )
    db_session.add(cdn)
    await db_session.commit()
    return cdn


@pytest.mark.asyncio
async def test_annotate_cdn_writes_cert_and_remark(
    client, admin_user, cdn_row, db_session,
):
    _, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/cdn-domains/{cdn_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={"cert_id": "cert-123", "remark": "renewed 2026-05"},
    )
    assert resp.status_code == 200, resp.text
    await db_session.refresh(cdn_row)
    assert cdn_row.cert_id == "cert-123"
    assert cdn_row.remark == "renewed 2026-05"
    assert cdn_row.last_modified_at is not None


# ─────────────────────────────────────────────────────────────────────
# annotate_domain — grade / possessor / remark
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def domain_row(db_session: AsyncSession, integ: Integration) -> CmdbDomain:
    d = CmdbDomain(
        integration_id=integ.id,
        domain_name="legacy.example.com",
        registrar="legacy",
        status="active",
    )
    db_session.add(d)
    await db_session.commit()
    return d


@pytest.mark.asyncio
async def test_annotate_domain_writes_grade_possessor_remark(
    client, admin_user, domain_row, db_session,
):
    _, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/domains/{domain_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={"grade": "S", "possessor": "PingCAP Inc.", "remark": "core brand domain"},
    )
    assert resp.status_code == 200, resp.text
    await db_session.refresh(domain_row)
    assert domain_row.grade == "S"
    assert domain_row.possessor == "PingCAP Inc."
    assert domain_row.remark == "core brand domain"
    assert domain_row.last_modified_at is not None


# ─────────────────────────────────────────────────────────────────────
# annotate_object_storage — owner_raw + audit_logs trail
# ─────────────────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def oss_row(db_session: AsyncSession, integ: Integration) -> CmdbObjectStorage:
    oss = CmdbObjectStorage(
        integration_id=integ.id,
        bucket_name="pytest-oss-bucket",
        region="cn-hongkong",
        status="active",
        owner_raw="legacy-team-finance",
    )
    db_session.add(oss)
    await db_session.commit()
    return oss


@pytest.mark.asyncio
async def test_annotate_object_storage_writes_audit_with_before_after(
    client, admin_user, oss_row, db_session,
):
    from sqlalchemy import select

    _, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/object-storage/{oss_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={"owner_raw": "platform-data"},
    )
    assert resp.status_code == 200, resp.text
    await db_session.refresh(oss_row)
    assert oss_row.owner_raw == "platform-data"

    rows = (
        await db_session.execute(
            select(AuditLog).where(
                AuditLog.action == "cmdb.object_storage.annotate",
                AuditLog.resource_id == str(oss_row.id),
            ),
        )
    ).scalars().all()
    assert len(rows) == 1
    log = rows[0]
    assert log.resource_type == "cmdb_object_storage"
    assert log.status == "success"
    assert log.detail["before"]["owner_raw"] == "legacy-team-finance"
    assert log.detail["after"]["owner_raw"] == "platform-data"
    assert log.detail["patch"] == {"owner_raw": "platform-data"}


@pytest.mark.asyncio
async def test_annotate_object_storage_sets_owner_person_id(
    client, admin_user, oss_row, db_session,
):
    """Migration 366: OSS owner parity with LB — PATCH owner_person_id
    sets the canonical FK and the response resolves owner_label +
    leaver-meta at serialize time (mirrors test_annotate_lb_*)."""
    person, token = admin_user
    resp = await client.patch(
        f"/api/v1/cmdb/object-storage/{oss_row.id}",
        headers={"Authorization": f"Bearer {token}"},
        json={"owner_person_id": str(person.id)},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()["data"]
    assert body["owner_person_id"] == str(person.id)
    # owner_label is resolved through _augment_response, not stored.
    assert body["owner_label"] and person.display_name in body["owner_label"]
    # Leaver-meta surfaces so PersonTag can render employment status.
    assert body["owner_employment_status"] == person.employment_status

    await db_session.refresh(oss_row)
    assert oss_row.owner_person_id == str(person.id)
    # Mirrors annotate_load_balancer: when the picker sends owner_person_id
    # without owner_raw, _resolve_owner_input backfills owner_raw with the
    # resolved person's email/display_name as the audit fallback.
    assert oss_row.owner_raw == (person.work_email or person.display_name)