from __future__ import annotations

from uuid import uuid4

import pytest
from langflow.services.database.models.deployment.model import Deployment
from langflow.services.database.models.deployment_provider_account.model import (
    DeploymentProviderAccount,
    DeploymentProviderKey,
)
from langflow.services.database.models.flow.model import Flow
from langflow.services.database.models.flow_version.model import FlowVersion
from langflow.services.database.models.flow_version_deployment_attachment.crud import (
    DeploymentAttachmentConflictError,
    SnapshotFlowVersionConflictError,
    count_attachments_by_deployment_ids,
    delete_deployment_attachment,
    delete_deployment_attachments_by_deployment_id,
    delete_deployment_attachments_by_keys,
    get_deployment_attachment,
    list_attachments_by_deployment_ids,
    list_attachments_for_flow_with_provider_info,
    list_deployment_attachments,
    list_deployment_attachments_for_flow_version_ids,
    update_deployment_attachment_provider_snapshot_id,
    update_flow_version_by_provider_snapshot_id,
)
from langflow.services.database.models.flow_version_deployment_attachment.crud import (
    create_deployment_attachment as _create_deployment_attachment,
)
from langflow.services.database.models.flow_version_deployment_attachment.model import (
    FlowVersionDeploymentAttachment,
)
from langflow.services.database.models.flow_version_deployment_attachment.schema import (
    DeploymentAttachmentKeyBatch,
)
from langflow.services.database.models.folder.model import Folder
from langflow.services.database.models.user.model import User
from lfx.services.adapters.deployment.schema import DeploymentType
from sqlalchemy import event
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import StaticPool
from sqlmodel import SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession

_TEST_PASSWORD = "hashed"  # noqa: S105  # pragma: allowlist secret


def _default_snapshot_id(*, flow_version_id, deployment_id) -> str:
    return f"snap-{deployment_id}-{flow_version_id}"


async def create_deployment_attachment(
    db: AsyncSession,
    *,
    user_id,
    flow_version_id,
    deployment_id,
    provider_snapshot_id: str | None = None,
):
    """Test-only wrapper that auto-generates provider_snapshot_id when omitted."""
    return await _create_deployment_attachment(
        db,
        user_id=user_id,
        flow_version_id=flow_version_id,
        deployment_id=deployment_id,
        provider_snapshot_id=provider_snapshot_id
        or _default_snapshot_id(
            flow_version_id=flow_version_id,
            deployment_id=deployment_id,
        ),
    )


@pytest.fixture(name="db_engine")
def db_engine_fixture():
    engine = create_async_engine(
        "sqlite+aiosqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

    @event.listens_for(engine.sync_engine, "connect")
    def _enable_fk(dbapi_connection, connection_record):  # noqa: ARG001
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

    return engine


@pytest.fixture(name="db")
async def db_fixture(db_engine):
    async with db_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    async with AsyncSession(db_engine, expire_on_commit=False) as session:
        yield session
    async with db_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)
    await db_engine.dispose()


@pytest.fixture
async def user(db: AsyncSession) -> User:
    u = User(username="testuser", password=_TEST_PASSWORD, is_active=True)
    db.add(u)
    await db.commit()
    await db.refresh(u)
    return u


@pytest.fixture
async def folder(db: AsyncSession, user: User) -> Folder:
    f = Folder(name="test-project", user_id=user.id)
    db.add(f)
    await db.commit()
    await db.refresh(f)
    return f


@pytest.fixture
async def provider_account(db: AsyncSession, user: User) -> DeploymentProviderAccount:
    acct = DeploymentProviderAccount(
        user_id=user.id,
        provider_tenant_id="tenant-1",
        provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
        name="test-provider",
        provider_url="https://provider.example.com",
        api_key="encrypted-value",  # pragma: allowlist secret
    )
    db.add(acct)
    await db.commit()
    await db.refresh(acct)
    return acct


@pytest.fixture
async def deployment(
    db: AsyncSession,
    user: User,
    folder: Folder,
    provider_account: DeploymentProviderAccount,
) -> Deployment:
    d = Deployment(
        user_id=user.id,
        project_id=folder.id,
        deployment_provider_account_id=provider_account.id,
        resource_key="rk-1",
        name="test-deployment",
        deployment_type=DeploymentType.AGENT,
    )
    db.add(d)
    await db.commit()
    await db.refresh(d)
    return d


@pytest.fixture
async def flow(db: AsyncSession, user: User, folder: Folder) -> Flow:
    f = Flow(
        id=uuid4(),
        name="test-flow",
        user_id=user.id,
        folder_id=folder.id,
        data={"nodes": [], "edges": []},
    )
    db.add(f)
    await db.commit()
    await db.refresh(f)
    return f


@pytest.fixture
async def flow_version(db: AsyncSession, user: User, flow: Flow) -> FlowVersion:
    fv = FlowVersion(
        flow_id=flow.id,
        user_id=user.id,
        version_number=1,
        data={"nodes": [], "edges": []},
    )
    db.add(fv)
    await db.commit()
    await db.refresh(fv)
    return fv


@pytest.mark.asyncio
class TestCreateDeploymentAttachment:
    async def test_create_happy_path(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="snap-1",
        )
        await db.commit()

        assert att.id is not None
        assert att.user_id == user.id
        assert att.flow_version_id == flow_version.id
        assert att.deployment_id == deployment.id
        assert att.provider_snapshot_id == "snap-1"
        assert att.created_at is not None

    async def test_create_without_snapshot_id_raises(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        with pytest.raises(ValueError, match="provider_snapshot_id must not be empty"):
            await _create_deployment_attachment(
                db,
                user_id=user.id,
                flow_version_id=flow_version.id,
                deployment_id=deployment.id,
                provider_snapshot_id=None,
            )

    async def test_create_duplicate_raises_value_error(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        await db.commit()

        with pytest.raises(DeploymentAttachmentConflictError, match="Attachment conflicts with an existing record"):
            await create_deployment_attachment(
                db,
                user_id=user.id,
                flow_version_id=flow_version.id,
                deployment_id=deployment.id,
            )


@pytest.mark.asyncio
class TestGetDeploymentAttachment:
    async def test_get_existing(self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment):
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        await db.commit()

        result = await get_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert result is not None
        assert result.flow_version_id == flow_version.id

    async def test_get_nonexistent_returns_none(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        result = await get_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert result is None

    async def test_get_wrong_user_returns_none(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        await db.commit()

        other_user = User(username="other", password=_TEST_PASSWORD, is_active=True)
        db.add(other_user)
        await db.commit()
        await db.refresh(other_user)

        result = await get_deployment_attachment(
            db,
            user_id=other_user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert result is None


@pytest.mark.asyncio
class TestListDeploymentAttachments:
    async def test_list_ordered_by_created_at(self, db: AsyncSession, user: User, flow: Flow, deployment: Deployment):
        fv1 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=3, data={})
        db.add_all([fv1, fv2])
        await db.commit()
        await db.refresh(fv1)
        await db.refresh(fv2)

        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv1.id, deployment_id=deployment.id)
        await db.commit()
        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv2.id, deployment_id=deployment.id)
        await db.commit()

        results = await list_deployment_attachments(db, user_id=user.id, deployment_id=deployment.id)
        assert len(results) == 2
        assert results[0].flow_version_id == fv1.id
        assert results[1].flow_version_id == fv2.id

    async def test_list_empty(self, db: AsyncSession, user: User, deployment: Deployment):
        results = await list_deployment_attachments(db, user_id=user.id, deployment_id=deployment.id)
        assert results == []


@pytest.mark.asyncio
class TestListDeploymentAttachmentsForFlowVersionIds:
    async def test_filters_by_flow_version_ids(self, db: AsyncSession, user: User, flow: Flow, deployment: Deployment):
        fv1 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=3, data={})
        fv3 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=4, data={})
        db.add_all([fv1, fv2, fv3])
        await db.commit()
        await db.refresh(fv1)
        await db.refresh(fv2)
        await db.refresh(fv3)

        for fv in [fv1, fv2, fv3]:
            await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv.id, deployment_id=deployment.id)
        await db.commit()

        results = await list_deployment_attachments_for_flow_version_ids(
            db,
            user_id=user.id,
            deployment_id=deployment.id,
            flow_version_ids=[fv1.id, fv3.id],
        )
        assert len(results) == 2
        returned_fv_ids = {r.flow_version_id for r in results}
        assert returned_fv_ids == {fv1.id, fv3.id}

    async def test_empty_flow_version_ids_returns_empty(self, db: AsyncSession, user: User, deployment: Deployment):
        results = await list_deployment_attachments_for_flow_version_ids(
            db,
            user_id=user.id,
            deployment_id=deployment.id,
            flow_version_ids=[],
        )
        assert results == []


@pytest.mark.asyncio
class TestDeleteDeploymentAttachment:
    async def test_delete_existing(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()

        count = await delete_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert count == 1

    async def test_delete_nonexistent_returns_zero(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        count = await delete_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert count == 0


@pytest.mark.asyncio
class TestUpdateDeploymentAttachmentProviderSnapshotId:
    async def test_update_snapshot_id(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()

        assert att.provider_snapshot_id is not None

        updated = await update_deployment_attachment_provider_snapshot_id(
            db, attachment=att, provider_snapshot_id="snap-new"
        )
        await db.commit()

        assert updated.provider_snapshot_id == "snap-new"

    async def test_blank_snapshot_id_raises(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="snap-old",
        )
        await db.commit()

        with pytest.raises(ValueError, match="provider_snapshot_id must not be empty"):
            await update_deployment_attachment_provider_snapshot_id(
                db,
                attachment=att,
                provider_snapshot_id="   ",
            )

    async def test_same_snapshot_id_is_noop(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="snap-same",
        )
        await db.commit()

        updated = await update_deployment_attachment_provider_snapshot_id(
            db,
            attachment=att,
            provider_snapshot_id="snap-same",
        )
        await db.commit()

        assert updated.id == att.id
        assert updated.provider_snapshot_id == "snap-same"


@pytest.mark.asyncio
class TestUpdateFlowVersionByProviderSnapshotId:
    async def test_updates_all_rows_for_snapshot_id(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
        folder: Folder,
        provider_account: DeploymentProviderAccount,
    ):
        flow_version_2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        deployment_2 = Deployment(
            user_id=user.id,
            project_id=folder.id,
            deployment_provider_account_id=provider_account.id,
            resource_key="rk-2",
            name="test-deployment-2",
            deployment_type=DeploymentType.AGENT,
        )
        db.add_all([flow_version_2, deployment_2])
        await db.commit()
        await db.refresh(flow_version_2)
        await db.refresh(deployment_2)

        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="tool-A",
        )
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment_2.id,
            provider_snapshot_id="tool-A",
        )
        await db.commit()

        updated_count = await update_flow_version_by_provider_snapshot_id(
            db,
            user_id=user.id,
            provider_snapshot_id="tool-A",
            flow_version_id=flow_version_2.id,
        )
        await db.commit()

        assert updated_count == 2

        rows = (
            await db.exec(
                select(FlowVersionDeploymentAttachment).where(
                    FlowVersionDeploymentAttachment.user_id == user.id,
                    FlowVersionDeploymentAttachment.provider_snapshot_id == "tool-A",
                )
            )
        ).all()
        assert len(rows) == 2
        assert all(row.flow_version_id == flow_version_2.id for row in rows)

    async def test_returns_zero_when_snapshot_id_not_found(self, db: AsyncSession, user: User):
        updated_count = await update_flow_version_by_provider_snapshot_id(
            db,
            user_id=user.id,
            provider_snapshot_id="missing",
            flow_version_id=uuid4(),
        )
        assert updated_count == 0


@pytest.mark.asyncio
class TestDeleteDeploymentAttachmentsByDeploymentId:
    async def test_delete_all_for_deployment(self, db: AsyncSession, user: User, flow: Flow, deployment: Deployment):
        fv1 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=3, data={})
        db.add_all([fv1, fv2])
        await db.commit()
        await db.refresh(fv1)
        await db.refresh(fv2)

        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv1.id, deployment_id=deployment.id)
        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv2.id, deployment_id=deployment.id)
        await db.commit()

        count = await delete_deployment_attachments_by_deployment_id(db, user_id=user.id, deployment_id=deployment.id)
        assert count == 2

        remaining = await list_deployment_attachments(db, user_id=user.id, deployment_id=deployment.id)
        assert remaining == []

    async def test_delete_none_returns_zero(self, db: AsyncSession, user: User, deployment: Deployment):
        count = await delete_deployment_attachments_by_deployment_id(db, user_id=user.id, deployment_id=deployment.id)
        assert count == 0


@pytest.mark.asyncio
class TestDeleteDeploymentAttachmentsByKeys:
    async def test_delete_exact_pairs_avoids_cartesian_cross_delete(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        deployment: Deployment,
        folder: Folder,
        provider_account: DeploymentProviderAccount,
    ):
        """Regression: deleting two pairs must not delete cross-combination rows."""
        d2 = Deployment(
            user_id=user.id,
            project_id=folder.id,
            deployment_provider_account_id=provider_account.id,
            resource_key="rk-keys-2",
            name="deploy-keys-2",
            deployment_type=DeploymentType.AGENT,
        )
        db.add(d2)

        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        fv3 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=3, data={})
        db.add_all([fv2, fv3])
        await db.commit()
        await db.refresh(d2)
        await db.refresh(fv2)
        await db.refresh(fv3)

        dep1_target = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=fv2.id,
            deployment_id=deployment.id,
            provider_snapshot_id="snap-target-1",
        )
        dep2_target = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=fv3.id,
            deployment_id=d2.id,
            provider_snapshot_id="snap-target-2",
        )
        dep1_cross = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=fv3.id,
            deployment_id=deployment.id,
            provider_snapshot_id="snap-cross-1",
        )
        dep2_cross = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=fv2.id,
            deployment_id=d2.id,
            provider_snapshot_id="snap-cross-2",
        )
        await db.commit()

        deleted_count = await delete_deployment_attachments_by_keys(
            db,
            user_id=user.id,
            attachment_key_batch=DeploymentAttachmentKeyBatch(
                keys=[
                    {"deployment_id": deployment.id, "flow_version_id": fv2.id},
                    {"deployment_id": d2.id, "flow_version_id": fv3.id},
                ]
            ),
        )
        await db.commit()

        remaining = await list_attachments_by_deployment_ids(db, user_id=user.id, deployment_ids=[deployment.id, d2.id])
        remaining_ids = {row.id for row in remaining}
        assert deleted_count == 2
        assert dep1_target.id not in remaining_ids
        assert dep2_target.id not in remaining_ids
        assert dep1_cross.id in remaining_ids
        assert dep2_cross.id in remaining_ids


@pytest.mark.asyncio
class TestListAttachmentsByDeploymentIds:
    async def test_list_across_deployments(
        self,
        db: AsyncSession,
        user: User,
        flow_version: FlowVersion,
        deployment: Deployment,
        folder: Folder,
        provider_account: DeploymentProviderAccount,
    ):
        d2 = Deployment(
            user_id=user.id,
            project_id=folder.id,
            deployment_provider_account_id=provider_account.id,
            resource_key="rk-2",
            name="deploy-2",
        )
        db.add(d2)
        await db.commit()
        await db.refresh(d2)

        fv2 = FlowVersion(flow_id=flow_version.flow_id, user_id=user.id, version_number=2, data={})
        db.add(fv2)
        await db.commit()
        await db.refresh(fv2)

        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv2.id, deployment_id=d2.id)
        await db.commit()

        results = await list_attachments_by_deployment_ids(db, user_id=user.id, deployment_ids=[deployment.id, d2.id])
        assert len(results) == 2

    async def test_empty_deployment_ids_returns_empty(self, db: AsyncSession, user: User):
        results = await list_attachments_by_deployment_ids(db, user_id=user.id, deployment_ids=[])
        assert results == []


@pytest.mark.asyncio
class TestListAttachmentsForFlowWithProviderInfo:
    async def test_returns_tuples_with_provider_info(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
        provider_account: DeploymentProviderAccount,
    ):
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="snap-1",
        )
        await db.commit()

        results = await list_attachments_for_flow_with_provider_info(db, user_id=user.id, flow_ids=[flow.id])
        assert len(results) == 1
        attachment, provider_account_id, provider_key = results[0]
        assert attachment.flow_version_id == flow_version.id
        assert provider_account_id == provider_account.id
        assert provider_key == DeploymentProviderKey.WATSONX_ORCHESTRATE

    async def test_no_attachments_returns_empty(self, db: AsyncSession, user: User, flow: Flow):
        results = await list_attachments_for_flow_with_provider_info(db, user_id=user.id, flow_ids=[flow.id])
        assert results == []

    async def test_empty_flow_ids_returns_empty(self, db: AsyncSession, user: User):
        results = await list_attachments_for_flow_with_provider_info(db, user_id=user.id, flow_ids=[])
        assert results == []

    async def test_multiple_versions_same_flow(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
        provider_account: DeploymentProviderAccount,  # noqa: ARG002
    ):
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        db.add(fv2)
        await db.commit()
        await db.refresh(fv2)

        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()
        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv2.id, deployment_id=deployment.id)
        await db.commit()

        results = await list_attachments_for_flow_with_provider_info(db, user_id=user.id, flow_ids=[flow.id])
        assert len(results) == 2

    async def test_multiple_flows_returns_attachments_for_each_flow(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
    ):
        second_flow = Flow(
            id=uuid4(),
            name="test-flow-2",
            user_id=user.id,
            folder_id=flow.folder_id,
            data={"nodes": [], "edges": []},
        )
        db.add(second_flow)
        await db.commit()
        await db.refresh(second_flow)

        second_flow_version = FlowVersion(
            flow_id=second_flow.id,
            user_id=user.id,
            version_number=1,
            data={"nodes": [], "edges": []},
        )
        db.add(second_flow_version)
        await db.commit()
        await db.refresh(second_flow_version)

        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=second_flow_version.id, deployment_id=deployment.id
        )
        await db.commit()

        results = await list_attachments_for_flow_with_provider_info(
            db,
            user_id=user.id,
            flow_ids=[flow.id, second_flow.id],
        )

        assert len(results) == 2
        flow_version_ids = {attachment.flow_version_id for attachment, _, _ in results}
        assert flow_version_ids == {flow_version.id, second_flow_version.id}


@pytest.mark.asyncio
class TestCountAttachmentsByDeploymentIds:
    async def test_count_distinct_flow_versions(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
    ):
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        db.add(fv2)
        await db.commit()
        await db.refresh(fv2)

        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await create_deployment_attachment(db, user_id=user.id, flow_version_id=fv2.id, deployment_id=deployment.id)
        await db.commit()

        counts = await count_attachments_by_deployment_ids(db, user_id=user.id, deployment_ids=[deployment.id])
        assert counts == {deployment.id: 2}

    async def test_empty_deployment_ids_returns_empty_dict(self, db: AsyncSession, user: User):
        counts = await count_attachments_by_deployment_ids(db, user_id=user.id, deployment_ids=[])
        assert counts == {}

    async def test_deployment_with_no_attachments_returns_zero_count(
        self, db: AsyncSession, user: User, deployment: Deployment
    ):
        counts = await count_attachments_by_deployment_ids(db, user_id=user.id, deployment_ids=[deployment.id])
        assert counts[deployment.id] == 0


@pytest.mark.asyncio
class TestSnapshotFlowVersionConflict:
    """Ensure a provider_snapshot_id can only be linked to one flow version.

    Valid:   (FV=1, tool=A, deploy=X) + (FV=1, tool=A, deploy=Y)  — same FV, different deployments
    Invalid: (FV=1, tool=A, deploy=X) + (FV=2, tool=A, deploy=Y)  — different FV, same tool
    """

    async def test_create_rejects_same_tool_different_flow_version(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
        folder: Folder,
        provider_account: DeploymentProviderAccount,
    ):
        # First attachment: FV1, tool-A, deployment-1
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="tool-A",
        )
        await db.commit()

        # Second flow version + deployment
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        db.add(fv2)
        d2 = Deployment(
            user_id=user.id,
            project_id=folder.id,
            deployment_provider_account_id=provider_account.id,
            resource_key="rk-2",
            name="deploy-2",
            deployment_type=DeploymentType.AGENT,
        )
        db.add(d2)
        await db.commit()
        await db.refresh(fv2)
        await db.refresh(d2)

        # Should reject: FV2, tool-A, deployment-2 — same tool, different flow version
        with pytest.raises(SnapshotFlowVersionConflictError, match="already attached to a different flow version"):
            await create_deployment_attachment(
                db,
                user_id=user.id,
                flow_version_id=fv2.id,
                deployment_id=d2.id,
                provider_snapshot_id="tool-A",
            )

    async def test_create_allows_same_tool_same_flow_version_different_deployment(
        self,
        db: AsyncSession,
        user: User,
        flow_version: FlowVersion,
        deployment: Deployment,
        folder: Folder,
        provider_account: DeploymentProviderAccount,
    ):
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="tool-A",
        )
        await db.commit()

        d2 = Deployment(
            user_id=user.id,
            project_id=folder.id,
            deployment_provider_account_id=provider_account.id,
            resource_key="rk-2",
            name="deploy-2",
            deployment_type=DeploymentType.AGENT,
        )
        db.add(d2)
        await db.commit()
        await db.refresh(d2)

        # Should succeed: same FV, same tool, different deployment
        att2 = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=d2.id,
            provider_snapshot_id="tool-A",
        )
        await db.commit()
        assert att2.provider_snapshot_id == "tool-A"

    async def test_update_rejects_same_tool_different_flow_version(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
        folder: Folder,
        provider_account: DeploymentProviderAccount,
    ):
        # FV1 owns tool-A
        await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id="tool-A",
        )
        await db.commit()

        # FV2 has no tool yet
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        db.add(fv2)
        d2 = Deployment(
            user_id=user.id,
            project_id=folder.id,
            deployment_provider_account_id=provider_account.id,
            resource_key="rk-2",
            name="deploy-2",
            deployment_type=DeploymentType.AGENT,
        )
        db.add(d2)
        await db.commit()
        await db.refresh(fv2)
        await db.refresh(d2)

        att2 = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=fv2.id,
            deployment_id=d2.id,
        )
        await db.commit()

        # Should reject: updating FV2's attachment to use tool-A (owned by FV1)
        with pytest.raises(SnapshotFlowVersionConflictError, match="already attached to a different flow version"):
            await update_deployment_attachment_provider_snapshot_id(
                db,
                attachment=att2,
                provider_snapshot_id="tool-A",
            )

    async def test_create_auto_generates_snapshot_ids_when_omitted(
        self,
        db: AsyncSession,
        user: User,
        flow: Flow,
        flow_version: FlowVersion,
        deployment: Deployment,
    ):
        """Helper wrapper should synthesize non-empty snapshot IDs when omitted."""
        fv2 = FlowVersion(flow_id=flow.id, user_id=user.id, version_number=2, data={})
        db.add(fv2)
        await db.commit()
        await db.refresh(fv2)

        att1 = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        await db.commit()
        att2 = await create_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=fv2.id,
            deployment_id=deployment.id,
        )
        await db.commit()

        assert att1.provider_snapshot_id == _default_snapshot_id(
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert att2.provider_snapshot_id == _default_snapshot_id(
            flow_version_id=fv2.id,
            deployment_id=deployment.id,
        )


@pytest.mark.asyncio
class TestCascadeDeletes:
    async def test_cascade_on_deployment_delete(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()
        att_id = att.id

        await db.delete(deployment)
        await db.commit()

        stmt = select(FlowVersionDeploymentAttachment).where(FlowVersionDeploymentAttachment.id == att_id)
        assert (await db.exec(stmt)).first() is None

    async def test_cascade_on_flow_version_delete(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()
        att_id = att.id

        await db.delete(flow_version)
        await db.commit()

        stmt = select(FlowVersionDeploymentAttachment).where(FlowVersionDeploymentAttachment.id == att_id)
        assert (await db.exec(stmt)).first() is None

    async def test_cascade_on_user_delete(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        att = await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()
        att_id = att.id

        await db.delete(user)
        await db.commit()

        stmt = select(FlowVersionDeploymentAttachment).where(FlowVersionDeploymentAttachment.id == att_id)
        assert (await db.exec(stmt)).first() is None


@pytest.mark.asyncio
class TestUserScoping:
    async def test_other_user_cannot_list(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()

        other_user = User(username="other", password=_TEST_PASSWORD, is_active=True)
        db.add(other_user)
        await db.commit()
        await db.refresh(other_user)

        results = await list_deployment_attachments(db, user_id=other_user.id, deployment_id=deployment.id)
        assert results == []

    async def test_other_user_cannot_get(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()

        other_user = User(username="other2", password=_TEST_PASSWORD, is_active=True)
        db.add(other_user)
        await db.commit()
        await db.refresh(other_user)

        result = await get_deployment_attachment(
            db,
            user_id=other_user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert result is None

    async def test_other_user_cannot_delete(
        self, db: AsyncSession, user: User, flow_version: FlowVersion, deployment: Deployment
    ):
        await create_deployment_attachment(
            db, user_id=user.id, flow_version_id=flow_version.id, deployment_id=deployment.id
        )
        await db.commit()

        other_user = User(username="other3", password=_TEST_PASSWORD, is_active=True)
        db.add(other_user)
        await db.commit()
        await db.refresh(other_user)

        count = await delete_deployment_attachment(
            db,
            user_id=other_user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert count == 0

        original = await get_deployment_attachment(
            db,
            user_id=user.id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
        )
        assert original is not None
