tegwick
5143864a86
- Created extensible BaseQueryParadigm interface with standardized QueryResult format - Implemented QueryParadigmRegistry for paradigm discovery and management - Added 5 working paradigms: SQL, FTS, GraphQL, JSONPath, Natural Language - Documented 9 additional paradigms: QBE, Batch Manipulation, Visual Query Builder, REST API, NoSQL, UNIX Pipeline, XPath/XQuery, RAG, Data Transformation - Integrated full CLI interface: list, search, show, exec, categories commands - Added comprehensive test suite with 23 test cases covering all components - Auto-registration system enables easy addition of new paradigms - Organized paradigms by category (structural, textual, semantic, visual, procedural, network) and complexity (beginner, intermediate, advanced) 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
333 lines
11 KiB
Python
333 lines
11 KiB
Python
"""
|
|
Tests for query paradigm system - Issue #62
|
|
"""
|
|
|
|
import pytest
|
|
import json
|
|
from markitect.query_paradigms.registry import registry
|
|
from markitect.query_paradigms.base import BaseQueryParadigm, QueryResult
|
|
from markitect.query_paradigms.paradigms.sql_paradigm import SQLQueryParadigm
|
|
from markitect.query_paradigms.paradigms.fts_paradigm import FullTextSearchParadigm
|
|
from markitect.query_paradigms.paradigms.qbe_paradigm import QueryByExampleParadigm
|
|
|
|
|
|
class TestQueryParadigmRegistry:
|
|
"""Test the query paradigm registry system."""
|
|
|
|
def test_registry_has_paradigms(self):
|
|
"""Test that paradigms are automatically registered."""
|
|
paradigms = registry.list_all()
|
|
assert len(paradigms) >= 14 # We expect at least 14 paradigms
|
|
|
|
# Check that key paradigms are present
|
|
paradigm_names = [p.name for p in paradigms]
|
|
assert "SQL" in paradigm_names
|
|
assert "FTS" in paradigm_names
|
|
assert "GraphQL" in paradigm_names
|
|
assert "Natural Language" in paradigm_names
|
|
|
|
def test_get_paradigm_by_name(self):
|
|
"""Test retrieving paradigms by name."""
|
|
sql_paradigm = registry.get("SQL")
|
|
assert sql_paradigm is not None
|
|
assert sql_paradigm.name == "SQL"
|
|
assert sql_paradigm.category == "structural"
|
|
|
|
# Test case insensitive lookup
|
|
fts_paradigm = registry.get("fts")
|
|
assert fts_paradigm is not None
|
|
assert fts_paradigm.name == "FTS"
|
|
|
|
def test_get_nonexistent_paradigm(self):
|
|
"""Test getting a paradigm that doesn't exist."""
|
|
result = registry.get("NonExistentParadigm")
|
|
assert result is None
|
|
|
|
def test_list_by_category(self):
|
|
"""Test filtering paradigms by category."""
|
|
structural = registry.list_by_category("structural")
|
|
assert len(structural) > 0
|
|
|
|
for paradigm in structural:
|
|
assert paradigm.category == "structural"
|
|
|
|
textual = registry.list_by_category("textual")
|
|
assert len(textual) > 0
|
|
|
|
for paradigm in textual:
|
|
assert paradigm.category == "textual"
|
|
|
|
def test_list_by_complexity(self):
|
|
"""Test filtering paradigms by complexity."""
|
|
beginner = registry.list_by_complexity("beginner")
|
|
assert len(beginner) > 0
|
|
|
|
for paradigm in beginner:
|
|
assert paradigm.complexity == "beginner"
|
|
|
|
def test_search_paradigms(self):
|
|
"""Test searching paradigms by query."""
|
|
# Search by name
|
|
sql_results = registry.search_paradigms("SQL")
|
|
assert len(sql_results) > 0
|
|
assert any(p.name == "SQL" for p in sql_results)
|
|
|
|
# Search by description
|
|
visual_results = registry.search_paradigms("visual")
|
|
assert len(visual_results) > 0
|
|
assert any("visual" in p.description.lower() for p in visual_results)
|
|
|
|
# Search for non-existent term
|
|
empty_results = registry.search_paradigms("xyznonexistent")
|
|
assert len(empty_results) == 0
|
|
|
|
def test_get_categories(self):
|
|
"""Test getting all available categories."""
|
|
categories = registry.get_categories()
|
|
assert isinstance(categories, list)
|
|
assert len(categories) > 0
|
|
assert "structural" in categories
|
|
assert "textual" in categories
|
|
assert "semantic" in categories
|
|
|
|
def test_get_complexity_levels(self):
|
|
"""Test getting all complexity levels."""
|
|
levels = registry.get_complexity_levels()
|
|
assert isinstance(levels, list)
|
|
assert len(levels) > 0
|
|
assert "beginner" in levels
|
|
assert "intermediate" in levels
|
|
assert "advanced" in levels
|
|
|
|
|
|
class TestSQLParadigm:
|
|
"""Test the SQL query paradigm."""
|
|
|
|
def test_paradigm_properties(self):
|
|
"""Test SQL paradigm basic properties."""
|
|
paradigm = SQLQueryParadigm()
|
|
assert paradigm.name == "SQL"
|
|
assert paradigm.category == "structural"
|
|
assert paradigm.complexity == "intermediate"
|
|
assert "database" in paradigm.description.lower()
|
|
|
|
def test_validate_query(self):
|
|
"""Test SQL query validation."""
|
|
paradigm = SQLQueryParadigm()
|
|
|
|
# Valid queries
|
|
valid, error = paradigm.validate_query("SELECT * FROM files")
|
|
assert valid
|
|
assert error is None
|
|
|
|
valid, error = paradigm.validate_query("SELECT name FROM files WHERE author = 'Alice'")
|
|
assert valid
|
|
|
|
# Invalid queries
|
|
valid, error = paradigm.validate_query("")
|
|
assert not valid
|
|
assert error is not None
|
|
|
|
valid, error = paradigm.validate_query(" ")
|
|
assert not valid
|
|
|
|
def test_get_examples(self):
|
|
"""Test SQL paradigm examples."""
|
|
paradigm = SQLQueryParadigm()
|
|
examples = paradigm.get_examples()
|
|
|
|
assert isinstance(examples, list)
|
|
assert len(examples) > 0
|
|
|
|
for example in examples:
|
|
assert "name" in example
|
|
assert "description" in example
|
|
assert "query" in example
|
|
assert isinstance(example["query"], str)
|
|
|
|
def test_get_syntax_help(self):
|
|
"""Test SQL syntax help."""
|
|
paradigm = SQLQueryParadigm()
|
|
help_text = paradigm.get_syntax_help()
|
|
|
|
assert isinstance(help_text, str)
|
|
assert len(help_text) > 0
|
|
assert "SELECT" in help_text
|
|
|
|
|
|
class TestFTSParadigm:
|
|
"""Test the Full Text Search paradigm."""
|
|
|
|
def test_paradigm_properties(self):
|
|
"""Test FTS paradigm basic properties."""
|
|
paradigm = FullTextSearchParadigm()
|
|
assert paradigm.name == "FTS"
|
|
assert paradigm.category == "textual"
|
|
assert paradigm.complexity == "beginner"
|
|
assert "search" in paradigm.description.lower()
|
|
|
|
def test_validate_query(self):
|
|
"""Test FTS query validation."""
|
|
paradigm = FullTextSearchParadigm()
|
|
|
|
# Valid queries
|
|
valid, error = paradigm.validate_query("documentation")
|
|
assert valid
|
|
assert error is None
|
|
|
|
valid, error = paradigm.validate_query("API AND documentation")
|
|
assert valid
|
|
|
|
valid, error = paradigm.validate_query('"getting started"')
|
|
assert valid
|
|
|
|
# Invalid queries
|
|
valid, error = paradigm.validate_query("")
|
|
assert not valid
|
|
assert error is not None
|
|
|
|
def test_get_examples(self):
|
|
"""Test FTS paradigm examples."""
|
|
paradigm = FullTextSearchParadigm()
|
|
examples = paradigm.get_examples()
|
|
|
|
assert isinstance(examples, list)
|
|
assert len(examples) > 0
|
|
|
|
# Check for expected example types
|
|
example_names = [ex["name"] for ex in examples]
|
|
assert "Simple search" in example_names
|
|
assert "Boolean search" in example_names
|
|
|
|
|
|
class TestQueryByExampleParadigm:
|
|
"""Test the Query By Example paradigm (documentation-only)."""
|
|
|
|
def test_paradigm_properties(self):
|
|
"""Test QBE paradigm basic properties."""
|
|
paradigm = QueryByExampleParadigm()
|
|
assert paradigm.name == "Query By Example"
|
|
assert paradigm.category == "visual"
|
|
assert paradigm.complexity == "beginner"
|
|
assert "template" in paradigm.description.lower()
|
|
|
|
def test_validate_query(self):
|
|
"""Test QBE query validation."""
|
|
paradigm = QueryByExampleParadigm()
|
|
|
|
# Valid JSON templates
|
|
valid, error = paradigm.validate_query('{"author": "Alice"}')
|
|
assert valid
|
|
assert error is None
|
|
|
|
valid, error = paradigm.validate_query('{"tags": ["tutorial"], "type": "markdown"}')
|
|
assert valid
|
|
|
|
# Invalid queries
|
|
valid, error = paradigm.validate_query("")
|
|
assert not valid
|
|
assert error is not None
|
|
|
|
valid, error = paradigm.validate_query("not json")
|
|
assert not valid
|
|
assert "JSON" in error
|
|
|
|
valid, error = paradigm.validate_query('["not", "an", "object"]')
|
|
assert not valid
|
|
assert "object" in error
|
|
|
|
def test_execute_returns_not_implemented(self):
|
|
"""Test that QBE execution returns not implemented error."""
|
|
paradigm = QueryByExampleParadigm()
|
|
result = paradigm.execute('{"author": "Alice"}')
|
|
|
|
assert isinstance(result, QueryResult)
|
|
assert not result.success
|
|
assert result.error_message is not None
|
|
assert "not yet implemented" in result.error_message.lower()
|
|
assert result.metadata["status"] == "not_implemented"
|
|
|
|
def test_get_syntax_help(self):
|
|
"""Test QBE syntax help."""
|
|
paradigm = QueryByExampleParadigm()
|
|
help_text = paradigm.get_syntax_help()
|
|
|
|
assert isinstance(help_text, str)
|
|
assert len(help_text) > 0
|
|
assert "JSON" in help_text
|
|
assert "template" in help_text.lower()
|
|
|
|
|
|
class TestQueryResult:
|
|
"""Test the QueryResult data structure."""
|
|
|
|
def test_query_result_creation(self):
|
|
"""Test creating a QueryResult."""
|
|
result = QueryResult(
|
|
paradigm="Test",
|
|
query="test query",
|
|
execution_time_ms=10.5,
|
|
result_count=3,
|
|
results=[{"id": 1}, {"id": 2}, {"id": 3}],
|
|
metadata={"type": "test"},
|
|
success=True
|
|
)
|
|
|
|
assert result.paradigm == "Test"
|
|
assert result.query == "test query"
|
|
assert result.execution_time_ms == 10.5
|
|
assert result.result_count == 3
|
|
assert len(result.results) == 3
|
|
assert result.metadata["type"] == "test"
|
|
assert result.success is True
|
|
assert result.error_message is None
|
|
|
|
def test_query_result_with_error(self):
|
|
"""Test creating a QueryResult with error."""
|
|
result = QueryResult(
|
|
paradigm="Test",
|
|
query="bad query",
|
|
execution_time_ms=1.0,
|
|
result_count=0,
|
|
results=[],
|
|
metadata={},
|
|
success=False,
|
|
error_message="Query failed"
|
|
)
|
|
|
|
assert not result.success
|
|
assert result.error_message == "Query failed"
|
|
assert result.result_count == 0
|
|
|
|
|
|
class TestBaseQueryParadigm:
|
|
"""Test the base query paradigm interface."""
|
|
|
|
def test_cannot_instantiate_base_class(self):
|
|
"""Test that BaseQueryParadigm cannot be instantiated directly."""
|
|
with pytest.raises(TypeError):
|
|
BaseQueryParadigm()
|
|
|
|
def test_paradigm_interface(self):
|
|
"""Test that paradigms implement the required interface."""
|
|
paradigm = SQLQueryParadigm()
|
|
|
|
# Test all required properties
|
|
assert hasattr(paradigm, 'name')
|
|
assert hasattr(paradigm, 'description')
|
|
assert hasattr(paradigm, 'category')
|
|
assert hasattr(paradigm, 'complexity')
|
|
|
|
# Test all required methods
|
|
assert hasattr(paradigm, 'execute')
|
|
assert hasattr(paradigm, 'get_examples')
|
|
assert hasattr(paradigm, 'validate_query')
|
|
assert hasattr(paradigm, 'get_syntax_help')
|
|
|
|
# Test optional methods
|
|
assert hasattr(paradigm, 'can_translate_from')
|
|
assert hasattr(paradigm, 'translate_query')
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__]) |