"""MCP ``server_destroy`` locks the row like its HTTP twin (2026-06-20).

``destroy`` is irreversible (cloud-side TerminateInstances). The HTTP twin
``_execute_server_lifecycle`` (cmdb/router.py) loads the row with
``SELECT ... WHERE deleted_at IS NULL FOR UPDATE`` precisely so two
concurrent callers (MCP+UI, or a double-fire) can't both pass the
``deleted_at IS NULL`` read and both terminate the same instance.

The MCP path previously used a plain ``rt.db.get(...)`` with no lock and no
``deleted_at IS NULL`` predicate in the query. This test pins the
observable guarantee the lock provides: once a server is destroyed (row
soft-deleted), a second ``server_destroy`` re-reads, no longer matches
``deleted_at IS NULL``, raises ``NotFound`` and does NOT fire a second
provider terminate.
"""
from __future__ import annotations

import uuid
from unittest.mock import AsyncMock

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

from app.core.exceptions import NotFoundError
from app.domains.cmdb.models import CmdbServer
from app.integrations.models import Integration
from app.mcp import context as _mcp_ctx
from app.mcp import tools as _tools

pytestmark = pytest.mark.asyncio


@pytest.fixture()
def _patch_factory(db_engine, monkeypatch):
    from sqlalchemy.ext.asyncio import async_sessionmaker

    from app.core import database as _db

    monkeypatch.setattr(
        _db,
        "async_session_factory",
        async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False),
    )


def _mock_provider() -> AsyncMock:
    provider = AsyncMock()
    provider.power_on = AsyncMock()
    provider.power_off = AsyncMock()
    provider.restart_server = AsyncMock()
    provider.destroy_server = AsyncMock()
    provider.close = AsyncMock(return_value=None)
    return provider


async def _make_server(db: AsyncSession) -> CmdbServer:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="ucloud",
        name=f"pytest-mcp-destroy-{uuid.uuid4().hex[:6]}",
        slug=f"pytest-mcp-destroy-{uuid.uuid4().hex[:8]}",
        enabled=True,
        is_primary=False,
        capabilities=["SERVERS"],
    )
    db.add(integ)
    await db.commit()
    srv = CmdbServer(
        integration_id=integ.id,
        instance_id="uhost-mcp-destroy",
        name="mcp-destroy-host",
        status="running",
        region="cn-bj2",
        project_id="org-mcp-destroy",
    )
    db.add(srv)
    await db.commit()
    return srv


async def test_second_destroy_is_notfound_and_does_not_refire(
    cloud_admin_user, db_session, _patch_factory, monkeypatch,
):
    admin, _ = cloud_admin_user
    srv = await _make_server(db_session)

    provider = _mock_provider()

    async def _fake_load(db, integration_id):  # noqa: ANN001
        return provider

    # The MCP tool imports _load_provider_for_dns from the router module at
    # call time, so patch it at the source.
    monkeypatch.setattr(
        "app.domains.cmdb.router._load_provider_for_dns", _fake_load,
    )

    token = _mcp_ctx.mcp_user_id.set(admin.id)
    try:
        res = await _tools.server_destroy(server_id=str(srv.id))
        assert res is not None
        # First destroy fired the provider terminate exactly once.
        assert provider.destroy_server.await_count == 1

        # Row is soft-deleted.
        row = (
            await db_session.execute(
                text("SELECT deleted_at FROM cmdb_servers WHERE id = :id"),
                {"id": str(srv.id)},
            )
        ).first()
        assert row is not None and row[0] is not None

        # Second destroy must NOT re-fire the provider — the locked,
        # deleted_at-filtered read no longer matches, so it 404s.
        with pytest.raises(NotFoundError):
            await _tools.server_destroy(server_id=str(srv.id))
        assert provider.destroy_server.await_count == 1
    finally:
        _mcp_ctx.mcp_user_id.reset(token)


async def _make_prod_server(db: AsyncSession) -> CmdbServer:
    """A server whose integration has an ACTIVE prod cloud scope."""
    from app.domains.cmdb.models import CmdbCloudScope

    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="ucloud",
        name=f"pytest-mcp-prodlock-{uuid.uuid4().hex[:6]}",
        slug=f"pytest-mcp-prodlock-{uuid.uuid4().hex[:8]}",
        enabled=True, is_primary=False, capabilities=["SERVERS"],
    )
    db.add(integ)
    await db.flush()
    db.add(CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integ.id),
        scope_type="project", external_id="org-mcp-prod",
        display_name="mcp prod account", environment="prod",
    ))
    srv = CmdbServer(
        integration_id=integ.id, instance_id="uhost-mcp-prodlock",
        name="mcp-prodlock-host", status="running", region="cn-bj2",
        project_id="org-mcp-prod",
    )
    db.add(srv)
    await db.commit()
    return srv


async def test_mcp_destroy_blocked_by_prod_write_lock(
    cloud_admin_user, db_session, _patch_factory, monkeypatch,
):
    """Regression (2026-07-03 audit): the MCP server-lifecycle tools must
    honour the prod cloud-write kill-switch, exactly like the HTTP twin.
    A prod server + engaged lock (default-locked) must refuse destroy and
    NOT fire the provider terminate."""
    from app.core.exceptions import AppException

    admin, _ = cloud_admin_user
    srv = await _make_prod_server(db_session)  # default = writes locked
    provider = _mock_provider()

    async def _fake_load(db, integration_id):  # noqa: ANN001
        return provider

    monkeypatch.setattr(
        "app.domains.cmdb.router._load_provider_for_dns", _fake_load,
    )

    token = _mcp_ctx.mcp_user_id.set(admin.id)
    try:
        with pytest.raises(AppException) as exc:
            await _tools.server_destroy(server_id=str(srv.id))
        assert exc.value.code == "PROD_CLOUD_WRITES_LOCKED"
    finally:
        _mcp_ctx.mcp_user_id.reset(token)

    # The irreversible provider terminate must NOT have fired.
    assert provider.destroy_server.await_count == 0
    # Row untouched.
    fresh = await db_session.get(CmdbServer, srv.id)
    assert fresh is not None and fresh.deleted_at is None