test_query_builder.py•7.25 kB
"""
Tests for QueryBuilder
License: Mozilla Public License 2.0
"""
import pytest
import sys
import os
# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from edgelake_mcp.query import QueryBuilder
class TestQueryBuilder:
"""Test cases for QueryBuilder"""
def setup_method(self):
"""Setup test fixtures"""
self.builder = QueryBuilder()
def test_simple_select(self):
"""Test basic SELECT * query"""
args = {
"database": "testdb",
"table": "users",
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM users LIMIT 10"
def test_select_with_columns(self):
"""Test SELECT with specific columns"""
args = {
"database": "testdb",
"table": "users",
"select": ["id", "name", "email"],
"limit": 5
}
query = self.builder.build_query(args)
assert query == "SELECT id, name, email FROM users LIMIT 5"
def test_where_clause(self):
"""Test WHERE clause"""
args = {
"database": "testdb",
"table": "users",
"where": "is_active = true AND age > 18",
"limit": 20
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM users WHERE is_active = true AND age > 18 LIMIT 20"
def test_group_by(self):
"""Test GROUP BY clause"""
args = {
"database": "testdb",
"table": "orders",
"select": ["customer_id", "COUNT(*) as count"],
"group_by": ["customer_id"],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT customer_id, COUNT(*) as count FROM orders GROUP BY customer_id LIMIT 10"
def test_order_by_asc(self):
"""Test ORDER BY ASC"""
args = {
"database": "testdb",
"table": "products",
"order_by": [{"column": "price", "direction": "ASC"}],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM products ORDER BY price ASC LIMIT 10"
def test_order_by_desc(self):
"""Test ORDER BY DESC"""
args = {
"database": "testdb",
"table": "products",
"order_by": [{"column": "created_at", "direction": "DESC"}],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM products ORDER BY created_at DESC LIMIT 10"
def test_multiple_order_by(self):
"""Test multiple ORDER BY columns"""
args = {
"database": "testdb",
"table": "products",
"order_by": [
{"column": "category"},
{"column": "price", "direction": "DESC"}
],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM products ORDER BY category ASC, price DESC LIMIT 10"
def test_include_tables(self):
"""Test include_tables (JOIN)"""
args = {
"database": "testdb",
"table": "orders",
"include_tables": ["customers", "products"],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM orders, customers, products LIMIT 10"
def test_cross_database_join(self):
"""Test cross-database JOIN"""
args = {
"database": "sales",
"table": "orders",
"include_tables": ["inventory.products"],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM orders, inventory.products LIMIT 10"
def test_extend_fields(self):
"""Test extended metadata fields"""
args = {
"database": "testdb",
"table": "events",
"extend_fields": ["+ip", "+hostname", "@table_name"],
"select": ["timestamp", "event_type"],
"limit": 10
}
query = self.builder.build_query(args)
assert query == "SELECT +ip, +hostname, @table_name, timestamp, event_type FROM events LIMIT 10"
def test_complex_query(self):
"""Test complex query with all features"""
args = {
"database": "iot_data",
"table": "sensor_readings",
"select": ["device_id", "AVG(temperature) as avg_temp", "COUNT(*) as count"],
"where": "timestamp >= '2025-01-01' AND device_type = 'temp'",
"group_by": ["device_id"],
"order_by": [{"column": "avg_temp", "direction": "DESC"}],
"limit": 20
}
query = self.builder.build_query(args)
expected = ("SELECT device_id, AVG(temperature) as avg_temp, COUNT(*) as count "
"FROM sensor_readings "
"WHERE timestamp >= '2025-01-01' AND device_type = 'temp' "
"GROUP BY device_id "
"ORDER BY avg_temp DESC "
"LIMIT 20")
assert query == expected
def test_validate_query_valid(self):
"""Test query validation with valid input"""
args = {
"database": "testdb",
"table": "users"
}
is_valid, error = self.builder.validate_query(args)
assert is_valid is True
assert error is None
def test_validate_query_missing_database(self):
"""Test validation fails when database missing"""
args = {
"table": "users"
}
is_valid, error = self.builder.validate_query(args)
assert is_valid is False
assert "database" in error
def test_validate_query_missing_table(self):
"""Test validation fails when table missing"""
args = {
"database": "testdb"
}
is_valid, error = self.builder.validate_query(args)
assert is_valid is False
assert "table" in error
def test_validate_query_invalid_limit(self):
"""Test validation fails with invalid limit"""
args = {
"database": "testdb",
"table": "users",
"limit": -5
}
is_valid, error = self.builder.validate_query(args)
assert is_valid is False
assert "limit" in error
def test_validate_query_invalid_order_by(self):
"""Test validation fails with invalid order_by"""
args = {
"database": "testdb",
"table": "users",
"order_by": [{"direction": "DESC"}] # Missing column
}
is_valid, error = self.builder.validate_query(args)
assert is_valid is False
assert "order_by" in error
def test_no_limit(self):
"""Test query without LIMIT"""
args = {
"database": "testdb",
"table": "users",
"limit": None
}
query = self.builder.build_query(args)
assert query == "SELECT * FROM users"
assert "LIMIT" not in query
if __name__ == "__main__":
pytest.main([__file__, "-v"])