From a2eae0381b3be80fcec8dd348614558da0088b73 Mon Sep 17 00:00:00 2001 From: Tarcio Date: Thu, 21 May 2026 14:17:43 -0300 Subject: [PATCH] feat(triggers): read-mostly API surface (aggregator + delete + bulk) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The frontend list view + bulk-delete needs a small HTTP surface over the in-flow trigger model. This commit adds it in three slim layers: NEW services/triggers/queries.py list_triggers_for_user(session, user_id) → list[TriggerInstance] Walks user-owned flows whose data is non-null, scans for CronTrigger nodes via discovery, joins each with its most-recent trigger_job rows. Output dataclass carries (flow_id, flow_name, component_id, cron_expression, timezone, max_attempts, next_fire_at, last_finished_status, last_finished_at) — enough for the table without further round-trips. O(F) flows + O(F) job batches. Volume bound by hand-edits in the canvas; query count traded for simplicity. NEW services/triggers/removal.py remove_cron_trigger_node(flow_data, component_id) remove_all_cron_trigger_nodes(flow_data) Pure dict mutations — no DB. Return new flow_data plus a boolean / list of removed ids. Both prune edges referencing the removed nodes so the resulting graph is still valid. Routes invoke these, write the result back via the standard _patch_flow helper, which fires the lifecycle hook that cancels the queued trigger_job rows for the removed components. No manual cascade in the API layer. REWRITTEN api/v1/triggers.py GET /api/v1/triggers list[TriggerInstance] for the current user DELETE /api/v1/triggers/{flow_id}/{component_id} strip one node + 204; 404 when not found DELETE /api/v1/triggers bulk strip every CronTrigger node from every owned flow; returns {flows_updated, components_removed} GET /api/v1/triggers/{flow_id}/{component_id}/jobs recent trigger_job rows for one component; optional ?status_filter, default limit 50 No POST / PATCH / PUT — the canvas is the only editing surface. Routes are thin: ownership check + helper call + response shape. NEW tests/unit/services/triggers/test_queries.py (6 tests) empty user, multi-flow aggregation, user isolation, next-fire computation, last-terminal computation, flows without data. NEW tests/unit/services/triggers/test_removal.py (10 tests) single-remove: missing flow_data, missing component, id collision with non-CronTrigger node, multi-node selectivity, edge pruning, input immutability. Bulk-remove: empty case, multi-strip with edge pruning, missing edges, malformed nodes list. 53/53 tests pass, ruff clean. The /triggers route is now ready for the frontend list view in Etapa G. --- src/backend/base/langflow/api/v1/triggers.py | 217 +++++++++++++++++- .../langflow/services/triggers/queries.py | 167 ++++++++++++++ .../langflow/services/triggers/removal.py | 122 ++++++++++ .../unit/services/triggers/test_queries.py | 190 +++++++++++++++ .../unit/services/triggers/test_removal.py | 131 +++++++++++ 5 files changed, 822 insertions(+), 5 deletions(-) create mode 100644 src/backend/base/langflow/services/triggers/queries.py create mode 100644 src/backend/base/langflow/services/triggers/removal.py create mode 100644 src/backend/tests/unit/services/triggers/test_queries.py create mode 100644 src/backend/tests/unit/services/triggers/test_removal.py diff --git a/src/backend/base/langflow/api/v1/triggers.py b/src/backend/base/langflow/api/v1/triggers.py index b411c7eee1..9c5e584840 100644 --- a/src/backend/base/langflow/api/v1/triggers.py +++ b/src/backend/base/langflow/api/v1/triggers.py @@ -1,11 +1,218 @@ """Read-mostly HTTP surface for the in-flow trigger feature. -Filled in a later commit. For now this is just the router shell so -the API package keeps importing cleanly while the schema refactor -takes effect — the trigger configuration moved into ``flow.data`` and -the old CRUD endpoints no longer apply. +Creation lives in the flow canvas: a user drops a CronTrigger +component, fills in its inputs, saves the flow. This module surfaces +the resulting state for a management UI: + + GET /api/v1/triggers + One row per CronTrigger component across the current user's + flows. Combines live config (read from ``flow.data``) with + the most recent ``trigger_job`` rows so the list view can + render "next fire" and "last run" columns. + + DELETE /api/v1/triggers/{flow_id}/{component_id} + Strip a single CronTrigger node from a flow. The lifecycle + hook on the subsequent save cancels its queued job rows. + Returns 404 when the flow or the component cannot be + found for the current user. + + DELETE /api/v1/triggers + Bulk strip: remove every CronTrigger node from every flow + the current user owns. Idempotent — re-running on a + cleaned-up set is a no-op and still returns 200. + + GET /api/v1/triggers/{flow_id}/{component_id}/jobs + Recent ``trigger_job`` rows for one component, descending + ``scheduled_at``. Used by the per-trigger history drawer. + +There are no POST / PATCH / PUT endpoints by design: the canvas is the +single editing surface for trigger configuration. """ -from fastapi import APIRouter +from __future__ import annotations + +from dataclasses import asdict +from datetime import datetime +from typing import TYPE_CHECKING +from uuid import UUID + +from fastapi import APIRouter, HTTPException, status +from sqlmodel import select + +from langflow.api.utils import CurrentActiveUser, DbSession +from langflow.services.database.models.flow.model import Flow +from langflow.services.database.models.jobs.model import JobStatus +from langflow.services.database.models.triggers.crud import list_jobs_for_flow +from langflow.services.database.models.triggers.model import TriggerJobRead +from langflow.services.triggers.queries import ( + TriggerInstance, + list_triggers_for_user, +) +from langflow.services.triggers.removal import ( + remove_all_cron_trigger_nodes, + remove_cron_trigger_node, +) + +if TYPE_CHECKING: + from sqlmodel.ext.asyncio.session import AsyncSession + router = APIRouter(prefix="/triggers", tags=["Triggers"]) + + +# --------------------------------------------------------------------------- # +# Response schemas (kept minimal & inline — no separate Pydantic dance) +# --------------------------------------------------------------------------- # + + +def _serialize_instance(instance: TriggerInstance) -> dict: + """``dataclasses.asdict`` plus JSON-friendly types for datetimes/UUIDs.""" + raw = asdict(instance) + return { + **raw, + "flow_id": str(instance.flow_id), + "next_fire_at": _iso(instance.next_fire_at), + "last_finished_at": _iso(instance.last_finished_at), + "last_finished_status": ( + instance.last_finished_status.value if instance.last_finished_status else None + ), + } + + +def _iso(value: datetime | None) -> str | None: + return value.isoformat() if value else None + + +# --------------------------------------------------------------------------- # +# Helpers shared by the routes +# --------------------------------------------------------------------------- # + + +async def _get_owned_flow( + session: AsyncSession, + flow_id: UUID, + user_id: UUID, +) -> Flow: + """Fetch a flow restricted to ``user_id``; 404 on miss. + + Centralises the ownership check so every route enforces it the + same way and the response payloads do not need to redo the + cross-user → 404 mapping inline. + """ + statement = select(Flow).where(Flow.id == flow_id, Flow.user_id == user_id) + flow = (await session.exec(statement)).first() + if flow is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Flow not found") + return flow + + +async def _persist_flow_data_change(session: AsyncSession, flow: Flow) -> None: + """Save the in-memory flow.data mutation back through the same hooks the API uses. + + Importing the helper inside the function avoids a hard module-load + coupling between this route and ``flows_helpers`` (which itself + imports from the triggers services package). + """ + from langflow.api.v1.flows_helpers import _patch_flow + from langflow.services.database.models.flow.model import FlowUpdate + + update = FlowUpdate(data=flow.data) + from langflow.services.deps import get_storage_service + + storage_service = get_storage_service() + await _patch_flow( + session=session, + db_flow=flow, + flow=update, + user_id=flow.user_id, + storage_service=storage_service, + ) + + +# --------------------------------------------------------------------------- # +# Routes +# --------------------------------------------------------------------------- # + + +@router.get("", response_model=list[dict]) +async def list_triggers( + *, + session: DbSession, + current_user: CurrentActiveUser, +) -> list[dict]: + """All CronTrigger components the current user has across all flows.""" + instances = await list_triggers_for_user(session, current_user.id) + return [_serialize_instance(i) for i in instances] + + +@router.delete("/{flow_id}/{component_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_trigger( + *, + session: DbSession, + current_user: CurrentActiveUser, + flow_id: UUID, + component_id: str, +) -> None: + """Strip one CronTrigger node from a flow. + + The post-save lifecycle hook downgrades the trigger's queued jobs + to ``cancelled``, so no manual cascade is needed here. + """ + flow = await _get_owned_flow(session, flow_id, current_user.id) + new_data, was_removed = remove_cron_trigger_node(flow.data, component_id) + if not was_removed: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Trigger component not found in this flow", + ) + flow.data = new_data + await _persist_flow_data_change(session, flow) + + +@router.delete("", response_model=dict, status_code=status.HTTP_200_OK) +async def delete_all_triggers( + *, + session: DbSession, + current_user: CurrentActiveUser, +) -> dict: + """Remove every CronTrigger component from every flow the user owns. + + Returns ``{"flows_updated": int, "components_removed": int}`` so the + UI can show a meaningful confirmation toast. + """ + statement = select(Flow).where(Flow.user_id == current_user.id) + flows = list((await session.exec(statement)).all()) + + flows_updated = 0 + components_removed = 0 + for flow in flows: + new_data, removed_ids = remove_all_cron_trigger_nodes(flow.data) + if not removed_ids: + continue + flow.data = new_data + await _persist_flow_data_change(session, flow) + flows_updated += 1 + components_removed += len(removed_ids) + + return {"flows_updated": flows_updated, "components_removed": components_removed} + + +@router.get("/{flow_id}/{component_id}/jobs", response_model=list[TriggerJobRead]) +async def list_trigger_jobs( + *, + session: DbSession, + current_user: CurrentActiveUser, + flow_id: UUID, + component_id: str, + status_filter: JobStatus | None = None, + limit: int = 50, +): + """Recent trigger_jobs for one component on one flow.""" + await _get_owned_flow(session, flow_id, current_user.id) + return await list_jobs_for_flow( + session, + flow_id, + component_id=component_id, + status=status_filter, + limit=limit, + ) diff --git a/src/backend/base/langflow/services/triggers/queries.py b/src/backend/base/langflow/services/triggers/queries.py new file mode 100644 index 0000000000..a18fa2d91d --- /dev/null +++ b/src/backend/base/langflow/services/triggers/queries.py @@ -0,0 +1,167 @@ +"""Read-only aggregator queries for the triggers HTTP surface. + +Walks every flow the current user owns, surfaces the CronTrigger +components present in their saved JSON, and joins each component +with its most recent ``trigger_job`` row(s). The HTTP layer hands +the result straight to the frontend list view. + +Kept separate from the route handlers so the query logic can be +exercised in unit tests without standing up a FastAPI client. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from sqlmodel import col, select + +from langflow.services.database.models.flow.model import Flow +from langflow.services.database.models.jobs.model import JobStatus +from langflow.services.database.models.triggers.model import TriggerJob +from langflow.services.triggers.discovery import ( + CronTriggerConfig, + find_cron_trigger_configs, +) + +if TYPE_CHECKING: + from datetime import datetime + from uuid import UUID + + from sqlmodel.ext.asyncio.session import AsyncSession + + +@dataclass(frozen=True) +class TriggerInstance: + """One CronTrigger component instance surfaced to the API. + + Combines the component config (live, from ``flow.data``) with two + cached views into the work queue: + + * ``next_fire_at`` — when the worker will pick up the next queued + job for this component, if any. + * ``last_finished_status`` / ``last_finished_at`` — the most + recent terminal job (completed / failed / cancelled / timed_out) + for this component, for the "last run" column in the UI. + + All fields are derived; nothing here is persisted in this shape. + """ + + flow_id: UUID + flow_name: str + component_id: str + cron_expression: str + timezone: str + max_attempts: int + next_fire_at: datetime | None + last_finished_status: JobStatus | None + last_finished_at: datetime | None + + +_TERMINAL_STATES: tuple[JobStatus, ...] = ( + JobStatus.COMPLETED, + JobStatus.FAILED, + JobStatus.CANCELLED, + JobStatus.TIMED_OUT, +) + + +async def _user_flows_with_data(session: AsyncSession, user_id: UUID) -> list[Flow]: + """Return user-owned flows that have ``flow.data`` populated. + + Filters NULL/empty data server-side so the Python iteration only + sees rows worth inspecting. + """ + statement = select(Flow).where( + Flow.user_id == user_id, + col(Flow.data).is_not(None), + ) + result = await session.exec(statement) + return list(result.all()) + + +async def _jobs_for_components( + session: AsyncSession, + flow_id: UUID, + component_ids: list[str], +) -> dict[str, list[TriggerJob]]: + """Group existing trigger_job rows by component_id for one flow. + + Returns ``{component_id: [TriggerJob, ...]}`` ordered with the most + recently-scheduled jobs first so the aggregator can grab "next + queued" and "last finished" with simple list comprehensions. + """ + if not component_ids: + return {} + statement = ( + select(TriggerJob) + .where( + TriggerJob.flow_id == flow_id, + col(TriggerJob.component_id).in_(component_ids), + ) + .order_by(col(TriggerJob.scheduled_at).desc()) + ) + result = await session.exec(statement) + by_component: dict[str, list[TriggerJob]] = {cid: [] for cid in component_ids} + for job in result.all(): + by_component.setdefault(job.component_id, []).append(job) + return by_component + + +def _summarise( + flow: Flow, + config: CronTriggerConfig, + jobs: list[TriggerJob], +) -> TriggerInstance: + """Combine one component's config with its job history into a TriggerInstance.""" + next_fire = next( + (j.scheduled_at for j in reversed(jobs) if j.status == JobStatus.QUEUED), + None, + ) + last_terminal = next( + (j for j in jobs if j.status in _TERMINAL_STATES), + None, + ) + return TriggerInstance( + flow_id=flow.id, + flow_name=flow.name, + component_id=config.component_id, + cron_expression=config.cron_expression, + timezone=config.timezone, + max_attempts=config.max_attempts, + next_fire_at=next_fire, + last_finished_status=last_terminal.status if last_terminal else None, + last_finished_at=last_terminal.finished_at if last_terminal else None, + ) + + +async def list_triggers_for_user( + session: AsyncSession, + user_id: UUID, +) -> list[TriggerInstance]: + """Surface every CronTrigger component owned by ``user_id``. + + O(F) flows + O(F) job batches — one query per flow that has + triggers. The volume of triggers a single user creates is bounded + by hand-editing the canvas, so we trade query count for simplicity. + """ + flows = await _user_flows_with_data(session, user_id) + + instances: list[TriggerInstance] = [] + for flow in flows: + configs = find_cron_trigger_configs(flow.data) + if not configs: + continue + jobs_by_component = await _jobs_for_components( + session, + flow.id, + [c.component_id for c in configs], + ) + instances.extend( + _summarise(flow, config, jobs_by_component.get(config.component_id, [])) + for config in configs + ) + + # Stable order: flow name first, then component id within the flow. + instances.sort(key=lambda i: (i.flow_name.lower(), i.component_id)) + return instances diff --git a/src/backend/base/langflow/services/triggers/removal.py b/src/backend/base/langflow/services/triggers/removal.py new file mode 100644 index 0000000000..6375d3fdf0 --- /dev/null +++ b/src/backend/base/langflow/services/triggers/removal.py @@ -0,0 +1,122 @@ +"""Strip CronTrigger nodes out of ``flow.data``. + +Used by the bulk and single-trigger DELETE endpoints. The function is +pure on the input dict — it returns a new ``flow.data`` plus the list +of removed component ids. The route handler writes the new data back +to the flow row and then calls the standard lifecycle hook, which +takes care of cancelling the now-orphaned queued ``trigger_job`` rows. + +Keeping the mutation pure (no DB access) makes the deletion logic +trivially unit-testable and keeps the API handlers free of dict +gymnastics. +""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +from langflow.services.triggers.discovery import CRON_TRIGGER_TYPE + + +def remove_cron_trigger_node( + flow_data: dict[str, Any] | None, + component_id: str, +) -> tuple[dict[str, Any], bool]: + """Return ``(new_flow_data, was_removed)`` after dropping a single node. + + Also prunes any edge whose source or target was the removed node, + so saving the result through the standard flow update path leaves + a valid graph for downstream consumers. The caller decides what + to do with ``was_removed`` (typically: 404 when False). + """ + if not flow_data: + return flow_data or {"nodes": [], "edges": []}, False + + new_data = deepcopy(flow_data) + nodes = new_data.get("nodes") + if not isinstance(nodes, list): + return new_data, False + + survivors = [n for n in nodes if not _matches(n, component_id)] + was_removed = len(survivors) != len(nodes) + new_data["nodes"] = survivors + + if was_removed: + new_data["edges"] = _prune_edges(new_data.get("edges"), {component_id}) + + return new_data, was_removed + + +def remove_all_cron_trigger_nodes( + flow_data: dict[str, Any] | None, +) -> tuple[dict[str, Any], list[str]]: + """Strip every CronTrigger node from ``flow_data`` in one pass. + + Returns ``(new_flow_data, removed_component_ids)``. ``removed_*`` + is the list of node ids that disappeared, in the order they + appeared in the original graph — useful for bulk-delete responses + that want to surface what was actually removed. + """ + if not flow_data: + return flow_data or {"nodes": [], "edges": []}, [] + + new_data = deepcopy(flow_data) + nodes = new_data.get("nodes") + if not isinstance(nodes, list): + return new_data, [] + + removed_ids: list[str] = [] + survivors: list[dict[str, Any]] = [] + for node in nodes: + if _is_cron_trigger_node(node): + node_id = node.get("id") if isinstance(node, dict) else None + if isinstance(node_id, str): + removed_ids.append(node_id) + continue + survivors.append(node) + + new_data["nodes"] = survivors + if removed_ids: + new_data["edges"] = _prune_edges(new_data.get("edges"), set(removed_ids)) + return new_data, removed_ids + + +# --------------------------------------------------------------------------- # +# internal helpers +# --------------------------------------------------------------------------- # + + +def _matches(node: object, component_id: str) -> bool: + """Return True when ``node`` is the CronTrigger with the given id.""" + return ( + isinstance(node, dict) + and node.get("id") == component_id + and _is_cron_trigger_node(node) + ) + + +def _is_cron_trigger_node(node: object) -> bool: + """Same matcher used by ``discovery.find_cron_trigger_nodes``. + + Guarded inline so this module stays import-free from heavy + discovery state — just the constant. + """ + if not isinstance(node, dict): + return False + data = node.get("data") + return isinstance(data, dict) and data.get("type") == CRON_TRIGGER_TYPE + + +def _prune_edges(edges: Any, removed_ids: set[str]) -> list[Any]: + """Drop any edge that references a removed node id.""" + if not isinstance(edges, list): + return [] + return [ + edge + for edge in edges + if not ( + isinstance(edge, dict) + and (edge.get("source") in removed_ids or edge.get("target") in removed_ids) + ) + ] diff --git a/src/backend/tests/unit/services/triggers/test_queries.py b/src/backend/tests/unit/services/triggers/test_queries.py new file mode 100644 index 0000000000..d184a59a2a --- /dev/null +++ b/src/backend/tests/unit/services/triggers/test_queries.py @@ -0,0 +1,190 @@ +"""Unit tests for the read aggregator (services.triggers.queries).""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +import pytest +from langflow.services.database.models.flow.model import Flow +from langflow.services.database.models.jobs.model import JobStatus +from langflow.services.database.models.triggers import TriggerJob +from langflow.services.database.models.user.model import User +from langflow.services.triggers.queries import list_triggers_for_user + + +def _cron_node(component_id: str, *, cron: str = "*/5 * * * *", tz: str = "UTC") -> dict: + return { + "id": component_id, + "type": "genericNode", + "data": { + "type": "CronTrigger", + "node": { + "template": { + "cron_expression": {"value": cron}, + "timezone": {"value": tz}, + "max_attempts": {"value": 3}, + }, + }, + }, + } + + +async def _seed_user(session, *, suffix: str = "") -> User: + user = User( + id=uuid4(), + username=f"u-{uuid4().hex[:6]}{suffix}", + password="x", # noqa: S106 + is_active=True, + ) + session.add(user) + await session.commit() + return user + + +async def _seed_flow(session, *, user_id, name: str, nodes: list[dict]) -> Flow: + flow = Flow(id=uuid4(), name=name, data={"nodes": nodes, "edges": []}, user_id=user_id) + session.add(flow) + await session.commit() + return flow + + +@pytest.mark.asyncio +async def test_list_returns_empty_for_user_without_triggers(async_session): + user = await _seed_user(async_session) + instances = await list_triggers_for_user(async_session, user.id) + assert instances == [] + + +@pytest.mark.asyncio +async def test_list_surfaces_one_instance_per_component(async_session): + user = await _seed_user(async_session) + await _seed_flow( + async_session, + user_id=user.id, + name="flow-a", + nodes=[ + _cron_node("CronTrigger-a1", cron="*/5 * * * *"), + _cron_node("CronTrigger-a2", cron="0 9 * * *"), + ], + ) + await _seed_flow( + async_session, + user_id=user.id, + name="flow-b", + nodes=[_cron_node("CronTrigger-b1", cron="0 0 * * *", tz="Europe/London")], + ) + + instances = await list_triggers_for_user(async_session, user.id) + by_id = {(i.flow_name, i.component_id): i for i in instances} + assert set(by_id.keys()) == { + ("flow-a", "CronTrigger-a1"), + ("flow-a", "CronTrigger-a2"), + ("flow-b", "CronTrigger-b1"), + } + london = by_id[("flow-b", "CronTrigger-b1")] + assert london.cron_expression == "0 0 * * *" + assert london.timezone == "Europe/London" + + +@pytest.mark.asyncio +async def test_list_isolates_users(async_session): + """A user must never see another user's triggers.""" + alice = await _seed_user(async_session, suffix="a") + bob = await _seed_user(async_session, suffix="b") + await _seed_flow(async_session, user_id=alice.id, name="alice-flow", nodes=[_cron_node("CronTrigger-a")]) + await _seed_flow(async_session, user_id=bob.id, name="bob-flow", nodes=[_cron_node("CronTrigger-b")]) + + alice_view = await list_triggers_for_user(async_session, alice.id) + bob_view = await list_triggers_for_user(async_session, bob.id) + + assert [i.component_id for i in alice_view] == ["CronTrigger-a"] + assert [i.component_id for i in bob_view] == ["CronTrigger-b"] + + +@pytest.mark.asyncio +async def test_list_picks_next_fire_from_queued_job(async_session): + user = await _seed_user(async_session) + flow = await _seed_flow( + async_session, + user_id=user.id, + name="alpha", + nodes=[_cron_node("CronTrigger-x")], + ) + next_fire = datetime.now(timezone.utc) + timedelta(minutes=4) + async_session.add( + TriggerJob( + id=uuid4(), + flow_id=flow.id, + component_id="CronTrigger-x", + status=JobStatus.QUEUED, + scheduled_at=next_fire, + attempt=1, + max_attempts=3, + ), + ) + await async_session.commit() + + instances = await list_triggers_for_user(async_session, user.id) + assert len(instances) == 1 + # SQLAlchemy may strip the tz when storing into TIMESTAMP, so + # compare the wall-clock instant tolerantly. + assert instances[0].next_fire_at is not None + delta = abs((instances[0].next_fire_at.replace(tzinfo=None) - next_fire.replace(tzinfo=None)).total_seconds()) + assert delta < 1 + + +@pytest.mark.asyncio +async def test_list_picks_last_terminal_job(async_session): + user = await _seed_user(async_session) + flow = await _seed_flow( + async_session, + user_id=user.id, + name="alpha", + nodes=[_cron_node("CronTrigger-x")], + ) + older = datetime.now(timezone.utc) - timedelta(hours=2) + newer = datetime.now(timezone.utc) - timedelta(minutes=2) + async_session.add( + TriggerJob( + id=uuid4(), + flow_id=flow.id, + component_id="CronTrigger-x", + status=JobStatus.COMPLETED, + scheduled_at=older, + started_at=older, + finished_at=older, + attempt=1, + max_attempts=3, + ), + ) + async_session.add( + TriggerJob( + id=uuid4(), + flow_id=flow.id, + component_id="CronTrigger-x", + status=JobStatus.FAILED, + scheduled_at=newer, + started_at=newer, + finished_at=newer, + attempt=1, + max_attempts=3, + error="boom", + ), + ) + await async_session.commit() + + instances = await list_triggers_for_user(async_session, user.id) + assert len(instances) == 1 + assert instances[0].last_finished_status == JobStatus.FAILED + + +@pytest.mark.asyncio +async def test_list_skips_flows_without_data(async_session): + user = await _seed_user(async_session) + # ``data=None`` flow — should be skipped by the server-side filter. + flow = Flow(id=uuid4(), name="empty", data=None, user_id=user.id) + async_session.add(flow) + await async_session.commit() + instances = await list_triggers_for_user(async_session, user.id) + assert instances == [] diff --git a/src/backend/tests/unit/services/triggers/test_removal.py b/src/backend/tests/unit/services/triggers/test_removal.py new file mode 100644 index 0000000000..ac48d5446e --- /dev/null +++ b/src/backend/tests/unit/services/triggers/test_removal.py @@ -0,0 +1,131 @@ +"""Unit tests for the pure removal helpers (no DB, no I/O).""" + +from __future__ import annotations + +from copy import deepcopy + +from langflow.services.triggers.removal import ( + remove_all_cron_trigger_nodes, + remove_cron_trigger_node, +) + + +def _cron_node(component_id: str) -> dict: + return { + "id": component_id, + "type": "genericNode", + "data": {"type": "CronTrigger", "node": {"template": {}}}, + } + + +def _chat_node(component_id: str = "ChatInput-c") -> dict: + return { + "id": component_id, + "type": "genericNode", + "data": {"type": "ChatInput", "node": {"template": {}}}, + } + + +def _flow(nodes: list[dict], edges: list[dict] | None = None) -> dict: + return {"nodes": nodes, "edges": edges or []} + + +# --------------------------------------------------------------------------- # +# remove_cron_trigger_node (single) +# --------------------------------------------------------------------------- # + + +def test_remove_single_returns_false_for_none_flow_data(): + result, removed = remove_cron_trigger_node(None, "CronTrigger-x") + assert removed is False + assert result == {"nodes": [], "edges": []} + + +def test_remove_single_returns_false_when_component_not_present(): + flow = _flow([_chat_node()]) + result, removed = remove_cron_trigger_node(flow, "CronTrigger-x") + assert removed is False + assert result["nodes"] == flow["nodes"] + + +def test_remove_single_does_not_match_a_non_cron_node_with_same_id(): + """Safety: id collisions never reach non-CronTrigger nodes. + + Even if another node accidentally has an id colliding with the target, + only nodes whose ``data.type`` is ``CronTrigger`` should be touched. + """ + colliding = {"id": "CronTrigger-x", "type": "genericNode", "data": {"type": "ChatInput"}} + flow = _flow([colliding]) + result, removed = remove_cron_trigger_node(flow, "CronTrigger-x") + assert removed is False + assert result["nodes"] == [colliding] + + +def test_remove_single_drops_only_the_targeted_node(): + flow = _flow([_chat_node(), _cron_node("CronTrigger-x"), _cron_node("CronTrigger-y")]) + result, removed = remove_cron_trigger_node(flow, "CronTrigger-x") + assert removed is True + ids = [n["id"] for n in result["nodes"]] + assert ids == ["ChatInput-c", "CronTrigger-y"] + + +def test_remove_single_prunes_edges_referencing_the_removed_node(): + flow = _flow( + [_cron_node("CronTrigger-x"), _chat_node()], + edges=[ + {"source": "CronTrigger-x", "target": "ChatInput-c"}, + {"source": "ChatInput-c", "target": "OtherNode"}, + ], + ) + result, removed = remove_cron_trigger_node(flow, "CronTrigger-x") + assert removed is True + assert result["edges"] == [{"source": "ChatInput-c", "target": "OtherNode"}] + + +def test_remove_single_does_not_mutate_the_input(): + flow = _flow([_cron_node("CronTrigger-x")]) + snapshot = deepcopy(flow) + remove_cron_trigger_node(flow, "CronTrigger-x") + assert flow == snapshot + + +# --------------------------------------------------------------------------- # +# remove_all_cron_trigger_nodes (bulk) +# --------------------------------------------------------------------------- # + + +def test_remove_all_returns_empty_list_when_no_triggers_present(): + flow = _flow([_chat_node()]) + result, removed_ids = remove_all_cron_trigger_nodes(flow) + assert removed_ids == [] + assert result["nodes"] == flow["nodes"] + + +def test_remove_all_strips_every_cron_node(): + flow = _flow( + [_chat_node(), _cron_node("CronTrigger-1"), _cron_node("CronTrigger-2")], + edges=[ + {"source": "CronTrigger-1", "target": "ChatInput-c"}, + {"source": "CronTrigger-2", "target": "ChatInput-c"}, + {"source": "ChatInput-c", "target": "Other"}, + ], + ) + result, removed_ids = remove_all_cron_trigger_nodes(flow) + assert removed_ids == ["CronTrigger-1", "CronTrigger-2"] + assert [n["id"] for n in result["nodes"]] == ["ChatInput-c"] + assert result["edges"] == [{"source": "ChatInput-c", "target": "Other"}] + + +def test_remove_all_handles_missing_edges_gracefully(): + flow = {"nodes": [_cron_node("CronTrigger-x")]} + result, removed_ids = remove_all_cron_trigger_nodes(flow) + assert removed_ids == ["CronTrigger-x"] + assert result["nodes"] == [] + assert result["edges"] == [] + + +def test_remove_all_returns_input_when_node_list_is_malformed(): + flow = {"nodes": "not-a-list"} + result, removed_ids = remove_all_cron_trigger_nodes(flow) + assert removed_ids == [] + assert result == {"nodes": "not-a-list"}