import tempfile
import uuid
from uuid import UUID

from fastapi import status
from httpx import AsyncClient


async def _attach_deployment_to_flow(*, user_id: UUID, flow_id: UUID, project_id: UUID) -> None:
    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_version.model import FlowVersion
    from langflow.services.database.models.flow_version_deployment_attachment.model import (
        FlowVersionDeploymentAttachment,
    )
    from langflow.services.deps import session_scope
    from lfx.services.adapters.deployment.schema import DeploymentType

    async with session_scope() as session:
        provider = DeploymentProviderAccount(
            user_id=user_id,
            name=f"provider-{flow_id.hex[:8]}",
            provider_tenant_id="tenant-1",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_url=f"https://provider-{flow_id.hex[:8]}.example.com",
            api_key="encrypted-value",  # pragma: allowlist secret
        )
        session.add(provider)
        await session.flush()

        deployment = Deployment(
            user_id=user_id,
            project_id=project_id,
            deployment_provider_account_id=provider.id,
            resource_key=f"rk-{flow_id.hex[:8]}",
            name=f"deployment-{flow_id.hex[:8]}",
            deployment_type=DeploymentType.AGENT,
        )
        session.add(deployment)
        await session.flush()

        flow_version = FlowVersion(
            flow_id=flow_id,
            user_id=user_id,
            version_number=1,
            data={"nodes": [], "edges": []},
        )
        session.add(flow_version)
        await session.flush()

        attachment = FlowVersionDeploymentAttachment(
            user_id=user_id,
            flow_version_id=flow_version.id,
            deployment_id=deployment.id,
            provider_snapshot_id=f"snapshot-{flow_id.hex[:8]}",
        )
        session.add(attachment)
        await session.commit()


async def test_create_flow(client: AsyncClient, logged_in_headers):
    # Use relative path - absolute paths outside allowed directory are rejected
    flow_filename = f"{uuid.uuid4()}.json"
    basic_case = {
        "name": "string",
        "description": "string",
        "icon": "string",
        "icon_bg_color": "#ff00ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "endpoint_name": "string",
        "tags": ["string"],
        "folder_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
        "fs_path": flow_filename,
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    result = response.json()

    assert response.status_code == status.HTTP_201_CREATED
    assert isinstance(result, dict), "The result must be a dictionary"
    assert "data" in result, "The result must have a 'data' key"
    assert "description" in result, "The result must have a 'description' key"
    assert "endpoint_name" in result, "The result must have a 'endpoint_name' key"
    assert "folder_id" in result, "The result must have a 'folder_id' key"
    assert "gradient" in result, "The result must have a 'gradient' key"
    assert "icon" in result, "The result must have a 'icon' key"
    assert "icon_bg_color" in result, "The result must have a 'icon_bg_color' key"
    assert "id" in result, "The result must have a 'id' key"
    assert "is_component" in result, "The result must have a 'is_component' key"
    assert "name" in result, "The result must have a 'name' key"
    assert "tags" in result, "The result must have a 'tags' key"
    assert "updated_at" in result, "The result must have a 'updated_at' key"
    assert "user_id" in result, "The result must have a 'user_id' key"
    assert "webhook" in result, "The result must have a 'webhook' key"


async def test_read_flows(client: AsyncClient, logged_in_headers):
    params = {
        "remove_example_flows": False,
        "components_only": False,
        "get_all": True,
        "header_flows": False,
        "page": 1,
        "size": 50,
    }
    response = await client.get("api/v1/flows/", params=params, headers=logged_in_headers)
    result = response.json()

    assert response.status_code == status.HTTP_200_OK
    assert isinstance(result, list), "The result must be a list"


async def test_get_flows_with_malformed_bearer_token_returns_401(client: AsyncClient):
    """CT-010: GET /api/v1/flows with malformed Bearer token must return 401 Unauthorized."""
    headers = {"Authorization": "Bearer invalid.token.here"}
    response = await client.get("api/v1/flows/", headers=headers)
    assert response.status_code == status.HTTP_401_UNAUTHORIZED
    data = response.json()
    assert "detail" in data
    assert "token" in data["detail"].lower() or "credential" in data["detail"].lower()


async def test_read_flow(client: AsyncClient, logged_in_headers):
    basic_case = {
        "name": "string",
        "description": "string",
        "icon": "string",
        "icon_bg_color": "#ff00ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "endpoint_name": "string",
        "tags": ["string"],
        "folder_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    }
    response_ = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    id_ = response_.json()["id"]
    response = await client.get(f"api/v1/flows/{id_}", headers=logged_in_headers)
    result = response.json()

    assert response.status_code == status.HTTP_200_OK
    assert isinstance(result, dict), "The result must be a dictionary"
    assert "data" in result, "The result must have a 'data' key"
    assert "description" in result, "The result must have a 'description' key"
    assert "endpoint_name" in result, "The result must have a 'endpoint_name' key"
    assert "folder_id" in result, "The result must have a 'folder_id' key"
    assert "gradient" in result, "The result must have a 'gradient' key"
    assert "icon" in result, "The result must have a 'icon' key"
    assert "icon_bg_color" in result, "The result must have a 'icon_bg_color' key"
    assert "id" in result, "The result must have a 'id' key"
    assert "is_component" in result, "The result must have a 'is_component' key"
    assert "name" in result, "The result must have a 'name' key"
    assert "tags" in result, "The result must have a 'tags' key"
    assert "updated_at" in result, "The result must have a 'updated_at' key"
    assert "user_id" in result, "The result must have a 'user_id' key"
    assert "webhook" in result, "The result must have a 'webhook' key"


async def test_update_flow(client: AsyncClient, logged_in_headers):
    name = "first_name"
    updated_name = "second_name"
    basic_case = {
        "description": "string",
        "icon": "string",
        "icon_bg_color": "#ff00ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "endpoint_name": "string",
        "tags": ["string"],
        "folder_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    }
    basic_case["name"] = name
    response_ = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    id_ = response_.json()["id"]

    # Use relative path - absolute paths outside allowed directory are rejected
    flow_filename = f"{uuid.uuid4()!s}.json"
    basic_case["name"] = updated_name
    basic_case["fs_path"] = flow_filename

    response = await client.patch(f"api/v1/flows/{id_}", json=basic_case, headers=logged_in_headers)
    result = response.json()

    assert isinstance(result, dict), "The result must be a dictionary"
    assert "data" in result, "The result must have a 'data' key"
    assert "description" in result, "The result must have a 'description' key"
    assert "endpoint_name" in result, "The result must have a 'endpoint_name' key"
    assert "folder_id" in result, "The result must have a 'folder_id' key"
    assert "gradient" in result, "The result must have a 'gradient' key"
    assert "icon" in result, "The result must have a 'icon' key"
    assert "icon_bg_color" in result, "The result must have a 'icon_bg_color' key"
    assert "id" in result, "The result must have a 'id' key"
    assert "is_component" in result, "The result must have a 'is_component' key"
    assert "name" in result, "The result must have a 'name' key"
    assert "tags" in result, "The result must have a 'tags' key"
    assert "updated_at" in result, "The result must have a 'updated_at' key"
    assert "user_id" in result, "The result must have a 'user_id' key"
    assert "webhook" in result, "The result must have a 'webhook' key"
    assert result["name"] == updated_name, "The name must be updated"


async def test_patch_flow_keeps_existing_endpoint_when_not_provided(client: AsyncClient, logged_in_headers):
    """Test that PATCH preserves endpoint_name when the field is omitted."""
    initial_flow = {
        "name": "patch_endpoint_flow",
        "endpoint_name": "keep_patch_endpoint",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    response = await client.patch(
        f"api/v1/flows/{flow_id}",
        json={"name": "patch_endpoint_flow_updated"},
        headers=logged_in_headers,
    )

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["endpoint_name"] == "keep_patch_endpoint"


async def test_patch_flow_allows_clearing_endpoint_with_null(client: AsyncClient, logged_in_headers):
    """Test that PATCH clears endpoint_name when it is explicitly set to null."""
    initial_flow = {
        "name": "patch_clear_endpoint_flow",
        "endpoint_name": "clear_patch_endpoint",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    response = await client.patch(
        f"api/v1/flows/{flow_id}",
        json={"endpoint_name": None},
        headers=logged_in_headers,
    )

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["endpoint_name"] is None


async def test_patch_flow_updates_access_and_action_fields(client: AsyncClient, logged_in_headers):
    """PATCH should persist public-sharing and MCP action metadata fields."""
    create_response = await client.post(
        "api/v1/flows/",
        json={"name": "patch_access_type_flow", "data": {}},
        headers=logged_in_headers,
    )
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    response = await client.patch(
        f"api/v1/flows/{flow_id}",
        json={
            "access_type": "PUBLIC",
            "action_name": "shared_action",
            "action_description": "Shared flow action",
        },
        headers=logged_in_headers,
    )

    assert response.status_code == status.HTTP_200_OK
    result = response.json()
    assert result["access_type"] == "PUBLIC"
    assert result["action_name"] == "shared_action"
    assert result["action_description"] == "Shared flow action"


async def test_create_flows(client: AsyncClient, logged_in_headers):
    amount_flows = 10
    basic_case = {
        "description": "string",
        "icon": "string",
        "icon_bg_color": "#ff00ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "tags": ["string"],
    }
    cases = []
    for i in range(amount_flows):
        case = basic_case.copy()
        case["name"] = f"string_{i}"
        case["endpoint_name"] = f"string_{i}"
        cases.append(case)

    response = await client.post("api/v1/flows/batch/", json={"flows": cases}, headers=logged_in_headers)
    result = response.json()

    assert response.status_code == status.HTTP_201_CREATED
    assert isinstance(result, list), "The result must be a list"
    assert len(result) == amount_flows, "The result must have the same amount of flows"


async def test_create_flows_with_explicit_folder(client: AsyncClient, logged_in_headers):
    project_response = await client.post(
        "api/v1/projects/",
        json={"name": "batch-folder-target", "description": "", "flows_list": [], "components_list": []},
        headers=logged_in_headers,
    )
    assert project_response.status_code == status.HTTP_201_CREATED
    project_id = project_response.json()["id"]

    amount_flows = 3
    basic_case = {
        "description": "string",
        "icon": "string",
        "icon_bg_color": "#ff00ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "tags": ["string"],
        "folder_id": project_id,
    }
    cases = []
    for i in range(amount_flows):
        case = basic_case.copy()
        case["name"] = f"string_folder_{i}"
        case["endpoint_name"] = f"string_folder_{i}"
        cases.append(case)

    response = await client.post("api/v1/flows/batch/", json={"flows": cases}, headers=logged_in_headers)
    result = response.json()

    assert response.status_code == status.HTTP_201_CREATED
    assert isinstance(result, list), "The result must be a list"
    assert len(result) == amount_flows, "The result must have the same amount of flows"
    assert all(item["folder_id"] == project_id for item in result), "All flows must be created in the target folder"


async def test_read_basic_examples(client: AsyncClient, logged_in_headers):
    response = await client.get("api/v1/flows/basic_examples/", headers=logged_in_headers)
    result = response.json()

    assert response.status_code == status.HTTP_200_OK
    assert isinstance(result, list), "The result must be a list"
    assert len(result) > 0, "The result must have at least one flow"


async def test_read_flows_user_isolation(client: AsyncClient, logged_in_headers, active_user):
    """Test that read_flows returns only flows from the current user."""
    from uuid import uuid4

    from langflow.services.auth.utils import get_password_hash
    from langflow.services.database.models.user.model import User
    from langflow.services.deps import session_scope

    # Create a second user
    other_user_id = uuid4()
    async with session_scope() as session:
        other_user = User(
            id=other_user_id,
            username="other_test_user",
            password=get_password_hash("testpassword"),
            is_active=True,
            is_superuser=False,
        )
        session.add(other_user)
        await session.commit()
        await session.refresh(other_user)

    # Login as the other user to get headers
    login_data = {"username": "other_test_user", "password": "testpassword"}  # pragma: allowlist secret
    response = await client.post("api/v1/login", data=login_data)
    assert response.status_code == 200
    tokens = response.json()
    other_user_headers = {"Authorization": f"Bearer {tokens['access_token']}"}

    # Create flows for the first user (active_user)
    flow_user1_1 = {
        "name": "user1_flow_1",
        "description": "Flow 1 for user 1",
        "icon": "string",
        "icon_bg_color": "#ff00ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "endpoint_name": "user1_flow_1_endpoint",
        "tags": ["user1"],
        "folder_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    }

    flow_user1_2 = {
        "name": "user1_flow_2",
        "description": "Flow 2 for user 1",
        "icon": "string",
        "icon_bg_color": "#00ff00",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "endpoint_name": "user1_flow_2_endpoint",
        "tags": ["user1"],
        "folder_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    }

    # Create flows for the second user
    flow_user2_1 = {
        "name": "user2_flow_1",
        "description": "Flow 1 for user 2",
        "icon": "string",
        "icon_bg_color": "#0000ff",
        "gradient": "string",
        "data": {},
        "is_component": False,
        "webhook": False,
        "endpoint_name": "user2_flow_1_endpoint",
        "tags": ["user2"],
        "folder_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    }

    # Create flows using the appropriate user headers
    response1 = await client.post("api/v1/flows/", json=flow_user1_1, headers=logged_in_headers)
    assert response1.status_code == status.HTTP_201_CREATED

    response2 = await client.post("api/v1/flows/", json=flow_user1_2, headers=logged_in_headers)
    assert response2.status_code == status.HTTP_201_CREATED

    response3 = await client.post("api/v1/flows/", json=flow_user2_1, headers=other_user_headers)
    assert response3.status_code == status.HTTP_201_CREATED

    # Test read_flows for user 1 - should only return user 1's flows
    params = {
        "remove_example_flows": True,  # Exclude example flows to focus on our test flows
        "components_only": False,
        "get_all": True,
        "header_flows": False,
        "page": 1,
        "size": 50,
    }

    response_user1 = await client.get("api/v1/flows/", params=params, headers=logged_in_headers)
    result_user1 = response_user1.json()

    assert response_user1.status_code == status.HTTP_200_OK
    assert isinstance(result_user1, list), "The result must be a list"

    # Verify only user 1's flows are returned
    user1_flow_names = [flow["name"] for flow in result_user1]
    assert "user1_flow_1" in user1_flow_names, "User 1's first flow should be returned"
    assert "user1_flow_2" in user1_flow_names, "User 1's second flow should be returned"
    assert "user2_flow_1" not in user1_flow_names, "User 2's flow should not be returned for user 1"

    # Verify all returned flows belong to user 1
    for flow in result_user1:
        assert str(flow["user_id"]) == str(active_user.id), f"Flow {flow['name']} should belong to user 1"

    # Test read_flows for user 2 - should only return user 2's flows
    response_user2 = await client.get("api/v1/flows/", params=params, headers=other_user_headers)
    result_user2 = response_user2.json()

    assert response_user2.status_code == status.HTTP_200_OK
    assert isinstance(result_user2, list), "The result must be a list"

    # Verify only user 2's flows are returned
    user2_flow_names = [flow["name"] for flow in result_user2]
    assert "user2_flow_1" in user2_flow_names, "User 2's flow should be returned"
    assert "user1_flow_1" not in user2_flow_names, "User 1's first flow should not be returned for user 2"
    assert "user1_flow_2" not in user2_flow_names, "User 1's second flow should not be returned for user 2"

    # Verify all returned flows belong to user 2
    for flow in result_user2:
        assert str(flow["user_id"]) == str(other_user_id), f"Flow {flow['name']} should belong to user 2"

    # Cleanup: Delete the other user
    async with session_scope() as session:
        user = await session.get(User, other_user_id)
        if user:
            await session.delete(user)
            await session.commit()


async def test_create_flow_rejects_absolute_path_outside_allowed_directory(client: AsyncClient, logged_in_headers):
    """Test that absolute paths outside the allowed directory are rejected."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "/etc/passwd",  # Absolute path outside allowed directory should be rejected
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert "within" in response.json()["detail"].lower() or "outside" in response.json()["detail"].lower()


async def test_create_flow_rejects_directory_traversal(client: AsyncClient, logged_in_headers):
    """Test that directory traversal sequences are rejected."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "../../etc/passwd",  # Directory traversal should be rejected
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert (
        "directory traversal" in response.json()["detail"].lower()
        or "absolute paths" in response.json()["detail"].lower()
    )


async def test_create_flow_rejects_null_bytes(client: AsyncClient, logged_in_headers):
    """Test that null bytes in paths are rejected."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "file\x00name.json",  # Null byte should be rejected
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert "absolute paths" in response.json()["detail"].lower() or "null" in response.json()["detail"].lower()


async def test_create_flow_rejects_windows_absolute_path_outside_allowed_directory(
    client: AsyncClient, logged_in_headers
):
    """Test that Windows-style absolute paths outside the allowed directory are rejected."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "C:\\Windows\\System32\\config\\sam",  # Windows absolute path outside
        # allowed directory should be rejected
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert "within" in response.json()["detail"].lower() or "outside" in response.json()["detail"].lower()


async def test_create_flow_accepts_relative_path(client: AsyncClient, logged_in_headers):
    """Test that valid relative paths are accepted."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "my_flow.json",  # Valid relative path
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_201_CREATED


async def test_create_flow_accepts_nested_relative_path(client: AsyncClient, logged_in_headers):
    """Test that nested relative paths are accepted."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "subfolder/my_flow.json",  # Valid nested relative path
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_201_CREATED


async def test_update_flow_rejects_absolute_path_outside_allowed_directory(client: AsyncClient, logged_in_headers):
    """Test that updating a flow with an absolute path outside allowed directory is rejected."""
    # First create a flow
    basic_case = {
        "name": "test_flow",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    # Try to update with absolute path outside allowed directory
    with tempfile.NamedTemporaryFile(delete=True) as temp_file:
        update_case = {
            "fs_path": temp_file.name,
        }
    update_response = await client.patch(f"api/v1/flows/{flow_id}", json=update_case, headers=logged_in_headers)
    assert update_response.status_code == status.HTTP_400_BAD_REQUEST
    assert "within" in update_response.json()["detail"].lower() or "outside" in update_response.json()["detail"].lower()


async def test_update_flow_accepts_relative_path(client: AsyncClient, logged_in_headers):
    """Test that updating a flow with a relative path is accepted."""
    # First create a flow
    basic_case = {
        "name": "test_flow",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    # Update with valid relative path
    update_case = {
        "fs_path": "updated_flow.json",
    }
    update_response = await client.patch(f"api/v1/flows/{flow_id}", json=update_case, headers=logged_in_headers)
    assert update_response.status_code == status.HTTP_200_OK


async def test_create_flow_rejects_empty_path(client: AsyncClient, logged_in_headers):
    """Test that empty fs_path is handled correctly (should be allowed as None).

    But empty string should fail validation.
    """
    # Empty string should fail validation if fs_path validation is called
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "",  # Empty string
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    # Empty string should be rejected by validation
    assert response.status_code == status.HTTP_400_BAD_REQUEST


async def test_create_flow_allows_none_path(client: AsyncClient, logged_in_headers):
    """Test that None/null fs_path is allowed (no file saving)."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        # fs_path not provided (None)
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_201_CREATED


async def test_create_flow_rejects_multiple_traversal(client: AsyncClient, logged_in_headers):
    """Test that multiple directory traversal sequences are rejected."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "../../../etc/passwd",  # Multiple traversals
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_400_BAD_REQUEST


async def test_create_flow_rejects_traversal_in_subpath(client: AsyncClient, logged_in_headers):
    """Test that directory traversal in subpaths is rejected."""
    basic_case = {
        "name": "test_flow",
        "data": {},
        "fs_path": "subfolder/../../etc/passwd",  # Traversal in subpath
    }
    response = await client.post("api/v1/flows/", json=basic_case, headers=logged_in_headers)
    assert response.status_code == status.HTTP_400_BAD_REQUEST


async def test_upload_flow_rejects_absolute_path(client: AsyncClient, logged_in_headers):
    """Test that uploading flows with absolute paths is rejected."""
    import json

    with tempfile.NamedTemporaryFile(delete=True) as temp_file:
        flow_data = {
            "name": "test_flow",
            "data": {},
            "fs_path": temp_file.name,  # Absolute path
        }
    # Create a JSON file content
    file_content = json.dumps({"flows": [flow_data]})

    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("flows.json", file_content, "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == status.HTTP_400_BAD_REQUEST


# PUT endpoint tests (upsert)


async def test_upsert_flow_creates_new_flow_with_specified_id(client: AsyncClient, logged_in_headers):
    """Test that PUT creates a new flow with the specified ID and returns 201."""
    specified_id = str(uuid.uuid4())
    flow_data = {
        "name": "upsert_new_flow",
        "description": "Created via upsert",
        "data": {},
    }

    response = await client.put(f"api/v1/flows/{specified_id}", json=flow_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_201_CREATED
    result = response.json()
    assert result["id"] == specified_id
    assert result["name"] == "upsert_new_flow"


async def test_upsert_flow_updates_existing_flow(client: AsyncClient, logged_in_headers):
    """Test that PUT updates an existing flow and returns 200."""
    # First create a flow via POST
    initial_flow = {
        "name": "initial_flow_name",
        "description": "initial description",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    # Now update via PUT
    updated_flow = {
        "name": "updated_flow_name",
        "description": "updated description",
        "data": {"nodes": [], "edges": []},
    }
    response = await client.put(f"api/v1/flows/{flow_id}", json=updated_flow, headers=logged_in_headers)

    assert response.status_code == status.HTTP_200_OK
    result = response.json()
    assert result["id"] == flow_id
    assert result["name"] == "updated_flow_name"
    assert result["description"] == "updated description"


async def test_upsert_flow_returns_404_for_other_users_flow(client: AsyncClient, logged_in_headers):
    """Test that PUT returns 404 when trying to upsert another user's flow (avoids leaking existence)."""
    from langflow.services.auth.utils import get_password_hash
    from langflow.services.database.models.user.model import User
    from langflow.services.deps import session_scope

    # Create another user
    other_user_id = uuid.uuid4()
    async with session_scope() as session:
        other_user = User(
            id=other_user_id,
            username="other_user_for_upsert_test",
            password=get_password_hash("testpassword"),
            is_active=True,
            is_superuser=False,
        )
        session.add(other_user)
        await session.commit()

    # Login as other user and create a flow
    login_data = {"username": "other_user_for_upsert_test", "password": "testpassword"}  # pragma: allowlist secret
    login_response = await client.post("api/v1/login", data=login_data)
    assert login_response.status_code == status.HTTP_200_OK, f"Login failed: {login_response.text}"
    other_user_headers = {"Authorization": f"Bearer {login_response.json()['access_token']}"}

    flow_data = {"name": "other_user_flow", "data": {}}
    create_response = await client.post("api/v1/flows/", json=flow_data, headers=other_user_headers)
    other_user_flow_id = create_response.json()["id"]

    # Try to upsert other user's flow with original user's credentials
    update_data = {"name": "trying_to_steal", "data": {}}
    response = await client.put(f"api/v1/flows/{other_user_flow_id}", json=update_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_404_NOT_FOUND
    assert "not found" in response.json()["detail"].lower()

    # Cleanup
    async with session_scope() as session:
        user = await session.get(User, other_user_id)
        if user:
            await session.delete(user)
            await session.commit()


async def test_upsert_flow_returns_400_for_invalid_folder_id(client: AsyncClient, logged_in_headers):
    """Test that PUT returns 400 when folder_id doesn't exist."""
    specified_id = str(uuid.uuid4())
    non_existent_folder_id = str(uuid.uuid4())
    flow_data = {
        "name": "flow_with_bad_folder",
        "data": {},
        "folder_id": non_existent_folder_id,
    }

    response = await client.put(f"api/v1/flows/{specified_id}", json=flow_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_400_BAD_REQUEST
    assert "folder not found" in response.json()["detail"].lower()


async def test_upsert_flow_returns_409_for_endpoint_name_conflict_on_create(client: AsyncClient, logged_in_headers):
    """Test that PUT returns 409 when endpoint_name conflicts during CREATE."""
    # First create a flow with a specific endpoint_name
    first_flow = {
        "name": "first_flow",
        "endpoint_name": "unique_endpoint",
        "data": {},
    }
    await client.post("api/v1/flows/", json=first_flow, headers=logged_in_headers)

    # Try to create new flow via PUT with same endpoint_name
    specified_id = str(uuid.uuid4())
    second_flow = {
        "name": "second_flow",
        "endpoint_name": "unique_endpoint",  # Same endpoint_name
        "data": {},
    }

    response = await client.put(f"api/v1/flows/{specified_id}", json=second_flow, headers=logged_in_headers)

    assert response.status_code == status.HTTP_409_CONFLICT
    assert "endpoint" in response.json()["detail"].lower()


async def test_upsert_flow_auto_renames_name_on_create_conflict(client: AsyncClient, logged_in_headers):
    """Test that PUT auto-renames name when it conflicts during CREATE."""
    # First create a flow with a specific name
    first_flow = {
        "name": "duplicate_name",
        "data": {},
    }
    await client.post("api/v1/flows/", json=first_flow, headers=logged_in_headers)

    # Create new flow via PUT with same name - should auto-rename
    specified_id = str(uuid.uuid4())
    second_flow = {
        "name": "duplicate_name",  # Same name
        "data": {},
    }

    response = await client.put(f"api/v1/flows/{specified_id}", json=second_flow, headers=logged_in_headers)

    assert response.status_code == status.HTTP_201_CREATED
    result = response.json()
    assert result["name"] == "duplicate_name (1)"  # Auto-renamed


async def test_upsert_flow_returns_409_for_name_conflict_on_update(client: AsyncClient, logged_in_headers):
    """Test that PUT returns 409 when name conflicts with another flow during UPDATE."""
    # Create two flows
    first_flow = {"name": "flow_one", "data": {}}
    second_flow = {"name": "flow_two", "data": {}}

    await client.post("api/v1/flows/", json=first_flow, headers=logged_in_headers)
    second_response = await client.post("api/v1/flows/", json=second_flow, headers=logged_in_headers)
    second_flow_id = second_response.json()["id"]

    # Try to update second flow to have first flow's name
    update_data = {"name": "flow_one", "data": {}}  # Conflict with first flow

    response = await client.put(f"api/v1/flows/{second_flow_id}", json=update_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_409_CONFLICT
    assert "name" in response.json()["detail"].lower()


async def test_upsert_flow_returns_409_for_endpoint_conflict_on_update(client: AsyncClient, logged_in_headers):
    """Test that PUT returns 409 when endpoint_name conflicts with another flow during UPDATE."""
    # Create two flows with different endpoint names
    first_flow = {"name": "endpoint_flow_one", "endpoint_name": "endpoint_one", "data": {}}
    second_flow = {"name": "endpoint_flow_two", "endpoint_name": "endpoint_two", "data": {}}

    await client.post("api/v1/flows/", json=first_flow, headers=logged_in_headers)
    second_response = await client.post("api/v1/flows/", json=second_flow, headers=logged_in_headers)
    second_flow_id = second_response.json()["id"]

    # Try to update second flow to have first flow's endpoint_name
    update_data = {"name": "endpoint_flow_two", "endpoint_name": "endpoint_one", "data": {}}

    response = await client.put(f"api/v1/flows/{second_flow_id}", json=update_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_409_CONFLICT
    assert "endpoint" in response.json()["detail"].lower()


async def test_upsert_flow_keeps_existing_folder_on_update_when_not_provided(client: AsyncClient, logged_in_headers):
    """Test that PUT keeps existing folder_id when not provided during UPDATE."""
    # Create a flow (will be assigned to default folder)
    initial_flow = {"name": "folder_test_flow", "data": {}}
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    flow_id = create_response.json()["id"]
    original_folder_id = create_response.json()["folder_id"]

    # Update via PUT without providing folder_id
    update_data = {"name": "folder_test_flow_updated", "data": {}}

    response = await client.put(f"api/v1/flows/{flow_id}", json=update_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_200_OK
    result = response.json()
    assert result["folder_id"] == original_folder_id  # Folder unchanged


async def test_upsert_flow_keeps_existing_endpoint_when_not_provided(client: AsyncClient, logged_in_headers):
    """Test that PUT preserves endpoint_name when it is omitted during update."""
    initial_flow = {
        "name": "upsert_endpoint_flow",
        "endpoint_name": "keep_upsert_endpoint",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    response = await client.put(
        f"api/v1/flows/{flow_id}",
        json={"name": "upsert_endpoint_flow_updated", "data": {}},
        headers=logged_in_headers,
    )

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["endpoint_name"] == "keep_upsert_endpoint"


async def test_upsert_flow_allows_clearing_endpoint_with_null(client: AsyncClient, logged_in_headers):
    """Test that PUT clears endpoint_name when it is explicitly set to null."""
    initial_flow = {
        "name": "upsert_clear_endpoint_flow",
        "endpoint_name": "clear_upsert_endpoint",
        "data": {},
    }
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    assert create_response.status_code == status.HTTP_201_CREATED
    flow_id = create_response.json()["id"]

    response = await client.put(
        f"api/v1/flows/{flow_id}",
        json={"name": "upsert_clear_endpoint_flow", "endpoint_name": None, "data": {}},
        headers=logged_in_headers,
    )

    assert response.status_code == status.HTTP_200_OK
    assert response.json()["endpoint_name"] is None


async def test_upsert_flow_ignores_user_id_from_body(client: AsyncClient, logged_in_headers, active_user):
    """Test that PUT ignores user_id from body and uses current user."""
    specified_id = str(uuid.uuid4())
    fake_user_id = str(uuid.uuid4())
    flow_data = {
        "name": "security_test_flow",
        "data": {},
        "user_id": fake_user_id,  # Should be ignored
    }

    response = await client.put(f"api/v1/flows/{specified_id}", json=flow_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_201_CREATED
    result = response.json()
    assert result["user_id"] == str(active_user.id)  # Should be current user, not fake
    assert result["user_id"] != fake_user_id


async def test_upsert_flow_allows_updating_own_flow_name(client: AsyncClient, logged_in_headers):
    """Test that PUT allows updating a flow to keep the same name (no conflict with itself)."""
    # Create a flow
    initial_flow = {"name": "self_update_flow", "description": "initial", "data": {}}
    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
    flow_id = create_response.json()["id"]

    # Update the flow keeping the same name but changing description
    update_data = {"name": "self_update_flow", "description": "updated", "data": {}}

    response = await client.put(f"api/v1/flows/{flow_id}", json=update_data, headers=logged_in_headers)

    assert response.status_code == status.HTTP_200_OK
    result = response.json()
    assert result["name"] == "self_update_flow"
    assert result["description"] == "updated"


async def test_delete_flow_with_deployed_versions_returns_409(client: AsyncClient, logged_in_headers, active_user):
    flow_resp = await client.post(
        "api/v1/flows/",
        json={"name": "deployed-delete-flow", "data": {"nodes": [], "edges": []}},
        headers=logged_in_headers,
    )
    assert flow_resp.status_code == status.HTTP_201_CREATED
    flow_payload = flow_resp.json()
    flow_id = UUID(flow_payload["id"])
    source_project_id = UUID(flow_payload["folder_id"])

    await _attach_deployment_to_flow(
        user_id=active_user.id,
        flow_id=flow_id,
        project_id=source_project_id,
    )

    delete_resp = await client.delete(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
    assert delete_resp.status_code == status.HTTP_409_CONFLICT
    assert "cannot be deleted because it has deployed versions" in delete_resp.json()["detail"].lower()


async def test_bulk_delete_with_deployed_flow_returns_409(client: AsyncClient, logged_in_headers, active_user):
    deployed_flow_resp = await client.post(
        "api/v1/flows/",
        json={"name": "deployed-bulk-flow", "data": {"nodes": [], "edges": []}},
        headers=logged_in_headers,
    )
    assert deployed_flow_resp.status_code == status.HTTP_201_CREATED
    deployed_payload = deployed_flow_resp.json()
    deployed_flow_id = UUID(deployed_payload["id"])
    source_project_id = UUID(deployed_payload["folder_id"])

    undeployed_flow_resp = await client.post(
        "api/v1/flows/",
        json={"name": "undeployed-bulk-flow", "data": {"nodes": [], "edges": []}},
        headers=logged_in_headers,
    )
    assert undeployed_flow_resp.status_code == status.HTTP_201_CREATED
    undeployed_flow_id = undeployed_flow_resp.json()["id"]

    await _attach_deployment_to_flow(
        user_id=active_user.id,
        flow_id=deployed_flow_id,
        project_id=source_project_id,
    )

    delete_resp = await client.request(
        "DELETE",
        "api/v1/flows/",
        json=[str(deployed_flow_id), undeployed_flow_id],
        headers=logged_in_headers,
    )
    assert delete_resp.status_code == status.HTTP_409_CONFLICT
    assert "cannot be deleted because it has deployed versions" in delete_resp.json()["detail"].lower()


async def test_patch_flow_folder_move_with_deployed_versions_returns_409(
    client: AsyncClient, logged_in_headers, active_user
):
    flow_resp = await client.post(
        "api/v1/flows/",
        json={"name": "deployed-patch-flow", "data": {"nodes": [], "edges": []}},
        headers=logged_in_headers,
    )
    assert flow_resp.status_code == status.HTTP_201_CREATED
    flow_payload = flow_resp.json()
    flow_id = flow_payload["id"]
    source_project_id = UUID(flow_payload["folder_id"])

    project_resp = await client.post(
        "api/v1/projects/",
        json={"name": "patch-target-project", "description": "", "flows_list": [], "components_list": []},
        headers=logged_in_headers,
    )
    assert project_resp.status_code == status.HTTP_201_CREATED
    target_project_id = project_resp.json()["id"]

    await _attach_deployment_to_flow(
        user_id=active_user.id,
        flow_id=UUID(flow_id),
        project_id=source_project_id,
    )

    patch_resp = await client.patch(
        f"api/v1/flows/{flow_id}",
        json={"folder_id": target_project_id},
        headers=logged_in_headers,
    )
    assert patch_resp.status_code == status.HTTP_409_CONFLICT
    assert "cannot be moved to another project" in patch_resp.json()["detail"].lower()


async def test_upsert_flow_folder_move_with_deployed_versions_returns_409(
    client: AsyncClient, logged_in_headers, active_user
):
    flow_resp = await client.post(
        "api/v1/flows/",
        json={"name": "deployed-put-flow", "data": {"nodes": [], "edges": []}},
        headers=logged_in_headers,
    )
    assert flow_resp.status_code == status.HTTP_201_CREATED
    flow_payload = flow_resp.json()
    flow_id = flow_payload["id"]
    source_project_id = UUID(flow_payload["folder_id"])

    project_resp = await client.post(
        "api/v1/projects/",
        json={"name": "put-target-project", "description": "", "flows_list": [], "components_list": []},
        headers=logged_in_headers,
    )
    assert project_resp.status_code == status.HTTP_201_CREATED
    target_project_id = project_resp.json()["id"]

    await _attach_deployment_to_flow(
        user_id=active_user.id,
        flow_id=UUID(flow_id),
        project_id=source_project_id,
    )

    put_resp = await client.put(
        f"api/v1/flows/{flow_id}",
        json={
            "name": "deployed-put-flow-updated",
            "data": {"nodes": [], "edges": []},
            "folder_id": target_project_id,
        },
        headers=logged_in_headers,
    )
    assert put_resp.status_code == status.HTTP_409_CONFLICT
    assert "cannot be moved to another project" in put_resp.json()["detail"].lower()
