"""Regression: ``sync_scopes_for_integration`` commits in batches
(prod 2026-04-24 15:14:48 UTC).

Bug context: GCP tenants with 100+ projects used to hold InnoDB row
locks on ``cmdb_cloud_scopes`` for the entire driver-list loop because
the function did one ``session.commit()`` at the very end. When a
concurrent path (admin "立即执行" button, or ``sync_grants_for_integration``'s
auto-owner adoption) also tried to write the same rows, they'd wait
on the locked rows until MySQL's 50-second ``innodb_lock_wait_timeout``
fired and the whole sync aborted with
``(asyncmy.errors.OperationalError) (1205, 'Lock wait timeout exceeded')``.

The fix commits every ``BATCH_SIZE`` scopes during the upsert loop +
an extra commit after the mark-and-close sweep. This test locks in:

* All scopes end up in the DB regardless of batch boundaries
  (e.g. exactly at the boundary, one below, one above, far above).
* Multiple commits are observed — the hold time is amortized. This
  is the part that actually prevents the lock timeout, so we assert
  it explicitly rather than trusting code shape.
* The mark-and-close sweep still runs and reports ``closed`` count.
"""
from __future__ import annotations

import uuid
from contextlib import asynccontextmanager
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.lifecycle_service import sync_scopes_for_integration
from app.domains.cmdb.models import (
    SCOPE_STATUS_ACTIVE,
    CmdbCloudScope,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import CloudScopeInfo
from app.integrations.models import Integration


async def _make_integration(
    db: AsyncSession, *, provider: str = "gcp",
) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-scope-batch-test",
        slug=f"{provider}-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=[],
    )
    db.add(row)
    await db.commit()
    return row


@pytest.mark.asyncio
async def test_same_provider_scope_owner_is_inherited_by_a_second_connector(
    db_session: AsyncSession,
) -> None:
    owner = IdentityPerson(
        id=uuid.uuid4(),
        source_system="test",
        hris_person_id=f"owner-{uuid.uuid4().hex[:8]}",
        work_email=f"owner-{uuid.uuid4().hex[:8]}@example.test",
        display_name="Cloud owner",
    )
    db_session.add(owner)
    await db_session.commit()
    primary = await _make_integration(db_session, provider="tencent")
    dns = await _make_integration(db_session, provider="tencent")
    account_id = "100012345678"
    db_session.add(CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(primary.id),
        scope_type="tencent_account",
        external_id=account_id,
        display_name="Tencent primary",
        status=SCOPE_STATUS_ACTIVE,
        owner_person_id=str(owner.id),
    ))
    await db_session.commit()
    driver = _FakeDriver([CloudScopeInfo(
        scope_type="tencent_account",
        external_id=account_id,
        display_name="Tencent DNS",
        status="active",
        raw={},
    )])

    with _patch_session_factory(db_session), _patch_driver(driver):
        await sync_scopes_for_integration(str(dns.id))

    inherited = await db_session.scalar(select(CmdbCloudScope).where(
        CmdbCloudScope.integration_id == str(dns.id),
        CmdbCloudScope.external_id == account_id,
    ))
    assert inherited is not None
    assert str(inherited.owner_person_id) == str(owner.id)


def _patch_session_factory(db_session: AsyncSession):
    @asynccontextmanager
    async def _factory():
        yield db_session
    return patch(
        "app.domains.cmdb.lifecycle_service.async_session_factory",
        new=_factory,
    )


class _FakeDriver:
    """Minimal stand-in for any cloud driver — only exposes
    ``list_scopes`` + ``close``, which is all the service touches."""

    def __init__(self, scopes: list[CloudScopeInfo]) -> None:
        self._scopes = scopes
        self.close = AsyncMock()

    async def list_scopes(self):
        return list(self._scopes)


def _patch_driver(driver: _FakeDriver):
    async def _loader(session, integration):
        return driver
    return patch(
        "app.domains.cmdb.lifecycle_service._load_cloud_driver",
        new=_loader,
    )


def _make_scope_info(i: int) -> CloudScopeInfo:
    return CloudScopeInfo(
        scope_type="gcp_project",
        external_id=f"gcp-proj-{i:04d}",
        display_name=f"project-{i}",
        status="active",
        raw={"index": i},
    )


@pytest.mark.asyncio
@pytest.mark.parametrize("n_scopes", [1, 49, 50, 51, 110, 250])
async def test_all_scopes_upserted_regardless_of_batch_boundary(
    db_session: AsyncSession, n_scopes: int,
) -> None:
    """Every scope returned by the driver must end up in the DB —
    boundary cases where the loop ends mid-batch are where the
    tail commit would matter most if it were missing."""
    integration = await _make_integration(db_session)
    driver = _FakeDriver([_make_scope_info(i) for i in range(n_scopes)])

    with _patch_session_factory(db_session), _patch_driver(driver):
        summary = await sync_scopes_for_integration(str(integration.id))

    assert summary["upserted"] == n_scopes
    count = await db_session.scalar(
        select(func.count(CmdbCloudScope.id)).where(
            CmdbCloudScope.integration_id == str(integration.id),
        )
    )
    assert count == n_scopes


@pytest.mark.asyncio
async def test_commit_happens_in_batches_not_once_at_end(
    db_session: AsyncSession,
) -> None:
    """Lock-hold time is what actually fails in prod — assert we're
    calling commit multiple times during a 110-scope run, not just
    once at the end (which was the pre-fix behavior)."""
    integration = await _make_integration(db_session)
    driver = _FakeDriver([_make_scope_info(i) for i in range(110)])

    commit_calls = 0
    original_commit = db_session.commit

    async def counting_commit() -> None:
        nonlocal commit_calls
        commit_calls += 1
        await original_commit()

    with _patch_session_factory(db_session), _patch_driver(driver):
        with patch.object(db_session, "commit", side_effect=counting_commit):
            await sync_scopes_for_integration(str(integration.id))

    # Expected commits for 110 scopes with BATCH_SIZE=50:
    #   2 mid-loop commits (at 50 and 100)
    #   + 1 tail commit after the upsert loop
    #   + 1 commit after the mark-and-close sweep
    #   + 1 commit after the unowned-incident loop
    # = 5 total. Allow ≥ 4 to tolerate implementation tweaks while still
    # proving we're NOT doing a single end-of-function commit.
    assert commit_calls >= 4, (
        f"Expected multiple commits for 110-scope run, got {commit_calls}"
    )


@pytest.mark.asyncio
async def test_orphan_scope_marked_closed_not_deleted(
    db_session: AsyncSession,
) -> None:
    """Mark-and-close sweep still runs after the batch-commit split —
    a scope in the DB that's missing from the driver's latest list
    gets ``status='closed'``, NOT soft-deleted (billing history
    depends on the FK still resolving).

    The pre-seeded orphan carries an OLD ``last_seen_sync_job_id`` to
    match the realistic shape of a row that was once observed by sync
    and is now missing — necessary after the 2026-05-28 sweep-predicate
    change that narrowed the close-sweep to rows previously observed by
    sync. A NULL ``last_seen_sync_job_id`` orphan would NOT be auto-
    closed (admin-only resource, not a sync-observed cloud row).
    """
    integration = await _make_integration(db_session)

    # Pre-seed an existing scope that the driver will no longer report.
    # Stamp a prior sync_job_id so the new sweep predicate
    # ``last_seen_sync_job_id IS NOT NULL`` matches — modeling a real
    # scope that was once cloud-observable.
    orphan_id = uuid.uuid4()
    prior_sync_job_id = uuid.uuid4()
    orphan = CmdbCloudScope(
        id=orphan_id,
        integration_id=str(integration.id),
        scope_type="gcp_project",
        external_id="gcp-orphan-001",
        display_name="orphan",
        status=SCOPE_STATUS_ACTIVE,
        last_seen_sync_job_id=str(prior_sync_job_id),
    )
    db_session.add(orphan)
    await db_session.commit()

    # Four returned + one orphan = exactly 20% drop, which is within
    # the default tolerance and must still permit the close sweep.
    driver = _FakeDriver([_make_scope_info(i) for i in range(1, 5)])
    with _patch_session_factory(db_session), _patch_driver(driver):
        summary = await sync_scopes_for_integration(str(integration.id))

    assert summary["closed"] == 1
    await db_session.refresh(orphan)
    assert orphan.status == "closed"
    assert orphan.deleted_at is None


@pytest.mark.asyncio
async def test_small_scope_snapshot_over_tolerance_refuses_close_sweep(
    db_session: AsyncSession,
) -> None:
    """A 2 → 1 snapshot is a 50% drop, not an acceptable 20% drop."""
    integration = await _make_integration(db_session)
    prior_job_id = str(uuid.uuid4())
    scopes = []
    for index in range(2):
        scope = CmdbCloudScope(
            id=uuid.uuid4(),
            integration_id=str(integration.id),
            scope_type="gcp_project",
            external_id=f"small-tenant-{index}",
            display_name=f"small tenant {index}",
            status=SCOPE_STATUS_ACTIVE,
            last_seen_sync_job_id=prior_job_id,
        )
        db_session.add(scope)
        scopes.append(scope)
    await db_session.commit()

    survivor = CloudScopeInfo(
        scope_type="gcp_project",
        external_id="small-tenant-0",
        display_name="small tenant 0",
        status="active",
        raw={},
    )
    with _patch_session_factory(db_session), _patch_driver(_FakeDriver([survivor])):
        with pytest.raises(RuntimeError, match="refusing close-sweep"):
            await sync_scopes_for_integration(str(integration.id))

    await db_session.refresh(scopes[1])
    assert scopes[1].status == SCOPE_STATUS_ACTIVE


@pytest.mark.asyncio
async def test_scope_sync_does_not_close_admin_seeded_null_job_scope(
    db_session: AsyncSession,
) -> None:
    """Counter-test for the 2026-05-28 sweep-predicate change: a scope
    with NULL ``last_seen_sync_job_id`` (e.g. admin-created via the
    scope-request approval flow, future migration import) is
    structurally outside the cloud-sync's authority and MUST NOT be
    auto-closed when the driver doesn't return it. Otherwise the
    legacy-CMDB → server-sync collision incident would replicate to
    scopes.
    """
    integration = await _make_integration(db_session)

    admin_seeded_id = uuid.uuid4()
    admin_seeded = CmdbCloudScope(
        id=admin_seeded_id,
        integration_id=str(integration.id),
        scope_type="gcp_project",
        external_id="gcp-admin-001",
        display_name="admin-seeded",
        status=SCOPE_STATUS_ACTIVE,
        # No last_seen_sync_job_id — admin created this directly.
    )
    db_session.add(admin_seeded)
    await db_session.commit()

    driver = _FakeDriver([_make_scope_info(1)])  # not gcp-admin-001
    with _patch_session_factory(db_session), _patch_driver(driver):
        summary = await sync_scopes_for_integration(str(integration.id))

    assert summary["closed"] == 0, (
        "admin-seeded scope (NULL last_seen_sync_job_id) must NOT be "
        "auto-closed by cloud-sync — same protection as the resource "
        "sweep in cmdb/sync.py."
    )
    await db_session.refresh(admin_seeded)
    assert admin_seeded.status == SCOPE_STATUS_ACTIVE
    assert admin_seeded.deleted_at is None