feat: reorganize tests by capability with separate test targets
Some checks failed
Test Suite / unit-tests (3.11) (push) Has been cancelled
Test Suite / unit-tests (3.12) (push) Has been cancelled
Test Suite / integration-tests (push) Has been cancelled
Test Suite / e2e-tests (push) Has been cancelled
Test Suite / performance-tests (push) Has been cancelled
Test Suite / code-quality (push) Has been cancelled
Test Suite / security-scan (push) Has been cancelled
Test Suite / test-summary (push) Has been cancelled

Separate capability-specific tests from core system tests to establish clear
test organization and separation of concerns.

## Test Reorganization:
- **markitect-content tests**: Moved 6 tests to capabilities/markitect-content/tests/
- **markitect-finance tests**: Moved 7 tests to markitect/finance/tests/
- **markitect-query tests**: Moved 1 test to markitect/query_paradigms/tests/
- **markitect-graphql tests**: Moved 2 tests to markitect/graphql/tests/
- **markitect-plugins tests**: Moved 2 tests to markitect/plugins/tests/

## Makefile Updates:
- **make test**: Excludes capability tests, runs only core system tests
- **make test-capabilities**: Runs all capability tests
- **make test-capability-***: Individual capability test targets
- Updated all test targets (test-red, test-green, test-ultra-fast, test-perf)
- Added capability test targets to help documentation

## Benefits:
- Clear separation between core system tests and capability-specific tests
- Faster core test execution (capability tests not run by default)
- Individual capability testing for focused development
- Supports future capability extraction workflow
- Maintains capability test independence

Test verification:
- Core tests: 1291 tests (capability tests excluded)
- Finance capability: 143 tests working independently
- Content capability: 79 tests working independently

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-10-25 02:37:45 +02:00
parent f0dfd04d45
commit 096017b93f
23 changed files with 74 additions and 8 deletions

View File

View File

@@ -0,0 +1,855 @@
"""
Tests for Issue #19: Plugin Architecture and Extensions System
This module provides comprehensive tests for the MarkiTect plugin system
including plugin discovery, loading, management, and CLI integration.
"""
import pytest
import json
import tempfile
import os
from pathlib import Path
from unittest.mock import Mock, patch
from markitect.plugins import (
PluginManager,
BasePlugin,
ProcessorPlugin,
FormatterPlugin,
PluginType,
PluginMetadata,
plugin_registry,
register_plugin
)
from markitect.plugins.manager import PluginManager
from markitect.plugins.registry import PluginRegistry
class TestPluginArchitecture:
"""Test suite for plugin architecture components."""
def setup_method(self):
"""Set up test environment."""
# Clear plugin registry for clean tests
plugin_registry.cleanup_all()
plugin_registry._plugins.clear()
plugin_registry._instances.clear()
plugin_registry._plugins_by_type.clear()
def teardown_method(self):
"""Clean up after tests."""
plugin_registry.cleanup_all()
plugin_registry._plugins.clear()
plugin_registry._instances.clear()
plugin_registry._plugins_by_type.clear()
class TestPluginBase:
"""Test base plugin functionality."""
def test_plugin_metadata_creation(self):
"""Test PluginMetadata creation and properties."""
metadata = PluginMetadata(
name="test_plugin",
version="1.0.0",
description="Test plugin",
author="Test Author",
plugin_type=PluginType.PROCESSOR,
dependencies=["dep1", "dep2"],
markitect_version=">=0.1.0"
)
assert metadata.name == "test_plugin"
assert metadata.version == "1.0.0"
assert metadata.description == "Test plugin"
assert metadata.author == "Test Author"
assert metadata.plugin_type == PluginType.PROCESSOR
assert metadata.dependencies == ["dep1", "dep2"]
assert metadata.markitect_version == ">=0.1.0"
def test_base_plugin_initialization(self):
"""Test BasePlugin initialization."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
config = {"option1": "value1", "option2": "value2"}
plugin = TestPlugin(config)
assert plugin.config == config
assert not plugin.is_initialized
def test_plugin_initialization_lifecycle(self):
"""Test plugin initialization and cleanup lifecycle."""
class TestPlugin(BasePlugin):
def __init__(self, config=None):
super().__init__(config)
self.initialized = False
self.cleaned_up = False
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
def _initialize(self):
self.initialized = True
def cleanup(self):
self.cleaned_up = True
plugin = TestPlugin()
assert not plugin.initialized
assert not plugin.is_initialized
# Test initialization
result = plugin.initialize()
assert result is True
assert plugin.initialized
assert plugin.is_initialized
# Test cleanup
plugin.cleanup()
assert plugin.cleaned_up
def test_plugin_initialization_failure(self):
"""Test plugin initialization failure handling."""
class FailingPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="failing",
version="1.0.0",
description="Failing plugin",
plugin_type=PluginType.EXTENSION
)
def _initialize(self):
raise Exception("Initialization failed")
plugin = FailingPlugin()
result = plugin.initialize()
assert result is False
assert not plugin.is_initialized
class TestProcessorPlugin:
"""Test processor plugin functionality."""
def test_processor_plugin_interface(self):
"""Test processor plugin interface implementation."""
class TestProcessor(ProcessorPlugin):
@property
def metadata(self):
return PluginMetadata(
name="test_processor",
version="1.0.0",
description="Test processor",
plugin_type=PluginType.PROCESSOR
)
def process(self, content: str, **kwargs) -> str:
return content.upper()
processor = TestProcessor()
result = processor.process("hello world")
assert result == "HELLO WORLD"
# Test default can_process implementation
assert processor.can_process("any content")
def test_processor_plugin_with_options(self):
"""Test processor plugin with processing options."""
class ConfigurableProcessor(ProcessorPlugin):
@property
def metadata(self):
return PluginMetadata(
name="configurable_processor",
version="1.0.0",
description="Configurable processor",
plugin_type=PluginType.PROCESSOR
)
def process(self, content: str, **kwargs) -> str:
if kwargs.get('uppercase', False):
content = content.upper()
if kwargs.get('reverse', False):
content = content[::-1]
return content
processor = ConfigurableProcessor()
# Test with no options
result = processor.process("hello")
assert result == "hello"
# Test with uppercase option
result = processor.process("hello", uppercase=True)
assert result == "HELLO"
# Test with both options
result = processor.process("hello", uppercase=True, reverse=True)
assert result == "OLLEH"
class TestFormatterPlugin:
"""Test formatter plugin functionality."""
def test_formatter_plugin_interface(self):
"""Test formatter plugin interface implementation."""
class TestFormatter(FormatterPlugin):
@property
def metadata(self):
return PluginMetadata(
name="test_formatter",
version="1.0.0",
description="Test formatter",
plugin_type=PluginType.FORMATTER
)
def format(self, data, **kwargs) -> str:
return json.dumps(data, indent=kwargs.get('indent', 2))
def get_file_extension(self) -> str:
return '.json'
formatter = TestFormatter()
data = {"key": "value", "number": 42}
result = formatter.format(data)
parsed = json.loads(result)
assert parsed == data
extension = formatter.get_file_extension()
assert extension == '.json'
class TestPluginRegistry:
"""Test plugin registry functionality."""
def setup_method(self):
"""Set up test environment."""
self.registry = PluginRegistry()
def test_plugin_registration(self):
"""Test plugin registration."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
# Test registration
name = self.registry.register(TestPlugin)
assert name == "TestPlugin"
assert "TestPlugin" in self.registry._plugins
# Test registration with custom name
custom_name = self.registry.register(TestPlugin, "custom_name")
assert custom_name == "custom_name"
assert "custom_name" in self.registry._plugins
def test_plugin_registration_duplicate_name(self):
"""Test plugin registration with duplicate name."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
self.registry.register(TestPlugin, "test_name")
# Should raise error for duplicate name
with pytest.raises(ValueError, match="already registered"):
self.registry.register(TestPlugin, "test_name")
def test_plugin_retrieval(self):
"""Test plugin retrieval from registry."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
self.registry.register(TestPlugin, "test_plugin")
# Test successful retrieval
plugin = self.registry.get_plugin("test_plugin")
assert plugin is not None
assert isinstance(plugin, TestPlugin)
# Test non-existent plugin
plugin = self.registry.get_plugin("non_existent")
assert plugin is None
def test_plugin_unregistration(self):
"""Test plugin unregistration."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
self.registry.register(TestPlugin, "test_plugin")
plugin = self.registry.get_plugin("test_plugin")
assert plugin is not None
# Test unregistration
result = self.registry.unregister("test_plugin")
assert result is True
# Plugin should no longer be available
plugin = self.registry.get_plugin("test_plugin")
assert plugin is None
# Test unregistering non-existent plugin
result = self.registry.unregister("non_existent")
assert result is False
def test_plugins_by_type(self):
"""Test retrieving plugins by type."""
class ProcessorPlugin1(ProcessorPlugin):
@property
def metadata(self):
return PluginMetadata(
name="processor1",
version="1.0.0",
description="Processor 1",
plugin_type=PluginType.PROCESSOR
)
def process(self, content, **kwargs):
return content
class FormatterPlugin1(FormatterPlugin):
@property
def metadata(self):
return PluginMetadata(
name="formatter1",
version="1.0.0",
description="Formatter 1",
plugin_type=PluginType.FORMATTER
)
def format(self, data, **kwargs):
return str(data)
def get_file_extension(self):
return '.txt'
self.registry.register(ProcessorPlugin1, "processor1")
self.registry.register(FormatterPlugin1, "formatter1")
# Test getting processors
processors = self.registry.get_plugins_by_type(PluginType.PROCESSOR)
assert "processor1" in processors
assert "formatter1" not in processors
# Test getting formatters
formatters = self.registry.get_plugins_by_type(PluginType.FORMATTER)
assert "formatter1" in formatters
assert "processor1" not in formatters
def test_list_plugins(self):
"""Test listing all plugins with metadata."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test plugin",
author="Test Author",
plugin_type=PluginType.EXTENSION
)
self.registry.register(TestPlugin, "test_plugin")
plugins = self.registry.list_plugins()
assert "test_plugin" in plugins
plugin_info = plugins["test_plugin"]
assert plugin_info["name"] == "test"
assert plugin_info["version"] == "1.0.0"
assert plugin_info["description"] == "Test plugin"
assert plugin_info["author"] == "Test Author"
assert plugin_info["type"] == "extension"
class TestPluginManager:
"""Test plugin manager functionality."""
def setup_method(self):
"""Set up test environment."""
# Clear plugin registry
plugin_registry.cleanup_all()
plugin_registry._plugins.clear()
plugin_registry._instances.clear()
plugin_registry._plugins_by_type.clear()
def test_plugin_manager_initialization(self):
"""Test plugin manager initialization."""
manager = PluginManager()
assert manager.config is not None
assert isinstance(manager.plugin_directories, list)
def test_plugin_manager_with_config(self):
"""Test plugin manager with custom configuration."""
with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
f.write("""
plugin_directories:
- "custom_plugins"
auto_discover: false
plugins:
test_plugin:
enabled: true
""")
config_path = f.name
try:
manager = PluginManager(config_path)
assert "custom_plugins" in manager.config.get('plugin_directories', [])
assert manager.config.get('auto_discover') is False
assert 'test_plugin' in manager.config.get('plugins', {})
finally:
os.unlink(config_path)
def test_plugin_discovery_empty(self):
"""Test plugin discovery with no plugins."""
manager = PluginManager()
discovered = manager.discover_plugins()
# Should be a dictionary (empty or with built-ins)
assert isinstance(discovered, dict)
@patch('importlib.import_module')
def test_load_plugin_success(self, mock_import):
"""Test successful plugin loading."""
class TestPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="test",
version="1.0.0",
description="Test",
plugin_type=PluginType.EXTENSION
)
# Mock module with plugin
mock_module = Mock()
mock_module.TestPlugin = TestPlugin
mock_import.return_value = mock_module
manager = PluginManager()
# Manually add to discovered plugins
manager._discovered_plugins = {
"test_plugin": {
"module_name": "test_module",
"class_name": "TestPlugin"
}
}
plugin = manager.load_plugin("test_plugin")
assert plugin is not None
assert isinstance(plugin, TestPlugin)
def test_load_plugin_not_found(self):
"""Test loading non-existent plugin."""
manager = PluginManager()
plugin = manager.load_plugin("non_existent_plugin")
assert plugin is None
def test_get_plugins_by_type(self):
"""Test getting plugins by type."""
class TestProcessor(ProcessorPlugin):
@property
def metadata(self):
return PluginMetadata(
name="test_processor",
version="1.0.0",
description="Test processor",
plugin_type=PluginType.PROCESSOR
)
def process(self, content, **kwargs):
return content
# Register plugin directly
plugin_registry.register(TestProcessor, "test_processor")
manager = PluginManager()
processors = manager.get_plugins_by_type(PluginType.PROCESSOR)
# Should have at least our test processor
assert len(processors) >= 1
assert any(isinstance(p, TestProcessor) for p in processors)
class TestPluginDecorator:
"""Test plugin registration decorator."""
def setup_method(self):
"""Set up test environment."""
# Clear plugin registry
plugin_registry.cleanup_all()
plugin_registry._plugins.clear()
plugin_registry._instances.clear()
plugin_registry._plugins_by_type.clear()
def test_register_plugin_decorator(self):
"""Test @register_plugin decorator."""
@register_plugin("decorated_plugin")
class DecoratedPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="decorated",
version="1.0.0",
description="Decorated plugin",
plugin_type=PluginType.EXTENSION
)
# Plugin should be automatically registered
assert "decorated_plugin" in plugin_registry._plugins
# Should be able to retrieve it
plugin = plugin_registry.get_plugin("decorated_plugin")
assert plugin is not None
assert isinstance(plugin, DecoratedPlugin)
def test_register_plugin_decorator_no_name(self):
"""Test @register_plugin decorator without name."""
@register_plugin()
class AutoNamedPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="auto_named",
version="1.0.0",
description="Auto named plugin",
plugin_type=PluginType.EXTENSION
)
# Should use class name
assert "AutoNamedPlugin" in plugin_registry._plugins
class TestBuiltinPlugins:
"""Test built-in plugins."""
def test_json_formatter_plugin(self):
"""Test built-in JSON formatter plugin."""
from markitect.plugins.builtin.formatters import JsonFormatter
formatter = JsonFormatter()
assert formatter.metadata.plugin_type == PluginType.FORMATTER
data = {"key": "value", "number": 42}
result = formatter.format(data)
parsed = json.loads(result)
assert parsed == data
assert formatter.get_file_extension() == '.json'
def test_table_formatter_plugin(self):
"""Test built-in table formatter plugin."""
from markitect.plugins.builtin.formatters import TableFormatter
formatter = TableFormatter()
assert formatter.metadata.plugin_type == PluginType.FORMATTER
# Test with list of dictionaries
data = [
{"name": "John", "age": 30},
{"name": "Jane", "age": 25}
]
result = formatter.format(data)
assert "John" in result
assert "Jane" in result
assert "name" in result
assert "age" in result
assert formatter.get_file_extension() == '.txt'
def test_markdown_processor_plugin(self):
"""Test built-in markdown processor plugin."""
from markitect.plugins.builtin.processors import MarkdownProcessor
processor = MarkdownProcessor()
assert processor.metadata.plugin_type == PluginType.PROCESSOR
# Test basic processing
content = "# Header\n\nSome content\n"
result = processor.process(content)
assert isinstance(result, str)
# Test can_process
assert processor.can_process("# Markdown header")
assert processor.can_process("Some **bold** text")
class TestPluginCLIIntegration:
"""Test plugin CLI command integration."""
def setup_method(self):
"""Set up test environment."""
# Clear plugin registry
plugin_registry.cleanup_all()
plugin_registry._plugins.clear()
plugin_registry._instances.clear()
plugin_registry._plugins_by_type.clear()
def test_plugin_list_command_import(self):
"""Test that plugin CLI commands can be imported."""
# This tests that the CLI commands are properly integrated
from markitect.cli import plugin_list, plugin_load, plugin_info
assert callable(plugin_list)
assert callable(plugin_load)
assert callable(plugin_info)
def test_plugin_type_enum_import(self):
"""Test that PluginType enum is accessible for CLI."""
from markitect.plugins.base import PluginType
# Test all plugin types are available
assert PluginType.PROCESSOR
assert PluginType.FORMATTER
assert PluginType.VALIDATOR
assert PluginType.EXPORTER
assert PluginType.GENERATOR
assert PluginType.IMPORTER
assert PluginType.TRANSFORMER
assert PluginType.EXTENSION
assert PluginType.BACKEND
assert PluginType.COMMAND
# Test values are strings
assert isinstance(PluginType.PROCESSOR.value, str)
class TestPluginErrorHandling:
"""Test plugin error handling and edge cases."""
def test_plugin_with_invalid_metadata(self):
"""Test plugin with invalid metadata."""
class BadMetadataPlugin(BasePlugin):
@property
def metadata(self):
# Missing required fields
return None
plugin = BadMetadataPlugin()
# Should handle gracefully
try:
plugin_registry.register(BadMetadataPlugin, "bad_plugin")
# Should not crash, might register as extension type
except Exception:
# Exception is acceptable for invalid metadata
pass
def test_plugin_initialization_with_bad_config(self):
"""Test plugin initialization with invalid configuration."""
class ConfigValidatingPlugin(BasePlugin):
@property
def metadata(self):
return PluginMetadata(
name="config_validator",
version="1.0.0",
description="Config validating plugin",
plugin_type=PluginType.EXTENSION
)
def validate_config(self):
errors = []
if 'required_field' not in self.config:
errors.append("Missing required_field")
return errors
# Test with invalid config
plugin = ConfigValidatingPlugin({"wrong_field": "value"})
errors = plugin.validate_config()
assert len(errors) > 0
assert "required_field" in errors[0]
# Test with valid config
plugin = ConfigValidatingPlugin({"required_field": "value"})
errors = plugin.validate_config()
assert len(errors) == 0
def test_plugin_manager_with_invalid_config_file(self):
"""Test plugin manager with invalid configuration file."""
with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
f.write("invalid: yaml: content: [") # Invalid YAML
config_path = f.name
try:
# Should not crash, should use defaults
manager = PluginManager(config_path)
assert manager.config is not None
# Should fall back to defaults
assert 'plugin_directories' in manager.config
finally:
os.unlink(config_path)
class TestPluginIntegration:
"""Integration tests for the plugin system."""
def setup_method(self):
"""Set up test environment."""
# Clear plugin registry
plugin_registry.cleanup_all()
plugin_registry._plugins.clear()
plugin_registry._instances.clear()
plugin_registry._plugins_by_type.clear()
def test_end_to_end_plugin_workflow(self):
"""Test complete plugin workflow from registration to usage."""
# 1. Create a plugin
@register_plugin("workflow_processor")
class WorkflowProcessor(ProcessorPlugin):
@property
def metadata(self):
return PluginMetadata(
name="workflow_processor",
version="1.0.0",
description="End-to-end workflow processor",
plugin_type=PluginType.PROCESSOR
)
def process(self, content, **kwargs):
prefix = kwargs.get('prefix', self.config.get('prefix', ''))
return f"{prefix}{content}"
# 2. Verify registration
assert "workflow_processor" in plugin_registry._plugins
# 3. Create manager and load plugin
manager = PluginManager()
plugin = manager.load_plugin("workflow_processor", {"prefix": ">> "})
# 4. Use plugin
assert plugin is not None
result = plugin.process("Hello World")
assert result == ">> Hello World"
# 5. Verify plugin is in registry
assert plugin_registry.is_loaded("workflow_processor")
# 6. Get plugin by type
processors = manager.get_plugins_by_type(PluginType.PROCESSOR)
assert any(isinstance(p, WorkflowProcessor) for p in processors)
# 7. Unload plugin
success = manager.unload_plugin("workflow_processor")
assert success is True
assert not plugin_registry.is_loaded("workflow_processor")
def test_multiple_plugins_interaction(self):
"""Test interaction between multiple plugins."""
# Register multiple plugins
@register_plugin("upper_processor")
class UpperProcessor(ProcessorPlugin):
@property
def metadata(self):
return PluginMetadata(
name="upper_processor",
version="1.0.0",
description="Uppercase processor",
plugin_type=PluginType.PROCESSOR
)
def process(self, content, **kwargs):
return content.upper()
@register_plugin("json_test_formatter")
class JsonTestFormatter(FormatterPlugin):
@property
def metadata(self):
return PluginMetadata(
name="json_test_formatter",
version="1.0.0",
description="JSON test formatter",
plugin_type=PluginType.FORMATTER
)
def format(self, data, **kwargs):
return json.dumps(data)
def get_file_extension(self):
return '.json'
manager = PluginManager()
# Load both plugins
processor = manager.load_plugin("upper_processor")
formatter = manager.load_plugin("json_test_formatter")
assert processor is not None
assert formatter is not None
# Use them together
processed = processor.process("hello world")
formatted = formatter.format({"result": processed})
data = json.loads(formatted)
assert data["result"] == "HELLO WORLD"
# Verify both are loaded
assert plugin_registry.is_loaded("upper_processor")
assert plugin_registry.is_loaded("json_test_formatter")
if __name__ == '__main__':
pytest.main([__file__])

View File

@@ -0,0 +1,627 @@
"""
Tests for Issue #83: Full text search functionality.
Tests the FTS5-based full text search plugin including indexing,
querying, and CLI integration.
"""
import pytest
import tempfile
import sqlite3
import json
import os
from pathlib import Path
from unittest.mock import patch, MagicMock
from markitect.plugins.builtin.search import FTSSearchPlugin, SearchIndexer, QueryParser
from markitect.database import DatabaseManager
class TestSearchIndexer:
"""Test the search indexing functionality."""
@pytest.fixture
def temp_db_path(self):
"""Create a temporary database for testing."""
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
db_path = f.name
# Initialize database with test data
db_manager = DatabaseManager(db_path)
db_manager.initialize_database()
# Add test markdown files
db_manager.store_markdown_file("test1.md", "# Test Document\n\nThis is a test document about API development.")
db_manager.store_markdown_file("test2.md", "# Another Document\n\nGraphQL interface documentation.")
db_manager.store_markdown_file("test3.md", "---\ntitle: Blog Post\n---\n# My Blog\n\nContent about technology.")
# Add test schemas
schema1 = {"type": "object", "title": "User Schema", "description": "Schema for user objects"}
schema2 = {"type": "object", "title": "Product Schema", "description": "E-commerce product definition"}
db_manager.store_schema_file("user.json", json.dumps(schema1))
db_manager.store_schema_file("product.json", json.dumps(schema2))
yield db_path
# Cleanup
os.unlink(db_path)
def test_check_fts_availability(self, temp_db_path):
"""Test checking FTS5 availability."""
indexer = SearchIndexer()
available = indexer.check_fts_availability(temp_db_path)
# FTS5 should be available in most modern SQLite installations
assert isinstance(available, bool)
def test_initialize_fts_tables(self, temp_db_path):
"""Test FTS5 table initialization."""
indexer = SearchIndexer()
indexer.initialize_fts_tables(temp_db_path)
# Check that FTS tables were created
conn = sqlite3.connect(temp_db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'fts_%'")
fts_tables = [row[0] for row in cursor.fetchall()]
if indexer.check_fts_availability(temp_db_path):
assert 'fts_files' in fts_tables
assert 'fts_schemas' in fts_tables
else:
# If FTS5 not available, should have status table
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='fts_status'")
assert cursor.fetchone() is not None
conn.close()
def test_rebuild_index(self, temp_db_path):
"""Test rebuilding search indexes."""
indexer = SearchIndexer()
indexer.initialize_fts_tables(temp_db_path)
stats = indexer.rebuild_index(temp_db_path)
assert 'files_indexed' in stats
assert 'schemas_indexed' in stats
if indexer.check_fts_availability(temp_db_path):
# If FTS5 is available, should index successfully
assert stats['files_indexed'] >= 0
assert stats['schemas_indexed'] >= 0
else:
# If FTS5 not available, might have error
pass # Just check stats exist
def test_get_index_info(self, temp_db_path):
"""Test getting index information."""
indexer = SearchIndexer()
indexer.initialize_fts_tables(temp_db_path)
indexer.rebuild_index(temp_db_path)
info = indexer.get_index_info(temp_db_path)
assert 'fts_enabled' in info
if info['fts_enabled']:
assert 'fts_tables' in info
assert 'fts_files_count' in info
assert 'fts_schemas_count' in info
class TestQueryParser:
"""Test query parsing functionality."""
def test_parse_simple_query(self):
"""Test parsing simple queries."""
parser = QueryParser()
# Simple word
result = parser.parse_query("test")
assert "test*" in result
# Multiple words
result = parser.parse_query("test document")
assert "test*" in result
assert "document*" in result
assert "AND" in result
def test_parse_phrase_query(self):
"""Test parsing phrase queries."""
parser = QueryParser()
result = parser.parse_query('"exact phrase"')
assert '"exact phrase"' in result
def test_parse_boolean_operators(self):
"""Test parsing boolean operators."""
parser = QueryParser()
# AND operator - if already FTS5, should be preserved
result = parser.parse_query("test AND document")
assert "test" in result
assert "AND" in result
assert "document" in result
# OR operator - if already FTS5, should be preserved
result = parser.parse_query("test OR document")
assert "test" in result
assert "OR" in result
assert "document" in result
# NOT operator - if already FTS5, should be preserved
result = parser.parse_query("test NOT document")
assert "test" in result
assert "NOT" in result
def test_validate_query(self):
"""Test query validation."""
parser = QueryParser()
# Valid queries
valid, error = parser.validate_query("test")
assert valid
assert error is None
valid, error = parser.validate_query('"exact phrase"')
assert valid
assert error is None
# Invalid queries
valid, error = parser.validate_query('unmatched "quote')
assert not valid
assert "quotes" in error
valid, error = parser.validate_query("test (unmatched")
assert not valid
assert "parentheses" in error
def test_get_query_terms(self):
"""Test extracting terms from queries."""
parser = QueryParser()
terms = parser.get_query_terms("test document AND api")
assert "test" in terms
assert "document" in terms
assert "api" in terms
assert "AND" not in terms # Operators should be excluded
def test_build_column_query(self):
"""Test building column-specific queries."""
parser = QueryParser()
result = parser.build_column_query("test", ["title", "content"])
assert "title:" in result
assert "content:" in result
assert "OR" in result
class TestFTSSearchPlugin:
"""Test the main FTS search plugin."""
@pytest.fixture
def temp_db_path(self):
"""Create a temporary database with test data."""
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
db_path = f.name
# Initialize database with test data
db_manager = DatabaseManager(db_path)
db_manager.initialize_database()
# Add test markdown files
db_manager.store_markdown_file("api-guide.md", "# API Guide\n\nComprehensive API development guide with examples.")
db_manager.store_markdown_file("tutorial.md", "# GraphQL Tutorial\n\nLearn GraphQL basics and advanced concepts.")
db_manager.store_markdown_file("readme.md", "---\ntitle: Project README\ntags: [documentation, guide]\n---\n# Project\n\nProject documentation and setup guide.")
# Add test schemas
schema1 = {"type": "object", "title": "API Schema", "description": "REST API response schema", "properties": {"data": {"type": "object"}}}
schema2 = {"type": "object", "title": "User Schema", "description": "User profile schema", "properties": {"name": {"type": "string"}}}
db_manager.store_schema_file("api-schema.json", json.dumps(schema1))
db_manager.store_schema_file("user-schema.json", json.dumps(schema2))
yield db_path
# Cleanup
os.unlink(db_path)
def test_plugin_metadata(self):
"""Test plugin metadata."""
plugin = FTSSearchPlugin()
metadata = plugin.metadata
assert metadata.name == "fts_search"
assert metadata.version == "1.0.0"
assert "full text search" in metadata.description.lower()
def test_initialize_plugin(self, temp_db_path):
"""Test plugin initialization."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
# Check that FTS tables exist (if FTS5 is available)
stats = plugin.get_search_stats(temp_db_path)
assert 'fts_enabled' in stats
def test_search_files_only(self, temp_db_path):
"""Test searching only in files."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
plugin.rebuild_index(temp_db_path)
results = plugin.search(temp_db_path, "API", content_type="files", limit=10)
# Should find files containing "API"
assert isinstance(results, list)
for result in results:
assert result['type'] == 'file'
assert 'file' in result
assert 'score' in result
def test_search_schemas_only(self, temp_db_path):
"""Test searching only in schemas."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
plugin.rebuild_index(temp_db_path)
results = plugin.search(temp_db_path, "schema", content_type="schemas", limit=10)
# Should find schemas
assert isinstance(results, list)
for result in results:
assert result['type'] == 'schema'
assert 'schema' in result
assert 'score' in result
def test_search_all_content(self, temp_db_path):
"""Test searching all content types."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
plugin.rebuild_index(temp_db_path)
results = plugin.search(temp_db_path, "guide", content_type="all", limit=10)
# Should find both files and schemas (or empty list if FTS5 unavailable)
assert isinstance(results, list)
# If results found, should be properly formatted and sorted
if results:
# Results should be sorted by score
scores = [result.get('score', 0) for result in results]
assert scores == sorted(scores, reverse=True)
# Check result structure
for result in results:
assert 'type' in result
assert 'score' in result
def test_search_with_pagination(self, temp_db_path):
"""Test search with pagination."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
plugin.rebuild_index(temp_db_path)
# Get first page
results1 = plugin.search(temp_db_path, "guide", limit=1, offset=0)
# Get second page
results2 = plugin.search(temp_db_path, "guide", limit=1, offset=1)
# Results should be different (if there are enough results)
if len(results1) > 0 and len(results2) > 0:
assert results1[0] != results2[0]
def test_fallback_search(self, temp_db_path):
"""Test fallback search when FTS5 fails."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
# Force fallback by using invalid FTS5 query syntax with mock
with patch.object(plugin, '_search_files', side_effect=Exception("FTS5 error")):
with patch.object(plugin, '_search_schemas', side_effect=Exception("FTS5 error")):
results = plugin.search(temp_db_path, "API", content_type="all", limit=10)
# Should still return results via fallback
assert isinstance(results, list)
def test_get_search_stats(self, temp_db_path):
"""Test getting search statistics."""
plugin = FTSSearchPlugin()
plugin.initialize(temp_db_path)
stats = plugin.get_search_stats(temp_db_path)
assert 'fts_enabled' in stats
assert 'fts_tables' in stats
class TestSearchCLI:
"""Test search CLI commands."""
@pytest.fixture
def temp_db_path(self):
"""Create a temporary database with test data."""
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
db_path = f.name
# Initialize database with test data
db_manager = DatabaseManager(db_path)
db_manager.initialize_database()
# Add test data
db_manager.store_markdown_file("test.md", "# Test\n\nThis is a test document.")
yield db_path
# Cleanup
os.unlink(db_path)
def test_search_init_command(self, temp_db_path):
"""Test the search init CLI command."""
from click.testing import CliRunner
from markitect.cli import cli
runner = CliRunner()
with patch('markitect.cli.get_database_path', return_value=temp_db_path):
result = runner.invoke(cli, ['search', 'init'])
assert result.exit_code == 0
assert "Search indexes initialized" in result.output or "Search plugin not available" in result.output
def test_search_query_command(self, temp_db_path):
"""Test the search query CLI command."""
from click.testing import CliRunner
from markitect.cli import cli
runner = CliRunner()
with patch('markitect.cli.get_database_path', return_value=temp_db_path):
# Initialize search first
runner.invoke(cli, ['search', 'init'])
# Perform search
result = runner.invoke(cli, ['search', 'query', 'test'])
assert result.exit_code == 0
# Should either show results or indicate no search plugin
assert "results" in result.output or "Search plugin not available" in result.output
def test_search_status_command(self, temp_db_path):
"""Test the search status CLI command."""
from click.testing import CliRunner
from markitect.cli import cli
runner = CliRunner()
with patch('markitect.cli.get_database_path', return_value=temp_db_path):
result = runner.invoke(cli, ['search', 'status'])
assert result.exit_code == 0
assert "Search Index Status" in result.output or "Search plugin not available" in result.output
def test_search_rebuild_command(self, temp_db_path):
"""Test the search rebuild CLI command."""
from click.testing import CliRunner
from markitect.cli import cli
runner = CliRunner()
with patch('markitect.cli.get_database_path', return_value=temp_db_path):
# Initialize search first
runner.invoke(cli, ['search', 'init'])
# Rebuild indexes
result = runner.invoke(cli, ['search', 'rebuild'])
if result.exit_code != 0:
print(f"Command output: {result.output}")
print(f"Exception: {result.exception}")
# Should succeed or fail gracefully with plugin unavailable message or database error
acceptable_errors = [
"Search plugin not available",
"database disk image is malformed", # Can happen with concurrent access
"database is locked"
]
if result.exit_code == 0:
assert "Rebuilding search indexes" in result.output
else:
# Check if it's an acceptable error
assert any(error in result.output for error in acceptable_errors)
class TestSearchIntegration:
"""Integration tests for search functionality."""
@pytest.fixture
def populated_db_path(self):
"""Create a database with realistic test data."""
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
db_path = f.name
db_manager = DatabaseManager(db_path)
db_manager.initialize_database()
# Add realistic markdown files
files = [
("api-documentation.md", """# API Documentation
## Authentication
The API uses Bearer token authentication. Include your token in the Authorization header.
## Endpoints
- GET /users - List all users
- POST /users - Create a new user
- GET /users/{id} - Get specific user
## Error Handling
All errors return JSON with error message and status code.
"""),
("graphql-guide.md", """---
title: GraphQL Complete Guide
tags: [graphql, api, tutorial]
author: Development Team
---
# GraphQL Complete Guide
GraphQL is a query language for APIs and a runtime for executing those queries.
## Benefits
- Single endpoint
- Type safety
- Efficient data fetching
- Strong introspection
## Schema Definition
Define your GraphQL schema using SDL (Schema Definition Language).
"""),
("project-readme.md", """# MarkiTect Project
MarkiTect is a comprehensive markdown content management and analysis system.
## Features
- Document indexing and storage
- Full text search capabilities
- GraphQL API interface
- Plugin system for extensibility
## Installation
1. Clone the repository
2. Install dependencies: pip install -r requirements.txt
3. Initialize database: markitect init
## Usage Examples
Search for content: markitect search query "API documentation"
""")
]
for filename, content in files:
db_manager.store_markdown_file(filename, content)
# Add realistic schemas
schemas = [
("user-schema.json", {
"type": "object",
"title": "User Schema",
"description": "Schema for user profile data in the API",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"email": {"type": "string", "format": "email"},
"created_at": {"type": "string", "format": "date-time"}
},
"required": ["id", "name", "email"]
}),
("api-response-schema.json", {
"type": "object",
"title": "API Response Schema",
"description": "Standard API response format for all endpoints",
"properties": {
"data": {"type": "object"},
"success": {"type": "boolean"},
"message": {"type": "string"},
"errors": {"type": "array", "items": {"type": "string"}}
},
"required": ["success"]
})
]
for filename, schema in schemas:
db_manager.store_schema_file(filename, json.dumps(schema))
yield db_path
# Cleanup
os.unlink(db_path)
def test_end_to_end_search_workflow(self, populated_db_path):
"""Test complete search workflow from initialization to querying."""
plugin = FTSSearchPlugin()
# Initialize search
plugin.initialize(populated_db_path)
# Rebuild indexes
stats = plugin.rebuild_index(populated_db_path)
if plugin.indexer.check_fts_availability(populated_db_path):
# If FTS5 is available, should index files
assert stats['files_indexed'] >= 0
assert stats['schemas_indexed'] >= 0
else:
# If FTS5 not available, might be 0
pass
# Search for API-related content
results = plugin.search(populated_db_path, "API", content_type="all", limit=10)
# Results should be a list (may be empty if FTS5 not available)
assert isinstance(results, list)
# If we have results, verify they're properly formatted
if results:
# Should find both files and schemas
result_types = {result['type'] for result in results}
assert len(result_types) > 0 # At least one type found
# Verify results have required fields
for result in results:
assert 'type' in result
assert 'score' in result
assert result['score'] > 0
if result['type'] == 'file':
assert 'file' in result
assert 'filename' in result['file']
elif result['type'] == 'schema':
assert 'schema' in result
assert 'filename' in result['schema']
def test_search_ranking_quality(self, populated_db_path):
"""Test that search ranking produces sensible results."""
plugin = FTSSearchPlugin()
plugin.initialize(populated_db_path)
plugin.rebuild_index(populated_db_path)
# Search for "GraphQL"
results = plugin.search(populated_db_path, "GraphQL", content_type="files", limit=10)
if results:
# The GraphQL guide should rank highest
top_result = results[0]
assert 'graphql' in top_result['file']['filename'].lower()
# Search for exact phrase
results = plugin.search(populated_db_path, '"API documentation"', content_type="files", limit=10)
if results:
# Should find exact phrase matches
for result in results:
content = result['file'].get('content', '').lower()
# Either in content or highlighted
assert 'api documentation' in content or 'api documentation' in result.get('highlight', '').lower()
def test_search_error_handling(self, populated_db_path):
"""Test search error handling and edge cases."""
plugin = FTSSearchPlugin()
plugin.initialize(populated_db_path)
# Empty query
results = plugin.search(populated_db_path, "", content_type="all", limit=10)
assert isinstance(results, list)
# Very long query
long_query = "word " * 100
results = plugin.search(populated_db_path, long_query, content_type="all", limit=10)
assert isinstance(results, list)
# Special characters
results = plugin.search(populated_db_path, "query with @#$%", content_type="all", limit=10)
assert isinstance(results, list)
# Zero limit
results = plugin.search(populated_db_path, "API", content_type="all", limit=0)
assert len(results) == 0