diff --git a/src/lfx/PLUGGABLE_SERVICES.md b/src/lfx/PLUGGABLE_SERVICES.md new file mode 100644 index 0000000000..26a3517cec --- /dev/null +++ b/src/lfx/PLUGGABLE_SERVICES.md @@ -0,0 +1,336 @@ +# Pluggable Services System + +LFX now supports a pluggable service architecture that allows you to customize and extend service implementations without modifying core code. + +## Overview + +The pluggable services system supports **three** discovery mechanisms: + +1. **Decorator Registration** - Services self-register when imported (best for library use) +2. **Configuration Files** - Explicit service mapping in config files (best for CLI use) +3. **Entry Points** - Python package entry points (best for distributable plugins) + +**Discovery order** (services are discovered in this order, with later sources able to override earlier ones): +1. Entry points (lowest priority) +2. Decorator registration +3. Configuration files (highest priority) + +## Quick Start + +### For CLI Users (Config File Approach) + +Create an `lfx.toml` in your project root: + +```toml +[services] +database_service = "langflow.services.database.service:DatabaseService" +storage_service = "langflow.services.storage.local:LocalStorageService" +cache_service = "langflow.services.cache.service:ThreadingInMemoryCache" +``` + +Then run: +```bash +lfx serve my_flow.json +``` + +Services will be automatically discovered from: +1. The current working directory (looking for `lfx.toml` or `pyproject.toml`) +2. Or the config directory specified in your settings + +**Note**: Service discovery happens automatically on first service access. + +### For Library Users (Decorator Approach) + +In your service module: + +```python +from lfx.services import register_service +from lfx.services.base import Service +from lfx.services.schema import ServiceType + +@register_service(ServiceType.DATABASE_SERVICE) +class MyDatabaseService(Service): + name = "database_service" + + def __init__(self, settings_service): + self.settings = settings_service.settings + self.set_ready() + + async def teardown(self) -> None: + # Cleanup logic + pass +``` + +Simply importing this module will register the service. + +### For Plugin Developers (Entry Points Approach) + +In your plugin's `pyproject.toml`: + +```toml +[project.entry-points."lfx.services"] +database_service = "my_plugin.services:MyDatabaseService" +storage_service = "my_plugin.services:MyStorageService" +``` + +When your package is installed, LFX will automatically discover these services. + +## Configuration Details + +### Config File Locations + +LFX searches for configuration in this order: + +1. `./lfx.toml` (project-specific config) +2. `./pyproject.toml` under `[tool.lfx.services]` (Python project integration) + +### Config File Format + +**Standalone `lfx.toml`:** + +```toml +[services] +database_service = "package.module:ClassName" +storage_service = "package.module:ClassName" +``` + +**In `pyproject.toml`:** + +```toml +[tool.lfx.services] +database_service = "package.module:ClassName" +storage_service = "package.module:ClassName" +``` + +### Service Keys + +Service keys **must** match `ServiceType` enum values exactly: + +- `database_service` +- `storage_service` +- `cache_service` +- `chat_service` +- `session_service` +- `task_service` +- `store_service` +- `variable_service` +- `telemetry_service` +- `tracing_service` +- `state_service` +- `job_queue_service` +- `shared_component_cache_service` +- `mcp_composer_service` + +**Important:** `settings_service` is **not pluggable** and cannot be overridden. It is always created using the built-in factory and provides the foundational configuration for all other services. + +## Creating Custom Services + +### Step 1: Implement the Service Class + +```python +from lfx.services.base import Service + +class MyCustomStorageService(Service): + name = "storage_service" # Must match ServiceType value + + def __init__(self, settings_service): + """Dependencies are auto-injected based on __init__ signature.""" + self.settings = settings_service + self.set_ready() + + async def save_file(self, flow_id: str, file_name: str, data: bytes) -> None: + # Your implementation + pass + + async def get_file(self, flow_id: str, file_name: str) -> bytes: + # Your implementation + pass + + async def teardown(self) -> None: + # Cleanup when service manager shuts down + pass +``` + +### Step 2: Register the Service + +**Option A: Decorator (recommended for libraries)** + +```python +from lfx.services import register_service +from lfx.services.schema import ServiceType + +@register_service(ServiceType.STORAGE_SERVICE) +class MyCustomStorageService(Service): + ... +``` + +**Option B: Config File (recommended for CLI users)** + +Add to `lfx.toml`: +```toml +[services] +storage_service = "my_package.services:MyCustomStorageService" +``` + +**Option C: Entry Points (recommended for plugins)** + +Add to your `pyproject.toml`: +```toml +[project.entry-points."lfx.services"] +storage_service = "my_package.services:MyCustomStorageService" +``` + +### Step 3: Use the Service + +```python +from lfx.services.deps import get_storage_service + +storage = get_storage_service() +await storage.save_file("flow_123", "data.json", b'{"key": "value"}') +``` + +## Dependency Injection + +Services can depend on other services. Dependencies are resolved automatically based on `__init__` parameter type hints: + +```python +from lfx.services.base import Service +from lfx.services.settings.service import SettingsService + +class MyDatabaseService(Service): + name = "database_service" + + def __init__(self, settings_service: SettingsService, cache_service: CacheService): + """ + Dependencies are auto-injected: + - settings_service: SettingsService will be created first + - cache_service: CacheService will be created second + """ + self.settings = settings_service.settings + self.cache = cache_service + self.set_ready() +``` + +The ServiceManager will: +1. Detect dependencies from type hints +2. Create dependencies first (in topological order) +3. Pass them to your service's `__init__` + +## Override Priority + +When multiple registration mechanisms provide the same service, **config files have highest priority**: + +1. **Entry Points** (lowest priority) - discovered first, use `override=False` +2. **Decorator Registration** (medium priority) - run when modules import, use `override=True` by default +3. **Config Files** (highest priority) - discovered last, use `override=True` + +Example: +```python +# Entry point registers: MyPluginStorage (override=False, won't replace existing) +# Decorator registers: CustomStorageService (override=True, replaces entry point) +# Config file registers: LocalStorageService (override=True, replaces decorator) + +# Result: LocalStorageService wins (config file is highest priority) +``` + +You can control decorator override behavior: + +```python +@register_service(ServiceType.STORAGE_SERVICE, override=False) +class MyService(Service): + ... # Won't override if already registered (e.g., from config file loaded earlier) +``` + +**Note**: In practice, decorators typically run during module import before `discover_plugins()` is called, but config files are explicitly designed to override decorators for deployment-time flexibility. + +## Testing Custom Services + +```python +import pytest +from lfx.services.manager import get_service_manager +from lfx.services.schema import ServiceType + +def test_custom_service(): + # Register your test service + manager = get_service_manager() + manager.register_service_class( + ServiceType.STORAGE_SERVICE, + MyTestStorageService, + override=True + ) + + # Use the service + from lfx.services.deps import get_storage_service + storage = get_storage_service() + + assert isinstance(storage, MyTestStorageService) +``` + +## Troubleshooting + +### Settings Service Cannot Be Overridden + +**Error:** `ValueError: Settings service cannot be registered via plugins` + +**Cause:** You attempted to register `settings_service` in a config file or via decorator. + +**Solution:** Remove `settings_service` from your config. The settings service is foundational and always uses the built-in implementation. Its `config_dir` is used to discover other plugins. + +### Service Not Found + +**Error:** `NoServiceRegisteredError: No factory registered for database_service` + +**Solutions:** +1. Check service key matches `ServiceType` enum value exactly +2. Verify import path is correct: `"module.path:ClassName"` +3. Ensure config file is in the correct location: + - If settings service exists: `{settings.config_dir}/lfx.toml` + - Otherwise: `./lfx.toml` or `./pyproject.toml` in current directory +4. Enable debug logging: `export LANGFLOW_LOG_LEVEL=DEBUG` + +### Import Errors + +**Error:** `ModuleNotFoundError: No module named 'langflow'` + +**Solutions:** +1. Ensure langflow is installed: `pip install langflow` +2. Check if module path in config is correct +3. Verify the package is importable: `python -c "import langflow.services.database"` + +### Circular Dependencies + +Services should not depend on each other in a circular way: + +❌ **Bad:** +```python +class ServiceA(Service): + def __init__(self, service_b: ServiceB): ... + +class ServiceB(Service): + def __init__(self, service_a: ServiceA): ... +``` + +✅ **Good:** +```python +class ServiceA(Service): + def __init__(self, settings: SettingsService): ... + +class ServiceB(Service): + def __init__(self, settings: SettingsService, service_a: ServiceA): ... +``` + +## Examples + +See: +- `lfx.toml.example` - Example configuration file showing Langflow service registration +- `src/lfx/services/` - Minimal built-in service implementations +- `src/backend/base/langflow/services/` - Full-featured Langflow services + +## Architecture Benefits + +The pluggable service system provides: +- ✅ **Automatic discovery** - Services found from config files, decorators, or entry points +- ✅ **Lazy instantiation** - Services created only when first accessed +- ✅ **Dependency injection** - Service dependencies resolved automatically +- ✅ **Lifecycle management** - Proper teardown when service manager shuts down +- ✅ **Flexibility** - Swap implementations without code changes (via config) diff --git a/src/lfx/README.md b/src/lfx/README.md index e0217d0f82..8fc0db33ee 100644 --- a/src/lfx/README.md +++ b/src/lfx/README.md @@ -26,6 +26,18 @@ uv run lfx serve my_flow.json ## Key Features +### Pluggable Services + +lfx supports a pluggable service architecture that allows you to customize and extend its behavior. You can replace built-in services (storage, telemetry, tracing, etc.) with your own implementations or use Langflow's full-featured services. + +📖 **See [PLUGGABLE_SERVICES.md](./PLUGGABLE_SERVICES.md) for details** including: + +- Quick start guides for CLI users, library developers, and plugin authors +- Service registration via config files, decorators, and entry points +- Creating custom service implementations with dependency injection +- Using full-featured Langflow services in lfx +- Troubleshooting and migration guides + ### Flattened Component Access lfx now supports simplified component imports for better developer experience: diff --git a/src/lfx/lfx.toml.example b/src/lfx/lfx.toml.example new file mode 100644 index 0000000000..16d0137979 --- /dev/null +++ b/src/lfx/lfx.toml.example @@ -0,0 +1,44 @@ +# Langflow Service Configuration +# +# This file tells LFX to use Langflow's full-featured service implementations +# instead of the minimal/noop services built into LFX. +# +# Usage: +# 1. Copy this file to your LANGFLOW_CONFIG_DIR as lfx.toml +# 2. Or add to pyproject.toml under [tool.lfx.services] +# +# The services will be automatically discovered and loaded when you run: +# - langflow run +# - lfx serve (when run from langflow project) + +[services] +# Core Services - Full-featured Langflow implementations +database_service = "langflow.services.database.service:DatabaseService" +storage_service = "langflow.services.storage.local:LocalStorageService" +cache_service = "langflow.services.cache.service:ThreadingInMemoryCache" + +# Authentication & Session Management +auth_service = "langflow.services.auth.service:AuthService" +session_service = "langflow.services.session.service:SessionService" + +# Chat & Communication +chat_service = "langflow.services.chat.service:ChatService" + +# State & Data Management +state_service = "langflow.services.state.service:StateService" +variable_service = "langflow.services.variable.service:VariableService" + +# Background Processing +task_service = "langflow.services.task.service:TaskService" +job_queue_service = "langflow.services.job_queue.service:JobQueueService" + +# Marketplace & Components +store_service = "langflow.services.store.service:StoreService" +shared_component_cache_service = "langflow.services.shared_component_cache.service:SharedComponentCacheService" + +# Observability +telemetry_service = "langflow.services.telemetry.service:TelemetryService" +tracing_service = "langflow.services.tracing.service:TracingService" + +# Note: settings_service is NOT configurable - it always uses the built-in implementation +# Note: mcp_composer_service uses the lfx implementation by default diff --git a/src/lfx/src/lfx/services/__init__.py b/src/lfx/src/lfx/services/__init__.py index 2be2266dc0..374d6bc8fb 100644 --- a/src/lfx/src/lfx/services/__init__.py +++ b/src/lfx/src/lfx/services/__init__.py @@ -1,3 +1,5 @@ +"""LFX services module - pluggable service architecture for dependency injection.""" + from .interfaces import ( CacheServiceProtocol, ChatServiceProtocol, @@ -9,6 +11,7 @@ from .interfaces import ( ) from .manager import ServiceManager from .mcp_composer import MCPComposerService, MCPComposerServiceFactory +from .registry import register_service from .session import NoopSession __all__ = [ @@ -23,4 +26,5 @@ __all__ = [ "StorageServiceProtocol", "TracingServiceProtocol", "VariableServiceProtocol", + "register_service", ] diff --git a/src/lfx/src/lfx/services/manager.py b/src/lfx/src/lfx/services/manager.py index bef2ae9d5a..0e44e4a9e6 100644 --- a/src/lfx/src/lfx/services/manager.py +++ b/src/lfx/src/lfx/services/manager.py @@ -1,7 +1,10 @@ -"""ServiceManager extracted from langflow for lfx package. +"""ServiceManager with pluggable service discovery. -This maintains the same API and most of the functionality, but removes -langflow-specific auto-discovery to break dependencies. +Supports multiple discovery mechanisms: +1. Decorator-based registration (@register_service) +2. Config file (lfx.toml / pyproject.toml) +3. Entry points (Python packages) +4. Fallback to noop/minimal implementations """ from __future__ import annotations @@ -10,6 +13,7 @@ import asyncio import importlib import inspect import threading +from pathlib import Path from typing import TYPE_CHECKING from lfx.log.logger import logger @@ -22,18 +26,27 @@ if TYPE_CHECKING: class NoFactoryRegisteredError(Exception): - pass + """Raised when no factory is registered for a service type.""" + + +class NoServiceRegisteredError(Exception): + """Raised when no service or factory is registered for a service type.""" class ServiceManager: - """Manages the creation of different services.""" + """Manages the creation of different services with pluggable discovery.""" def __init__(self) -> None: + """Initialize the service manager with empty service and factory registries.""" self.services: dict[str, Service] = {} self.factories: dict[str, ServiceFactory] = {} + self.service_classes: dict[ServiceType, type[Service]] = {} # New: direct service class registry self._lock = threading.RLock() self.keyed_lock = KeyedMemoryLockManager() self.factory_registered = False + self._plugins_discovered = False + + # Always register settings service from lfx.services.settings.factory import SettingsServiceFactory self.register_factory(SettingsServiceFactory()) @@ -57,6 +70,39 @@ class ServiceManager: """Set the factory registered flag.""" self.factory_registered = True + def register_service_class( + self, + service_type: ServiceType, + service_class: type[Service], + *, + override: bool = True, + ) -> None: + """Register a service class directly (without factory). + + Args: + service_type: The service type enum value + service_class: The service class to register + override: Whether to override existing registration (default: True) + + Raises: + ValueError: If attempting to register the settings service (not allowed) + """ + # Settings service cannot be overridden via plugins + if service_type == ServiceType.SETTINGS_SERVICE: + msg = "Settings service cannot be registered via plugins. It is always created using the built-in factory." + logger.warning(msg) + raise ValueError(msg) + + if service_type in self.service_classes and not override: + logger.warning(f"Service {service_type.value} already registered. Use override=True to replace it.") + return + + if service_type in self.service_classes: + logger.debug(f"Overriding service registration for {service_type.value}") + + self.service_classes[service_type] = service_class + logger.debug(f"Registered service class: {service_type.value} -> {service_class.__name__}") + def register_factory( self, service_factory: ServiceFactory, @@ -75,6 +121,112 @@ class ServiceManager: def _create_service(self, service_name: ServiceType, default: ServiceFactory | None = None) -> None: """Create a new service given its name, handling dependencies.""" logger.debug(f"Create service {service_name}") + + # Settings service is special - always use factory, never from plugins + if service_name == ServiceType.SETTINGS_SERVICE: + self._create_service_from_factory(service_name, default) + return + + # Try plugin discovery first (if not already done) + if not self._plugins_discovered: + # Get config_dir from settings service if available + config_dir = None + if ServiceType.SETTINGS_SERVICE in self.services: + settings_service = self.services[ServiceType.SETTINGS_SERVICE] + if hasattr(settings_service, "settings") and settings_service.settings.config_dir: + config_dir = Path(settings_service.settings.config_dir) + + self.discover_plugins(config_dir) + + # Check if we have a direct service class registration (new system) + if service_name in self.service_classes: + self._create_service_from_class(service_name) + return + + # Fall back to factory-based creation (old system) + self._create_service_from_factory(service_name, default) + + def _create_service_from_class(self, service_name: ServiceType) -> None: + """Create a service from a registered service class (new plugin system).""" + service_class = self.service_classes[service_name] + logger.debug(f"Creating service from class: {service_name.value} -> {service_class.__name__}") + + # Inspect __init__ to determine dependencies + init_signature = inspect.signature(service_class.__init__) + dependencies = {} + + for param_name, param in init_signature.parameters.items(): + if param_name == "self": + continue + + # Try to resolve dependency from type hint first + dependency_type = None + if param.annotation != inspect.Parameter.empty: + dependency_type = self._resolve_service_type_from_annotation(param.annotation) + + # If type hint didn't work, try to resolve from parameter name + # E.g., param name "settings_service" -> ServiceType.SETTINGS_SERVICE + if not dependency_type: + try: + dependency_type = ServiceType(param_name) + except ValueError: + # Not a valid service type - skip this parameter if it has a default + # Otherwise let it fail during instantiation + if param.default == inspect.Parameter.empty: + # No default, can't resolve - will fail during instantiation + pass + continue + + if dependency_type: + # Recursively create dependency if not exists + if dependency_type not in self.services: + self._create_service(dependency_type) + dependencies[param_name] = self.services[dependency_type] + + # Create the service instance + try: + service_instance = service_class(**dependencies) + # Don't call set_ready() here - let the service control its own ready state + self.services[service_name] = service_instance + logger.debug(f"Service created successfully: {service_name.value}") + except Exception as exc: + logger.exception(f"Failed to create service {service_name.value}: {exc}") + raise + + def _resolve_service_type_from_annotation(self, annotation) -> ServiceType | None: + """Resolve a ServiceType from a type annotation. + + Args: + annotation: The type annotation from __init__ signature + + Returns: + ServiceType if resolvable, None otherwise + """ + # Handle string annotations (forward references) + annotation_name = annotation if isinstance(annotation, str) else getattr(annotation, "__name__", None) + + if not annotation_name: + return None + + # Try to match service class name to ServiceType + # E.g., "SettingsService" -> ServiceType.SETTINGS_SERVICE + for service_type in ServiceType: + # Check if registered service class matches + if service_type in self.service_classes: + registered_class = self.service_classes[service_type] + if registered_class.__name__ == annotation_name: + return service_type + + # Check if annotation name matches expected pattern + # E.g., "SettingsService" -> "settings_service" + expected_name = annotation_name.replace("Service", "").lower() + "_service" + if service_type.value == expected_name: + return service_type + + return None + + def _create_service_from_factory(self, service_name: ServiceType, default: ServiceFactory | None = None) -> None: + """Create a service from a factory (old system).""" self._validate_service_creation(service_name, default) if service_name == ServiceType.SETTINGS_SERVICE: @@ -163,6 +315,145 @@ class ServiceManager: return factories + def discover_plugins(self, config_dir: Path | None = None) -> None: + """Discover and register service plugins from multiple sources. + + Discovery order (last wins): + 1. Entry points (installed packages) + 2. Config files (lfx.toml / pyproject.toml) + 3. Decorator-registered services (already in self.service_classes) + + Args: + config_dir: Directory to search for config files. + If None, tries to use settings_service.settings.config_dir, + then falls back to current working directory. + + Note: + The settings service cannot be overridden via plugins and is always + created using the built-in factory. + """ + if self._plugins_discovered: + logger.debug("Plugins already discovered, skipping...") + return + + # Get config_dir from settings service if not provided + if config_dir is None and ServiceType.SETTINGS_SERVICE in self.services: + settings_service = self.services[ServiceType.SETTINGS_SERVICE] + if hasattr(settings_service, "settings") and settings_service.settings.config_dir: + config_dir = Path(settings_service.settings.config_dir) + + logger.debug(f"Starting plugin discovery (config_dir: {config_dir or 'cwd'})...") + + # 1. Discover from entry points + self._discover_from_entry_points() + + # 2. Discover from config files + self._discover_from_config(config_dir) + + self._plugins_discovered = True + logger.debug(f"Plugin discovery complete. Registered services: {list(self.service_classes.keys())}") + + def _discover_from_entry_points(self) -> None: + """Discover services from Python entry points.""" + from importlib.metadata import entry_points + + eps = entry_points(group="lfx.services") + + for ep in eps: + try: + service_class = ep.load() + # Entry point name should match ServiceType enum value + service_type = ServiceType(ep.name) + self.register_service_class(service_type, service_class, override=False) + logger.debug(f"Loaded service from entry point: {ep.name}") + except (ValueError, AttributeError) as exc: + logger.warning(f"Failed to load entry point {ep.name}: {exc}") + except Exception as exc: # noqa: BLE001 + logger.debug(f"Error loading entry point {ep.name}: {exc}") + + def _discover_from_config(self, config_dir: Path | None = None) -> None: + """Discover services from config files (lfx.toml / pyproject.toml).""" + config_dir = Path.cwd() if config_dir is None else Path(config_dir) + + # Try lfx.toml first + lfx_config = config_dir / "lfx.toml" + if lfx_config.exists(): + self._load_config_file(lfx_config) + return + + # Try pyproject.toml with [tool.lfx.services] + pyproject_config = config_dir / "pyproject.toml" + if pyproject_config.exists(): + self._load_pyproject_config(pyproject_config) + + def _load_config_file(self, config_path: Path) -> None: + """Load services from lfx.toml config file.""" + try: + import tomllib as tomli # Python 3.11+ + except ImportError: + import tomli # Python 3.10 + + try: + with config_path.open("rb") as f: + config = tomli.load(f) + + services = config.get("services", {}) + for service_key, service_path in services.items(): + self._register_service_from_path(service_key, service_path) + + logger.debug(f"Loaded {len(services)} services from {config_path}") + except Exception as exc: # noqa: BLE001 + logger.warning(f"Failed to load config from {config_path}: {exc}") + + def _load_pyproject_config(self, config_path: Path) -> None: + """Load services from pyproject.toml [tool.lfx.services] section.""" + try: + import tomllib as tomli # Python 3.11+ + except ImportError: + import tomli # Python 3.10 + + try: + with config_path.open("rb") as f: + config = tomli.load(f) + + services = config.get("tool", {}).get("lfx", {}).get("services", {}) + for service_key, service_path in services.items(): + self._register_service_from_path(service_key, service_path) + + if services: + logger.debug(f"Loaded {len(services)} services from {config_path}") + except Exception as exc: # noqa: BLE001 + logger.warning(f"Failed to load config from {config_path}: {exc}") + + def _register_service_from_path(self, service_key: str, service_path: str) -> None: + """Register a service from a module:class path string. + + Args: + service_key: ServiceType enum value (e.g., "database_service") + service_path: Import path (e.g., "langflow.services.database.service:DatabaseService") + """ + try: + # Validate service_key matches ServiceType enum + service_type = ServiceType(service_key) + except ValueError: + logger.warning(f"Invalid service key '{service_key}' - must match ServiceType enum value") + return + + try: + # Parse module:class format + if ":" not in service_path: + logger.warning(f"Invalid service path '{service_path}' - must be 'module:class' format") + return + + module_path, class_name = service_path.split(":", 1) + module = importlib.import_module(module_path) + service_class = getattr(module, class_name) + + self.register_service_class(service_type, service_class, override=True) + logger.debug(f"Registered service from config: {service_key} -> {service_path}") + except Exception as exc: # noqa: BLE001 + logger.warning(f"Failed to register service {service_key} from {service_path}: {exc}") + # Global variables for lazy initialization _service_manager: ServiceManager | None = None diff --git a/src/lfx/src/lfx/services/registry.py b/src/lfx/src/lfx/services/registry.py new file mode 100644 index 0000000000..02876a7c2f --- /dev/null +++ b/src/lfx/src/lfx/services/registry.py @@ -0,0 +1,52 @@ +"""Service registration decorator for pluggable services. + +Allows services to self-register with the service manager using a decorator. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, TypeVar + +from lfx.log.logger import logger + +if TYPE_CHECKING: + from lfx.services.base import Service + from lfx.services.schema import ServiceType + +ServiceT = TypeVar("ServiceT", bound="Service") + + +def register_service(service_type: ServiceType, *, override: bool = True): + """Decorator to register a service class with the service manager. + + Usage: + @register_service(ServiceType.DATABASE_SERVICE) + class DatabaseService(Service): + name = "database_service" + ... + + Args: + service_type: The ServiceType enum value for this service + override: Whether to override existing registrations (default: True) + + Returns: + Decorator function that registers the service class + """ + + def decorator(service_class: type[ServiceT]) -> type[ServiceT]: + """Register the service class and return it unchanged.""" + try: + from lfx.services.manager import get_service_manager + + service_manager = get_service_manager() + service_manager.register_service_class(service_type, service_class, override=override) + logger.debug(f"Registered service via decorator: {service_type.value} -> {service_class.__name__}") + except ValueError: + # Re-raise ValueError (used for settings service protection) + raise + except Exception as exc: # noqa: BLE001 + logger.warning(f"Failed to register service {service_type.value} from decorator: {exc}") + + return service_class + + return decorator diff --git a/src/lfx/src/lfx/services/storage/local.py b/src/lfx/src/lfx/services/storage/local.py index bac3f8c4f4..68bfdc0751 100644 --- a/src/lfx/src/lfx/services/storage/local.py +++ b/src/lfx/src/lfx/services/storage/local.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING from aiofile import async_open from lfx.log.logger import logger +from lfx.services.base import Service from lfx.services.storage.service import StorageService if TYPE_CHECKING: @@ -18,7 +19,7 @@ if TYPE_CHECKING: EXPECTED_PATH_PARTS = 2 # Path format: "flow_id/filename" -class LocalStorageService(StorageService): +class LocalStorageService(StorageService, Service): """A service class for handling local file storage operations.""" def __init__( @@ -53,6 +54,10 @@ class LocalStorageService(StorageService): flow_id, file_name = parts return self.build_full_path(flow_id, file_name) + async def teardown(self) -> None: + """Teardown the storage service.""" + # No cleanup needed for local storage + def build_full_path(self, flow_id: str, file_name: str) -> str: """Build the full path of a file in the local storage.""" return str(self.data_dir / flow_id / file_name) @@ -216,7 +221,3 @@ class LocalStorageService(StorageService): raise else: return file_size_stat.st_size - - async def teardown(self) -> None: - """Perform any cleanup operations when the service is being torn down.""" - # No specific teardown actions required for local diff --git a/src/lfx/src/lfx/services/telemetry/__init__.py b/src/lfx/src/lfx/services/telemetry/__init__.py new file mode 100644 index 0000000000..adbd11711e --- /dev/null +++ b/src/lfx/src/lfx/services/telemetry/__init__.py @@ -0,0 +1,5 @@ +"""Telemetry service for lfx package.""" + +from .service import TelemetryService + +__all__ = ["TelemetryService"] diff --git a/src/lfx/src/lfx/services/telemetry/base.py b/src/lfx/src/lfx/services/telemetry/base.py new file mode 100644 index 0000000000..d5afa250c8 --- /dev/null +++ b/src/lfx/src/lfx/services/telemetry/base.py @@ -0,0 +1,86 @@ +"""Abstract base class for telemetry services.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from lfx.services.base import Service + +if TYPE_CHECKING: + from pydantic import BaseModel + + +class BaseTelemetryService(Service, ABC): + """Abstract base class for telemetry services. + + Defines the minimal interface that all telemetry service implementations + must provide, whether minimal (LFX) or full-featured (Langflow). + """ + + @abstractmethod + def __init__(self): + """Initialize the telemetry service.""" + super().__init__() + + @abstractmethod + async def send_telemetry_data(self, payload: BaseModel, path: str | None = None) -> None: + """Send telemetry data to the telemetry backend. + + Args: + payload: The telemetry payload to send + path: Optional path to append to the base URL + """ + + @abstractmethod + async def log_package_run(self, payload: BaseModel) -> None: + """Log a package run event. + + Args: + payload: Run payload containing run information + """ + + @abstractmethod + async def log_package_shutdown(self) -> None: + """Log a package shutdown event.""" + + @abstractmethod + async def log_package_version(self) -> None: + """Log the package version information.""" + + @abstractmethod + async def log_package_playground(self, payload: BaseModel) -> None: + """Log a playground interaction event. + + Args: + payload: Playground payload containing interaction information + """ + + @abstractmethod + async def log_package_component(self, payload: BaseModel) -> None: + """Log a component usage event. + + Args: + payload: Component payload containing component information + """ + + @abstractmethod + async def log_exception(self, exc: Exception, context: str) -> None: + """Log an unhandled exception. + + Args: + exc: The exception that occurred + context: Context where exception occurred + """ + + @abstractmethod + def start(self) -> None: + """Start the telemetry service.""" + + @abstractmethod + async def stop(self) -> None: + """Stop the telemetry service.""" + + @abstractmethod + async def flush(self) -> None: + """Flush any pending telemetry data.""" diff --git a/src/lfx/src/lfx/services/telemetry/service.py b/src/lfx/src/lfx/services/telemetry/service.py new file mode 100644 index 0000000000..72844b45de --- /dev/null +++ b/src/lfx/src/lfx/services/telemetry/service.py @@ -0,0 +1,100 @@ +"""Lightweight telemetry service for LFX package.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from lfx.log.logger import logger +from lfx.services.telemetry.base import BaseTelemetryService + +if TYPE_CHECKING: + from pydantic import BaseModel + + +class TelemetryService(BaseTelemetryService): + """Minimal telemetry service implementation for LFX. + + This is a lightweight implementation that logs telemetry events + but does not send data to any external service. For full telemetry + functionality, use the Langflow TelemetryService. + """ + + def __init__(self): + """Initialize the telemetry service with do-not-track enabled.""" + super().__init__() + self.do_not_track = True # Minimal implementation never sends data + self.set_ready() + + @property + def name(self) -> str: + """Service name identifier. + + Returns: + str: The service name. + """ + return "telemetry_service" + + async def send_telemetry_data(self, payload: BaseModel, path: str | None = None) -> None: # noqa: ARG002 + """Log telemetry data (minimal implementation - no actual sending). + + Args: + payload: The telemetry payload + path: Optional path + """ + logger.debug(f"Telemetry event (not sent): {path}") + + async def log_package_run(self, payload: BaseModel) -> None: # noqa: ARG002 + """Log a package run event. + + Args: + payload: Run payload + """ + logger.debug("Telemetry: package run") + + async def log_package_shutdown(self) -> None: + """Log a package shutdown event.""" + logger.debug("Telemetry: package shutdown") + + async def log_package_version(self) -> None: + """Log the package version.""" + logger.debug("Telemetry: package version") + + async def log_package_playground(self, payload: BaseModel) -> None: # noqa: ARG002 + """Log a playground interaction. + + Args: + payload: Playground payload + """ + logger.debug("Telemetry: playground interaction") + + async def log_package_component(self, payload: BaseModel) -> None: # noqa: ARG002 + """Log a component usage. + + Args: + payload: Component payload + """ + logger.debug("Telemetry: component usage") + + async def log_exception(self, exc: Exception, context: str) -> None: + """Log an unhandled exception. + + Args: + exc: The exception + context: Exception context + """ + logger.debug(f"Telemetry: exception in {context}: {exc.__class__.__name__}") + + def start(self) -> None: + """Start the telemetry service (minimal implementation - noop).""" + logger.debug("Telemetry service started (minimal mode)") + + async def stop(self) -> None: + """Stop the telemetry service (minimal implementation - noop).""" + logger.debug("Telemetry service stopped") + + async def flush(self) -> None: + """Flush pending telemetry (minimal implementation - noop).""" + + async def teardown(self) -> None: + """Teardown the telemetry service.""" + await self.stop() diff --git a/src/lfx/src/lfx/services/tracing/__init__.py b/src/lfx/src/lfx/services/tracing/__init__.py index 3e0a19bc78..e4a3dbea00 100644 --- a/src/lfx/src/lfx/services/tracing/__init__.py +++ b/src/lfx/src/lfx/services/tracing/__init__.py @@ -1 +1 @@ -# Tracing service for lfx package +"""Tracing service for lfx package - minimal implementation.""" diff --git a/src/lfx/src/lfx/services/tracing/base.py b/src/lfx/src/lfx/services/tracing/base.py new file mode 100644 index 0000000000..148a19d093 --- /dev/null +++ b/src/lfx/src/lfx/services/tracing/base.py @@ -0,0 +1,119 @@ +"""Abstract base class for tracing services.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any + +from lfx.services.base import Service + +if TYPE_CHECKING: + from uuid import UUID + + from langchain.callbacks.base import BaseCallbackHandler + + from lfx.custom.custom_component.component import Component + + +class BaseTracingService(Service, ABC): + """Abstract base class for tracing services. + + Defines the minimal interface that all tracing service implementations + must provide, whether minimal (LFX) or full-featured (Langflow). + """ + + @abstractmethod + def __init__(self): + """Initialize the tracing service.""" + super().__init__() + + @abstractmethod + async def start_tracers( + self, + run_id: UUID, + run_name: str, + user_id: str | None, + session_id: str | None, + project_name: str | None = None, + ) -> None: + """Start tracers for a graph run. + + Args: + run_id: Unique identifier for the run + run_name: Name of the run + user_id: User identifier (optional) + session_id: Session identifier (optional) + project_name: Project name (optional) + """ + + @abstractmethod + async def end_tracers(self, outputs: dict, error: Exception | None = None) -> None: + """End tracers for a graph run. + + Args: + outputs: Output data from the run + error: Exception if run failed (optional) + """ + + @abstractmethod + @asynccontextmanager + async def trace_component( + self, + component: Component, + trace_name: str, + inputs: dict[str, Any], + metadata: dict[str, Any] | None = None, + ): + """Context manager for tracing a component execution. + + Args: + component: The component being traced + trace_name: Name for the trace + inputs: Input data to the component + metadata: Additional metadata (optional) + + Yields: + Self for method chaining + """ + + @abstractmethod + def add_log(self, trace_name: str, log: Any) -> None: + """Add a log entry to the current trace. + + Args: + trace_name: Name of the trace + log: Log data to add + """ + + @abstractmethod + def set_outputs( + self, + trace_name: str, + outputs: dict[str, Any], + output_metadata: dict[str, Any] | None = None, + ) -> None: + """Set outputs for the current trace. + + Args: + trace_name: Name of the trace + outputs: Output data + output_metadata: Additional output metadata (optional) + """ + + @abstractmethod + def get_langchain_callbacks(self) -> list[BaseCallbackHandler]: + """Get LangChain callback handlers for tracing. + + Returns: + List of callback handlers + """ + + @property + @abstractmethod + def project_name(self) -> str | None: + """Get the current project name. + + Returns: + Project name or None if not set + """ diff --git a/src/lfx/src/lfx/services/tracing/service.py b/src/lfx/src/lfx/services/tracing/service.py index 32b32ef0c0..53ecfc15ac 100644 --- a/src/lfx/src/lfx/services/tracing/service.py +++ b/src/lfx/src/lfx/services/tracing/service.py @@ -1,21 +1,134 @@ -"""Lightweight tracing service for lfx package.""" +"""Lightweight tracing service for LFX package.""" -from lfx.services.base import Service +# ruff: noqa: ARG002 +from __future__ import annotations + +from contextlib import asynccontextmanager +from typing import TYPE_CHECKING, Any + +from lfx.log.logger import logger +from lfx.services.tracing.base import BaseTracingService + +if TYPE_CHECKING: + from uuid import UUID + + from langchain.callbacks.base import BaseCallbackHandler + + from lfx.custom.custom_component.component import Component -class TracingService(Service): - """Lightweight tracing service.""" +class TracingService(BaseTracingService): + """Minimal tracing service implementation for LFX. + + This is a lightweight implementation that logs trace events + but does not integrate with external tracing services. For full + tracing functionality (LangSmith, LangFuse, etc.), use the + Langflow TracingService. + """ + + def __init__(self): + """Initialize the tracing service.""" + super().__init__() + self.deactivated = False + self.set_ready() @property def name(self) -> str: + """Service name identifier. + + Returns: + str: The service name. + """ return "tracing_service" - def log(self, message: str, **kwargs) -> None: # noqa: ARG002 - """Log a message with optional metadata.""" - # Lightweight implementation - just log basic info - from lfx.log.logger import logger + async def start_tracers( + self, + run_id: UUID, + run_name: str, + user_id: str | None, + session_id: str | None, + project_name: str | None = None, + ) -> None: + """Start tracers (minimal implementation - just logs). - logger.debug(f"Trace: {message}") + Args: + run_id: Run identifier + run_name: Run name + user_id: User identifier + session_id: Session identifier + project_name: Project name + """ + logger.debug(f"Trace started: {run_name}") + + async def end_tracers(self, outputs: dict, error: Exception | None = None) -> None: + """End tracers (minimal implementation - just logs). + + Args: + outputs: Output data + error: Exception if any + """ + logger.debug("Trace ended") + + @asynccontextmanager + async def trace_component( + self, + component: Component, + trace_name: str, + inputs: dict[str, Any], + metadata: dict[str, Any] | None = None, + ): + """Trace a component (minimal implementation). + + Args: + component: Component to trace + trace_name: Trace name + inputs: Input data + metadata: Metadata + """ + logger.debug(f"Tracing component: {trace_name}") + yield self + + def add_log(self, trace_name: str, log: Any) -> None: + """Add a log entry (minimal implementation - just logs). + + Args: + trace_name: Trace name + log: Log data + """ + logger.debug(f"Trace log: {trace_name}") + + def set_outputs( + self, + trace_name: str, + outputs: dict[str, Any], + output_metadata: dict[str, Any] | None = None, + ) -> None: + """Set outputs (minimal implementation - noop). + + Args: + trace_name: Trace name + outputs: Output data + output_metadata: Output metadata + """ + logger.debug(f"Trace outputs set: {trace_name}") + + def get_langchain_callbacks(self) -> list[BaseCallbackHandler]: + """Get LangChain callbacks (minimal implementation - empty list). + + Returns: + Empty list (no callbacks in minimal implementation) + """ + return [] + + @property + def project_name(self) -> str | None: + """Get project name (minimal implementation - returns None). + + Returns: + None + """ + return None async def teardown(self) -> None: """Teardown the tracing service.""" + logger.debug("Tracing service teardown") diff --git a/src/lfx/src/lfx/services/variable/__init__.py b/src/lfx/src/lfx/services/variable/__init__.py new file mode 100644 index 0000000000..52dc76643a --- /dev/null +++ b/src/lfx/src/lfx/services/variable/__init__.py @@ -0,0 +1,5 @@ +"""Variable service for lfx package.""" + +from .service import VariableService + +__all__ = ["VariableService"] diff --git a/src/lfx/src/lfx/services/variable/service.py b/src/lfx/src/lfx/services/variable/service.py new file mode 100644 index 0000000000..d0e1aef8d1 --- /dev/null +++ b/src/lfx/src/lfx/services/variable/service.py @@ -0,0 +1,83 @@ +"""Minimal variable service for lfx package with in-memory storage and environment fallback.""" + +import os + +from lfx.log.logger import logger +from lfx.services.base import Service + + +class VariableService(Service): + """Minimal variable service with in-memory storage and environment fallback. + + This is a lightweight implementation for LFX that maintains in-memory + variables and falls back to environment variables for reads. No database storage. + """ + + name = "variable_service" + + def __init__(self) -> None: + """Initialize the variable service.""" + super().__init__() + self._variables: dict[str, str] = {} + self.set_ready() + logger.debug("Variable service initialized (env vars only)") + + def get_variable(self, name: str, **kwargs) -> str | None: # noqa: ARG002 + """Get a variable value. + + First checks in-memory cache, then environment variables. + + Args: + name: Variable name + **kwargs: Additional arguments (ignored in minimal implementation) + + Returns: + Variable value or None if not found + """ + # Check in-memory first + if name in self._variables: + return self._variables[name] + + # Fall back to environment variable + value = os.getenv(name) + if value: + logger.debug(f"Variable '{name}' loaded from environment") + return value + + def set_variable(self, name: str, value: str, **kwargs) -> None: # noqa: ARG002 + """Set a variable value (in-memory only). + + Args: + name: Variable name + value: Variable value + **kwargs: Additional arguments (ignored in minimal implementation) + """ + self._variables[name] = value + logger.debug(f"Variable '{name}' set (in-memory only)") + + def delete_variable(self, name: str, **kwargs) -> None: # noqa: ARG002 + """Delete a variable (from in-memory cache only). + + Args: + name: Variable name + **kwargs: Additional arguments (ignored in minimal implementation) + """ + if name in self._variables: + del self._variables[name] + logger.debug(f"Variable '{name}' deleted (from in-memory cache)") + + def list_variables(self, **kwargs) -> list[str]: # noqa: ARG002 + """List all variables (in-memory only). + + Args: + **kwargs: Additional arguments (ignored in minimal implementation) + + Returns: + List of variable names + """ + return list(self._variables.keys()) + + async def teardown(self) -> None: + """Teardown the variable service.""" + self._variables.clear() + logger.debug("Variable service teardown") diff --git a/src/lfx/tests/unit/services/__init__.py b/src/lfx/tests/unit/services/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/lfx/tests/unit/services/test_decorator_registration.py b/src/lfx/tests/unit/services/test_decorator_registration.py new file mode 100644 index 0000000000..d8ebc5b233 --- /dev/null +++ b/src/lfx/tests/unit/services/test_decorator_registration.py @@ -0,0 +1,137 @@ +"""Tests for decorator-based service registration.""" + +import pytest +from lfx.services.base import Service +from lfx.services.manager import ServiceManager +from lfx.services.schema import ServiceType +from lfx.services.storage.local import LocalStorageService +from lfx.services.telemetry.service import TelemetryService +from lfx.services.tracing.service import TracingService + + +@pytest.fixture +def clean_manager(): + """Create a fresh ServiceManager for testing decorators.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + +class TestDecoratorRegistration: + """Tests for @register_service decorator with real services.""" + + def test_decorator_registers_real_storage_service(self, clean_manager): + """Test that decorator registers real LocalStorageService.""" + # Use direct registration to simulate decorator (since decorator uses singleton) + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService, override=True) + + assert ServiceType.STORAGE_SERVICE in clean_manager.service_classes + assert clean_manager.service_classes[ServiceType.STORAGE_SERVICE] == LocalStorageService + + # Verify we can actually create and use the service + storage = clean_manager.get(ServiceType.STORAGE_SERVICE) + assert isinstance(storage, LocalStorageService) + assert storage.ready is True + + @pytest.mark.asyncio + async def test_decorator_registers_real_telemetry_service(self, clean_manager): + """Test that decorator registers real TelemetryService.""" + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService, override=True) + + assert ServiceType.TELEMETRY_SERVICE in clean_manager.service_classes + assert clean_manager.service_classes[ServiceType.TELEMETRY_SERVICE] == TelemetryService + + # Verify service works + telemetry = clean_manager.get(ServiceType.TELEMETRY_SERVICE) + assert isinstance(telemetry, TelemetryService) + await telemetry.log_package_version() # Should not raise + + def test_decorator_with_override_false_preserves_first(self, clean_manager): + """Test decorator with override=False preserves first registration.""" + # Register first service + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService, override=True) + + # Try to register second service with override=False + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TracingService, override=False) + + # Should still be first service + assert clean_manager.service_classes[ServiceType.TELEMETRY_SERVICE] == TelemetryService + + def test_decorator_with_override_true_replaces(self, clean_manager): + """Test decorator with override=True replaces existing.""" + # Register first service + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService, override=True) + + # Replace with second service + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TracingService, override=True) + + # Should be second service + assert clean_manager.service_classes[ServiceType.TELEMETRY_SERVICE] == TracingService + + def test_cannot_decorate_settings_service(self, clean_manager): + """Test that decorating settings service raises ValueError.""" + with pytest.raises(ValueError, match="Settings service cannot be registered"): + clean_manager.register_service_class(ServiceType.SETTINGS_SERVICE, LocalStorageService) + + def test_decorator_with_custom_service_class(self, clean_manager): + """Test decorator with a custom service implementation.""" + + class CustomTracingService(Service): + @property + def name(self) -> str: + return "tracing_service" + + def __init__(self): + super().__init__() + self.messages = [] + self.set_ready() + + def add_log(self, trace_name: str, log: dict): + self.messages.append(f"{trace_name}: {log}") + + async def teardown(self) -> None: + self.messages.clear() + + clean_manager.register_service_class(ServiceType.TRACING_SERVICE, CustomTracingService, override=True) + + # Verify registration + assert clean_manager.service_classes[ServiceType.TRACING_SERVICE] == CustomTracingService + + # Verify we can use it + tracing = clean_manager.get(ServiceType.TRACING_SERVICE) + assert isinstance(tracing, CustomTracingService) + tracing.add_log("test_trace", {"message": "test message"}) + assert len(tracing.messages) == 1 + + def test_decorator_preserves_class_functionality(self, clean_manager): + """Test that decorator preserves all class functionality.""" + clean_manager.register_service_class(ServiceType.VARIABLE_SERVICE, LocalStorageService, override=True) + + # Class should still be usable directly (not just through manager) + direct_instance = LocalStorageService() + assert direct_instance.ready is True + assert direct_instance.name == "storage_service" + + def test_multiple_decorators_on_different_services(self, clean_manager): + """Test registering multiple different services.""" + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService, override=True) + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService, override=True) + clean_manager.register_service_class(ServiceType.TRACING_SERVICE, TracingService, override=True) + + # All should be registered + assert len(clean_manager.service_classes) == 3 + assert ServiceType.STORAGE_SERVICE in clean_manager.service_classes + assert ServiceType.TELEMETRY_SERVICE in clean_manager.service_classes + assert ServiceType.TRACING_SERVICE in clean_manager.service_classes + + # All should be creatable + storage = clean_manager.get(ServiceType.STORAGE_SERVICE) + telemetry = clean_manager.get(ServiceType.TELEMETRY_SERVICE) + tracing = clean_manager.get(ServiceType.TRACING_SERVICE) + + assert isinstance(storage, LocalStorageService) + assert isinstance(telemetry, TelemetryService) + assert isinstance(tracing, TracingService) diff --git a/src/lfx/tests/unit/services/test_edge_cases.py b/src/lfx/tests/unit/services/test_edge_cases.py new file mode 100644 index 0000000000..a31b1aefae --- /dev/null +++ b/src/lfx/tests/unit/services/test_edge_cases.py @@ -0,0 +1,498 @@ +"""Edge case tests for pluggable service system.""" + +import pytest +from lfx.services.base import Service +from lfx.services.manager import ServiceManager +from lfx.services.schema import ServiceType + + +class TestCircularDependencyDetection: + """Test detection and handling of circular dependencies.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_self_circular_dependency(self, clean_manager): + """Test service that depends on itself.""" + + class SelfCircularService(Service): + @property + def name(self) -> str: + return "self_circular" + + def __init__(self, storage_service): + super().__init__() + self.storage = storage_service + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, SelfCircularService) + + # Should raise RecursionError or TypeError (missing required argument) + with pytest.raises((RecursionError, RuntimeError, TypeError)): + clean_manager.get(ServiceType.STORAGE_SERVICE) + + +class TestServiceLifecycle: + """Test service lifecycle management.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_service_ready_state(self, clean_manager): + """Test service ready state tracking.""" + + class SlowInitService(Service): + @property + def name(self) -> str: + return "slow_service" + + def __init__(self): + super().__init__() + # Don't set ready immediately + + def complete_init(self): + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, SlowInitService) + + service = clean_manager.get(ServiceType.STORAGE_SERVICE) + + # Should not be ready yet + assert service.ready is False + + # Complete initialization + service.complete_init() + assert service.ready is True + + @pytest.mark.asyncio + async def test_service_teardown_called(self, clean_manager): + """Test that teardown is called on services.""" + teardown_called = [] + + class TeardownTrackingService(Service): + name = "tracking_service" + + def __init__(self): + super().__init__() + self.set_ready() + + async def teardown(self) -> None: + teardown_called.append(True) + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, TeardownTrackingService) + + # Create service + clean_manager.get(ServiceType.STORAGE_SERVICE) + + # Teardown + await clean_manager.teardown() + + # Should have been called + assert len(teardown_called) == 1 + + @pytest.mark.asyncio + async def test_multiple_teardowns_safe(self, clean_manager): + """Test that calling teardown multiple times is safe.""" + + class SimpleService(Service): + name = "simple_service" + + def __init__(self): + super().__init__() + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, SimpleService) + clean_manager.get(ServiceType.STORAGE_SERVICE) + + # Teardown multiple times - should not raise + await clean_manager.teardown() + await clean_manager.teardown() + + +class TestConfigParsingEdgeCases: + """Test edge cases in configuration parsing.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_empty_config_file(self, clean_manager, tmp_path): + """Test empty configuration file.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text("") + + # Should not raise + clean_manager.discover_plugins(config_dir) + + assert len(clean_manager.service_classes) == 0 + + def test_config_with_no_services_section(self, clean_manager, tmp_path): + """Test config file with no [services] section.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[other_section] +key = "value" +""" + ) + + # Should not raise + clean_manager.discover_plugins(config_dir) + + assert len(clean_manager.service_classes) == 0 + + def test_config_with_empty_services_section(self, clean_manager, tmp_path): + """Test config with empty [services] section.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +""" + ) + + # Should not raise + clean_manager.discover_plugins(config_dir) + + assert len(clean_manager.service_classes) == 0 + + def test_config_with_malformed_import_path(self, clean_manager, tmp_path): + """Test config with malformed import path.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "invalid_path_without_colon" +""" + ) + + # Should not raise, just log warning + clean_manager.discover_plugins(config_dir) + + assert ServiceType.STORAGE_SERVICE not in clean_manager.service_classes + + def test_config_with_too_many_colons(self, clean_manager, tmp_path): + """Test config with too many colons in import path.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "module:submodule:class:extra" +""" + ) + + # Should not raise, just log warning + clean_manager.discover_plugins(config_dir) + + assert ServiceType.STORAGE_SERVICE not in clean_manager.service_classes + + +class TestServiceRegistrationEdgeCases: + """Test edge cases in service registration.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_register_non_service_class(self, clean_manager): + """Test registering a class that doesn't inherit from Service.""" + + class NotAService: + @property + def name(self) -> str: + return "not_service" + + def __init__(self): + pass + + async def teardown(self) -> None: + pass + + # Should not raise during registration + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, NotAService) + + # Gets created successfully but doesn't have service methods + service = clean_manager.get(ServiceType.STORAGE_SERVICE) + assert service is not None + # But won't have ready attribute since it doesn't inherit from Service + assert not hasattr(service, "_ready") + + def test_register_abstract_service(self, clean_manager): + """Test registering an abstract service class.""" + from abc import ABC, abstractmethod + + class AbstractService(Service, ABC): + name = "abstract_service" + + @abstractmethod + def do_something(self): + pass + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, AbstractService) + + # Should fail due to abstract methods + with pytest.raises(TypeError): + clean_manager.get(ServiceType.STORAGE_SERVICE) + + def test_register_same_service_multiple_times_with_override(self, clean_manager): + """Test registering same service type multiple times with override.""" + + class Service1(Service): + name = "service1" + + async def teardown(self) -> None: + pass + + class Service2(Service): + name = "service2" + + async def teardown(self) -> None: + pass + + class Service3(Service): + name = "service3" + + async def teardown(self) -> None: + pass + + # Register multiple times + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, Service1, override=True) + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, Service2, override=True) + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, Service3, override=True) + + # Should have the last one + assert clean_manager.service_classes[ServiceType.STORAGE_SERVICE] == Service3 + + +class TestDependencyInjectionEdgeCases: + """Test edge cases in dependency injection.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_service_with_optional_dependencies(self, clean_manager): + """Test service with optional parameters.""" + + class ServiceWithOptional(Service): + name = "optional_service" + + def __init__(self, settings_service=None): + super().__init__() + self.settings = settings_service + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, ServiceWithOptional) + + service = clean_manager.get(ServiceType.STORAGE_SERVICE) + + # Should have settings injected + from lfx.services.settings.service import SettingsService + + assert isinstance(service.settings, SettingsService) + + def test_service_with_no_init_params(self, clean_manager): + """Test service that takes no init parameters.""" + + class NoParamService(Service): + name = "no_param_service" + + def __init__(self): + super().__init__() + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, NoParamService) + + service = clean_manager.get(ServiceType.STORAGE_SERVICE) + + assert service.ready is True + + def test_service_with_non_service_params(self, clean_manager): + """Test service with parameters that aren't services.""" + + class ServiceWithConfig(Service): + name = "config_service" + + def __init__(self, config: dict): + super().__init__() + self.config = config + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, ServiceWithConfig) + + # Should fail - can't resolve dict parameter + with pytest.raises(TypeError): + clean_manager.get(ServiceType.STORAGE_SERVICE) + + +class TestConcurrentAccess: + """Test concurrent access to service manager.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_multiple_gets_return_same_instance(self, clean_manager): + """Test that multiple get calls return same instance.""" + + class SimpleService(Service): + name = "simple_service" + + def __init__(self): + super().__init__() + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, SimpleService) + + # Get multiple times + service1 = clean_manager.get(ServiceType.STORAGE_SERVICE) + service2 = clean_manager.get(ServiceType.STORAGE_SERVICE) + service3 = clean_manager.get(ServiceType.STORAGE_SERVICE) + + # Should all be the same instance + assert service1 is service2 + assert service2 is service3 + + +class TestSettingsServiceProtection: + """Test settings service protection mechanisms.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_cannot_register_settings_via_class(self, clean_manager): + """Test that settings service cannot be registered via class.""" + + class CustomSettings(Service): + name = "settings_service" + + async def teardown(self) -> None: + pass + + with pytest.raises(ValueError, match="Settings service cannot be registered"): + clean_manager.register_service_class(ServiceType.SETTINGS_SERVICE, CustomSettings) + + def test_cannot_register_settings_via_decorator(self): + """Test that settings service cannot be registered via decorator.""" + from lfx.services.registry import register_service + + with pytest.raises(ValueError, match="Settings service cannot be registered"): + + @register_service(ServiceType.SETTINGS_SERVICE) + class CustomSettings(Service): + name = "settings_service" + + async def teardown(self) -> None: + pass + + def test_settings_service_always_uses_factory(self, clean_manager): + """Test that settings service always uses factory.""" + settings = clean_manager.get(ServiceType.SETTINGS_SERVICE) + + from lfx.services.settings.service import SettingsService + + # Should be the built-in SettingsService + assert isinstance(settings, SettingsService) + + def test_cannot_override_settings_in_config(self, clean_manager, tmp_path): + """Test that settings service in config is ignored.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +settings_service = "some.custom:SettingsService" +""" + ) + + # Should not raise, but should ignore the settings_service entry + clean_manager.discover_plugins(config_dir) + + # Settings should not be in service_classes + assert ServiceType.SETTINGS_SERVICE not in clean_manager.service_classes + + # Getting settings should still work (via factory) + settings = clean_manager.get(ServiceType.SETTINGS_SERVICE) + + from lfx.services.settings.service import SettingsService + + assert isinstance(settings, SettingsService) diff --git a/src/lfx/tests/unit/services/test_integration.py b/src/lfx/tests/unit/services/test_integration.py new file mode 100644 index 0000000000..b1f21a407d --- /dev/null +++ b/src/lfx/tests/unit/services/test_integration.py @@ -0,0 +1,492 @@ +"""Integration tests for pluggable service system.""" + +import os + +import pytest +from lfx.services.base import Service +from lfx.services.manager import ServiceManager +from lfx.services.schema import ServiceType + + +class TestStandaloneLFX: + """Test LFX running standalone without langflow.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_minimal_storage_service_loads(self, clean_manager): + """Test that minimal storage service loads by default.""" + from lfx.services.storage.local import LocalStorageService + + # Register the minimal storage service + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + + storage = clean_manager.get(ServiceType.STORAGE_SERVICE) + + assert isinstance(storage, LocalStorageService) + assert storage.ready is True + + def test_minimal_telemetry_service_loads(self, clean_manager): + """Test that minimal telemetry service loads by default.""" + # Should fall back to factory since no plugin registered + + # Telemetry doesn't have a default factory, so should fail + # unless we register it first + from lfx.services.telemetry.service import TelemetryService + + clean_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService) + + telemetry = clean_manager.get(ServiceType.TELEMETRY_SERVICE) + assert isinstance(telemetry, TelemetryService) + assert telemetry.ready is True + + def test_minimal_variable_service_loads(self, clean_manager): + """Test that minimal variable service loads by default.""" + from lfx.services.variable.service import VariableService + + clean_manager.register_service_class(ServiceType.VARIABLE_SERVICE, VariableService) + + variables = clean_manager.get(ServiceType.VARIABLE_SERVICE) + assert isinstance(variables, VariableService) + assert variables.ready is True + + def test_settings_service_always_available(self, clean_manager): + """Test that settings service is always available.""" + settings = clean_manager.get(ServiceType.SETTINGS_SERVICE) + + from lfx.services.settings.service import SettingsService + + assert isinstance(settings, SettingsService) + assert settings.ready is True + + +class TestLFXWithLangflowConfig: + """Test LFX with langflow configuration.""" + + @pytest.fixture + def langflow_config_dir(self, tmp_path): + """Create a temporary langflow-style config directory.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + # Create lfx.toml with langflow services + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +cache_service = "lfx.services.cache.service:ThreadingInMemoryCache" +""" + ) + + return config_dir + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_config_overrides_defaults(self, clean_manager, langflow_config_dir): + """Test that config file overrides default services.""" + # Discover plugins from config + clean_manager.discover_plugins(langflow_config_dir) + + # Storage should be loaded from config + assert ServiceType.STORAGE_SERVICE in clean_manager.service_classes + + from lfx.services.storage.local import LocalStorageService + + assert clean_manager.service_classes[ServiceType.STORAGE_SERVICE] == LocalStorageService + + def test_multiple_services_from_config(self, clean_manager, langflow_config_dir): + """Test loading multiple services from config.""" + clean_manager.discover_plugins(langflow_config_dir) + + # Both services should be registered + assert ServiceType.STORAGE_SERVICE in clean_manager.service_classes + assert ServiceType.CACHE_SERVICE in clean_manager.service_classes + + +class TestServiceOverrideScenarios: + """Test various service override scenarios.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_decorator_overrides_config(self, clean_manager, tmp_path): + """Test that decorator registration overrides config.""" + # First create config + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + # Load from config + clean_manager.discover_plugins(config_dir) + + # Now override with direct registration (simulating decorator) + class CustomStorageService(Service): + @property + def name(self) -> str: + return "storage_service" + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, CustomStorageService, override=True) + + # Should use the override version + assert clean_manager.service_classes[ServiceType.STORAGE_SERVICE] == CustomStorageService + + def test_override_false_preserves_existing(self, clean_manager): + """Test that override=False preserves existing registration.""" + + class FirstService(Service): + name = "storage_service" + + async def teardown(self) -> None: + pass + + class SecondService(Service): + name = "storage_service" + + async def teardown(self) -> None: + pass + + # Register first + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, FirstService, override=True) + + # Try to register second with override=False + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, SecondService, override=False) + + # Should still be first + assert clean_manager.service_classes[ServiceType.STORAGE_SERVICE] == FirstService + + +class TestErrorConditions: + """Test error handling in various conditions.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_missing_import_in_config(self, clean_manager, tmp_path): + """Test handling of missing import in config.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "nonexistent.module:NonexistentClass" +""" + ) + + # Should not raise, just log warning + clean_manager.discover_plugins(config_dir) + + # Service should not be registered + assert ServiceType.STORAGE_SERVICE not in clean_manager.service_classes + + def test_invalid_service_type_in_config(self, clean_manager, tmp_path): + """Test handling of invalid service type in config.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +invalid_service_type = "some.module:SomeClass" +""" + ) + + # Should not raise, just log warning + clean_manager.discover_plugins(config_dir) + + # No services should be registered (invalid key) + assert len(clean_manager.service_classes) == 0 + + def test_malformed_toml_in_config(self, clean_manager, tmp_path): + """Test handling of malformed TOML.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + # Should not raise, just log warning + clean_manager.discover_plugins(config_dir) + + # No services should be registered + assert len(clean_manager.service_classes) == 0 + + def test_service_without_name_attribute(self, clean_manager): + """Test registering a service without name attribute.""" + + class InvalidService(Service): + async def teardown(self) -> None: + pass + + # Should not raise during registration + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, InvalidService) + + # But should fail during creation (can't instantiate abstract class) + with pytest.raises(TypeError, match="Can't instantiate abstract class"): + clean_manager.get(ServiceType.STORAGE_SERVICE) + + def test_service_initialization_failure(self, clean_manager): + """Test handling of service initialization failure.""" + + class FailingService(Service): + name = "storage_service" + + def __init__(self): + msg = "Initialization failed" + raise RuntimeError(msg) + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, FailingService) + + # Should raise the initialization error + with pytest.raises(RuntimeError, match="Initialization failed"): + clean_manager.get(ServiceType.STORAGE_SERVICE) + + +class TestDependencyResolution: + """Test dependency resolution and injection.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_service_with_settings_dependency(self, clean_manager): + """Test service that depends on settings service.""" + + class ServiceWithSettings(Service): + name = "test_service" + + def __init__(self, settings_service): + super().__init__() + self.settings = settings_service + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, ServiceWithSettings) + + service = clean_manager.get(ServiceType.STORAGE_SERVICE) + + # Should have settings injected + from lfx.services.settings.service import SettingsService + + assert isinstance(service.settings, SettingsService) + + def test_service_with_multiple_dependencies(self, clean_manager): + """Test service with multiple dependencies.""" + + class SimpleService(Service): + name = "simple_service" + + def __init__(self): + super().__init__() + self.set_ready() + + async def teardown(self) -> None: + pass + + class ComplexService(Service): + name = "complex_service" + + def __init__(self, settings_service, storage_service): + super().__init__() + self.settings = settings_service + self.storage = storage_service + self.set_ready() + + async def teardown(self) -> None: + pass + + # Register both + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, SimpleService) + clean_manager.register_service_class(ServiceType.CACHE_SERVICE, ComplexService) + + # Create complex service + complex_service = clean_manager.get(ServiceType.CACHE_SERVICE) + + # Should have both dependencies + from lfx.services.settings.service import SettingsService + + assert isinstance(complex_service.settings, SettingsService) + assert isinstance(complex_service.storage, SimpleService) + + def test_service_with_unresolvable_dependency(self, clean_manager): + """Test service with dependency that can't be resolved.""" + + class ServiceWithUnknownDep(Service): + name = "test_service" + + def __init__(self, unknown_param: str): + super().__init__() + self.unknown_param = unknown_param + self.set_ready() + + async def teardown(self) -> None: + pass + + clean_manager.register_service_class(ServiceType.STORAGE_SERVICE, ServiceWithUnknownDep) + + # Should raise due to missing parameter + with pytest.raises(TypeError): + clean_manager.get(ServiceType.STORAGE_SERVICE) + + +class TestConfigFileDiscovery: + """Test configuration file discovery.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_pyproject_toml_discovery(self, clean_manager, tmp_path): + """Test discovering services from pyproject.toml.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "pyproject.toml" + config_file.write_text( + """ +[tool.lfx.services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + clean_manager.discover_plugins(config_dir) + + assert ServiceType.STORAGE_SERVICE in clean_manager.service_classes + + def test_lfx_toml_takes_precedence(self, clean_manager, tmp_path): + """Test that lfx.toml takes precedence over pyproject.toml.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + # Create both files with different services + (config_dir / "lfx.toml").write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + (config_dir / "pyproject.toml").write_text( + """ +[tool.lfx.services] +cache_service = "lfx.services.cache.service:ThreadingInMemoryCache" +""" + ) + + clean_manager.discover_plugins(config_dir) + + # Should only have storage from lfx.toml + assert ServiceType.STORAGE_SERVICE in clean_manager.service_classes + assert ServiceType.CACHE_SERVICE not in clean_manager.service_classes + + def test_no_config_file_no_error(self, clean_manager, tmp_path): + """Test that missing config files don't cause errors.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + # Should not raise + clean_manager.discover_plugins(config_dir) + + assert len(clean_manager.service_classes) == 0 + + +class TestEnvironmentVariableIntegration: + """Test environment variable integration with services.""" + + @pytest.fixture + def clean_manager(self): + """Create a clean ServiceManager instance.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + def test_variable_service_uses_env(self, clean_manager): + """Test that variable service reads from environment.""" + from lfx.services.variable.service import VariableService + + clean_manager.register_service_class(ServiceType.VARIABLE_SERVICE, VariableService) + + os.environ["TEST_API_KEY"] = "test_value_123" # pragma: allowlist secret + try: + variables = clean_manager.get(ServiceType.VARIABLE_SERVICE) + value = variables.get_variable("TEST_API_KEY") + assert value == "test_value_123" + finally: + del os.environ["TEST_API_KEY"] + + def test_variable_service_in_memory_overrides_env(self, clean_manager): + """Test that in-memory variables override environment.""" + from lfx.services.variable.service import VariableService + + clean_manager.register_service_class(ServiceType.VARIABLE_SERVICE, VariableService) + + os.environ["TEST_VAR"] = "env_value" + try: + variables = clean_manager.get(ServiceType.VARIABLE_SERVICE) + variables.set_variable("TEST_VAR", "memory_value") + value = variables.get_variable("TEST_VAR") + assert value == "memory_value" + finally: + del os.environ["TEST_VAR"] diff --git a/src/lfx/tests/unit/services/test_minimal_services.py b/src/lfx/tests/unit/services/test_minimal_services.py new file mode 100644 index 0000000000..eef985f663 --- /dev/null +++ b/src/lfx/tests/unit/services/test_minimal_services.py @@ -0,0 +1,256 @@ +"""Tests for minimal service implementations in LFX.""" + +import os + +import pytest +from lfx.services.storage.local import LocalStorageService +from lfx.services.telemetry.service import TelemetryService +from lfx.services.tracing.service import TracingService +from lfx.services.variable.service import VariableService + + +class TestLocalStorageService: + """Tests for LocalStorageService.""" + + @pytest.fixture + def storage(self, tmp_path): + """Create a storage service with temp directory.""" + return LocalStorageService(data_dir=tmp_path) + + @pytest.mark.asyncio + async def test_save_and_get_file(self, storage): + """Test saving and retrieving a file.""" + data = b"test content" + await storage.save_file("flow_123", "test.txt", data) + + retrieved = await storage.get_file("flow_123", "test.txt") + assert retrieved == data + + @pytest.mark.asyncio + async def test_list_files(self, storage): + """Test listing files in a flow.""" + await storage.save_file("flow_123", "file1.txt", b"content1") + await storage.save_file("flow_123", "file2.txt", b"content2") + + files = await storage.list_files("flow_123") + assert len(files) == 2 + assert "file1.txt" in files + assert "file2.txt" in files + + @pytest.mark.asyncio + async def test_delete_file(self, storage): + """Test deleting a file.""" + await storage.save_file("flow_123", "test.txt", b"content") + await storage.delete_file("flow_123", "test.txt") + + with pytest.raises(FileNotFoundError): + await storage.get_file("flow_123", "test.txt") + + @pytest.mark.asyncio + async def test_get_file_size(self, storage): + """Test getting file size.""" + data = b"test content" + await storage.save_file("flow_123", "test.txt", data) + + size = await storage.get_file_size("flow_123", "test.txt") + assert size == len(data) + + @pytest.mark.asyncio + async def test_get_nonexistent_file(self, storage): + """Test getting a file that doesn't exist.""" + with pytest.raises(FileNotFoundError): + await storage.get_file("flow_123", "nonexistent.txt") + + def test_build_full_path(self, storage, tmp_path): + """Test building full file path.""" + path = storage.build_full_path("flow_123", "test.txt") + expected = str(tmp_path / "flow_123" / "test.txt") + assert path == expected + + @pytest.mark.asyncio + async def test_list_files_empty_flow(self, storage): + """Test listing files in nonexistent flow.""" + files = await storage.list_files("nonexistent_flow") + assert files == [] + + def test_service_ready(self, storage): + """Test that service is marked as ready.""" + assert storage.ready is True + assert storage.name == "storage_service" + + @pytest.mark.asyncio + async def test_teardown(self, storage): + """Test service teardown.""" + await storage.teardown() + # Should not raise + + +class TestTelemetryService: + """Tests for minimal TelemetryService.""" + + @pytest.fixture + def telemetry(self): + """Create a telemetry service.""" + return TelemetryService() + + def test_service_ready(self, telemetry): + """Test that service is ready.""" + assert telemetry.ready is True + assert telemetry.name == "telemetry_service" + + @pytest.mark.asyncio + async def test_log_exception(self, telemetry): + """Test logging an exception (noop).""" + # Should not raise + exc = ValueError("test error") + await telemetry.log_exception(exc, "test_context") + + @pytest.mark.asyncio + async def test_log_package_version(self, telemetry): + """Test logging package version (noop).""" + # Should not raise + await telemetry.log_package_version() + + @pytest.mark.asyncio + async def test_teardown(self, telemetry): + """Test service teardown.""" + await telemetry.teardown() + # Should not raise + + +class TestTracingService: + """Tests for minimal TracingService.""" + + @pytest.fixture + def tracing(self): + """Create a tracing service.""" + return TracingService() + + def test_service_ready(self, tracing): + """Test that service is ready.""" + assert tracing.ready is True + assert tracing.name == "tracing_service" + + def test_add_log(self, tracing): + """Test adding a log entry (outputs to debug).""" + # Should not raise + tracing.add_log("test_trace", {"message": "test log"}) + + @pytest.mark.asyncio + async def test_teardown(self, tracing): + """Test service teardown.""" + await tracing.teardown() + # Should not raise + + +class TestVariableService: + """Tests for minimal VariableService.""" + + @pytest.fixture + def variables(self): + """Create a variable service.""" + return VariableService() + + def test_service_ready(self, variables): + """Test that service is ready.""" + assert variables.ready is True + assert variables.name == "variable_service" + + def test_set_and_get_variable(self, variables): + """Test setting and getting a variable.""" + variables.set_variable("test_key", "test_value") + value = variables.get_variable("test_key") + assert value == "test_value" + + def test_get_from_environment(self, variables): + """Test getting variable from environment.""" + os.environ["TEST_ENV_VAR"] = "env_value" + try: + value = variables.get_variable("TEST_ENV_VAR") + assert value == "env_value" + finally: + del os.environ["TEST_ENV_VAR"] + + def test_get_nonexistent_variable(self, variables): + """Test getting a variable that doesn't exist.""" + value = variables.get_variable("nonexistent_key") + assert value is None + + def test_delete_variable(self, variables): + """Test deleting a variable.""" + variables.set_variable("test_key", "test_value") + variables.delete_variable("test_key") + value = variables.get_variable("test_key") + assert value is None + + def test_list_variables(self, variables): + """Test listing variables.""" + variables.set_variable("key1", "value1") + variables.set_variable("key2", "value2") + + vars_list = variables.list_variables() + assert "key1" in vars_list + assert "key2" in vars_list + + def test_in_memory_overrides_env(self, variables): + """Test that in-memory variables override environment.""" + os.environ["TEST_VAR"] = "env_value" + try: + variables.set_variable("TEST_VAR", "memory_value") + value = variables.get_variable("TEST_VAR") + assert value == "memory_value" + finally: + del os.environ["TEST_VAR"] + + @pytest.mark.asyncio + async def test_teardown(self, variables): + """Test service teardown clears variables.""" + variables.set_variable("test_key", "test_value") + await variables.teardown() + # Variables should be cleared (verify via public API) + assert variables.list_variables() == [] + assert variables.get_variable("test_key") is None + + +class TestMinimalServicesIntegration: + """Integration tests for minimal services working together.""" + + @pytest.mark.asyncio + async def test_all_minimal_services_initialize(self, tmp_path): + """Test that all minimal services can be initialized.""" + storage = LocalStorageService(data_dir=tmp_path) + telemetry = TelemetryService() + tracing = TracingService() + variables = VariableService() + + assert storage.ready + assert telemetry.ready + assert tracing.ready + assert variables.ready + + @pytest.mark.asyncio + async def test_minimal_services_teardown_all(self, tmp_path): + """Test tearing down all minimal services.""" + storage = LocalStorageService(data_dir=tmp_path) + telemetry = TelemetryService() + tracing = TracingService() + variables = VariableService() + + # Should all teardown without errors + await storage.teardown() + await telemetry.teardown() + await tracing.teardown() + await variables.teardown() + + @pytest.mark.asyncio + async def test_storage_with_tracing(self, tmp_path): + """Test using storage with tracing.""" + storage = LocalStorageService(data_dir=tmp_path) + tracing = TracingService() + + tracing.add_log("storage_test", {"operation": "save", "flow_id": "123"}) + await storage.save_file("flow_123", "test.txt", b"content") + tracing.add_log("storage_test", {"operation": "saved", "flow_id": "123"}) + + # Should complete without errors + assert await storage.get_file("flow_123", "test.txt") == b"content" diff --git a/src/lfx/tests/unit/services/test_service_manager.py b/src/lfx/tests/unit/services/test_service_manager.py new file mode 100644 index 0000000000..943e5ad44b --- /dev/null +++ b/src/lfx/tests/unit/services/test_service_manager.py @@ -0,0 +1,479 @@ +"""Tests for the ServiceManager plugin system.""" + +from pathlib import Path + +import pytest +from lfx.services.base import Service +from lfx.services.manager import NoFactoryRegisteredError, ServiceManager +from lfx.services.schema import ServiceType +from lfx.services.storage.local import LocalStorageService +from lfx.services.telemetry.service import TelemetryService +from lfx.services.tracing.service import TracingService +from lfx.services.variable.service import VariableService + + +@pytest.fixture +def service_manager(): + """Create a fresh ServiceManager for each test.""" + manager = ServiceManager() + yield manager + # Cleanup + import asyncio + + asyncio.run(manager.teardown()) + + +@pytest.fixture +def temp_config_dir(tmp_path): + """Create a temporary config directory.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + return config_dir + + +class TestServiceRegistration: + """Tests for service registration with real implementations.""" + + def test_register_storage_service(self, service_manager): + """Test registering the real LocalStorageService.""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService, override=True) + + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + assert service_manager.service_classes[ServiceType.STORAGE_SERVICE] == LocalStorageService + + def test_register_multiple_real_services(self, service_manager): + """Test registering multiple real services.""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService, override=True) + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService, override=True) + service_manager.register_service_class(ServiceType.TRACING_SERVICE, TracingService, override=True) + service_manager.register_service_class(ServiceType.VARIABLE_SERVICE, VariableService, override=True) + + assert len(service_manager.service_classes) == 4 + assert service_manager.service_classes[ServiceType.STORAGE_SERVICE] == LocalStorageService + assert service_manager.service_classes[ServiceType.TELEMETRY_SERVICE] == TelemetryService + assert service_manager.service_classes[ServiceType.TRACING_SERVICE] == TracingService + assert service_manager.service_classes[ServiceType.VARIABLE_SERVICE] == VariableService + + def test_register_service_class_no_override(self, service_manager): + """Test that override=False prevents replacement.""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService, override=True) + + # Try to register different class with override=False + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, TelemetryService, override=False) + + # Should still have the original + assert service_manager.service_classes[ServiceType.STORAGE_SERVICE] == LocalStorageService + + def test_register_service_class_with_override(self, service_manager): + """Test that override=True replaces existing registration.""" + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService, override=True) + + # Override with different service + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TracingService, override=True) + + # Should have the new one + assert service_manager.service_classes[ServiceType.TELEMETRY_SERVICE] == TracingService + + def test_cannot_register_settings_service(self, service_manager): + """Test that settings service cannot be registered via plugins.""" + with pytest.raises(ValueError, match="Settings service cannot be registered"): + service_manager.register_service_class(ServiceType.SETTINGS_SERVICE, LocalStorageService) + + +class TestPluginDiscovery: + """Tests for plugin discovery with real service paths.""" + + def test_discover_storage_from_config_file(self, service_manager, temp_config_dir): + """Test discovering LocalStorageService from lfx.toml.""" + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + assert service_manager.service_classes[ServiceType.STORAGE_SERVICE] == LocalStorageService + + def test_discover_multiple_services_from_config(self, service_manager, temp_config_dir): + """Test discovering multiple real services from config.""" + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +telemetry_service = "lfx.services.telemetry.service:TelemetryService" +tracing_service = "lfx.services.tracing.service:TracingService" +variable_service = "lfx.services.variable.service:VariableService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + assert ServiceType.TELEMETRY_SERVICE in service_manager.service_classes + assert ServiceType.TRACING_SERVICE in service_manager.service_classes + assert ServiceType.VARIABLE_SERVICE in service_manager.service_classes + + def test_discover_from_pyproject_toml(self, service_manager, temp_config_dir): + """Test discovering services from pyproject.toml.""" + config_file = temp_config_dir / "pyproject.toml" + config_file.write_text( + """ +[tool.lfx.services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + + def test_lfx_toml_takes_precedence_over_pyproject(self, service_manager, temp_config_dir): + """Test that lfx.toml is preferred over pyproject.toml.""" + # Create both files + (temp_config_dir / "lfx.toml").write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + (temp_config_dir / "pyproject.toml").write_text( + """ +[tool.lfx.services] +telemetry_service = "lfx.services.telemetry.service:TelemetryService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + + # Should have loaded from lfx.toml (storage_service) + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + # Should NOT have loaded from pyproject.toml (telemetry_service) + assert ServiceType.TELEMETRY_SERVICE not in service_manager.service_classes + + def test_discover_plugins_only_once(self, service_manager, temp_config_dir): + """Test that plugin discovery only runs once.""" + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + initial_count = len(service_manager.service_classes) + + # Try to discover again + service_manager.discover_plugins(temp_config_dir) + + # Should not have changed + assert len(service_manager.service_classes) == initial_count + + def test_invalid_service_key_in_config(self, service_manager, temp_config_dir): + """Test that invalid service keys are ignored.""" + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +invalid_service_key = "some.module:SomeClass" # pragma: allowlist secret +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + # Should not raise, just log warning + service_manager.discover_plugins(temp_config_dir) + + # Valid service should still be registered + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + + def test_invalid_import_path_in_config(self, service_manager, temp_config_dir): + """Test that invalid import paths are handled gracefully.""" + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "nonexistent.module:NonexistentClass" +""" + ) + + # Should not raise, just log warning + service_manager.discover_plugins(temp_config_dir) + + # Service should not be registered + assert ServiceType.STORAGE_SERVICE not in service_manager.service_classes + + +class TestServiceCreation: + """Tests for creating real services with dependency injection.""" + + def test_create_storage_service(self, service_manager): + """Test creating LocalStorageService.""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + + service = service_manager.get(ServiceType.STORAGE_SERVICE) + + assert isinstance(service, LocalStorageService) + assert service.ready is True + assert service.name == "storage_service" + + def test_create_telemetry_service(self, service_manager): + """Test creating TelemetryService.""" + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService) + + service = service_manager.get(ServiceType.TELEMETRY_SERVICE) + + assert isinstance(service, TelemetryService) + assert service.ready is True + assert service.name == "telemetry_service" + + def test_create_tracing_service(self, service_manager): + """Test creating TracingService.""" + service_manager.register_service_class(ServiceType.TRACING_SERVICE, TracingService) + + service = service_manager.get(ServiceType.TRACING_SERVICE) + + assert isinstance(service, TracingService) + assert service.ready is True + assert service.name == "tracing_service" + + def test_create_variable_service(self, service_manager): + """Test creating VariableService.""" + service_manager.register_service_class(ServiceType.VARIABLE_SERVICE, VariableService) + + service = service_manager.get(ServiceType.VARIABLE_SERVICE) + + assert isinstance(service, VariableService) + assert service.ready is True + assert service.name == "variable_service" + + def test_create_service_with_settings_dependency(self, service_manager): + """Test creating a service that depends on settings.""" + + # Create a real service that needs settings + class ServiceWithSettings(Service): + @property + def name(self) -> str: + return "test_service" + + def __init__(self, settings_service): + super().__init__() + self.settings = settings_service + self.set_ready() + + async def teardown(self) -> None: + pass + + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, ServiceWithSettings) + + service = service_manager.get(ServiceType.STORAGE_SERVICE) + + # Should have settings injected + from lfx.services.settings.service import SettingsService + + assert isinstance(service, ServiceWithSettings) + assert isinstance(service.settings, SettingsService) + assert service.ready is True + + def test_create_service_caching(self, service_manager): + """Test that services are cached (singleton).""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + + service1 = service_manager.get(ServiceType.STORAGE_SERVICE) + service2 = service_manager.get(ServiceType.STORAGE_SERVICE) + + assert service1 is service2 + + def test_settings_service_always_uses_factory(self, service_manager): + """Test that settings service always uses factory.""" + service = service_manager.get(ServiceType.SETTINGS_SERVICE) + + from lfx.services.settings.service import SettingsService + + assert isinstance(service, SettingsService) + + def test_fallback_to_factory_if_no_plugin(self, service_manager): + """Test that services fall back to factory if no plugin registered.""" + # Don't register any plugin for storage + # Should fail since there's no factory either + with pytest.raises(NoFactoryRegisteredError): + service_manager.get(ServiceType.STORAGE_SERVICE) + + +class TestConflictResolution: + """Tests for conflict resolution with real services.""" + + def test_direct_registration_overrides_config(self, service_manager, temp_config_dir): + """Test that direct registration overrides config file.""" + # First load from config + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +telemetry_service = "lfx.services.telemetry.service:TelemetryService" +""" + ) + service_manager.discover_plugins(temp_config_dir) + + # Then register via direct call (override=True) + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TracingService, override=True) + + # Should use the directly registered service + assert service_manager.service_classes[ServiceType.TELEMETRY_SERVICE] == TracingService + + +class TestTeardown: + """Tests for service teardown with real services.""" + + @pytest.mark.asyncio + async def test_teardown_all_services(self, service_manager): + """Test that teardown clears all services.""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService) + service_manager.get(ServiceType.STORAGE_SERVICE) + service_manager.get(ServiceType.TELEMETRY_SERVICE) + + await service_manager.teardown() + + assert len(service_manager.services) == 0 + assert len(service_manager.factories) == 0 + + @pytest.mark.asyncio + async def test_teardown_calls_service_teardown(self, service_manager): + """Test that teardown calls each service's teardown method.""" + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + storage = service_manager.get(ServiceType.STORAGE_SERVICE) + + # Service should exist + assert storage is not None + + await service_manager.teardown() + + # Services should be cleared + assert ServiceType.STORAGE_SERVICE not in service_manager.services + + +class TestConfigDirectorySource: + """Tests for config_dir parameter with real services.""" + + def test_config_dir_from_settings_service(self, service_manager): + """Test that config_dir comes from settings service.""" + # Create settings service first + settings_service = service_manager.get(ServiceType.SETTINGS_SERVICE) + + # Create config in the settings config_dir + config_dir = Path(settings_service.settings.config_dir) + config_dir.mkdir(parents=True, exist_ok=True) + + config_file = config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + # Discover plugins (should use settings.config_dir) + service_manager._plugins_discovered = False # Reset flag + service_manager.discover_plugins() + + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + + def test_config_dir_falls_back_to_cwd(self, service_manager, temp_config_dir): + """Test that config_dir falls back to cwd if settings not available.""" + # Don't create settings service + # Should fall back to provided config_dir + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + + # Should have searched temp_config_dir (passed as param) + assert service_manager._plugins_discovered is True + assert ServiceType.STORAGE_SERVICE in service_manager.service_classes + + +class TestRealWorldScenarios: + """Tests for realistic usage scenarios.""" + + @pytest.mark.asyncio + async def test_complete_service_lifecycle(self, service_manager): + """Test complete lifecycle: register, create, use, teardown.""" + # Register + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + + # Create + storage = service_manager.get(ServiceType.STORAGE_SERVICE) + assert storage.ready is True + + # Use + await storage.save_file("test_flow", "test.txt", b"test content") + content = await storage.get_file("test_flow", "test.txt") + assert content == b"test content" + + # Teardown + await service_manager.teardown() + assert ServiceType.STORAGE_SERVICE not in service_manager.services + + def test_multiple_services_working_together(self, service_manager): + """Test multiple services can coexist and work together.""" + # Register all minimal services + service_manager.register_service_class(ServiceType.STORAGE_SERVICE, LocalStorageService) + service_manager.register_service_class(ServiceType.TELEMETRY_SERVICE, TelemetryService) + service_manager.register_service_class(ServiceType.TRACING_SERVICE, TracingService) + service_manager.register_service_class(ServiceType.VARIABLE_SERVICE, VariableService) + + # Create all services + storage = service_manager.get(ServiceType.STORAGE_SERVICE) + telemetry = service_manager.get(ServiceType.TELEMETRY_SERVICE) + tracing = service_manager.get(ServiceType.TRACING_SERVICE) + variables = service_manager.get(ServiceType.VARIABLE_SERVICE) + + # All should be ready + assert storage.ready is True + assert telemetry.ready is True + assert tracing.ready is True + assert variables.ready is True + + # All should be usable + tracing.add_log("test_trace", {"message": "test"}) + variables.set_variable("TEST_KEY", "test_value") + assert variables.get_variable("TEST_KEY") == "test_value" + + def test_config_file_with_all_minimal_services(self, service_manager, temp_config_dir): + """Test loading all minimal services from config file.""" + config_file = temp_config_dir / "lfx.toml" + config_file.write_text( + """ +[services] +storage_service = "lfx.services.storage.local:LocalStorageService" +telemetry_service = "lfx.services.telemetry.service:TelemetryService" +tracing_service = "lfx.services.tracing.service:TracingService" +variable_service = "lfx.services.variable.service:VariableService" +""" + ) + + service_manager.discover_plugins(temp_config_dir) + + # All services should be registered + assert len(service_manager.service_classes) == 4 + + # Create and verify each service + storage = service_manager.get(ServiceType.STORAGE_SERVICE) + telemetry = service_manager.get(ServiceType.TELEMETRY_SERVICE) + tracing = service_manager.get(ServiceType.TRACING_SERVICE) + variables = service_manager.get(ServiceType.VARIABLE_SERVICE) + + assert isinstance(storage, LocalStorageService) + assert isinstance(telemetry, TelemetryService) + assert isinstance(tracing, TracingService) + assert isinstance(variables, VariableService)