mirror of
https://github.com/hanzoai/flow.git
synced 2026-10-17 01:29:18 +00:00
fix: Add plugin to load routes (#11759)
* add plugin to load routes * Update src/backend/base/langflow/main.py Co-authored-by: Gabriel Luiz Freitas Almeida <[email protected]> * Update src/backend/base/langflow/main.py Co-authored-by: Gabriel Luiz Freitas Almeida <[email protected]> * avoid overwriting exisiting routes * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * Create tables needed for SSO * update migration phase * update sso path and fix ruff errors * resolve review comments * add more tests * [autofix.ci] apply automated fixes * fix ruff errors * [autofix.ci] apply automated fixes * fix ruff errors * fix: reduce Node.js heap size to 4GB in Docker builds to prevent OOM The Vite frontend build was configured with --max-old-space-size=12288 (12GB), which exceeds available RAM on ARM64 CI runners, causing the build process to be OOM-killed during the transform phase. Reduced to 4GB (4096MB) which is sufficient for the Vite build and prevents OOM kills in memory-constrained Docker BuildKit environments. * fix: avoid redundant recursive chown on /app in backend Dockerfile The recursive chown -R on /app was re-owning the entire .venv (~2.6GB, 40k+ files) which was already correctly owned via COPY --chown=1000:0. This was causing the build to be killed on ARM64 runners. Changed to non-recursive chown on /app since only the directory itself needs ownership set. /app/data still gets recursive chown (it's empty). * fix: add Docker cleanup between image builds to prevent disk full The 40GB ARM64 runner runs out of disk when building 3 Docker images sequentially. Each image (main ~8GB layers, backend ~5GB, frontend) accumulates build cache and layers that exhaust the disk. Added cleanup steps between builds that: - Remove the tested image (no longer needed) - Prune all unused Docker data and buildx cache - Log disk usage before/after for debugging --------- Co-authored-by: Gabriel Luiz Freitas Almeida <[email protected]> Co-authored-by: Hanzo Dev <[email protected]> Co-authored-by: vijay kumar katuri <[email protected]>
This commit is contained in:
Himavarsha
committed by
GitHub
GitHub
co-authored by
Gabriel Luiz Freitas Almeida
Hanzo Dev
vijay kumar katuri
Gabriel Luiz Freitas Almeida
Hanzo Dev
vijay kumar katuri
parent
1b570d75da
commit
b2a3162b60
+90
@@ -0,0 +1,90 @@
|
||||
"""add SSO plugin tables sso_user_profile and sso_config
|
||||
|
||||
Revision ID: 2ef4ca4016a3
|
||||
Revises: 369268b9af8b
|
||||
Create Date: 2026-02-22 23:15:53.139035
|
||||
|
||||
Phase: EXPAND
|
||||
"""
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
import sqlmodel
|
||||
from alembic import op
|
||||
|
||||
from langflow.utils import migration
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "2ef4ca4016a3"
|
||||
down_revision: str | None = "369268b9af8b"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
if not migration.table_exists("sso_config", conn):
|
||||
op.create_table(
|
||||
"sso_config",
|
||||
sa.Column("id", sa.Uuid(), nullable=False),
|
||||
sa.Column("provider", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("provider_name", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("enabled", sa.Boolean(), nullable=False),
|
||||
sa.Column("enforce_sso", sa.Boolean(), nullable=False),
|
||||
sa.Column("client_id", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("client_secret_encrypted", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("discovery_url", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("redirect_uri", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("scopes", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("email_claim", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("username_claim", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("user_id_claim", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("token_endpoint", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("authorization_endpoint", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("jwks_uri", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("issuer", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=False),
|
||||
sa.Column("created_by", sa.Uuid(), nullable=True),
|
||||
sa.ForeignKeyConstraint(["created_by"], ["user.id"], ondelete="SET NULL"),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
if not migration.table_exists("sso_user_profile", conn):
|
||||
op.create_table(
|
||||
"sso_user_profile",
|
||||
sa.Column("id", sa.Uuid(), nullable=False),
|
||||
sa.Column("user_id", sa.Uuid(), nullable=False),
|
||||
sa.Column("sso_provider", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("sso_user_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column("email", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
|
||||
sa.Column("sso_last_login_at", sa.DateTime(), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
with op.batch_alter_table("sso_user_profile", schema=None) as batch_op:
|
||||
batch_op.create_index(batch_op.f("ix_sso_user_profile_email"), ["email"], unique=False)
|
||||
batch_op.create_index(
|
||||
batch_op.f("uq_sso_user_profile_provider_user"),
|
||||
["sso_provider", "sso_user_id"],
|
||||
unique=True,
|
||||
)
|
||||
batch_op.create_index(batch_op.f("ix_sso_user_profile_user_id"), ["user_id"], unique=True)
|
||||
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
if migration.table_exists("sso_user_profile", conn):
|
||||
with op.batch_alter_table("sso_user_profile", schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f("ix_sso_user_profile_user_id"))
|
||||
batch_op.drop_index(batch_op.f("uq_sso_user_profile_provider_user"))
|
||||
batch_op.drop_index(batch_op.f("ix_sso_user_profile_email"))
|
||||
op.drop_table("sso_user_profile")
|
||||
if migration.table_exists("sso_config", conn):
|
||||
op.drop_table("sso_config")
|
||||
# ### end Alembic commands ###
|
||||
@@ -38,6 +38,7 @@ from langflow.initial_setup.setup import (
|
||||
sync_flows_from_fs,
|
||||
)
|
||||
from langflow.middleware import ContentSizeLimitMiddleware
|
||||
from langflow.plugin_routes import load_plugin_routes
|
||||
from langflow.services.deps import (
|
||||
get_queue_service,
|
||||
get_service,
|
||||
@@ -518,6 +519,12 @@ def create_app():
|
||||
app.include_router(health_check_router)
|
||||
app.include_router(log_router)
|
||||
|
||||
# --- plugin hook ---
|
||||
# Discover and register additional routers provided by plugins
|
||||
# via the ``langflow.plugins`` entry-point group.
|
||||
# Each entry point must expose a callable ``register(app: FastAPI) -> None``.
|
||||
load_plugin_routes(app)
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def exception_handler(_request: Request, exc: Exception):
|
||||
if isinstance(exc, HTTPException):
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
"""Plugin route discovery and registration with conflict protection.
|
||||
|
||||
Plugins register via the ``langflow.plugins`` entry-point group. They receive
|
||||
a wrapper so they cannot overwrite or shadow existing Langflow routes.
|
||||
"""
|
||||
|
||||
from importlib.metadata import entry_points
|
||||
|
||||
from fastapi import FastAPI
|
||||
from lfx.log.logger import logger
|
||||
|
||||
|
||||
def _get_route_keys(app: FastAPI) -> set[tuple[str, str]]:
|
||||
"""Collect (path, method) for all routes already on the app.
|
||||
|
||||
Used to build the reserved set before loading plugins so that plugin
|
||||
routes cannot overwrite or shadow existing Langflow routes.
|
||||
"""
|
||||
keys: set[tuple[str, str]] = set()
|
||||
for route in app.router.routes:
|
||||
if hasattr(route, "path") and hasattr(route, "methods"):
|
||||
for method in route.methods:
|
||||
if method != "HEAD": # often same as GET
|
||||
keys.add((route.path, method))
|
||||
elif hasattr(route, "path") and hasattr(route, "path_regex"):
|
||||
# Mount or similar: reserve path for all methods
|
||||
keys.add((route.path, "*"))
|
||||
return keys
|
||||
|
||||
|
||||
class _PluginAppWrapper:
|
||||
"""Wrapper around the real FastAPI app that only allows adding routes.
|
||||
|
||||
- Rejects adding a route if (path, method) is already reserved (no shadowing).
|
||||
- Does not expose router/routes so plugins cannot remove or reorder routes.
|
||||
"""
|
||||
|
||||
def __init__(self, app: FastAPI, reserved: set[tuple[str, str]]) -> None:
|
||||
self._app = app
|
||||
self._reserved = set(reserved)
|
||||
|
||||
def _check_and_reserve(self, path: str, methods: set[str]) -> None:
|
||||
for method in methods:
|
||||
if method == "HEAD":
|
||||
continue
|
||||
key = (path, method)
|
||||
if key in self._reserved:
|
||||
msg = f"Plugin route conflicts with existing route: {path} [{method}]"
|
||||
raise ValueError(msg)
|
||||
self._reserved.add(key)
|
||||
|
||||
def include_router(self, router, prefix: str = "", **kwargs) -> None:
|
||||
# Effective prefix: include_router(prefix=) + router's own prefix (e.g. APIRouter(prefix="/sso"))
|
||||
router_prefix = getattr(router, "prefix", "") or ""
|
||||
base = (prefix or "") + router_prefix
|
||||
|
||||
for route in router.routes:
|
||||
if hasattr(route, "path") and hasattr(route, "methods"):
|
||||
full_path = base + route.path
|
||||
self._check_and_reserve(full_path, set(route.methods))
|
||||
elif hasattr(route, "path"):
|
||||
full_path = base + route.path
|
||||
self._check_and_reserve(full_path, {"*"})
|
||||
self._app.include_router(router, prefix=prefix, **kwargs)
|
||||
|
||||
def get(self, path: str, **kwargs):
|
||||
self._check_and_reserve(path, {"GET"})
|
||||
return self._app.get(path, **kwargs)
|
||||
|
||||
def post(self, path: str, **kwargs):
|
||||
self._check_and_reserve(path, {"POST"})
|
||||
return self._app.post(path, **kwargs)
|
||||
|
||||
def put(self, path: str, **kwargs):
|
||||
self._check_and_reserve(path, {"PUT"})
|
||||
return self._app.put(path, **kwargs)
|
||||
|
||||
def delete(self, path: str, **kwargs):
|
||||
self._check_and_reserve(path, {"DELETE"})
|
||||
return self._app.delete(path, **kwargs)
|
||||
|
||||
def patch(self, path: str, **kwargs):
|
||||
self._check_and_reserve(path, {"PATCH"})
|
||||
return self._app.patch(path, **kwargs)
|
||||
|
||||
def on_event(self, event_type: str):
|
||||
return self._app.on_event(event_type)
|
||||
|
||||
# Expose other delegates plugins might need (no route mutation)
|
||||
@property
|
||||
def openapi(self):
|
||||
return self._app.openapi
|
||||
|
||||
def add_api_route(self, path: str, endpoint, **kwargs):
|
||||
methods = kwargs.get("methods", ["GET"])
|
||||
self._check_and_reserve(path, set(methods))
|
||||
return self._app.add_api_route(path, endpoint, **kwargs)
|
||||
|
||||
|
||||
def load_plugin_routes(app: FastAPI) -> None:
|
||||
"""Discover and register additional routers from enterprise plugins.
|
||||
|
||||
Plugins register themselves via the ``langflow.plugins`` entry-point group.
|
||||
Each entry point must expose a callable with the signature::
|
||||
|
||||
def register(app: FastAPI) -> None: ...
|
||||
|
||||
"""
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
|
||||
eps = entry_points(group="langflow.plugins")
|
||||
for ep in sorted(eps, key=lambda e: e.name):
|
||||
try:
|
||||
plugin_register = ep.load()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.error(
|
||||
"Failed to load plugin entry point '%s' (broken import or missing dependency)",
|
||||
ep.name,
|
||||
exc_info=True,
|
||||
)
|
||||
continue
|
||||
|
||||
try:
|
||||
plugin_register(wrapper)
|
||||
logger.info(f"Loaded plugin: {ep.name}")
|
||||
except ValueError as e:
|
||||
logger.warning(
|
||||
"Plugin '%s' rejected (route conflict): %s",
|
||||
ep.name,
|
||||
e,
|
||||
exc_info=True,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.error(
|
||||
"Plugin '%s' failed during registration",
|
||||
ep.name,
|
||||
exc_info=True,
|
||||
)
|
||||
@@ -1,4 +1,5 @@
|
||||
from .api_key import ApiKey
|
||||
from .auth import SSOConfig, SSOUserProfile
|
||||
from .file import File
|
||||
from .flow import Flow
|
||||
from .folder import Folder
|
||||
@@ -15,6 +16,8 @@ __all__ = [
|
||||
"Folder",
|
||||
"Job",
|
||||
"MessageTable",
|
||||
"SSOConfig",
|
||||
"SSOUserProfile",
|
||||
"TransactionTable",
|
||||
"User",
|
||||
"Variable",
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
from .sso import SSOConfig, SSOUserProfile
|
||||
|
||||
__all__ = [
|
||||
"SSOConfig",
|
||||
"SSOUserProfile",
|
||||
]
|
||||
@@ -0,0 +1,76 @@
|
||||
"""SSO plugin tables.
|
||||
|
||||
These tables are used by the SSO plugin for identity and provider configuration.
|
||||
Migrations are managed by Langflow (OSS); the plugin must not create or
|
||||
migrate these tables.
|
||||
|
||||
Plugins must use these tables via the models exported from
|
||||
``langflow.services.database.models`` (e.g. ``SSOUserProfile``, ``SSOConfig``).
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from uuid import uuid4
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import Column, ForeignKey, UniqueConstraint
|
||||
from sqlmodel import Field, SQLModel
|
||||
|
||||
from langflow.schema.serialize import UUIDstr
|
||||
|
||||
|
||||
class SSOUserProfile(SQLModel, table=True): # type: ignore[call-arg]
|
||||
"""SSO profile per user. Used by the SSO plugin for JIT provisioning and login."""
|
||||
|
||||
__tablename__ = "sso_user_profile"
|
||||
__table_args__ = (UniqueConstraint("sso_provider", "sso_user_id", name="uq_sso_user_profile_provider_user"),)
|
||||
|
||||
id: UUIDstr = Field(default_factory=uuid4, primary_key=True)
|
||||
user_id: UUIDstr = Field(
|
||||
sa_column=Column(
|
||||
sa.Uuid(),
|
||||
ForeignKey("user.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
unique=True,
|
||||
index=True,
|
||||
)
|
||||
)
|
||||
sso_provider: str = Field()
|
||||
sso_user_id: str = Field()
|
||||
email: str | None = Field(default=None, index=True)
|
||||
sso_last_login_at: datetime | None = Field(default=None)
|
||||
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
|
||||
|
||||
class SSOConfig(SQLModel, table=True): # type: ignore[call-arg]
|
||||
"""SSO provider configuration (persisted in DB). Used by the SSO plugin."""
|
||||
|
||||
__tablename__ = "sso_config"
|
||||
|
||||
id: UUIDstr = Field(default_factory=uuid4, primary_key=True)
|
||||
provider: str = Field(description="oidc, saml, ldap")
|
||||
provider_name: str = Field()
|
||||
enabled: bool = Field(default=True)
|
||||
enforce_sso: bool = Field(default=False)
|
||||
client_id: str | None = Field(default=None)
|
||||
client_secret_encrypted: str | None = Field(default=None)
|
||||
discovery_url: str | None = Field(default=None)
|
||||
redirect_uri: str | None = Field(default=None)
|
||||
scopes: str | None = Field(default="openid email profile")
|
||||
email_claim: str = Field(default="email")
|
||||
username_claim: str = Field(default="preferred_username")
|
||||
user_id_claim: str = Field(default="sub")
|
||||
token_endpoint: str | None = Field(default=None)
|
||||
authorization_endpoint: str | None = Field(default=None)
|
||||
jwks_uri: str | None = Field(default=None)
|
||||
issuer: str | None = Field(default=None)
|
||||
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
created_by: UUIDstr | None = Field(
|
||||
default=None,
|
||||
sa_column=Column(
|
||||
sa.Uuid(),
|
||||
ForeignKey("user.id", ondelete="SET NULL"),
|
||||
nullable=True,
|
||||
),
|
||||
)
|
||||
@@ -1,142 +1,144 @@
|
||||
"""Integration tests for database service with Windows + PostgreSQL fix.
|
||||
"""Integration tests for database service with Windows + PostgreSQL event loop.
|
||||
|
||||
Tests that the database service properly handles event loop configuration
|
||||
across different platforms and database types.
|
||||
Uses real settings objects and real SQLite engine where possible; only
|
||||
platform.system is patched to simulate Windows vs non-Windows. Project
|
||||
convention: avoid mocking in tests.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from langflow.services.database.service import DatabaseService
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
|
||||
class _TestSettings:
|
||||
"""Minimal settings-like object (no mocks) for DatabaseService init."""
|
||||
|
||||
database_url = "sqlite+aiosqlite:///:memory:"
|
||||
database_connection_retry = False
|
||||
sqlite_pragmas = {}
|
||||
db_driver_connection_settings = None
|
||||
db_connection_settings = {}
|
||||
db_connect_timeout = 5
|
||||
alembic_log_to_stdout = True
|
||||
alembic_log_file = "alembic.log"
|
||||
use_noop_database = False
|
||||
model_fields_set = set()
|
||||
|
||||
|
||||
class _TestAuthSettings:
|
||||
AUTO_LOGIN = False
|
||||
SUPERUSER = "admin"
|
||||
|
||||
|
||||
class _TestSettingsService:
|
||||
"""Minimal settings service (no mocks) for DatabaseService."""
|
||||
|
||||
settings = _TestSettings()
|
||||
auth_settings = _TestAuthSettings()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_settings_service():
|
||||
"""Real settings-like object with required attributes (no MagicMock)."""
|
||||
return _TestSettingsService()
|
||||
|
||||
|
||||
class TestDatabaseServiceWindowsPostgres:
|
||||
"""Test database service with Windows + PostgreSQL event loop configuration."""
|
||||
"""Database service with Windows + PostgreSQL event loop configuration."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings_service(self):
|
||||
"""Create a mock settings service for testing."""
|
||||
mock_service = MagicMock()
|
||||
mock_service.settings.database_url = "sqlite:///test.db"
|
||||
mock_service.settings.database_connection_retry = False
|
||||
mock_service.settings.sqlite_pragmas = {}
|
||||
mock_service.settings.db_driver_connection_settings = None
|
||||
mock_service.settings.db_connection_settings = {}
|
||||
mock_service.settings.alembic_log_to_stdout = True
|
||||
mock_service.settings.alembic_log_file = "alembic.log"
|
||||
return mock_service
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "postgresql://user:pass@localhost/db"}, clear=True)
|
||||
@patch("langflow.services.database.service.create_async_engine")
|
||||
@patch("langflow.services.database.service.configure_windows_postgres_event_loop")
|
||||
def test_windows_postgresql_configures_event_loop(
|
||||
self, mock_configure, mock_create_engine, mock_platform, mock_settings_service
|
||||
):
|
||||
"""Test that Windows + PostgreSQL configures the event loop correctly."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_postgresql_calls_configure_with_source(self, mock_platform, test_settings_service):
|
||||
"""Windows + PostgreSQL: configure_windows_postgres_event_loop is called with source."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_settings_service.settings.database_url = "postgresql://user:pass@localhost/db"
|
||||
mock_create_engine.return_value = MagicMock()
|
||||
mock_configure.return_value = True
|
||||
test_settings_service.settings.database_url = "postgresql://user:pass@localhost/db"
|
||||
with patch("langflow.services.database.service.create_async_engine") as create_engine:
|
||||
create_engine.return_value = create_async_engine(
|
||||
"sqlite+aiosqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
_ = DatabaseService(test_settings_service)
|
||||
create_engine.assert_called_once()
|
||||
mock_platform.assert_called()
|
||||
|
||||
_ = DatabaseService(mock_settings_service)
|
||||
mock_configure.assert_called_once_with(source="database_service")
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {}, clear=True)
|
||||
@patch("langflow.services.database.service.create_async_engine")
|
||||
def test_linux_postgresql_no_event_loop_change(self, mock_create_engine, mock_platform, mock_settings_service):
|
||||
"""Test that Linux + PostgreSQL doesn't change event loop."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_linux_sqlite_real_engine_no_event_loop_change(self, mock_platform, test_settings_service):
|
||||
"""Linux + SQLite: real engine creation, event loop unchanged (no mock of create_async_engine)."""
|
||||
mock_platform.return_value = "Linux"
|
||||
mock_settings_service.settings.database_url = "postgresql://user:pass@localhost/db"
|
||||
mock_create_engine.return_value = MagicMock()
|
||||
|
||||
test_settings_service.settings.database_url = "sqlite+aiosqlite://"
|
||||
original_policy = asyncio.get_event_loop_policy()
|
||||
_ = DatabaseService(mock_settings_service)
|
||||
|
||||
# Policy should remain unchanged
|
||||
service = DatabaseService(test_settings_service)
|
||||
|
||||
assert asyncio.get_event_loop_policy() is original_policy
|
||||
assert service.async_session_maker is not None
|
||||
assert service.database_url == "sqlite+aiosqlite://"
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {}, clear=True)
|
||||
@patch("langflow.services.database.service.create_async_engine")
|
||||
def test_macos_postgresql_no_event_loop_change(self, mock_create_engine, mock_platform, mock_settings_service):
|
||||
"""Test that macOS + PostgreSQL doesn't change event loop."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_macos_sqlite_real_engine_no_event_loop_change(self, mock_platform, test_settings_service):
|
||||
"""MacOS + SQLite: real engine, event loop unchanged."""
|
||||
mock_platform.return_value = "Darwin"
|
||||
mock_settings_service.settings.database_url = "postgresql://user:pass@localhost/db"
|
||||
mock_create_engine.return_value = MagicMock()
|
||||
|
||||
test_settings_service.settings.database_url = "sqlite+aiosqlite://"
|
||||
original_policy = asyncio.get_event_loop_policy()
|
||||
_ = DatabaseService(mock_settings_service)
|
||||
|
||||
# Policy should remain unchanged
|
||||
service = DatabaseService(test_settings_service)
|
||||
|
||||
assert asyncio.get_event_loop_policy() is original_policy
|
||||
assert service.async_session_maker is not None
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {}, clear=True)
|
||||
@patch("langflow.services.database.service.create_async_engine")
|
||||
@patch("langflow.services.database.service.configure_windows_postgres_event_loop")
|
||||
def test_windows_sqlite_no_event_loop_change(
|
||||
self, mock_configure, mock_create_engine, mock_platform, mock_settings_service
|
||||
):
|
||||
"""Test that Windows + SQLite doesn't change event loop."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_sqlite_real_engine_configure_called(self, mock_platform, test_settings_service):
|
||||
"""Windows + SQLite: real engine; configure is still called (returns False)."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_settings_service.settings.database_url = "sqlite:///test.db"
|
||||
mock_create_engine.return_value = MagicMock()
|
||||
mock_configure.return_value = False
|
||||
test_settings_service.settings.database_url = "sqlite+aiosqlite://"
|
||||
original_policy = asyncio.get_event_loop_policy()
|
||||
|
||||
_ = DatabaseService(mock_settings_service)
|
||||
mock_configure.assert_called_once_with(source="database_service")
|
||||
service = DatabaseService(test_settings_service)
|
||||
|
||||
def test_database_url_sanitization(self, mock_settings_service):
|
||||
"""Test that database URLs are properly sanitized."""
|
||||
assert asyncio.get_event_loop_policy() is original_policy
|
||||
assert service.async_session_maker is not None
|
||||
|
||||
def test_database_url_sanitization_real_settings(self, test_settings_service):
|
||||
"""Database URLs are sanitized (real settings, only create_engine patched for speed)."""
|
||||
test_cases = [
|
||||
("sqlite:///test.db", "sqlite+aiosqlite:///test.db"),
|
||||
("postgresql://user:pass@localhost/db", "postgresql+psycopg://user:pass@localhost/db"),
|
||||
("postgres://user:pass@localhost/db", "postgresql+psycopg://user:pass@localhost/db"),
|
||||
]
|
||||
|
||||
with patch("langflow.services.database.service.create_async_engine") as mock_create_engine:
|
||||
mock_create_engine.return_value = MagicMock()
|
||||
|
||||
with patch("langflow.services.database.service.create_async_engine") as create_engine:
|
||||
create_engine.return_value = create_async_engine(
|
||||
"sqlite+aiosqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
for input_url, expected_url in test_cases:
|
||||
mock_settings_service.settings.database_url = input_url
|
||||
service = DatabaseService(mock_settings_service)
|
||||
test_settings_service.settings.database_url = input_url
|
||||
service = DatabaseService(test_settings_service)
|
||||
assert service.database_url == expected_url
|
||||
|
||||
@patch("platform.system")
|
||||
def test_docker_environment_compatibility(self, mock_platform, mock_settings_service):
|
||||
"""Test that Docker environments work correctly."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_async_session_maker_available_after_init(self, mock_platform, test_settings_service):
|
||||
"""After init with SQLite, async_session_maker is usable (real engine)."""
|
||||
mock_platform.return_value = "Linux"
|
||||
os.environ["DOCKER_CONTAINER"] = "true"
|
||||
mock_settings_service.settings.database_url = "postgresql://postgres:5432/langflow"
|
||||
test_settings_service.settings.database_url = "sqlite+aiosqlite://"
|
||||
|
||||
with patch("langflow.services.database.service.create_async_engine") as mock_create_engine:
|
||||
mock_create_engine.return_value = MagicMock()
|
||||
service = DatabaseService(test_settings_service)
|
||||
|
||||
# Should not raise any errors
|
||||
service = DatabaseService(mock_settings_service)
|
||||
assert service.database_url == "postgresql+psycopg://postgres:5432/langflow"
|
||||
assert service.async_session_maker is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_operations_work_after_configuration(self, mock_settings_service):
|
||||
"""Test that async operations work correctly after event loop configuration."""
|
||||
mock_settings_service.settings.database_url = "sqlite:///test.db"
|
||||
async def test_async_operations_work_after_init(self, test_settings_service):
|
||||
"""Async session maker works after init (real engine, no mock)."""
|
||||
with patch("langflow.helpers.windows_postgres_helper.platform.system") as mock_platform:
|
||||
mock_platform.return_value = "Linux"
|
||||
service = DatabaseService(test_settings_service)
|
||||
|
||||
with patch("langflow.services.database.service.create_async_engine") as mock_create_engine:
|
||||
mock_engine = MagicMock()
|
||||
mock_create_engine.return_value = mock_engine
|
||||
|
||||
service = DatabaseService(mock_settings_service)
|
||||
|
||||
# Test that async session maker is properly configured
|
||||
assert service.async_session_maker is not None
|
||||
|
||||
# Simulate an async operation
|
||||
async def test_async():
|
||||
return True
|
||||
|
||||
result = await test_async()
|
||||
assert result is True
|
||||
async with service.async_session_maker() as session:
|
||||
assert session is not None
|
||||
await session.execute(text("SELECT 1"))
|
||||
|
||||
@@ -0,0 +1,248 @@
|
||||
"""Tests for plugin route discovery and conflict protection.
|
||||
|
||||
Ensures that plugins loaded via the langflow.plugins entry-point group
|
||||
cannot overwrite or shadow existing Langflow routes.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.routing import APIRouter
|
||||
from langflow.plugin_routes import (
|
||||
_get_route_keys,
|
||||
_PluginAppWrapper,
|
||||
load_plugin_routes,
|
||||
)
|
||||
|
||||
|
||||
class TestGetRouteKeys:
|
||||
"""Tests for _get_route_keys."""
|
||||
|
||||
def test_returns_default_routes_for_app_with_no_custom_routes(self):
|
||||
"""App with no custom routes still has FastAPI default OpenAPI routes in the set."""
|
||||
app = FastAPI()
|
||||
keys = _get_route_keys(app)
|
||||
# FastAPI adds /docs, /openapi.json, /redoc, etc. by default
|
||||
assert ("/openapi.json", "GET") in keys
|
||||
assert ("/docs", "GET") in keys
|
||||
|
||||
def test_collects_route_path_and_methods(self):
|
||||
"""All (path, method) pairs from app routes are collected."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return "ok"
|
||||
|
||||
@app.post("/api/action")
|
||||
def action():
|
||||
return None
|
||||
|
||||
keys = _get_route_keys(app)
|
||||
assert ("/health", "GET") in keys
|
||||
assert ("/api/action", "POST") in keys
|
||||
# HEAD is excluded to avoid false conflicts with GET
|
||||
assert ("/health", "HEAD") not in keys
|
||||
|
||||
def test_includes_mounts_as_wildcard(self):
|
||||
"""Mounts are reserved as (path, '*')."""
|
||||
app = FastAPI()
|
||||
app.mount("/static", MagicMock())
|
||||
keys = _get_route_keys(app)
|
||||
assert any(path == "/static" and method == "*" for path, method in keys)
|
||||
|
||||
|
||||
class TestPluginAppWrapper:
|
||||
"""Tests for _PluginAppWrapper: add-only, no overwrite."""
|
||||
|
||||
def test_allows_non_conflicting_route(self):
|
||||
"""Wrapper allows adding a route that does not conflict with reserved set."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/existing")
|
||||
def existing():
|
||||
return "core"
|
||||
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
|
||||
@wrapper.get("/plugin-only")
|
||||
def plugin_only():
|
||||
return "plugin"
|
||||
|
||||
# Real app should have both routes
|
||||
keys_after = _get_route_keys(app)
|
||||
assert ("/existing", "GET") in keys_after
|
||||
assert ("/plugin-only", "GET") in keys_after
|
||||
|
||||
def test_raises_on_conflicting_get(self):
|
||||
"""Adding a route with same (path, method) as reserved raises ValueError."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/api/login")
|
||||
def core_login():
|
||||
return "core"
|
||||
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
|
||||
with pytest.raises(ValueError, match="Plugin route conflicts with existing route: /api/login \\[GET\\]"):
|
||||
wrapper.get("/api/login")(lambda: "plugin")
|
||||
|
||||
def test_raises_on_conflicting_include_router(self):
|
||||
"""include_router with a route that conflicts with reserved raises ValueError."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/api/v1/flow")
|
||||
def core_flow():
|
||||
return "core"
|
||||
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.get("/flow")
|
||||
def plugin_flow():
|
||||
return "plugin"
|
||||
|
||||
with pytest.raises(ValueError, match="Plugin route conflicts with existing route"):
|
||||
wrapper.include_router(router, prefix="/api/v1")
|
||||
|
||||
def test_include_router_allows_non_conflicting_prefix(self):
|
||||
"""include_router with distinct prefix succeeds and reserves new paths."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return "ok"
|
||||
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
|
||||
router = APIRouter(prefix="/sso")
|
||||
|
||||
@router.get("/login")
|
||||
def sso_login():
|
||||
return "sso"
|
||||
|
||||
wrapper.include_router(router, prefix="/api/v1")
|
||||
|
||||
keys_after = _get_route_keys(app)
|
||||
assert ("/api/v1/sso/login", "GET") in keys_after
|
||||
|
||||
def test_on_event_delegates_without_conflict_check(self):
|
||||
"""on_event is delegated to the real app and does not check route conflicts."""
|
||||
app = FastAPI()
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
called = []
|
||||
|
||||
@wrapper.on_event("startup")
|
||||
def on_startup():
|
||||
called.append(True)
|
||||
|
||||
# Trigger lifespan startup to ensure handler is registered
|
||||
assert hasattr(wrapper._app, "router")
|
||||
|
||||
def test_second_plugin_cannot_overwrite_first_plugin_route(self):
|
||||
"""Reserved set is cumulative; second plugin cannot register same path."""
|
||||
app = FastAPI()
|
||||
reserved = _get_route_keys(app)
|
||||
wrapper = _PluginAppWrapper(app, reserved)
|
||||
|
||||
# First "plugin" adds /api/v1/sso/login
|
||||
wrapper.get("/api/v1/sso/login")(lambda: "first")
|
||||
|
||||
# Second plugin trying same path should fail
|
||||
with pytest.raises(ValueError, match=r"conflicts with existing route.*/api/v1/sso/login"):
|
||||
wrapper.get("/api/v1/sso/login")(lambda: "second")
|
||||
|
||||
|
||||
class TestLoadPluginRoutes:
|
||||
"""Tests for load_plugin_routes with mocked entry_points."""
|
||||
|
||||
def test_no_crash_when_no_plugins(self):
|
||||
"""When there are no entry points, load_plugin_routes does not crash."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return "ok"
|
||||
|
||||
with patch("langflow.plugin_routes.entry_points", return_value=[]):
|
||||
load_plugin_routes(app)
|
||||
|
||||
keys = _get_route_keys(app)
|
||||
assert ("/health", "GET") in keys
|
||||
|
||||
def test_plugin_that_registers_route_is_loaded(self):
|
||||
"""A plugin that registers a non-conflicting route is loaded successfully."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return "ok"
|
||||
|
||||
def register(app_like):
|
||||
@app_like.get("/api/v1/sso/login")
|
||||
def login():
|
||||
return "sso"
|
||||
|
||||
ep = MagicMock()
|
||||
ep.name = "enterprise"
|
||||
ep.load.return_value = register
|
||||
|
||||
with patch("langflow.plugin_routes.entry_points", return_value=[ep]):
|
||||
load_plugin_routes(app)
|
||||
|
||||
keys = _get_route_keys(app)
|
||||
assert ("/api/v1/sso/login", "GET") in keys
|
||||
|
||||
def test_plugin_with_conflict_is_skipped_app_continues(self):
|
||||
"""When a plugin tries to register a conflicting route, that plugin is skipped."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/api/v1/flow")
|
||||
def core_flow():
|
||||
return "core"
|
||||
|
||||
def conflicting_register(app_like):
|
||||
app_like.get("/api/v1/flow")(lambda: "plugin")
|
||||
|
||||
ep = MagicMock()
|
||||
ep.name = "bad_plugin"
|
||||
ep.load.return_value = conflicting_register
|
||||
|
||||
with patch("langflow.plugin_routes.entry_points", return_value=[ep]):
|
||||
load_plugin_routes(app)
|
||||
|
||||
# Core route must still be the only one at that path
|
||||
routes_at_path = [
|
||||
r for r in app.router.routes if getattr(r, "path", None) == "/api/v1/flow" and hasattr(r, "methods")
|
||||
]
|
||||
assert len(routes_at_path) == 1
|
||||
|
||||
def test_plugin_that_raises_is_skipped_app_continues(self):
|
||||
"""When a plugin raises a non-ValueError exception, it is skipped."""
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return "ok"
|
||||
|
||||
def broken_register(_app_like):
|
||||
err_msg = "plugin broken"
|
||||
raise RuntimeError(err_msg)
|
||||
|
||||
ep = MagicMock()
|
||||
ep.name = "broken_plugin"
|
||||
ep.load.return_value = broken_register
|
||||
|
||||
with patch("langflow.plugin_routes.entry_points", return_value=[ep]):
|
||||
load_plugin_routes(app)
|
||||
|
||||
# App still has core route
|
||||
keys = _get_route_keys(app)
|
||||
assert ("/health", "GET") in keys
|
||||
@@ -0,0 +1,187 @@
|
||||
"""Tests for SSO plugin models against a real database.
|
||||
|
||||
No mocks: uses in-memory SQLite with foreign keys enabled to verify
|
||||
CASCADE delete, unique constraints, and default values.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from langflow.services.database.models.auth.sso import SSOConfig, SSOUserProfile
|
||||
from langflow.services.database.models.user.model import User
|
||||
from sqlalchemy import event
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
from sqlalchemy.pool import StaticPool
|
||||
from sqlmodel import SQLModel, select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
# Placeholder for User.password in tests (not a real secret)
|
||||
_TEST_PASSWORD = "hashed" # noqa: S105
|
||||
|
||||
|
||||
@pytest.fixture(name="sso_db_engine")
|
||||
def sso_db_engine():
|
||||
"""Async engine with SQLite and foreign keys enabled (real DB, no mocks)."""
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
|
||||
@event.listens_for(engine.sync_engine, "connect")
|
||||
def set_sqlite_pragma(dbapi_connection, connection_record): # noqa: ARG001
|
||||
cursor = dbapi_connection.cursor()
|
||||
cursor.execute("PRAGMA foreign_keys=ON")
|
||||
cursor.close()
|
||||
|
||||
return engine
|
||||
|
||||
|
||||
@pytest.fixture(name="sso_async_session")
|
||||
async def sso_async_session(sso_db_engine):
|
||||
"""Async session with SSO and User tables created (real DB)."""
|
||||
async with sso_db_engine.begin() as conn:
|
||||
await conn.run_sync(SQLModel.metadata.create_all)
|
||||
async with AsyncSession(sso_db_engine, expire_on_commit=False) as session:
|
||||
yield session
|
||||
async with sso_db_engine.begin() as conn:
|
||||
await conn.run_sync(SQLModel.metadata.drop_all)
|
||||
await sso_db_engine.dispose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSSOUserProfile:
|
||||
"""SSOUserProfile model tests against real database."""
|
||||
|
||||
async def test_create_and_read_sso_user_profile(self, sso_async_session):
|
||||
"""Create and read SSOUserProfile records."""
|
||||
user = User(username="sso_user", password=_TEST_PASSWORD)
|
||||
sso_async_session.add(user)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(user)
|
||||
|
||||
profile = SSOUserProfile(
|
||||
user_id=user.id,
|
||||
sso_provider="oidc",
|
||||
sso_user_id="sub-123",
|
||||
email="[email protected]",
|
||||
)
|
||||
sso_async_session.add(profile)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(profile)
|
||||
|
||||
assert profile.id is not None
|
||||
assert profile.user_id == user.id
|
||||
assert profile.sso_provider == "oidc"
|
||||
assert profile.sso_user_id == "sub-123"
|
||||
assert profile.email == "[email protected]"
|
||||
assert profile.created_at is not None
|
||||
assert profile.updated_at is not None
|
||||
|
||||
async def test_user_id_unique_constraint(self, sso_async_session):
|
||||
"""Cannot create two SSO profiles for the same user."""
|
||||
user = User(username="unique_user", password=_TEST_PASSWORD)
|
||||
sso_async_session.add(user)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(user)
|
||||
|
||||
sso_async_session.add(SSOUserProfile(user_id=user.id, sso_provider="oidc", sso_user_id="sub-1"))
|
||||
await sso_async_session.commit()
|
||||
|
||||
duplicate = SSOUserProfile(user_id=user.id, sso_provider="saml", sso_user_id="sub-2")
|
||||
sso_async_session.add(duplicate)
|
||||
with pytest.raises(IntegrityError, match=r"UNIQUE constraint failed|unique constraint"):
|
||||
await sso_async_session.commit()
|
||||
|
||||
async def test_composite_unique_sso_provider_sso_user_id(self, sso_async_session):
|
||||
"""Same (sso_provider, sso_user_id) cannot be used for two different users."""
|
||||
user1 = User(username="user1", password=_TEST_PASSWORD)
|
||||
user2 = User(username="user2", password=_TEST_PASSWORD)
|
||||
sso_async_session.add(user1)
|
||||
sso_async_session.add(user2)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(user1)
|
||||
await sso_async_session.refresh(user2)
|
||||
|
||||
sso_async_session.add(SSOUserProfile(user_id=user1.id, sso_provider="oidc", sso_user_id="sub-123"))
|
||||
await sso_async_session.commit()
|
||||
|
||||
duplicate = SSOUserProfile(user_id=user2.id, sso_provider="oidc", sso_user_id="sub-123")
|
||||
sso_async_session.add(duplicate)
|
||||
with pytest.raises(IntegrityError, match=r"UNIQUE constraint failed|unique constraint"):
|
||||
await sso_async_session.commit()
|
||||
|
||||
async def test_cascade_delete_when_user_deleted(self, sso_async_session):
|
||||
"""Deleting user deletes associated SSOUserProfile (CASCADE)."""
|
||||
user = User(username="cascade_user", password=_TEST_PASSWORD)
|
||||
sso_async_session.add(user)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(user)
|
||||
|
||||
profile = SSOUserProfile(user_id=user.id, sso_provider="oidc", sso_user_id="sub-cascade")
|
||||
sso_async_session.add(profile)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(profile)
|
||||
profile_id = profile.id
|
||||
|
||||
await sso_async_session.delete(user)
|
||||
await sso_async_session.commit()
|
||||
|
||||
result = await sso_async_session.exec(select(SSOUserProfile).where(SSOUserProfile.id == profile_id))
|
||||
assert result.first() is None
|
||||
|
||||
async def test_default_timestamps_set(self, sso_async_session):
|
||||
"""created_at and updated_at are set on create."""
|
||||
user = User(username="ts_user", password=_TEST_PASSWORD)
|
||||
sso_async_session.add(user)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(user)
|
||||
|
||||
profile = SSOUserProfile(user_id=user.id, sso_provider="oidc", sso_user_id="sub-ts")
|
||||
sso_async_session.add(profile)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(profile)
|
||||
|
||||
assert profile.created_at is not None
|
||||
assert profile.updated_at is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSSOConfig:
|
||||
"""SSOConfig model tests against real database."""
|
||||
|
||||
async def test_create_and_read_sso_config(self, sso_async_session):
|
||||
"""Create and read SSOConfig."""
|
||||
config = SSOConfig(
|
||||
provider="oidc",
|
||||
provider_name="Test OIDC",
|
||||
)
|
||||
sso_async_session.add(config)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(config)
|
||||
|
||||
assert config.id is not None
|
||||
assert config.provider == "oidc"
|
||||
assert config.provider_name == "Test OIDC"
|
||||
assert config.enabled is True
|
||||
assert config.enforce_sso is False
|
||||
assert config.scopes == "openid email profile"
|
||||
assert config.email_claim == "email"
|
||||
assert config.username_claim == "preferred_username"
|
||||
assert config.user_id_claim == "sub"
|
||||
assert config.created_at is not None
|
||||
assert config.updated_at is not None
|
||||
|
||||
async def test_default_values(self, sso_async_session):
|
||||
"""Default values are applied when not specified."""
|
||||
config = SSOConfig(provider="oidc", provider_name="Default Test")
|
||||
sso_async_session.add(config)
|
||||
await sso_async_session.commit()
|
||||
await sso_async_session.refresh(config)
|
||||
|
||||
assert config.enabled is True
|
||||
assert config.enforce_sso is False
|
||||
assert config.scopes == "openid email profile"
|
||||
assert config.email_claim == "email"
|
||||
assert config.username_claim == "preferred_username"
|
||||
assert config.user_id_claim == "sub"
|
||||
assert config.created_by is None
|
||||
@@ -1,16 +1,15 @@
|
||||
"""Unit tests for Windows + PostgreSQL event loop configuration.
|
||||
|
||||
These tests ensure the fix works correctly across all platforms:
|
||||
- Windows with PostgreSQL (applies fix)
|
||||
- Windows with SQLite (no fix)
|
||||
- Linux (no fix)
|
||||
- macOS (no fix)
|
||||
- Docker (no fix)
|
||||
Tests use minimal mocking: only platform.system (and env) where needed to
|
||||
simulate Windows on non-Windows CI. asyncio policy get/set are exercised
|
||||
for real. Project convention: avoid mocking in tests.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
import platform
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from langflow.helpers.windows_postgres_helper import (
|
||||
@@ -19,8 +18,25 @@ from langflow.helpers.windows_postgres_helper import (
|
||||
configure_windows_postgres_event_loop,
|
||||
)
|
||||
|
||||
# Mock class for asyncio.WindowsSelectorEventLoopPolicy (only exists on Windows)
|
||||
MockWindowsSelectorPolicy = type("WindowsSelectorEventLoopPolicy", (), {})
|
||||
|
||||
# Subclass required so set_event_loop_policy() accepts our instance (real asyncio API).
|
||||
class MockWindowsSelectorPolicy(asyncio.AbstractEventLoopPolicy):
|
||||
"""Minimal policy for tests when WindowsSelectorEventLoopPolicy is not available."""
|
||||
|
||||
def get_event_loop(self):
|
||||
return asyncio.get_event_loop()
|
||||
|
||||
def set_event_loop(self, loop):
|
||||
pass
|
||||
|
||||
def new_event_loop(self):
|
||||
return asyncio.new_event_loop()
|
||||
|
||||
def get_child_watcher(self):
|
||||
return None
|
||||
|
||||
def set_child_watcher(self, watcher):
|
||||
pass
|
||||
|
||||
|
||||
class TestWindowsPostgresHelper:
|
||||
@@ -34,137 +50,94 @@ class TestWindowsPostgresHelper:
|
||||
asyncio.set_event_loop_policy(original_policy)
|
||||
|
||||
def test_constants_defined(self):
|
||||
"""Test that required constants are properly defined."""
|
||||
"""Required constants are defined."""
|
||||
assert LANGFLOW_DATABASE_URL == "LANGFLOW_DATABASE_URL"
|
||||
assert POSTGRESQL_PREFIXES == ("postgresql", "postgres")
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {}, clear=True)
|
||||
def test_non_windows_returns_false(self, mock_platform):
|
||||
"""Test that non-Windows systems return False and don't change event loop."""
|
||||
for system in ["Linux", "Darwin", "FreeBSD"]:
|
||||
mock_platform.return_value = system
|
||||
os.environ[LANGFLOW_DATABASE_URL] = "postgresql://user:pass@localhost/db"
|
||||
|
||||
@pytest.mark.skipif(platform.system() == "Windows", reason="Only run on non-Windows")
|
||||
def test_on_non_windows_with_postgres_url_returns_false_no_mock(self):
|
||||
"""On non-Windows, function returns False without mocking platform."""
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "postgresql://user:pass@localhost/db"}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is False
|
||||
|
||||
assert result is False
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {}, clear=True)
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_without_database_url_returns_false(self, mock_platform):
|
||||
"""Test Windows without DATABASE_URL returns False."""
|
||||
"""Windows without DATABASE_URL returns False."""
|
||||
mock_platform.return_value = "Windows"
|
||||
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is False
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "sqlite:///test.db"}, clear=True)
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_with_sqlite_returns_false(self, mock_platform):
|
||||
"""Test Windows with SQLite returns False and doesn't change event loop."""
|
||||
"""Windows with SQLite URL returns False."""
|
||||
mock_platform.return_value = "Windows"
|
||||
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "sqlite:///test.db"}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is False
|
||||
|
||||
@patch.object(asyncio, "WindowsSelectorEventLoopPolicy", MockWindowsSelectorPolicy, create=True)
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "postgresql://user:pass@localhost/db"}, clear=True)
|
||||
@patch("asyncio.get_event_loop_policy")
|
||||
@patch("asyncio.set_event_loop_policy")
|
||||
def test_windows_with_postgresql_sets_policy(self, mock_set_policy, mock_get_policy, mock_platform):
|
||||
"""Test Windows with PostgreSQL sets WindowsSelectorEventLoopPolicy."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_with_postgresql_sets_policy(self, mock_platform):
|
||||
"""Windows with PostgreSQL URL sets event loop policy (real get/set_event_loop_policy)."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_policy = MagicMock()
|
||||
mock_get_policy.return_value = mock_policy
|
||||
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "postgresql://user:pass@localhost/db"}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is True
|
||||
mock_set_policy.assert_called_once()
|
||||
args = mock_set_policy.call_args[0]
|
||||
assert len(args) == 1
|
||||
assert "WindowsSelectorEventLoopPolicy" in str(args[0].__class__)
|
||||
assert isinstance(asyncio.get_event_loop_policy(), MockWindowsSelectorPolicy)
|
||||
|
||||
@patch.object(asyncio, "WindowsSelectorEventLoopPolicy", MockWindowsSelectorPolicy, create=True)
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "postgres://user:pass@localhost/db"}, clear=True)
|
||||
@patch("asyncio.get_event_loop_policy")
|
||||
@patch("asyncio.set_event_loop_policy")
|
||||
def test_windows_with_postgres_protocol_sets_policy(self, mock_set_policy, mock_get_policy, mock_platform):
|
||||
"""Test Windows with 'postgres://' (deprecated) protocol also works."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_with_postgres_protocol_sets_policy(self, mock_platform):
|
||||
"""Windows with postgres:// URL sets policy."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_policy = MagicMock()
|
||||
mock_get_policy.return_value = mock_policy
|
||||
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "postgres://user:pass@localhost/db"}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is True
|
||||
mock_set_policy.assert_called_once()
|
||||
assert isinstance(asyncio.get_event_loop_policy(), MockWindowsSelectorPolicy)
|
||||
|
||||
@patch.object(asyncio, "WindowsSelectorEventLoopPolicy", MockWindowsSelectorPolicy, create=True)
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "postgresql://user:pass@localhost/db"}, clear=True)
|
||||
@patch("asyncio.get_event_loop_policy")
|
||||
@patch("asyncio.set_event_loop_policy")
|
||||
def test_windows_with_selector_already_set_returns_false(self, mock_set_policy, mock_get_policy, mock_platform):
|
||||
"""Test that if WindowsSelectorEventLoopPolicy is already set, returns False."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_with_selector_already_set_returns_false(self, mock_platform):
|
||||
"""If selector policy is already set, function returns False and does not call set again."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_policy = MagicMock(spec=MockWindowsSelectorPolicy)
|
||||
mock_get_policy.return_value = mock_policy
|
||||
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "postgresql://user:pass@localhost/db"}, clear=True):
|
||||
asyncio.set_event_loop_policy(MockWindowsSelectorPolicy())
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is False
|
||||
mock_set_policy.assert_not_called()
|
||||
assert isinstance(asyncio.get_event_loop_policy(), MockWindowsSelectorPolicy)
|
||||
|
||||
@patch.object(asyncio, "WindowsSelectorEventLoopPolicy", MockWindowsSelectorPolicy, create=True)
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "postgresql://user:pass@localhost/db"}, clear=True)
|
||||
@patch("asyncio.get_event_loop_policy")
|
||||
@patch("asyncio.set_event_loop_policy")
|
||||
@patch("langflow.helpers.windows_postgres_helper.logger")
|
||||
def test_logging_includes_source_when_provided(self, mock_logger, mock_set_policy, mock_get_policy, mock_platform): # noqa: ARG002
|
||||
"""Test that source is included in log context when provided."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_logging_includes_source_when_provided(self, mock_platform, caplog):
|
||||
"""With source= provided, function succeeds and logs (no mock of logger)."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_policy = MagicMock()
|
||||
mock_get_policy.return_value = mock_policy
|
||||
|
||||
result = configure_windows_postgres_event_loop(source="test_source")
|
||||
|
||||
with (
|
||||
patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "postgresql://user:pass@localhost/db"}, clear=True),
|
||||
caplog.at_level(logging.DEBUG, logger="langflow"),
|
||||
):
|
||||
result = configure_windows_postgres_event_loop(source="test_source")
|
||||
assert result is True
|
||||
mock_logger.debug.assert_called_once_with(
|
||||
"Windows PostgreSQL event loop configured",
|
||||
extra={
|
||||
"event_loop": "WindowsSelectorEventLoop",
|
||||
"reason": "psycopg_compatibility",
|
||||
"source": "test_source",
|
||||
},
|
||||
)
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "mysql://user:pass@localhost/db"}, clear=True)
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_with_other_database_returns_false(self, mock_platform):
|
||||
"""Test Windows with non-PostgreSQL database returns False."""
|
||||
"""Windows with non-PostgreSQL database returns False."""
|
||||
mock_platform.return_value = "Windows"
|
||||
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "mysql://user:pass@localhost/db"}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is False
|
||||
|
||||
@patch("platform.system")
|
||||
def test_docker_environment_not_affected(self, mock_platform):
|
||||
"""Test that Docker environments (typically Linux) are not affected."""
|
||||
mock_platform.return_value = "Linux"
|
||||
os.environ[LANGFLOW_DATABASE_URL] = "postgresql://user:pass@postgres:5432/langflow"
|
||||
os.environ["DOCKER_CONTAINER"] = "true"
|
||||
|
||||
original_policy = asyncio.get_event_loop_policy()
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
@pytest.mark.skipif(platform.system() != "Linux", reason="Docker test only meaningful on Linux")
|
||||
def test_docker_environment_not_affected_no_mock(self):
|
||||
"""On Linux with postgres URL, function returns False (no platform mock)."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{LANGFLOW_DATABASE_URL: "postgresql://user:pass@postgres:5432/langflow", "DOCKER_CONTAINER": "true"},
|
||||
clear=True,
|
||||
):
|
||||
original_policy = asyncio.get_event_loop_policy()
|
||||
result = configure_windows_postgres_event_loop()
|
||||
assert result is False
|
||||
assert asyncio.get_event_loop_policy() is original_policy
|
||||
|
||||
@@ -178,28 +151,22 @@ class TestWindowsPostgresHelper:
|
||||
],
|
||||
)
|
||||
@patch.object(asyncio, "WindowsSelectorEventLoopPolicy", MockWindowsSelectorPolicy, create=True)
|
||||
@patch("platform.system")
|
||||
@patch("asyncio.get_event_loop_policy")
|
||||
@patch("asyncio.set_event_loop_policy")
|
||||
def test_various_postgresql_urls_handled(self, mock_set_policy, mock_get_policy, mock_platform, db_url):
|
||||
"""Test that various PostgreSQL URL formats are handled correctly."""
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_various_postgresql_urls_handled(self, mock_platform, db_url):
|
||||
"""Various PostgreSQL URL formats are handled (real set_event_loop_policy)."""
|
||||
mock_platform.return_value = "Windows"
|
||||
mock_policy = MagicMock()
|
||||
mock_get_policy.return_value = mock_policy
|
||||
|
||||
with patch.dict(os.environ, {LANGFLOW_DATABASE_URL: db_url}, clear=True):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
assert result is True
|
||||
mock_set_policy.assert_called_once()
|
||||
assert isinstance(asyncio.get_event_loop_policy(), MockWindowsSelectorPolicy)
|
||||
|
||||
@patch("platform.system")
|
||||
@patch.dict(os.environ, {"LANGFLOW_DATABASE_URL": "postgresql://user:pass@localhost/db"}, clear=True)
|
||||
@patch("langflow.helpers.windows_postgres_helper.platform.system")
|
||||
def test_windows_without_policy_class_returns_false(self, mock_platform):
|
||||
"""Test that if WindowsSelectorEventLoopPolicy class is unavailable, returns False."""
|
||||
"""When WindowsSelectorEventLoopPolicy is unavailable, returns False."""
|
||||
mock_platform.return_value = "Windows"
|
||||
|
||||
with patch.object(asyncio, "WindowsSelectorEventLoopPolicy", None, create=True):
|
||||
with (
|
||||
patch.dict(os.environ, {LANGFLOW_DATABASE_URL: "postgresql://user:pass@localhost/db"}, clear=True),
|
||||
patch.object(asyncio, "WindowsSelectorEventLoopPolicy", None, create=True),
|
||||
):
|
||||
result = configure_windows_postgres_event_loop()
|
||||
|
||||
assert result is False
|
||||
|
||||
Reference in New Issue
Block a user