test_deployment_schemas.py

"""Tests for deployment API schemas.

Security invariants and validation behaviour that must not regress.
"""

from uuid import uuid4

import pytest
from langflow.api.v1.schemas.deployments import (
    DEPLOYMENT_DESCRIPTION_MAX_LENGTH,
    DeploymentConfigListResponse,
    DeploymentCreateRequest,
    DeploymentFlowVersionListItem,
    DeploymentFlowVersionListResponse,
    DeploymentListItem,
    DeploymentListResponse,
    DeploymentProviderAccountCreateRequest,
    DeploymentProviderAccountGetResponse,
    DeploymentProviderAccountUpdateRequest,
    DeploymentUpdateRequest,
    FlowIdsQuery,
)
from langflow.services.database.models.deployment_provider_account.schemas import DeploymentProviderKey
from pydantic import ValidationError

TEST_API_KEY = "key"  # pragma: allowlist secret

# ---------------------------------------------------------------------------
# Security: credentials must never appear in response schemas
# ---------------------------------------------------------------------------


class TestCredentialSecurity:
    """Ensure credentials are excluded from every response model."""

    def test_provider_account_response_excludes_api_key(self):
        """DeploymentProviderAccountGetResponse.model_fields must not contain api_key."""
        assert "api_key" not in DeploymentProviderAccountGetResponse.model_fields

    def test_provider_account_response_includes_provider_data(self):
        """DeploymentProviderAccountGetResponse includes non-sensitive provider metadata."""
        assert "provider_data" in DeploymentProviderAccountGetResponse.model_fields

    def test_provider_account_response_dump_excludes_credentials(self):
        """model_dump() on a response instance must never contain credential fields."""
        response = DeploymentProviderAccountGetResponse(
            id=uuid4(),
            name="staging",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_data={"url": "https://api.us-south.wxo.cloud.ibm.com", "tenant_id": "tenant-1"},
        )
        dumped = response.model_dump()
        assert "api_key" not in dumped
        assert dumped["provider_data"] == {"url": "https://api.us-south.wxo.cloud.ibm.com", "tenant_id": "tenant-1"}
        assert "api_key" not in (dumped["provider_data"] or {})


# ---------------------------------------------------------------------------
# NonEmptyStr validation
# ---------------------------------------------------------------------------


class TestProviderAccountName:
    def test_create_accepts_valid_name(self):
        account = DeploymentProviderAccountCreateRequest(
            name="production",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_data={"url": "https://api.us-south.wxo.cloud.ibm.com", "api_key": TEST_API_KEY},
        )
        assert account.name == "production"

    def test_create_strips_name_whitespace(self):
        account = DeploymentProviderAccountCreateRequest(
            name="  staging  ",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_data={"url": "https://api.us-south.wxo.cloud.ibm.com", "api_key": TEST_API_KEY},
        )
        assert account.name == "staging"

    def test_create_rejects_empty_name(self):
        with pytest.raises(ValidationError, match="name"):
            DeploymentProviderAccountCreateRequest(
                name="",
                provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
                provider_data={"url": "https://example.com", "api_key": TEST_API_KEY},
            )

    def test_create_rejects_whitespace_only_name(self):
        with pytest.raises(ValidationError, match="name"):
            DeploymentProviderAccountCreateRequest(
                name="   ",
                provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
                provider_data={"url": "https://example.com", "api_key": TEST_API_KEY},
            )

    def test_create_rejects_missing_name(self):
        with pytest.raises(ValidationError):
            DeploymentProviderAccountCreateRequest(
                provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
                provider_data={"url": "https://example.com", "api_key": TEST_API_KEY},
            )

    def test_update_accepts_name(self):
        update = DeploymentProviderAccountUpdateRequest(name="new-name")
        assert update.name == "new-name"

    def test_update_rejects_null_name(self):
        with pytest.raises(ValidationError, match=r"name.*cannot be set to null"):
            DeploymentProviderAccountUpdateRequest(name=None)

    def test_response_includes_name(self):
        assert "name" in DeploymentProviderAccountGetResponse.model_fields


# ---------------------------------------------------------------------------
# Provider-data contract boundary
# ---------------------------------------------------------------------------


class TestProviderAccountProviderDataBoundary:
    """Provider-specific fields belong under provider_data at the API boundary."""

    def test_create_accepts_provider_data_url(self):
        account = DeploymentProviderAccountCreateRequest(
            name="staging",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_data={
                "url": "https://api.us-south.wxo.cloud.ibm.com/v1",
                "api_key": TEST_API_KEY,
            },
        )
        assert account.provider_data["url"] == "https://api.us-south.wxo.cloud.ibm.com/v1"

    def test_update_rejects_url_field(self):
        with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
            DeploymentProviderAccountUpdateRequest(url="https://new.example.com/api")

    def test_update_rejects_tenant_id_field(self):
        with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
            DeploymentProviderAccountUpdateRequest(tenant_id="tenant-1")

    def test_create_rejects_top_level_tenant_id_field(self):
        with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
            DeploymentProviderAccountCreateRequest(
                name="staging",
                tenant_id="tenant-1",
                provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
                provider_data={"url": "https://api.us-south.wxo.cloud.ibm.com/v1", "api_key": TEST_API_KEY},
            )

    def test_create_rejects_top_level_url_field(self):
        with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
            DeploymentProviderAccountCreateRequest(
                name="staging",
                provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
                url="https://api.us-south.wxo.cloud.ibm.com/v1",
                provider_data={"api_key": TEST_API_KEY},
            )


class TestProviderKeyEnum:
    def test_accepts_valid_enum_value(self):
        account = DeploymentProviderAccountCreateRequest(
            name="staging",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_data={"url": "https://api.us-south.wxo.cloud.ibm.com", "api_key": TEST_API_KEY},
        )
        assert account.provider_key == DeploymentProviderKey.WATSONX_ORCHESTRATE

    def test_accepts_valid_string_value(self):
        account = DeploymentProviderAccountCreateRequest(
            name="staging",
            provider_key="watsonx-orchestrate",
            provider_data={"url": "https://api.us-south.wxo.cloud.ibm.com", "api_key": TEST_API_KEY},
        )
        assert account.provider_key == DeploymentProviderKey.WATSONX_ORCHESTRATE

    def test_rejects_invalid_provider_key(self):
        with pytest.raises(ValidationError):
            DeploymentProviderAccountCreateRequest(
                name="staging",
                provider_key="unknown-provider",
                provider_data={"url": "https://example.com", "api_key": TEST_API_KEY},
            )

    def test_rejects_empty_string(self):
        with pytest.raises(ValidationError):
            DeploymentProviderAccountCreateRequest(
                name="staging",
                provider_key="",
                provider_data={"url": "https://example.com", "api_key": TEST_API_KEY},
            )


class TestDeploymentUpdateRequest:
    def test_accepts_provider_data_only(self):
        payload = DeploymentUpdateRequest(provider_data={"mode": "dry_run"})
        assert payload.provider_data == {"mode": "dry_run"}

    def test_rejects_empty_payload(self):
        with pytest.raises(ValidationError, match="At least one of"):
            DeploymentUpdateRequest()

    def test_rejects_explicit_null_only_payload(self):
        with pytest.raises(ValidationError, match="At least one of"):
            DeploymentUpdateRequest(description=None)

    def test_rejects_description_over_max_length(self):
        with pytest.raises(ValidationError, match="at most"):
            DeploymentUpdateRequest(description="x" * (DEPLOYMENT_DESCRIPTION_MAX_LENGTH + 1))


class TestDeploymentSpecPayloadCompatibility:
    def test_create_request_rejects_provider_spec_dict(self):
        with pytest.raises(ValidationError, match="provider_spec"):
            DeploymentCreateRequest(
                provider_id=uuid4(),
                name="deployment",
                description="",
                type="agent",
                provider_spec={"region": "us-east-1", "size": "small"},
            )

    def test_create_request_accepts_provider_data_payload(self):
        request = DeploymentCreateRequest(
            provider_id=uuid4(),
            name="deployment",
            description="",
            type="agent",
            provider_data={"operations": []},
        )
        assert request.provider_data == {"operations": []}

    def test_create_request_rejects_description_over_max_length(self):
        with pytest.raises(ValidationError, match="at most"):
            DeploymentCreateRequest(
                provider_id=uuid4(),
                name="deployment",
                description="x" * (DEPLOYMENT_DESCRIPTION_MAX_LENGTH + 1),
                type="agent",
                provider_data={"operations": []},
            )


# ---------------------------------------------------------------------------
# DeploymentConfigListResponse / DeploymentSnapshotListResponse
# ---------------------------------------------------------------------------


class TestDeploymentConfigListResponse:
    def test_provider_data_contains_connections(self):
        response = DeploymentConfigListResponse(
            provider_data={
                "connections": [
                    {"id": "cfg_1", "name": "Config 1"},
                    {"id": "cfg_2", "name": "Config 2"},
                ],
                "scope": "shared",
            },
            page=1,
            size=20,
            total=2,
        )
        assert len(response.provider_data["connections"]) == 2
        assert response.provider_data["scope"] == "shared"
        assert response.page == 1
        assert response.total == 2

    def test_allows_null_provider_data(self):
        response = DeploymentConfigListResponse()
        assert response.provider_data is None
        assert response.page is None
        assert response.size is None
        assert response.total is None

    def test_has_provider_data_and_pagination_fields_only(self):
        assert set(DeploymentConfigListResponse.model_fields.keys()) == {
            "provider_data",
            "page",
            "size",
            "total",
        }


class TestDeploymentSnapshotListResponse:
    def test_provider_data_contains_tools(self):
        from langflow.api.v1.schemas.deployments import DeploymentSnapshotListResponse

        response = DeploymentSnapshotListResponse(
            provider_data={
                "tools": [
                    {"id": "tool-1", "name": "Tool 1"},
                    {"id": "tool-2", "name": "Tool 2"},
                ],
                "scope": "shared",
            },
            page=1,
            size=20,
            total=2,
        )
        assert len(response.provider_data["tools"]) == 2
        assert response.page == 1

    def test_allows_null_provider_data(self):
        from langflow.api.v1.schemas.deployments import DeploymentSnapshotListResponse

        response = DeploymentSnapshotListResponse()
        assert response.provider_data is None

    def test_has_provider_data_and_pagination_fields_only(self):
        from langflow.api.v1.schemas.deployments import DeploymentSnapshotListResponse

        assert set(DeploymentSnapshotListResponse.model_fields.keys()) == {
            "provider_data",
            "page",
            "size",
            "total",
        }


class TestDeploymentFlowVersionListSchemas:
    def test_flow_version_list_item_uses_attached_at(self):
        from datetime import datetime, timezone

        now = datetime.now(tz=timezone.utc)
        item = DeploymentFlowVersionListItem(
            id=uuid4(),
            flow_id=uuid4(),
            version_number=3,
            attached_at=now,
            provider_snapshot_id="tool-1",
            provider_data={"app_ids": ["cfg-1"]},
        )
        assert item.attached_at == now
        assert item.provider_snapshot_id == "tool-1"
        assert item.provider_data == {"app_ids": ["cfg-1"]}

    def test_flow_version_list_item_does_not_expose_description_or_created_at(self):
        assert "description" not in DeploymentFlowVersionListItem.model_fields
        assert "created_at" not in DeploymentFlowVersionListItem.model_fields

    def test_flow_version_list_response_wraps_items_with_pagination(self):
        response = DeploymentFlowVersionListResponse(
            flow_versions=[
                DeploymentFlowVersionListItem(
                    id=uuid4(),
                    flow_id=uuid4(),
                    version_number=1,
                    attached_at=None,
                    provider_snapshot_id=None,
                    provider_data=None,
                )
            ],
            page=2,
            size=5,
            total=9,
        )
        assert len(response.flow_versions) == 1
        assert response.page == 2
        assert response.size == 5
        assert response.total == 9


class TestDeploymentListResponse:
    def test_allows_provider_only_shape(self):
        response = DeploymentListResponse(provider_data={"deployments": []})
        assert response.deployments is None
        assert response.page is None
        assert response.size is None
        assert response.total is None
        assert response.provider_data == {"deployments": []}


# ---------------------------------------------------------------------------
# FlowIdsQuery validation
# ---------------------------------------------------------------------------


class TestFlowIdsQueryValidation:
    """Validate the FlowIdsQuery annotated type used for the list_deployments filter."""

    def test_none_passes_through(self):
        from pydantic import TypeAdapter

        adapter = TypeAdapter(FlowIdsQuery)
        assert adapter.validate_python(None) is None

    def test_single_valid_uuid(self):
        from pydantic import TypeAdapter

        uid = uuid4()
        adapter = TypeAdapter(FlowIdsQuery)
        result = adapter.validate_python([uid])
        assert result == [uid]

    def test_accepts_string_uuid(self):
        from pydantic import TypeAdapter

        uid = uuid4()
        adapter = TypeAdapter(FlowIdsQuery)
        result = adapter.validate_python([str(uid)])
        assert result == [uid]

    def test_rejects_more_than_one(self):
        from pydantic import TypeAdapter, ValidationError

        adapter = TypeAdapter(FlowIdsQuery)
        with pytest.raises(ValidationError, match="at most 1"):
            adapter.validate_python([uuid4(), uuid4()])

    def test_rejects_invalid_uuid(self):
        from pydantic import TypeAdapter, ValidationError

        adapter = TypeAdapter(FlowIdsQuery)
        with pytest.raises(ValidationError):
            adapter.validate_python(["not-a-uuid"])

    def test_rejects_empty_list(self):
        from pydantic import TypeAdapter, ValidationError

        adapter = TypeAdapter(FlowIdsQuery)
        with pytest.raises(ValidationError, match="flow_ids"):
            adapter.validate_python([])

    def test_deduplicates(self):
        from pydantic import TypeAdapter

        uid = uuid4()
        adapter = TypeAdapter(FlowIdsQuery)
        result = adapter.validate_python([uid, uid])
        assert result == [uid]


# ---------------------------------------------------------------------------
# DeploymentListItem.flow_version_ids
# ---------------------------------------------------------------------------


class TestDeploymentListItemFlowVersionIds:
    def _make_item(self, **kwargs):
        defaults = {
            "id": uuid4(),
            "provider_id": uuid4(),
            "provider_key": DeploymentProviderKey.WATSONX_ORCHESTRATE,
            "name": "dep",
            "type": "agent",
            "resource_key": "rk-1",
        }
        defaults.update(kwargs)
        return DeploymentListItem(**defaults)

    def test_defaults_to_none(self):
        item = self._make_item()
        assert item.flow_version_ids is None

    def test_accepts_flow_version_ids_list(self):
        fv_id = uuid4()
        item = self._make_item(
            flow_version_ids=[fv_id],
        )
        assert item.flow_version_ids == [fv_id]

    def test_accepts_empty_list(self):
        item = self._make_item(flow_version_ids=[])
        assert item.flow_version_ids == []


# ---------------------------------------------------------------------------
# DeploymentConfigListResponse — pagination & empty configs
# ---------------------------------------------------------------------------


class TestDeploymentConfigListResponsePagination:
    def test_rejects_page_less_than_one(self):
        with pytest.raises(ValidationError):
            DeploymentConfigListResponse(page=0, size=10, total=0)

    def test_rejects_size_less_than_one(self):
        with pytest.raises(ValidationError):
            DeploymentConfigListResponse(page=1, size=0, total=0)

    def test_rejects_negative_total(self):
        with pytest.raises(ValidationError):
            DeploymentConfigListResponse(page=1, size=10, total=-1)


# ---------------------------------------------------------------------------
# ExecutionCreateRequest — required fields, provider_data validation
# ---------------------------------------------------------------------------


class TestRunCreateRequest:
    def test_rejects_extra_fields(self):
        from langflow.api.v1.schemas.deployments import RunCreateRequest

        with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
            RunCreateRequest(provider_data={"input": "x"}, unknown_field="y")


# ---------------------------------------------------------------------------
# ExecutionCreateResponse — all fields including nullable provider_data
# ---------------------------------------------------------------------------


class TestRunCreateResponse:
    def test_required_deployment_id(self):
        from langflow.api.v1.schemas.deployments import RunCreateResponse

        with pytest.raises(ValidationError, match="deployment_id"):
            RunCreateResponse()


# ---------------------------------------------------------------------------
# SnapshotUpdateRequest — required fields
# ---------------------------------------------------------------------------


class TestSnapshotUpdateRequest:
    def test_requires_flow_version_id(self):
        from langflow.api.v1.schemas.deployments import SnapshotUpdateRequest

        with pytest.raises(ValidationError, match="flow_version_id"):
            SnapshotUpdateRequest()

    def test_rejects_extra_fields(self):
        from langflow.api.v1.schemas.deployments import SnapshotUpdateRequest

        with pytest.raises(ValidationError, match="Extra inputs are not permitted"):
            SnapshotUpdateRequest(flow_version_id=uuid4(), extra_field="bad")


# ---------------------------------------------------------------------------
# SnapshotUpdateResponse — response fields
# ---------------------------------------------------------------------------


class TestSnapshotUpdateResponse:
    def test_requires_flow_version_id(self):
        from langflow.api.v1.schemas.deployments import SnapshotUpdateResponse

        with pytest.raises(ValidationError, match="flow_version_id"):
            SnapshotUpdateResponse(provider_snapshot_id="snap-1")

    def test_requires_provider_snapshot_id(self):
        from langflow.api.v1.schemas.deployments import SnapshotUpdateResponse

        with pytest.raises(ValidationError, match="provider_snapshot_id"):
            SnapshotUpdateResponse(flow_version_id=uuid4())