jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

refactor(core): implement centralized dynamic lazy import system for components (#8932)

Browse source 
* feat: add import utilities for LangFlow components

- Introduced a new module `_importing.py` containing the `import_mod` function.
- This function dynamically imports attributes from specified modules, enhancing modularity and flexibility in component initialization.
- Comprehensive docstring added for clarity on usage and parameters.

* feat: implement dynamic imports for LangFlow components

- Added dynamic import functionality to various LangFlow components, allowing for lazy loading of attributes on access.
- Introduced  mapping in each component's  to manage imports efficiently.
- Enhanced error handling for import failures, providing clearer messages for missing attributes.
- Updated  method to reflect available attributes for better introspection and tab-completion support.
- Comprehensive docstrings added to improve documentation and usability.

* test: add comprehensive tests for dynamic imports and component accessibility

- Introduced integration tests for dynamic import functionality, ensuring components are discoverable and instantiable post-refactor.
- Added unit tests for the `_import_utils` module, validating the `import_mod` function's behavior and error handling.
- Implemented tests to confirm all component modules are importable and maintain backward compatibility with existing import patterns.
- Enhanced performance tests to measure lazy loading efficiency and memory usage during component access.
- Ensured that all components have the required attributes for dynamic loading and that circular imports are prevented.

* chore: update ruff pre-commit hook to version 0.12.2 in configuration file

* refactor: update warning handling for dynamic imports

- Moved the warning suppression for LangChainDeprecationWarning into the dynamic import context to ensure it only applies during the import process.
- This change enhances clarity and maintains the original functionality while improving the robustness of the import mechanism.

* test: enhance dynamic import integration tests for component attributes

- Removed unnecessary import of AgentComponent and added assertions to verify essential attributes of OpenAIModelComponent, including display_name, description, icon, and inputs.
- Ensured that each input field has the required attributes for better validation of component integrity during dynamic imports.

* refactor: update import paths for Message class in conversation utilities

- Changed the import of the Message class from langflow.field_typing to langflow.schema.message across multiple utility files, ensuring consistency and alignment with the updated module structure.
- This refactor enhances code clarity and maintains compatibility with the latest schema definitions.

* refactor: remove Vectara components from LangFlow

- Deleted the Vectara components module from the codebase, streamlining the component structure.
- This change helps to reduce complexity and maintain focus on core functionalities.

* refactor: remove Vectara references from LangFlow component imports

- Eliminated Vectara from both the import statements and dynamic imports mapping, streamlining the component structure.
- This change contributes to reducing complexity and maintaining focus on core functionalities within the LangFlow framework.

* [autofix.ci] apply automated fixes

* fix: remove 'vectara' from __all__ in components module

* refactor: improve error handling tests for dynamic imports

* test: add tests for ModuleNotFoundError handling with None and special module names

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Edwin Jose <edwin.jose@datastax.com>
This commit is contained in:
Gabriel Luiz Freitas Almeida 2025-08-05 17:48:59 -03:00 • committed by GitHub
commit 80ebe03d94
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
58 changed files with 3174 additions and 232 deletions
Download patch file Download diff file Expand all files Collapse all files

299
src/backend/tests/integration/test_dynamic_import_integration.py Normal file
View file

@ -0,0 +1,299 @@
"""Integration tests for dynamic import refactor.
Tests the dynamic import system in realistic usage scenarios to ensure
the refactor doesn't break existing functionality.
"""
import sys
import time
import pytest
from langflow.components.agents import AgentComponent
from langflow.components.data import APIRequestComponent
from langflow.components.openai import OpenAIModelComponent
class TestDynamicImportIntegration:
"""Integration tests for the dynamic import system."""
def test_component_discovery_still_works(self):
"""Test that component discovery mechanisms still work after refactor."""
# This tests that the existing component discovery logic
# can still find and load components
from langflow import components
# Test that we can discover components through the main module
openai_module = components.openai
assert hasattr(openai_module, "OpenAIModelComponent")
data_module = components.data
assert hasattr(data_module, "APIRequestComponent")
def test_existing_import_patterns_work(self):
"""Test that all existing import patterns continue to work."""
# Test direct imports
import langflow.components.data as data_comp
# Test module imports
import langflow.components.openai as openai_comp
# All should work
assert OpenAIModelComponent is not None
assert APIRequestComponent is not None
assert AgentComponent is not None
assert openai_comp.OpenAIModelComponent is not None
assert data_comp.APIRequestComponent is not None
def test_component_instantiation_works(self):
"""Test that components can still be instantiated normally."""
# Test that we can create component instances
# (Note: Some components may require specific initialization parameters)
from langflow.components.helpers import CalculatorComponent
# Should be able to access the class
assert CalculatorComponent is not None
assert callable(CalculatorComponent)
def test_template_creation_compatibility(self):
"""Test that template creation still works with dynamic imports."""
# Test accessing component attributes needed for templates
# Components should have all necessary attributes for template creation
assert hasattr(OpenAIModelComponent, "__name__")
assert hasattr(OpenAIModelComponent, "__module__")
assert hasattr(OpenAIModelComponent, "display_name")
assert isinstance(OpenAIModelComponent.display_name, str)
assert OpenAIModelComponent.display_name
assert hasattr(OpenAIModelComponent, "description")
assert isinstance(OpenAIModelComponent.description, str)
assert OpenAIModelComponent.description
assert hasattr(OpenAIModelComponent, "icon")
assert isinstance(OpenAIModelComponent.icon, str)
assert OpenAIModelComponent.icon
assert hasattr(OpenAIModelComponent, "inputs")
assert isinstance(OpenAIModelComponent.inputs, list)
assert len(OpenAIModelComponent.inputs) > 0
# Check that each input has required attributes
for input_field in OpenAIModelComponent.inputs:
assert hasattr(input_field, "name"), f"Input {input_field} missing 'name' attribute"
assert hasattr(input_field, "display_name"), f"Input {input_field} missing 'display_name' attribute"
def test_multiple_import_styles_same_result(self):
"""Test that different import styles yield the same component."""
# Import the same component in different ways
from langflow import components
from langflow.components.openai import OpenAIModelComponent as DirectImport
dynamic_import = components.openai.OpenAIModelComponent
import langflow.components.openai as openai_module
module_import = openai_module.OpenAIModelComponent
# All three should be the exact same class object
assert DirectImport is dynamic_import
assert dynamic_import is module_import
assert DirectImport is module_import
def test_startup_performance_improvement(self):
"""Test that startup time is improved with lazy loading."""
# This test measures the difference in import time
# Fresh modules to test startup behavior
modules_to_clean = [
"langflow.components.vectorstores",
"langflow.components.tools",
"langflow.components.langchain_utilities",
]
for module_name in modules_to_clean:
if module_name in sys.modules:
del sys.modules[module_name]
# Time the import of a large module
start_time = time.time()
from langflow.components import vectorstores
import_time = time.time() - start_time
# Import time should be very fast (just loading the __init__.py)
assert import_time < 0.1 # Should be well under 100ms
# Test that we can access a component (it may already be cached from previous tests)
# This is expected behavior in a test suite where components get cached
# Now access a component - this should trigger loading
start_time = time.time()
chroma_component = vectorstores.ChromaVectorStoreComponent
access_time = time.time() - start_time
assert chroma_component is not None
# Access time should still be reasonable
assert access_time < 2.0 # Should be under 2 seconds
def test_memory_usage_efficiency(self):
"""Test that memory usage is more efficient with lazy loading."""
from langflow.components import processing
# Count currently loaded components
initial_component_count = len([k for k in processing.__dict__ if k.endswith("Component")])
# Access just one component
combine_text = processing.CombineTextComponent
assert combine_text is not None
# At least one more component should be loaded now
after_one_access = len([k for k in processing.__dict__ if k.endswith("Component")])
assert after_one_access >= initial_component_count
# Access another component
split_text = processing.SplitTextComponent
assert split_text is not None
# Should have at least one more component loaded
after_two_access = len([k for k in processing.__dict__ if k.endswith("Component")])
assert after_two_access >= after_one_access
def test_error_handling_in_realistic_scenarios(self):
"""Test error handling in realistic usage scenarios."""
from langflow import components
# Test accessing non-existent component category
with pytest.raises(AttributeError):
_ = components.nonexistent_category
# Test accessing non-existent component in valid category
with pytest.raises(AttributeError):
_ = components.openai.NonExistentComponent
def test_ide_autocomplete_support(self):
"""Test that IDE autocomplete support still works."""
import langflow.components.openai as openai_components
from langflow import components
# __dir__ should return all available components/modules
main_dir = dir(components)
assert "openai" in main_dir
assert "data" in main_dir
assert "agents" in main_dir
openai_dir = dir(openai_components)
assert "OpenAIModelComponent" in openai_dir
assert "OpenAIEmbeddingsComponent" in openai_dir
def test_concurrent_access(self):
"""Test that concurrent access to components works correctly."""
import threading
from langflow.components import helpers
results = []
errors = []
def access_component():
try:
component = helpers.CalculatorComponent
results.append(component)
except Exception as e:
errors.append(e)
# Create multiple threads accessing the same component
threads = []
for _ in range(5):
thread = threading.Thread(target=access_component)
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
# Should have no errors
assert len(errors) == 0
assert len(results) == 5
# All results should be the same component class
first_result = results[0]
for result in results[1:]:
assert result is first_result
def test_circular_import_prevention(self):
"""Test that the refactor doesn't introduce circular imports."""
# This test ensures that importing components doesn't create
# circular dependency issues
# These imports should work without circular import errors
from langflow import components
from langflow.components import openai
# Access components in different orders
model1 = components.openai.OpenAIModelComponent
model2 = openai.OpenAIModelComponent
model3 = OpenAIModelComponent
# All should be the same
assert model1 is model2 is model3
def test_large_scale_component_access(self):
"""Test accessing many components doesn't cause issues."""
from langflow.components import vectorstores
# Access multiple components rapidly
components_accessed = []
component_names = [
"ChromaVectorStoreComponent",
"PineconeVectorStoreComponent",
"FaissVectorStoreComponent",
"WeaviateVectorStoreComponent",
"QdrantVectorStoreComponent",
]
for name in component_names:
if hasattr(vectorstores, name):
component = getattr(vectorstores, name)
components_accessed.append(component)
# Should have accessed multiple components without issues
assert len(components_accessed) > 0
# All should be different classes
assert len(set(components_accessed)) == len(components_accessed)
def test_component_metadata_preservation(self):
"""Test that component metadata is preserved after dynamic loading."""
# Component should have all expected metadata
assert hasattr(OpenAIModelComponent, "__name__")
assert hasattr(OpenAIModelComponent, "__module__")
assert hasattr(OpenAIModelComponent, "__doc__")
# Module path should be correct
assert "openai" in OpenAIModelComponent.__module__
def test_backwards_compatibility_comprehensive(self):
"""Comprehensive test of backwards compatibility."""
# Test all major import patterns that should still work
# 1. Direct component imports
from langflow.components.data import APIRequestComponent
assert AgentComponent is not None
assert APIRequestComponent is not None
# 2. Module imports
# 3. Main module access
import langflow.components as comp
import langflow.components.helpers as helpers_mod
import langflow.components.openai as openai_mod
# 4. Nested access
nested_component = comp.openai.OpenAIModelComponent
direct_component = openai_mod.OpenAIModelComponent
# All patterns should work and yield consistent results
assert openai_mod.OpenAIModelComponent is not None
assert helpers_mod.CalculatorComponent is not None
assert nested_component is direct_component
if __name__ == "__main__":
pytest.main([__file__, "-v"])

307
src/backend/tests/unit/components/test_all_modules_importable.py Normal file
View file

@ -0,0 +1,307 @@
"""Test to ensure all component modules are importable after dynamic import refactor.
This test validates that every component module can be imported successfully
and that all components listed in __all__ can be accessed.
"""
import importlib
import pytest
from langflow import components
class TestAllModulesImportable:
"""Test that all component modules are importable."""
def test_all_component_categories_importable(self):
"""Test that all component categories in __all__ can be imported."""
failed_imports = []
for category_name in components.__all__:
try:
category_module = getattr(components, category_name)
assert category_module is not None, f"Category {category_name} is None"
# Verify it's actually a module
assert hasattr(category_module, "__name__"), f"Category {category_name} is not a module"
except Exception as e:
failed_imports.append(f"{category_name}: {e!s}")
if failed_imports:
pytest.fail(f"Failed to import categories: {failed_imports}")
def test_all_components_in_categories_importable(self):
"""Test that all components in each category's __all__ can be imported."""
failed_imports = []
successful_imports = 0
for category_name in components.__all__:
try:
category_module = getattr(components, category_name)
if hasattr(category_module, "__all__"):
for component_name in category_module.__all__:
try:
component = getattr(category_module, component_name)
assert component is not None, f"Component {component_name} is None"
assert callable(component), f"Component {component_name} is not callable"
successful_imports += 1
except Exception as e:
failed_imports.append(f"{category_name}.{component_name}: {e!s}")
else:
# Category doesn't have __all__, skip
continue
except Exception as e:
failed_imports.append(f"Category {category_name}: {e!s}")
print(f"Successfully imported {successful_imports} components") # noqa: T201
if failed_imports:
print(f"Failed imports ({len(failed_imports)}):") # noqa: T201
for failure in failed_imports[:10]: # Show first 10 failures
print(f" - {failure}") # noqa: T201
if len(failed_imports) > 10:
print(f" ... and {len(failed_imports) - 10} more") # noqa: T201
pytest.fail(f"Failed to import {len(failed_imports)} components")
def test_dynamic_imports_mapping_complete(self):
"""Test that _dynamic_imports mapping is complete for all categories."""
failed_mappings = []
for category_name in components.__all__:
try:
category_module = getattr(components, category_name)
if hasattr(category_module, "__all__") and hasattr(category_module, "_dynamic_imports"):
category_all = set(category_module.__all__)
dynamic_imports_keys = set(category_module._dynamic_imports.keys())
# Check that all items in __all__ have corresponding _dynamic_imports entries
missing_in_dynamic = category_all - dynamic_imports_keys
if missing_in_dynamic:
failed_mappings.append(f"{category_name}: Missing in _dynamic_imports: {missing_in_dynamic}")
# Check that all _dynamic_imports keys are in __all__
missing_in_all = dynamic_imports_keys - category_all
if missing_in_all:
failed_mappings.append(f"{category_name}: Missing in __all__: {missing_in_all}")
except Exception as e:
failed_mappings.append(f"{category_name}: Error checking mappings: {e!s}")
if failed_mappings:
pytest.fail(f"Inconsistent mappings: {failed_mappings}")
def test_backward_compatibility_imports(self):
"""Test that traditional import patterns still work."""
# Test some key imports that should always work
traditional_imports = [
("langflow.components.openai", "OpenAIModelComponent"),
("langflow.components.anthropic", "AnthropicModelComponent"),
("langflow.components.data", "APIRequestComponent"),
("langflow.components.agents", "AgentComponent"),
("langflow.components.helpers", "CalculatorComponent"),
]
failed_imports = []
for module_name, component_name in traditional_imports:
try:
module = importlib.import_module(module_name)
component = getattr(module, component_name)
assert component is not None
assert callable(component)
except Exception as e:
failed_imports.append(f"{module_name}.{component_name}: {e!s}")
if failed_imports:
pytest.fail(f"Traditional imports failed: {failed_imports}")
def test_component_modules_have_required_attributes(self):
"""Test that component modules have required attributes for dynamic loading."""
failed_modules = []
for category_name in components.__all__:
try:
category_module = getattr(components, category_name)
# Check for required attributes
required_attrs = ["__all__"]
failed_modules.extend(
f"{category_name}: Missing required attribute {attr}"
for attr in required_attrs
if not hasattr(category_module, attr)
)
# Check that if it has dynamic imports, it has the pattern
if hasattr(category_module, "_dynamic_imports"):
if not hasattr(category_module, "__getattr__"):
failed_modules.append(f"{category_name}: Has _dynamic_imports but no __getattr__")
if not hasattr(category_module, "__dir__"):
failed_modules.append(f"{category_name}: Has _dynamic_imports but no __dir__")
except Exception as e:
failed_modules.append(f"{category_name}: Error checking attributes: {e!s}")
if failed_modules:
pytest.fail(f"Module attribute issues: {failed_modules}")
def test_no_circular_imports(self):
"""Test that there are no circular import issues."""
# Test importing in different orders to catch circular imports
import_orders = [
["agents", "data", "openai"],
["openai", "agents", "data"],
["data", "openai", "agents"],
]
for order in import_orders:
try:
for category_name in order:
category_module = getattr(components, category_name)
# Access a component to trigger dynamic import
if hasattr(category_module, "__all__") and category_module.__all__:
first_component_name = category_module.__all__[0]
getattr(category_module, first_component_name)
except Exception as e:
pytest.fail(f"Circular import issue with order {order}: {e!s}")
def test_component_access_caching(self):
"""Test that component access caching works correctly."""
# Access the same component multiple times and ensure caching works
test_cases = [
("openai", "OpenAIModelComponent"),
("data", "APIRequestComponent"),
("helpers", "CalculatorComponent"),
]
for category_name, component_name in test_cases:
category_module = getattr(components, category_name)
# First access
component1 = getattr(category_module, component_name)
# Component should now be cached in module globals
assert component_name in category_module.__dict__
# Second access should return the same object
component2 = getattr(category_module, component_name)
assert component1 is component2, f"Caching failed for {category_name}.{component_name}"
def test_error_handling_for_missing_components(self):
"""Test that appropriate errors are raised for missing components."""
test_cases = [
("openai", "NonExistentComponent"),
("data", "AnotherNonExistentComponent"),
]
for category_name, component_name in test_cases:
category_module = getattr(components, category_name)
with pytest.raises(AttributeError, match=f"has no attribute '{component_name}'"):
getattr(category_module, component_name)
def test_dir_functionality(self):
"""Test that __dir__ functionality works for all modules."""
# Test main components module
main_dir = dir(components)
assert "openai" in main_dir
assert "data" in main_dir
assert "agents" in main_dir
# Test category modules
for category_name in ["openai", "data", "helpers"]:
category_module = getattr(components, category_name)
category_dir = dir(category_module)
# Should include all components from __all__
if hasattr(category_module, "__all__"):
for component_name in category_module.__all__:
assert component_name in category_dir, f"{component_name} missing from dir({category_name})"
def test_module_metadata_preservation(self):
"""Test that module metadata is preserved after dynamic loading."""
test_components = [
("openai", "OpenAIModelComponent"),
("anthropic", "AnthropicModelComponent"),
("data", "APIRequestComponent"),
]
for category_name, component_name in test_components:
category_module = getattr(components, category_name)
component = getattr(category_module, component_name)
# Check that component has expected metadata
assert hasattr(component, "__name__")
assert hasattr(component, "__module__")
assert component.__name__ == component_name
assert category_name in component.__module__
class TestSpecificModulePatterns:
"""Test specific module patterns and edge cases."""
def test_empty_init_modules(self):
"""Test modules that might have empty __init__.py files."""
# These modules might have empty __init__.py files in the original structure
potentially_empty_modules = [
"chains",
"output_parsers",
"textsplitters",
"toolkits",
"link_extractors",
"documentloaders",
]
for module_name in potentially_empty_modules:
if module_name in components.__all__:
try:
module = getattr(components, module_name)
# Should be able to import even if empty
assert module is not None
except Exception as e:
pytest.fail(f"Failed to import potentially empty module {module_name}: {e}")
def test_platform_specific_imports(self):
"""Test platform-specific imports like NVIDIA Windows components."""
# Test NVIDIA module which has platform-specific logic
nvidia_module = components.nvidia
assert nvidia_module is not None
# Should have basic components regardless of platform
assert "NVIDIAModelComponent" in nvidia_module.__all__
# Should be able to access components
nvidia_model = nvidia_module.NVIDIAModelComponent
assert nvidia_model is not None
def test_large_modules_import_efficiently(self):
"""Test that large modules with many components import efficiently."""
import time
# Test large modules
large_modules = ["vectorstores", "processing", "langchain_utilities"]
for module_name in large_modules:
if module_name in components.__all__:
start_time = time.time()
module = getattr(components, module_name)
import_time = time.time() - start_time
# Initial import should be fast (just loading __init__.py)
assert import_time < 0.5, f"Module {module_name} took too long to import: {import_time}s"
# Should have components available
assert hasattr(module, "__all__")
assert len(module.__all__) > 0
if __name__ == "__main__":
pytest.main([__file__, "-v"])

297
src/backend/tests/unit/components/test_dynamic_imports.py Normal file
View file

@ -0,0 +1,297 @@
"""Tests for dynamic import refactor in langflow components.
This module tests the new langchain-style dynamic import system to ensure:
1. Lazy loading works correctly
2. Components are imported only when accessed
3. Caching works properly
4. Error handling for missing components
5. __dir__ functionality for IDE autocomplete
6. Backward compatibility with existing imports
"""
from unittest.mock import patch
import pytest
from langflow.components._importing import import_mod
class TestImportUtils:
"""Test the import_mod utility function."""
def test_import_mod_with_module_name(self):
"""Test importing specific attribute from a module."""
# Test importing a specific class from a module
result = import_mod("OpenAIModelComponent", "openai_chat_model", "langflow.components.openai")
assert result is not None
assert hasattr(result, "__name__")
assert "OpenAI" in result.__name__
def test_import_mod_without_module_name(self):
"""Test importing entire module when module_name is None."""
result = import_mod("agents", "__module__", "langflow.components")
assert result is not None
# Should return the agents module
assert hasattr(result, "__all__")
def test_import_mod_module_not_found(self):
"""Test error handling when module doesn't exist."""
with pytest.raises(ImportError, match="not found"):
import_mod("NonExistentComponent", "nonexistent_module", "langflow.components.openai")
def test_import_mod_attribute_not_found(self):
"""Test error handling when attribute doesn't exist in module."""
with pytest.raises(AttributeError):
import_mod("NonExistentComponent", "openai_chat_model", "langflow.components.openai")
class TestComponentDynamicImports:
"""Test dynamic import behavior in component modules."""
def test_main_components_module_dynamic_import(self):
"""Test that main components module imports submodules dynamically."""
# Import the main components module
from langflow import components
# Test that submodules are in __all__
assert "agents" in components.__all__
assert "data" in components.__all__
assert "openai" in components.__all__
# Access agents module - this should work via dynamic import
agents_module = components.agents
assert agents_module is not None
# Should be cached in globals after access
assert "agents" in components.__dict__
assert components.__dict__["agents"] is agents_module
# Second access should return cached version
agents_module_2 = components.agents
assert agents_module_2 is agents_module
def test_main_components_module_dir(self):
"""Test __dir__ functionality for main components module."""
from langflow import components
dir_result = dir(components)
# Should include all component categories
assert "agents" in dir_result
assert "data" in dir_result
assert "openai" in dir_result
assert "vectorstores" in dir_result
def test_main_components_module_missing_attribute(self):
"""Test error handling for non-existent component category."""
from langflow import components
with pytest.raises(AttributeError, match="has no attribute 'nonexistent_category'"):
_ = components.nonexistent_category
def test_category_module_dynamic_import(self):
"""Test dynamic import behavior in category modules like openai."""
import langflow.components.openai as openai_components
# Test that components are in __all__
assert "OpenAIModelComponent" in openai_components.__all__
assert "OpenAIEmbeddingsComponent" in openai_components.__all__
# Access component - this should work via dynamic import
openai_model = openai_components.OpenAIModelComponent
assert openai_model is not None
# Should be cached in globals after access
assert "OpenAIModelComponent" in openai_components.__dict__
assert openai_components.__dict__["OpenAIModelComponent"] is openai_model
# Second access should return cached version
openai_model_2 = openai_components.OpenAIModelComponent
assert openai_model_2 is openai_model
def test_category_module_dir(self):
"""Test __dir__ functionality for category modules."""
import langflow.components.openai as openai_components
dir_result = dir(openai_components)
assert "OpenAIModelComponent" in dir_result
assert "OpenAIEmbeddingsComponent" in dir_result
def test_category_module_missing_component(self):
"""Test error handling for non-existent component in category."""
import langflow.components.openai as openai_components
with pytest.raises(AttributeError, match="has no attribute 'NonExistentComponent'"):
_ = openai_components.NonExistentComponent
def test_multiple_category_modules(self):
"""Test dynamic imports work across multiple category modules."""
import langflow.components.anthropic as anthropic_components
import langflow.components.data as data_components
# Test different categories work independently
anthropic_model = anthropic_components.AnthropicModelComponent
api_request = data_components.APIRequestComponent
assert anthropic_model is not None
assert api_request is not None
# Test they're cached in their respective modules
assert "AnthropicModelComponent" in anthropic_components.__dict__
assert "APIRequestComponent" in data_components.__dict__
def test_backward_compatibility(self):
"""Test that existing import patterns still work."""
# These imports should work the same as before
from langflow.components.agents import AgentComponent
from langflow.components.data import APIRequestComponent
from langflow.components.openai import OpenAIModelComponent
assert OpenAIModelComponent is not None
assert APIRequestComponent is not None
assert AgentComponent is not None
def test_component_instantiation(self):
"""Test that dynamically imported components can be instantiated."""
from langflow.components import helpers
# Import component dynamically
calculator_class = helpers.CalculatorComponent
# Should be able to instantiate (even if it requires parameters)
assert callable(calculator_class)
assert hasattr(calculator_class, "__init__")
def test_import_error_handling(self):
"""Test error handling when import fails."""
import langflow.components.notdiamond as notdiamond_components
# Patch the import_mod function directly
with patch("langflow.components.notdiamond.import_mod") as mock_import_mod:
# Mock import_mod to raise ImportError
mock_import_mod.side_effect = ImportError("Module not found")
# Clear any cached attribute
if "NotDiamondComponent" in notdiamond_components.__dict__:
del notdiamond_components.__dict__["NotDiamondComponent"]
with pytest.raises(AttributeError, match="Could not import"):
_ = notdiamond_components.NotDiamondComponent
def test_consistency_check(self):
"""Test that __all__ and _dynamic_imports are consistent."""
import langflow.components.openai as openai_components
# All items in __all__ should have corresponding entries in _dynamic_imports
for component_name in openai_components.__all__:
assert component_name in openai_components._dynamic_imports
# All keys in _dynamic_imports should be in __all__
for component_name in openai_components._dynamic_imports:
assert component_name in openai_components.__all__
def test_type_checking_imports(self):
"""Test that TYPE_CHECKING imports work correctly with dynamic loading."""
# This test ensures that imports in TYPE_CHECKING blocks
# work correctly with the dynamic import system
import langflow.components.searchapi as searchapi_components
# Components should be available for dynamic loading
assert "SearchComponent" in searchapi_components.__all__
assert "SearchComponent" in searchapi_components._dynamic_imports
# Accessing should trigger dynamic import and caching
component = searchapi_components.SearchComponent
assert component is not None
assert "SearchComponent" in searchapi_components.__dict__
class TestPerformanceCharacteristics:
"""Test performance characteristics of dynamic imports."""
def test_lazy_loading_performance(self):
"""Test that components can be accessed and cached properly."""
from langflow.components import vectorstores
# Test that we can access a component
chroma = vectorstores.ChromaVectorStoreComponent
assert chroma is not None
# After access, it should be cached in the module's globals
assert "ChromaVectorStoreComponent" in vectorstores.__dict__
# Subsequent access should return the same cached object
chroma_2 = vectorstores.ChromaVectorStoreComponent
assert chroma_2 is chroma
def test_caching_behavior(self):
"""Test that components are cached after first access."""
from langflow.components import models
# First access
embedding_model_1 = models.EmbeddingModelComponent
# Second access should return the exact same object (cached)
embedding_model_2 = models.EmbeddingModelComponent
assert embedding_model_1 is embedding_model_2
def test_memory_usage_multiple_accesses(self):
"""Test memory behavior with multiple component accesses."""
from langflow.components import processing
# Access multiple components
components = []
component_names = ["CombineTextComponent", "SplitTextComponent", "JSONCleaner", "RegexExtractorComponent"]
for name in component_names:
component = getattr(processing, name)
components.append(component)
# Each should be cached
assert name in processing.__dict__
# All should be different classes
assert len(set(components)) == len(components)
class TestSpecialCases:
"""Test special cases and edge conditions."""
def test_empty_init_files(self):
"""Test that empty __init__.py files are handled gracefully."""
# Test accessing components from categories that might have empty __init__.py
from langflow import components
# These should work even if some categories have empty __init__.py files
agents = components.agents
assert agents is not None
def test_platform_specific_components(self):
"""Test platform-specific component handling (like NVIDIA Windows components)."""
import langflow.components.nvidia as nvidia_components
# NVIDIA components should be available
nvidia_model = nvidia_components.NVIDIAModelComponent
assert nvidia_model is not None
# Platform-specific components should be handled correctly
# (This test will pass regardless of platform since the import structure handles it)
assert "NVIDIAModelComponent" in nvidia_components.__all__
def test_import_structure_integrity(self):
"""Test that the import structure maintains integrity."""
from langflow import components
# Test that we can access nested components through the hierarchy
openai_model = components.openai.OpenAIModelComponent
data_api = components.data.APIRequestComponent
assert openai_model is not None
assert data_api is not None
# Test that both main module and submodules are properly cached
assert "openai" in components.__dict__
assert "data" in components.__dict__
if __name__ == "__main__":
# Run tests
pytest.main([__file__, "-v"])

176
src/backend/tests/unit/test_import_utils.py Normal file
View file

@ -0,0 +1,176 @@
"""Unit tests for the _import_utils module.
Tests the core import_mod function used throughout the dynamic import system.
"""
from unittest.mock import patch
import pytest
from langflow.components._importing import import_mod
class TestImportAttr:
"""Test the import_mod utility function in detail."""
def test_import_module_with_none_module_name(self):
"""Test importing a module when module_name is None."""
# This should import the module directly using the attr_name
result = import_mod("agents", None, "langflow.components")
# Should return the agents module
assert result is not None
assert hasattr(result, "__all__")
def test_import_module_with_module_name(self):
"""Test importing a module when module_name is __module__."""
# This should import the module directly using the attr_name
result = import_mod("agents", "__module__", "langflow.components")
# Should return the agents module
assert result is not None
assert hasattr(result, "__all__")
def test_import_modibute_from_module(self):
"""Test importing a specific attribute from a module."""
# Test importing a class from a specific module
result = import_mod("AnthropicModelComponent", "anthropic", "langflow.components.anthropic")
assert result is not None
assert hasattr(result, "__name__")
assert "Component" in result.__name__
def test_import_nonexistent_module(self):
"""Test error handling when module doesn't exist."""
with pytest.raises(ImportError, match="not found"):
import_mod("SomeComponent", "nonexistent_module", "langflow.components.openai")
def test_module_not_found_with_none_module_name(self):
"""Test ModuleNotFoundError handling when module_name is None."""
with pytest.raises(AttributeError, match="has no attribute"):
import_mod("nonexistent_module", None, "langflow.components")
def test_module_not_found_with_module_special_name(self):
"""Test ModuleNotFoundError handling when module_name is '__module__'."""
with pytest.raises(AttributeError, match="has no attribute"):
import_mod("nonexistent_module", "__module__", "langflow.components")
def test_import_nonexistent_attribute(self):
"""Test error handling when attribute doesn't exist in module."""
with pytest.raises(AttributeError):
import_mod("NonExistentComponent", "anthropic", "langflow.components.anthropic")
def test_import_with_none_package(self):
"""Test behavior when package is None."""
# This should raise TypeError because relative imports require a package
with pytest.raises(TypeError, match="package.*required"):
import_mod("something", "some_module", None)
def test_module_not_found_error_handling(self):
"""Test specific ModuleNotFoundError handling."""
with patch("importlib.import_module") as mock_import_module:
mock_import_module.side_effect = ModuleNotFoundError("No module named 'test'")
with pytest.raises(ImportError, match="not found"):
import_mod("TestComponent", "test_module", "test.package")
def test_getattr_error_handling(self):
"""Test AttributeError handling when getting attribute from module."""
# Test the case where the module exists but doesn't have the attribute
# Use a real module that exists
with pytest.raises(AttributeError):
# os module exists but doesn't have 'NonExistentAttribute'
import_mod("NonExistentAttribute", "path", "os")
def test_relative_import_behavior(self):
"""Test that relative imports are constructed correctly."""
# This test verifies the relative import logic
result = import_mod("helpers", "__module__", "langflow.components")
assert result is not None
def test_package_resolution(self):
"""Test that package parameter is used correctly."""
# Test with a known working package and module
result = import_mod("CalculatorComponent", "calculator_core", "langflow.components.helpers")
assert result is not None
assert callable(result)
def test_import_mod_with_special_module_name(self):
"""Test behavior with special module_name values."""
# Test with "__module__" - should import the attr_name as a module
result = import_mod("data", "__module__", "langflow.components")
assert result is not None
# Test with None - should also import the attr_name as a module
result2 = import_mod("data", None, "langflow.components")
assert result2 is not None
def test_error_message_formatting(self):
"""Test that error messages are properly formatted."""
with pytest.raises(ImportError) as exc_info:
import_mod("NonExistent", "nonexistent", "langflow.components")
error_msg = str(exc_info.value)
assert "langflow.components" in error_msg
assert "nonexistent" in error_msg
def test_return_value_types(self):
"""Test that import_mod returns appropriate types."""
# Test module import
module_result = import_mod("openai", "__module__", "langflow.components")
assert hasattr(module_result, "__name__")
# Test class import
class_result = import_mod("OpenAIModelComponent", "openai_chat_model", "langflow.components.openai")
assert callable(class_result)
assert hasattr(class_result, "__name__")
def test_caching_independence(self):
"""Test that import_mod doesn't interfere with Python's module caching."""
# Multiple calls should work consistently
result1 = import_mod("agents", "__module__", "langflow.components")
result2 = import_mod("agents", "__module__", "langflow.components")
# Should return the same module object (Python's import caching)
assert result1 is result2
class TestImportAttrEdgeCases:
"""Test edge cases and boundary conditions for import_mod."""
def test_empty_strings(self):
"""Test behavior with empty strings."""
with pytest.raises((ImportError, ValueError)):
import_mod("", "module", "package")
with pytest.raises((ImportError, ValueError)):
import_mod("attr", "", "package")
def test_whitespace_handling(self):
"""Test that whitespace in names is handled appropriately."""
with pytest.raises(ImportError):
import_mod("attr name", "module", "package")
def test_special_characters(self):
"""Test handling of special characters in names."""
with pytest.raises((ImportError, ValueError)):
import_mod("attr-name", "module", "package")
def test_unicode_names(self):
"""Test handling of unicode characters in names."""
with pytest.raises(ImportError):
import_mod("attß", "module", "package")
def test_very_long_names(self):
"""Test handling of very long module/attribute names."""
long_name = "a" * 1000
with pytest.raises(ImportError):
import_mod(long_name, "module", "package")
def test_numeric_names(self):
"""Test handling of numeric names."""
with pytest.raises(ImportError):
import_mod("123", "module", "package")
if __name__ == "__main__":
pytest.main([__file__, "-v"])