import io
import json
import zipfile
from datetime import datetime, timezone
from typing import NamedTuple
from uuid import UUID, uuid4

import orjson
import pytest
from httpx import AsyncClient
from langflow.api.v1.schemas import FlowListCreate, ResultDataResponse
from langflow.initial_setup.setup import load_starter_projects
from langflow.services.database.models.base import orjson_dumps
from langflow.services.database.models.flow import Flow, FlowCreate, FlowUpdate
from langflow.services.database.models.folder.model import FolderCreate
from langflow.services.database.utils import session_getter
from langflow.services.deps import get_db_service
from lfx.graph.utils import log_transaction, log_vertex_build


@pytest.fixture(scope="module")
def json_style():
    # class FlowStyleBase(SQLModel):
    # color: str = Field(index=True)
    # emoji: str = Field(index=False)
    # flow_id: UUID = Field(default=None, foreign_key="flow.id")
    return orjson_dumps(
        {
            "color": "red",
            "emoji": "👍",
        }
    )


@pytest.mark.usefixtures("active_user")
async def test_create_flow(client: AsyncClient, json_flow: str, logged_in_headers):
    flow = orjson.loads(json_flow)
    data = flow["data"]
    flow = FlowCreate(name=str(uuid4()), description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    assert response.json()["name"] == flow.name
    assert response.json()["data"] == flow.data
    # flow is optional so we can create a flow without a flow
    flow = FlowCreate(name=str(uuid4()))
    response = await client.post("api/v1/flows/", json=flow.model_dump(exclude_unset=True), headers=logged_in_headers)
    assert response.status_code == 201
    assert response.json()["name"] == flow.name
    assert response.json()["data"] == flow.data


@pytest.mark.usefixtures("active_user")
async def test_read_flows(client: AsyncClient, json_flow: str, logged_in_headers):
    flow_data = orjson.loads(json_flow)
    data = flow_data["data"]
    flow = FlowCreate(name=str(uuid4()), description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    assert response.json()["name"] == flow.name
    assert response.json()["data"] == flow.data

    flow = FlowCreate(name=str(uuid4()), description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    assert response.json()["name"] == flow.name
    assert response.json()["data"] == flow.data

    response = await client.get("api/v1/flows/", headers=logged_in_headers)
    assert response.status_code == 200
    assert len(response.json()) > 0


async def test_read_flows_pagination_with_params(client: AsyncClient, logged_in_headers):
    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"page": 3, "size": 10, "get_all": False}
    )
    assert response.status_code == 200
    assert response.json()["page"] == 3
    assert response.json()["size"] == 10
    assert response.json()["pages"] == 0
    assert response.json()["total"] == 0
    assert len(response.json()["items"]) == 0


async def test_read_flows_pagination_with_flows(client: AsyncClient, logged_in_headers):
    number_of_flows = 30
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    flow_ids = []
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
        flow_ids.append(response.json()["id"])

    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"page": 3, "size": 10, "get_all": False}
    )
    assert response.status_code == 200
    assert response.json()["page"] == 3
    assert response.json()["size"] == 10
    assert response.json()["pages"] == 3
    assert response.json()["total"] == number_of_flows
    assert len(response.json()["items"]) == 10

    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"page": 4, "size": 10, "get_all": False}
    )
    assert response.status_code == 200
    assert response.json()["page"] == 4
    assert response.json()["size"] == 10
    assert response.json()["pages"] == 3
    assert response.json()["total"] == number_of_flows
    assert len(response.json()["items"]) == 0


async def test_read_flows_custom_page_size(client: AsyncClient, logged_in_headers):
    number_of_flows = 30
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201

    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"page": 1, "size": 15, "get_all": False}
    )
    assert response.status_code == 200
    assert response.json()["page"] == 1
    assert response.json()["size"] == 15
    assert response.json()["pages"] == 2
    assert response.json()["total"] == number_of_flows
    assert len(response.json()["items"]) == 15


async def test_read_flows_invalid_page(client: AsyncClient, logged_in_headers):
    number_of_flows = 30
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    flow_ids = []
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
        flow_ids.append(response.json()["id"])

    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"page": 0, "size": 10, "get_all": False}
    )
    assert response.status_code == 422  # Assuming 422 is the status code for invalid input


async def test_read_flows_invalid_size(client: AsyncClient, logged_in_headers):
    number_of_flows = 30
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    flow_ids = []
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
        flow_ids.append(response.json()["id"])

    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"page": 1, "size": 0, "get_all": False}
    )
    assert response.status_code == 422  # Assuming 422 is the status code for invalid input


async def test_read_flows_no_pagination_params(client: AsyncClient, logged_in_headers):
    number_of_flows = 30
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201

    response = await client.get("api/v1/flows/", headers=logged_in_headers, params={"get_all": False})
    assert response.status_code == 200
    # Assert default pagination values, adjust these according to your API's default behavior
    assert response.json()["page"] == 1
    assert response.json()["size"] == 50
    assert response.json()["pages"] == 1
    assert response.json()["total"] == number_of_flows
    assert len(response.json()["items"]) == number_of_flows


async def test_read_flows_components_only_paginated(client: AsyncClient, logged_in_headers):
    number_of_flows = 10
    flows = [
        FlowCreate(name=f"Flow {i}", description="description", data={}, is_component=True)
        for i in range(number_of_flows)
    ]

    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201

    response = await client.get(
        "api/v1/flows/", headers=logged_in_headers, params={"components_only": True, "get_all": False}
    )

    assert response.status_code == 200
    response_json = response.json()
    assert response_json["total"] == 10
    assert response_json["pages"] == 1
    assert response_json["page"] == 1
    assert response_json["size"] == 50
    assert all(flow["is_component"] is True for flow in response_json["items"])


async def test_read_flows_components_only(client: AsyncClient, logged_in_headers):
    number_of_flows = 10
    flows = [
        FlowCreate(name=f"Flow {i}", description="description", data={}, is_component=True)
        for i in range(number_of_flows)
    ]
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
    response = await client.get("api/v1/flows/", headers=logged_in_headers, params={"components_only": True})
    assert response.status_code == 200
    response_json = response.json()
    assert all(flow["is_component"] is True for flow in response_json)


async def test_read_flow(client: AsyncClient, json_flow: str, logged_in_headers):
    flow = orjson.loads(json_flow)
    data = flow["data"]
    unique_name = str(uuid4())
    flow = FlowCreate(name=unique_name, description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
    flow_id = response.json()["id"]  # flow_id should be a UUID but is a string
    # turn it into a UUID
    flow_id = UUID(flow_id)

    response = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
    assert response.status_code == 200
    assert response.json()["name"] == flow.name
    assert response.json()["data"] == flow.data


@pytest.mark.usefixtures("active_user")
async def test_update_flow(client: AsyncClient, json_flow: str, logged_in_headers):
    flow = orjson.loads(json_flow)
    data = flow["data"]

    flow = FlowCreate(name="Test Flow", description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)

    flow_id = response.json()["id"]
    updated_flow = FlowUpdate(
        name="Updated Flow",
        description="updated description",
        data=data,
    )
    response = await client.patch(f"api/v1/flows/{flow_id}", json=updated_flow.model_dump(), headers=logged_in_headers)

    assert response.status_code == 200
    assert response.json()["name"] == updated_flow.name
    assert response.json()["description"] == updated_flow.description
    # assert response.json()["data"] == updated_flow.data


@pytest.mark.usefixtures("active_user")
async def test_delete_flow(client: AsyncClient, json_flow: str, logged_in_headers):
    flow = orjson.loads(json_flow)
    data = flow["data"]
    flow = FlowCreate(name="Test Flow", description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
    flow_id = response.json()["id"]
    response = await client.delete(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
    assert response.status_code == 200
    assert response.json()["message"] == "Flow deleted successfully"


@pytest.mark.usefixtures("active_user")
async def test_delete_flows(client: AsyncClient, logged_in_headers):
    # Create ten flows
    number_of_flows = 10
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    flow_ids = []
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
        flow_ids.append(response.json()["id"])

    response = await client.request("DELETE", "api/v1/flows/", headers=logged_in_headers, json=flow_ids)
    assert response.status_code == 200, response.content
    assert response.json().get("deleted") == number_of_flows


@pytest.mark.usefixtures("active_user")
async def test_delete_flows_with_transaction_and_build(client: AsyncClient, logged_in_headers):
    # Create ten flows
    number_of_flows = 10
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    flow_ids = []
    for flow in flows:
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
        flow_ids.append(response.json()["id"])

    class VertexTuple(NamedTuple):
        id: str

    # Create a transaction for each flow
    for flow_id in flow_ids:
        await log_transaction(
            str(flow_id), source=VertexTuple(id="vid"), target=VertexTuple(id="tid"), status="success"
        )

    # Create a build for each flow
    for flow_id in flow_ids:
        build = {
            "valid": True,
            "params": {},
            "data": ResultDataResponse(),
            "artifacts": {},
            "vertex_id": "vid",
            "flow_id": flow_id,
        }
        await log_vertex_build(
            flow_id=build["flow_id"],
            vertex_id=build["vertex_id"],
            valid=build["valid"],
            params=build["params"],
            data=build["data"],
            artifacts=build.get("artifacts"),
        )

    response = await client.request("DELETE", "api/v1/flows/", headers=logged_in_headers, json=flow_ids)
    assert response.status_code == 200, response.content
    assert response.json().get("deleted") == number_of_flows

    for flow_id in flow_ids:
        response = await client.request(
            "GET", "api/v1/monitor/transactions", params={"flow_id": flow_id}, headers=logged_in_headers
        )
        assert response.status_code == 200
        json_response = response.json()
        assert json_response["items"] == []

    for flow_id in flow_ids:
        response = await client.request(
            "GET", "api/v1/monitor/builds", params={"flow_id": flow_id}, headers=logged_in_headers
        )
        assert response.status_code == 200
        assert response.json() == {"vertex_builds": {}}


@pytest.mark.usefixtures("active_user")
async def test_delete_folder_with_flows_with_transaction_and_build(client: AsyncClient, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description", components_list=[], flows_list=[])

    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201, f"Expected status code 201, but got {response.status_code}"

    created_folder = response.json()
    folder_id = created_folder["id"]

    # Create ten flows
    number_of_flows = 10
    flows = [FlowCreate(name=f"Flow {i}", description="description", data={}) for i in range(number_of_flows)]
    flow_ids = []
    for flow in flows:
        flow.folder_id = folder_id
        response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
        assert response.status_code == 201
        flow_ids.append(response.json()["id"])

    class VertexTuple(NamedTuple):
        id: str
        params: dict

    # Create a transaction for each flow
    for flow_id in flow_ids:
        await log_transaction(
            str(flow_id),
            source=VertexTuple(id="vid", params={}),
            target=VertexTuple(id="tid", params={}),
            status="success",
        )

    # Create a build for each flow
    for flow_id in flow_ids:
        build = {
            "valid": True,
            "params": {},
            "data": ResultDataResponse(),
            "artifacts": {},
            "vertex_id": "vid",
            "flow_id": flow_id,
        }
        await log_vertex_build(
            flow_id=build["flow_id"],
            vertex_id=build["vertex_id"],
            valid=build["valid"],
            params=build["params"],
            data=build["data"],
            artifacts=build.get("artifacts"),
        )

    response = await client.request("DELETE", f"api/v1/projects/{folder_id}", headers=logged_in_headers)
    assert response.status_code == 204

    for flow_id in flow_ids:
        response = await client.request(
            "GET", "api/v1/monitor/transactions", params={"flow_id": flow_id}, headers=logged_in_headers
        )
        assert response.status_code == 200, response.json()
        json_response = response.json()
        assert json_response["items"] == []

    for flow_id in flow_ids:
        response = await client.request(
            "GET", "api/v1/monitor/builds", params={"flow_id": flow_id}, headers=logged_in_headers
        )
        assert response.status_code == 200
        assert response.json() == {"vertex_builds": {}}


async def test_get_flows_from_folder_pagination(client: AsyncClient, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description", components_list=[], flows_list=[])

    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201, f"Expected status code 201, but got {response.status_code}"

    created_folder = response.json()
    folder_id = created_folder["id"]

    response = await client.get(
        f"api/v1/projects/{folder_id}", headers=logged_in_headers, params={"page": 1, "size": 50}
    )
    assert response.status_code == 200
    assert response.json()["folder"]["name"] == folder_name
    assert response.json()["folder"]["description"] == "Test project description"
    assert response.json()["flows"]["page"] == 1
    assert response.json()["flows"]["size"] == 50
    assert response.json()["flows"]["pages"] == 0
    assert response.json()["flows"]["total"] == 0
    assert len(response.json()["flows"]["items"]) == 0


async def test_get_flows_from_folder_pagination_with_params(client: AsyncClient, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description", components_list=[], flows_list=[])

    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201, f"Expected status code 201, but got {response.status_code}"

    created_folder = response.json()
    folder_id = created_folder["id"]

    response = await client.get(
        f"api/v1/projects/{folder_id}", headers=logged_in_headers, params={"page": 3, "size": 10}
    )
    assert response.status_code == 200
    assert response.json()["folder"]["name"] == folder_name
    assert response.json()["folder"]["description"] == "Test project description"
    assert response.json()["flows"]["page"] == 3
    assert response.json()["flows"]["size"] == 10
    assert response.json()["flows"]["pages"] == 0
    assert response.json()["flows"]["total"] == 0
    assert len(response.json()["flows"]["items"]) == 0


@pytest.mark.usefixtures("session")
async def test_create_flows(client: AsyncClient, json_flow: str, logged_in_headers):
    flow = orjson.loads(json_flow)
    data = flow["data"]
    # Create test data
    flow_unique_name = str(uuid4())
    flow_2_unique_name = str(uuid4())
    flow_list = FlowListCreate(
        flows=[
            FlowCreate(name=flow_unique_name, description="description", data=data),
            FlowCreate(name=flow_2_unique_name, description="description", data=data),
        ]
    )
    # Make request to endpoint
    response = await client.post("api/v1/flows/batch/", json=flow_list.dict(), headers=logged_in_headers)
    # Check response status code
    assert response.status_code == 201
    # Check response data
    response_data = response.json()
    assert len(response_data) == 2
    assert flow_unique_name in response_data[0]["name"]
    assert response_data[0]["description"] == "description"
    assert response_data[0]["data"] == data
    assert response_data[1]["name"] == flow_2_unique_name
    assert response_data[1]["description"] == "description"
    assert response_data[1]["data"] == data


@pytest.mark.usefixtures("session")
async def test_upload_file(client: AsyncClient, json_flow: str, logged_in_headers):
    flow = orjson.loads(json_flow)
    data = flow["data"]
    # Create test data
    flow_unique_name = str(uuid4())
    flow_2_unique_name = str(uuid4())
    flow_list = FlowListCreate(
        flows=[
            FlowCreate(name=flow_unique_name, description="description", data=data),
            FlowCreate(name=flow_2_unique_name, description="description", data=data),
        ]
    )
    file_contents = orjson_dumps(flow_list.dict())
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("examples.json", file_contents, "application/json")},
        headers=logged_in_headers,
    )
    # Check response status code
    assert response.status_code == 201
    # Check response data
    response_data = response.json()
    assert len(response_data) == 2
    assert flow_unique_name in response_data[0]["name"]
    assert response_data[0]["description"] == "description"
    assert response_data[0]["data"] == data
    assert response_data[1]["name"] == flow_2_unique_name
    assert response_data[1]["description"] == "description"
    assert response_data[1]["data"] == data


@pytest.mark.usefixtures("session")
async def test_upload_file_flows_not_a_list_returns_422(client: AsyncClient, logged_in_headers):
    """Regression: { "flows": <non-list> } must return 422, not 500."""
    file_contents = orjson.dumps({"flows": "oops"})
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("bad.json", file_contents, "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 422
    assert "flows" in response.json()["detail"].lower()


@pytest.mark.usefixtures("session")
async def test_upload_file_flows_item_not_a_dict_returns_422(client: AsyncClient, logged_in_headers):
    """Regression: { "flows": [1] } must return 422, not 500."""
    file_contents = orjson.dumps({"flows": [1]})
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("bad.json", file_contents, "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 422
    assert "flows" in response.json()["detail"].lower()


@pytest.mark.usefixtures("session")
async def test_upload_file_flows_mixed_items_returns_422(client: AsyncClient, logged_in_headers):
    """Regression: { "flows": [{}, 42] } must return 422 at the bad item."""
    file_contents = orjson.dumps({"flows": [{}, 42]})
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("bad.json", file_contents, "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 422
    assert "flows[1]" in response.json()["detail"]


@pytest.mark.usefixtures("session")
async def test_download_file(
    client: AsyncClient,
    json_flow,
    active_user,
    logged_in_headers,
):
    flow = orjson.loads(json_flow)
    data = flow["data"]
    # Create test data
    flow_unique_name = str(uuid4())
    flow_2_unique_name = str(uuid4())
    flow_list = FlowListCreate(
        flows=[
            FlowCreate(name=flow_unique_name, description="description", data=data),
            FlowCreate(name=flow_2_unique_name, description="description", data=data),
        ]
    )
    db_manager = get_db_service()
    async with session_getter(db_manager) as _session:
        saved_flows = []
        for flow in flow_list.flows:
            flow.user_id = active_user.id
            db_flow = Flow.model_validate(flow.model_dump(exclude={"id"}))
            _session.add(db_flow)
            saved_flows.append(db_flow)
        await _session.commit()
        # Make request to endpoint inside the session context
        flow_ids = [str(db_flow.id) for db_flow in saved_flows]  # Convert UUIDs to strings
        flow_ids_json = json.dumps(flow_ids)
        response = await client.post(
            "api/v1/flows/download/",
            data=flow_ids_json,
            headers={**logged_in_headers, "Content-Type": "application/json"},
        )
    # Check response status code
    assert response.status_code == 200, response.json()
    # Check response data
    # Since the endpoint now returns a zip file, we need to check the content type and the filename in the headers
    assert response.headers["Content-Type"] == "application/x-zip-compressed"
    content_disposition = response.headers["Content-Disposition"]
    assert "attachment" in content_disposition
    assert 'filename="' in content_disposition
    assert "filename*=UTF-8''" in content_disposition


@pytest.mark.usefixtures("session")
async def test_download_flows_content_disposition_dual_param(client: AsyncClient, logged_in_headers):
    """Downloading multiple flows must produce a dual-param RFC 5987 Content-Disposition header.

    The ZIP filename is always timestamp-based (pure ASCII). This test verifies
    that both the legacy 'filename=' param and the RFC 5987 'filename*=' param
    are present and well-formed, as required for compatibility with both old
    and new HTTP clients.
    """
    from urllib.parse import unquote

    # Create two flows (names don't affect the ZIP filename, which is timestamp-based)
    chinese_names = ["龙流程", "测试下载"]
    flow_ids = []
    for name in chinese_names:
        create_resp = await client.post(
            "api/v1/flows/",
            json={"name": name, "description": "", "data": {}},
            headers=logged_in_headers,
        )
        assert create_resp.status_code == 201
        flow_ids.append(create_resp.json()["id"])

    download_response = await client.post(
        "api/v1/flows/download/",
        data=json.dumps(flow_ids),
        headers={**logged_in_headers, "Content-Type": "application/json"},
    )
    assert download_response.status_code == 200
    assert download_response.headers["Content-Type"] == "application/x-zip-compressed"

    content_disposition = download_response.headers["Content-Disposition"]
    # Must include both params: ASCII fallback and RFC 5987
    assert "attachment" in content_disposition
    assert 'filename="' in content_disposition
    assert "filename*=UTF-8''" in content_disposition
    # The RFC 5987 value must decode to a .zip filename
    rfc5987_value = content_disposition.split("filename*=UTF-8''")[-1].split(";")[0].strip()
    decoded = unquote(rfc5987_value)
    assert decoded.endswith(".zip")


@pytest.mark.usefixtures("session")
async def test_download_single_flow_returns_normalized_json(client: AsyncClient, logged_in_headers):
    """Downloading a single flow returns normalized JSON rather than a ZIP archive."""
    code_value = "print('hello')\nprint('world')"
    flow_payload = FlowCreate(
        name=str(uuid4()),
        description="single flow export",
        data={
            "nodes": [
                {
                    "id": "node-1",
                    "data": {
                        "node": {
                            "template": {
                                "code": {"type": "code", "value": code_value},
                                "api_key": {"name": "api_key", "password": True, "value": "super-secret"},
                            }
                        }
                    },
                }
            ],
            "edges": [],
        },
    )

    create_response = await client.post("api/v1/flows/", json=flow_payload.model_dump(), headers=logged_in_headers)
    assert create_response.status_code == 201
    flow_id = create_response.json()["id"]

    download_response = await client.post(
        "api/v1/flows/download/",
        data=json.dumps([flow_id]),
        headers={**logged_in_headers, "Content-Type": "application/json"},
    )
    assert download_response.status_code == 200
    assert download_response.headers["Content-Type"].startswith("application/json")

    downloaded = download_response.json()
    assert downloaded["name"] == flow_payload.name
    assert "updated_at" not in downloaded
    assert "user_id" not in downloaded
    assert "folder_id" not in downloaded
    assert "access_type" not in downloaded
    assert downloaded["data"]["nodes"][0]["data"]["node"]["template"]["code"]["value"] == [
        "print('hello')",
        "print('world')",
    ]
    assert downloaded["data"]["nodes"][0]["data"]["node"]["template"]["api_key"]["value"] is None


@pytest.mark.usefixtures("session")
async def test_upload_zip_file_to_flows(client: AsyncClient, json_flow: str, logged_in_headers):
    """Test uploading a ZIP file containing flow JSONs to the flows upload endpoint."""
    flow = orjson.loads(json_flow)
    data = flow["data"]

    flow_1_name = str(uuid4())
    flow_2_name = str(uuid4())
    flow_1 = {"name": flow_1_name, "description": "desc1", "data": data}
    flow_2 = {"name": flow_2_name, "description": "desc2", "data": data}

    # Create a ZIP file in memory with individual flow JSONs
    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr(f"{flow_1_name}.json", json.dumps(flow_1))
        zf.writestr(f"{flow_2_name}.json", json.dumps(flow_2))
    zip_buffer.seek(0)

    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("flows.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 2
    names = {r["name"] for r in response_data}
    assert flow_1_name in names
    assert flow_2_name in names


@pytest.mark.usefixtures("session")
async def test_upload_zip_file_to_projects(client: AsyncClient, json_flow: str, logged_in_headers):
    """Test uploading a ZIP file containing flow JSONs to the projects upload endpoint."""
    flow = orjson.loads(json_flow)
    data = flow["data"]

    flow_1_name = str(uuid4())
    flow_2_name = str(uuid4())
    flow_1 = {"name": flow_1_name, "description": "desc1", "data": data}
    flow_2 = {"name": flow_2_name, "description": "desc2", "data": data}

    # Create a ZIP file in memory
    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr(f"{flow_1_name}.json", json.dumps(flow_1))
        zf.writestr(f"{flow_2_name}.json", json.dumps(flow_2))
    zip_buffer.seek(0)

    response = await client.post(
        "api/v1/projects/upload/",
        files={"file": ("My Project.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 2
    # All flows should belong to the same folder
    folder_ids = {r["folder_id"] for r in response_data}
    assert len(folder_ids) == 1

    # Verify the project name was derived from the ZIP filename
    folder_id = folder_ids.pop()
    project_response = await client.get(f"api/v1/projects/{folder_id}", headers=logged_in_headers)
    assert project_response.status_code == 200
    assert project_response.json()["name"].startswith("My Project")


@pytest.mark.usefixtures("session")
async def test_upload_empty_zip_returns_400(client: AsyncClient, logged_in_headers):
    """Test that uploading a ZIP with no JSON files returns 400."""
    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr("readme.txt", "not a flow")
    zip_buffer.seek(0)

    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("empty.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert "No valid flow JSON files" in response.json()["detail"]


@pytest.mark.usefixtures("session")
async def test_download_then_upload_roundtrip(client: AsyncClient, json_flow: str, active_user, logged_in_headers):
    """Test that downloading flows as ZIP and re-uploading works end-to-end."""
    flow = orjson.loads(json_flow)
    data = flow["data"]

    flow_1_name = str(uuid4())
    flow_2_name = str(uuid4())
    flow_list = FlowListCreate(
        flows=[
            FlowCreate(name=flow_1_name, description="description", data=data),
            FlowCreate(name=flow_2_name, description="description", data=data),
        ]
    )

    # Create flows in DB
    db_manager = get_db_service()
    async with session_getter(db_manager) as _session:
        saved_flows = []
        for f in flow_list.flows:
            f.user_id = active_user.id
            f.id = uuid4()
            db_flow = Flow.model_validate(f, from_attributes=True)
            _session.add(db_flow)
            saved_flows.append(db_flow)
        await _session.commit()
        flow_ids = [str(db_flow.id) for db_flow in saved_flows]

    # Download as ZIP
    download_response = await client.post(
        "api/v1/flows/download/",
        data=json.dumps(flow_ids),
        headers={**logged_in_headers, "Content-Type": "application/json"},
    )
    assert download_response.status_code == 200
    assert download_response.headers["Content-Type"] == "application/x-zip-compressed"

    # Re-upload the ZIP
    upload_response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("flows.zip", download_response.content, "application/zip")},
        headers=logged_in_headers,
    )
    assert upload_response.status_code == 201
    uploaded = upload_response.json()
    assert len(uploaded) == 2


@pytest.mark.usefixtures("session")
async def test_upload_zip_with_invalid_json(client: AsyncClient, json_flow: str, logged_in_headers):
    """ZIP entries with invalid JSON are skipped; valid entries are still processed."""
    flow = orjson.loads(json_flow)
    data = flow["data"]
    valid_flow = {"name": "valid_flow", "description": "good", "data": data}

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr("valid.json", json.dumps(valid_flow))
        zf.writestr("broken.json", "{not valid json!!!}")
    zip_buffer.seek(0)

    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("mixed.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 1
    assert response_data[0]["name"] == "valid_flow"


@pytest.mark.usefixtures("session")
async def test_upload_zip_exceeding_max_entries(client: AsyncClient, json_flow: str, logged_in_headers, monkeypatch):
    """ZIP with more JSON entries than the limit raises 400."""
    import langflow.api.utils.zip_utils as zip_utils_mod

    monkeypatch.setattr(zip_utils_mod, "MAX_ZIP_ENTRIES", 3)

    flow = orjson.loads(json_flow)
    data = flow["data"]

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        for i in range(5):
            zf.writestr(f"flow_{i}.json", json.dumps({"name": f"flow_{i}", "data": data}))
    zip_buffer.seek(0)

    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("too_many.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert "exceeding the limit" in response.json()["detail"]


@pytest.mark.usefixtures("session")
async def test_upload_zip_with_oversized_entry(client: AsyncClient, json_flow: str, logged_in_headers, monkeypatch):
    """Entries exceeding size limit are skipped; smaller valid entries are processed."""
    import langflow.api.utils.zip_utils as zip_utils_mod

    flow = orjson.loads(json_flow)
    data = flow["data"]
    small_flow = {"name": "small_flow", "data": {"nodes": [], "edges": []}}
    big_flow = {"name": "big_flow", "data": data}

    # Build the ZIP first, then set the limit between the two entry sizes
    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr("small.json", json.dumps(small_flow))
        zf.writestr("big.json", json.dumps(big_flow))

    # Re-read to find the actual sizes and pick a limit between them
    with zipfile.ZipFile(io.BytesIO(zip_buffer.getvalue()), "r") as zf:
        sizes = {info.filename: info.file_size for info in zf.infolist()}
    limit = (sizes["small.json"] + sizes["big.json"]) // 2
    monkeypatch.setattr(zip_utils_mod, "MAX_ENTRY_UNCOMPRESSED_BYTES", limit)

    zip_buffer.seek(0)
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("oversized.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 1
    assert response_data[0]["name"] == "small_flow"


@pytest.mark.usefixtures("session")
async def test_upload_zip_with_mixed_valid_invalid(client: AsyncClient, json_flow: str, logged_in_headers, monkeypatch):
    """Mix of valid flows, invalid JSON, and oversized entries → only valid flows returned."""
    import langflow.api.utils.zip_utils as zip_utils_mod

    flow = orjson.loads(json_flow)
    data = flow["data"]
    valid_flow = {"name": "keeper", "data": {"nodes": [], "edges": []}}
    oversized_flow = {"name": "too_big", "data": data, "padding": "x" * 500}

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr("valid.json", json.dumps(valid_flow))
        zf.writestr("broken.json", "NOT JSON {{{")
        zf.writestr("huge.json", json.dumps(oversized_flow))
        zf.writestr("readme.txt", "ignored non-json")

    # Set limit between valid entry size and oversized entry size
    with zipfile.ZipFile(io.BytesIO(zip_buffer.getvalue()), "r") as zf:
        sizes = {info.filename: info.file_size for info in zf.infolist()}
    limit = (sizes["valid.json"] + sizes["huge.json"]) // 2
    monkeypatch.setattr(zip_utils_mod, "MAX_ENTRY_UNCOMPRESSED_BYTES", limit)

    zip_buffer.seek(0)
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("mixed.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 1
    assert response_data[0]["name"] == "keeper"


@pytest.mark.usefixtures("session")
async def test_upload_zip_to_projects_filename_none(client: AsyncClient, json_flow: str, logged_in_headers):
    """When filename has no stem (e.g. '.zip'), the project name defaults to 'Imported Project'."""
    flow = orjson.loads(json_flow)
    data = flow["data"]
    flow_data = {"name": "flow_none", "data": data}

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        zf.writestr("flow.json", json.dumps(flow_data))
    zip_buffer.seek(0)

    # filename=".zip" → rsplit gives ("", "zip") → "" is falsy → "Imported Project"
    response = await client.post(
        "api/v1/projects/upload/",
        files={"file": (".zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 1

    folder_id = response_data[0]["folder_id"]
    project_response = await client.get(f"api/v1/projects/{folder_id}", headers=logged_in_headers)
    assert project_response.status_code == 200
    assert project_response.json()["name"].startswith("Imported Project")


@pytest.mark.usefixtures("session")
async def test_upload_json_file_to_projects_rejoins_code_lines(client: AsyncClient, logged_in_headers):
    """Project JSON upload accepts exported code-as-lines format and stores code as a string."""
    project_name = f"JSON Project {uuid4()}"
    code_lines = ["print('alpha')", "print('beta')"]
    payload = {
        "folder_name": project_name,
        "folder_description": "json upload",
        "flows": [
            {
                "name": f"Flow {uuid4()}",
                "description": "imported from project json",
                "data": {
                    "nodes": [
                        {
                            "id": "node-1",
                            "data": {
                                "node": {
                                    "template": {
                                        "code": {"type": "code", "value": code_lines},
                                    }
                                }
                            },
                        }
                    ],
                    "edges": [],
                },
            }
        ],
    }

    response = await client.post(
        "api/v1/projects/upload/",
        files={"file": ("project.json", json.dumps(payload).encode("utf-8"), "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201, response.text
    response_data = response.json()
    assert len(response_data) == 1
    assert response_data[0]["folder_id"] is not None
    assert (
        response_data[0]["data"]["nodes"][0]["data"]["node"]["template"]["code"]["value"]
        == "print('alpha')\nprint('beta')"
    )

    project_response = await client.get(f"api/v1/projects/{response_data[0]['folder_id']}", headers=logged_in_headers)
    assert project_response.status_code == 200
    assert project_response.json()["name"].startswith(project_name)


@pytest.mark.usefixtures("session")
async def test_download_project_zip_sanitizes_flow_names(client: AsyncClient, json_flow: str, logged_in_headers):
    """Project ZIP downloads must sanitize flow names to prevent Zip Slip paths."""
    project_response = await client.post(
        "api/v1/projects/",
        json={"name": f"Download Project {uuid4()}", "description": "", "flows_list": [], "components_list": []},
        headers=logged_in_headers,
    )
    assert project_response.status_code == 201
    project_id = project_response.json()["id"]

    flow = orjson.loads(json_flow)
    create_response = await client.post(
        "api/v1/flows/",
        json={
            "name": "../escaped-flow",
            "description": "path traversal test",
            "folder_id": project_id,
            "data": flow["data"],
            "is_component": False,
        },
        headers=logged_in_headers,
    )
    assert create_response.status_code == 201

    download_response = await client.get(f"api/v1/projects/download/{project_id}", headers=logged_in_headers)
    assert download_response.status_code == 200

    with zipfile.ZipFile(io.BytesIO(download_response.content), "r") as zip_file:
        file_names = zip_file.namelist()
        assert "__escaped-flow.json" in file_names
        assert all("/" not in name and ".." not in name for name in file_names)


@pytest.mark.usefixtures("session")
async def test_upload_bad_zip_file_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading a corrupt/invalid ZIP file returns 400 with a descriptive error."""
    # Payload with an EOCD signature prepended by garbage. Python 3.10-3.13 treat
    # this as a ZIP and fail in ZipFile() construction ("not a valid ZIP archive");
    # Python 3.14's stricter zipfile.is_zipfile() rejects it and the route falls
    # through to the JSON branch ("Invalid JSON file"). Both paths return 400 with
    # a descriptive detail, which is the user-facing contract under test.
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as zf:
        zf.writestr("dummy.json", '{"name":"x"}')
    valid_zip = buf.getvalue()
    # Minimal EOCD is 22 bytes; keep it and prepend garbage
    corrupt_zip = b"garbage" * 10 + valid_zip[-22:]

    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("corrupt.zip", corrupt_zip, "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    detail = response.json()["detail"]
    assert "not a valid ZIP" in detail or "Invalid JSON file" in detail


@pytest.mark.usefixtures("session")
async def test_upload_no_file_to_flows_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading with no file to flows endpoint returns 400."""
    response = await client.post(
        "api/v1/flows/upload/",
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert response.json()["detail"] == "No file provided"


@pytest.mark.usefixtures("session")
async def test_upload_no_file_to_projects_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading with no file to projects endpoint returns 400."""
    response = await client.post(
        "api/v1/projects/upload/",
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert response.json()["detail"] == "No file provided"


@pytest.mark.usefixtures("session")
async def test_upload_empty_file_to_flows_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading an empty file to flows endpoint returns 400."""
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("empty.json", b"", "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert response.json()["detail"] == "The uploaded file is empty"


@pytest.mark.usefixtures("session")
async def test_upload_empty_file_to_projects_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading an empty file to projects endpoint returns 400."""
    response = await client.post(
        "api/v1/projects/upload/",
        files={"file": ("empty.json", b"", "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert response.json()["detail"] == "The uploaded file is empty"


@pytest.mark.usefixtures("session")
async def test_upload_invalid_json_to_flows_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading invalid JSON content to flows endpoint returns 400."""
    response = await client.post(
        "api/v1/flows/upload/",
        files={"file": ("bad.json", b"this is not json", "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert "Invalid JSON file" in response.json()["detail"]


@pytest.mark.usefixtures("session")
async def test_upload_invalid_json_to_projects_returns_400(client: AsyncClient, logged_in_headers):
    """Uploading invalid JSON content to projects endpoint returns 400."""
    response = await client.post(
        "api/v1/projects/upload/",
        files={"file": ("bad.json", b"this is not json", "application/json")},
        headers=logged_in_headers,
    )
    assert response.status_code == 400
    assert "Invalid JSON file" in response.json()["detail"]


@pytest.mark.usefixtures("session")
async def test_upload_zip_to_projects_batch_name_dedup(client: AsyncClient, json_flow: str, logged_in_headers):
    """Multiple flows with the same name get unique names within the batch."""
    flow = orjson.loads(json_flow)
    data = flow["data"]

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w") as zf:
        for i in range(3):
            zf.writestr(f"flow_{i}.json", json.dumps({"name": "duplicate_name", "data": data}))
    zip_buffer.seek(0)

    response = await client.post(
        "api/v1/projects/upload/",
        files={"file": ("dedup_test.zip", zip_buffer.getvalue(), "application/zip")},
        headers=logged_in_headers,
    )
    assert response.status_code == 201
    response_data = response.json()
    assert len(response_data) == 3
    names = [r["name"] for r in response_data]
    # All names must be unique
    assert len(set(names)) == 3


@pytest.mark.usefixtures("active_user")
async def test_create_flow_with_invalid_data(client: AsyncClient, logged_in_headers):
    flow = {"name": "a" * 256, "data": "Invalid flow data"}
    response = await client.post("api/v1/flows/", json=flow, headers=logged_in_headers)
    assert response.status_code == 422


@pytest.mark.usefixtures("active_user")
async def test_get_nonexistent_flow(client: AsyncClient, logged_in_headers):
    uuid = uuid4()
    response = await client.get(f"api/v1/flows/{uuid}", headers=logged_in_headers)
    assert response.status_code == 404


@pytest.mark.usefixtures("active_user")
async def test_update_flow_idempotency(client: AsyncClient, json_flow: str, logged_in_headers):
    flow_data = orjson.loads(json_flow)
    data = flow_data["data"]
    flow_data = FlowCreate(name="Test Flow", description="description", data=data)
    response = await client.post("api/v1/flows/", json=flow_data.dict(), headers=logged_in_headers)
    flow_id = response.json()["id"]
    updated_flow = FlowCreate(name="Updated Flow", description="description", data=data)
    response1 = await client.put(f"api/v1/flows/{flow_id}", json=updated_flow.model_dump(), headers=logged_in_headers)
    response2 = await client.put(f"api/v1/flows/{flow_id}", json=updated_flow.model_dump(), headers=logged_in_headers)
    assert response1.json() == response2.json()


@pytest.mark.usefixtures("active_user")
async def test_update_nonexistent_flow(client: AsyncClient, json_flow: str, logged_in_headers):
    flow_data = orjson.loads(json_flow)
    data = flow_data["data"]
    uuid = uuid4()
    updated_flow = FlowCreate(
        name="Updated Flow",
        description="description",
        data=data,
    )
    response = await client.patch(f"api/v1/flows/{uuid}", json=updated_flow.model_dump(), headers=logged_in_headers)
    assert response.status_code == 404, response.text


@pytest.mark.usefixtures("active_user")
async def test_delete_nonexistent_flow(client: AsyncClient, logged_in_headers):
    uuid = uuid4()
    response = await client.delete(f"api/v1/flows/{uuid}", headers=logged_in_headers)
    assert response.status_code == 404


@pytest.mark.usefixtures("active_user")
async def test_read_only_starter_projects(client: AsyncClient, logged_in_headers):
    response = await client.get("api/v1/flows/basic_examples/", headers=logged_in_headers)
    starter_projects = await load_starter_projects()
    assert response.status_code == 200
    assert len(response.json()) == len(starter_projects)


async def test_sqlite_pragmas():
    import asyncio
    import sqlite3
    from urllib.parse import unquote

    # PRAGMA queries don't work well through SQLModel's async session abstraction
    # They need direct database access, so we use sqlite3 directly
    db_service = get_db_service()
    database_url = db_service.database_url

    if not database_url.startswith("sqlite"):
        pytest.skip("This test only works with SQLite databases")

    # Extract the database path from the URL
    if "///" in database_url:
        db_path = database_url.split("///", 1)[1]
    elif "//" in database_url:
        db_path = database_url.split("//", 1)[1]
    else:
        pytest.skip("Could not extract database path from URL")

    db_path = unquote(db_path)

    def get_pragmas():
        conn = sqlite3.connect(db_path)
        try:
            conn.execute("PRAGMA journal_mode=wal;")
            journal_mode = conn.execute("PRAGMA journal_mode;").fetchone()[0]
            synchronous = conn.execute("PRAGMA synchronous;").fetchone()[0]
            return journal_mode, synchronous
        finally:
            conn.close()

    journal_mode, synchronous = await asyncio.to_thread(get_pragmas)
    assert journal_mode == "wal"
    assert synchronous in [0, 1, 2], f"Unexpected synchronous value: {synchronous}"


@pytest.mark.usefixtures("active_user")
async def test_read_folder(client: AsyncClient, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description")
    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    created_folder = response.json()
    folder_id = created_folder["id"]

    # Read the project
    response = await client.get(f"api/v1/projects/{folder_id}", headers=logged_in_headers)
    assert response.status_code == 200
    folder_data = response.json()
    assert folder_data["name"] == folder_name
    assert folder_data["description"] == "Test project description"
    assert "flows" in folder_data
    assert isinstance(folder_data["flows"], list)


@pytest.mark.usefixtures("active_user")
async def test_read_folder_with_pagination(client: AsyncClient, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description")
    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    created_folder = response.json()
    folder_id = created_folder["id"]

    # Read the project with pagination
    response = await client.get(
        f"api/v1/projects/{folder_id}", headers=logged_in_headers, params={"page": 1, "size": 10}
    )
    assert response.status_code == 200
    folder_data = response.json()
    assert isinstance(folder_data, dict)
    assert "folder" in folder_data
    assert "flows" in folder_data
    assert folder_data["folder"]["name"] == folder_name
    assert folder_data["folder"]["description"] == "Test project description"
    assert folder_data["flows"]["page"] == 1
    assert folder_data["flows"]["size"] == 10
    assert isinstance(folder_data["flows"]["items"], list)


@pytest.mark.usefixtures("active_user")
async def test_read_folder_with_flows(client: AsyncClient, json_flow: str, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    flow_name = f"Test Flow {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description")
    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    created_folder = response.json()
    folder_id = created_folder["id"]

    # Create a flow in the project
    flow_data = orjson.loads(json_flow)
    data = flow_data["data"]
    flow = FlowCreate(name=flow_name, description="description", data=data)
    flow.folder_id = folder_id
    response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201

    # Read the project with flows
    response = await client.get(f"api/v1/projects/{folder_id}", headers=logged_in_headers)
    assert response.status_code == 200
    folder_data = response.json()
    assert folder_data["name"] == folder_name
    assert folder_data["description"] == "Test project description"
    assert len(folder_data["flows"]) == 1
    assert folder_data["flows"][0]["name"] == flow_name


@pytest.mark.usefixtures("active_user")
async def test_read_nonexistent_folder(client: AsyncClient, logged_in_headers):
    nonexistent_id = str(uuid4())
    response = await client.get(f"api/v1/projects/{nonexistent_id}", headers=logged_in_headers)
    assert response.status_code == 404
    assert response.json()["detail"] == "Project not found"


@pytest.mark.usefixtures("active_user")
async def test_read_folder_with_search(client: AsyncClient, json_flow: str, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description")
    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    created_folder = response.json()
    folder_id = created_folder["id"]

    # Create two flows in the project
    flow_data = orjson.loads(json_flow)
    flow_name_1 = f"Test Flow 1 {uuid4()}"
    flow_name_2 = f"Another Flow {uuid4()}"

    flow1 = FlowCreate(
        name=flow_name_1, description="Test flow description", data=flow_data["data"], folder_id=folder_id
    )
    flow2 = FlowCreate(
        name=flow_name_2, description="Another flow description", data=flow_data["data"], folder_id=folder_id
    )
    flow1.folder_id = folder_id
    flow2.folder_id = folder_id
    await client.post("api/v1/flows/", json=flow1.model_dump(), headers=logged_in_headers)
    await client.post("api/v1/flows/", json=flow2.model_dump(), headers=logged_in_headers)

    # Read the project with search
    response = await client.get(
        f"api/v1/projects/{folder_id}", headers=logged_in_headers, params={"search": "Test", "page": 1, "size": 10}
    )
    assert response.status_code == 200
    folder_data = response.json()
    assert len(folder_data["flows"]["items"]) == 1
    assert folder_data["flows"]["items"][0]["name"] == flow_name_1


@pytest.mark.usefixtures("active_user")
async def test_read_folder_with_component_filter(client: AsyncClient, json_flow: str, logged_in_headers):
    # Create a new project
    folder_name = f"Test Project {uuid4()}"
    project = FolderCreate(name=folder_name, description="Test project description")
    response = await client.post("api/v1/projects/", json=project.model_dump(), headers=logged_in_headers)
    assert response.status_code == 201
    created_folder = response.json()
    folder_id = created_folder["id"]

    # Create a component flow in the project
    flow_data = orjson.loads(json_flow)
    component_flow_name = f"Component Flow {uuid4()}"
    component_flow = FlowCreate(
        name=component_flow_name,
        description="Component flow description",
        data=flow_data["data"],
        folder_id=folder_id,
        is_component=True,
    )
    component_flow.folder_id = folder_id
    await client.post("api/v1/flows/", json=component_flow.model_dump(), headers=logged_in_headers)

    # Read the project with component filter
    response = await client.get(
        f"api/v1/projects/{folder_id}", headers=logged_in_headers, params={"is_component": True, "page": 1, "size": 10}
    )
    assert response.status_code == 200
    folder_data = response.json()
    assert len(folder_data["flows"]["items"]) == 1
    assert folder_data["flows"]["items"][0]["name"] == component_flow_name
    assert folder_data["flows"]["items"][0]["is_component"] == True  # noqa: E712


def test_transaction_excludes_code_key(session):
    """Test that the code key is excluded from transaction inputs when logged to the database."""
    from langflow.services.database.models.transactions.model import TransactionTable

    # Create a flow to associate with the transaction
    flow = Flow(name=str(uuid4()), description="Test flow", data={})
    session.add(flow)
    session.commit()
    session.refresh(flow)

    # Create input data with a code key
    input_data = {"param1": "value1", "param2": "value2", "code": "print('Hello, world!')"}

    # Create a transaction with inputs containing a code key
    transaction = TransactionTable(
        timestamp=datetime.now(timezone.utc),
        vertex_id="test-vertex",
        target_id="test-target",
        inputs=input_data,
        outputs={"result": "success"},
        status="completed",
        flow_id=flow.id,
    )

    # Verify that the code key is removed during transaction creation
    assert transaction.inputs is not None
    assert "code" not in transaction.inputs
    assert "param1" in transaction.inputs
    assert "param2" in transaction.inputs

    # Add the transaction to the database
    session.add(transaction)
    session.commit()
    session.refresh(transaction)

    # Verify that the code key is not in the saved transaction inputs
    assert transaction.inputs is not None
    assert "code" not in transaction.inputs
    assert "param1" in transaction.inputs
    assert "param2" in transaction.inputs
