119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
"""Tests for the query module."""
|
|
|
|
import pytest
|
|
|
|
from configconverter.query import QueryEngine
|
|
from configconverter.exceptions import QueryError, ParseError
|
|
|
|
|
|
class TestQueryEngine:
|
|
"""Tests for the QueryEngine class."""
|
|
|
|
@pytest.fixture
|
|
def query_engine(self):
|
|
return QueryEngine()
|
|
|
|
def test_simple_key_query(self, query_engine):
|
|
content = '{"name": "test", "value": 123}'
|
|
result = query_engine.query(content, "name")
|
|
assert result == "test"
|
|
|
|
def test_nested_key_query(self, query_engine):
|
|
content = '{"server": {"host": "localhost", "port": 8080}}'
|
|
result = query_engine.query(content, "server.host")
|
|
assert result == "localhost"
|
|
|
|
def test_array_access(self, query_engine):
|
|
content = '{"items": [{"id": 1}, {"id": 2}, {"id": 3}]}'
|
|
result = query_engine.query(content, "items[0].id")
|
|
assert result == 1
|
|
|
|
def test_wildcard_query(self, query_engine):
|
|
content = '{"items": [{"id": 1}, {"id": 2}, {"id": 3}]}'
|
|
result = query_engine.query(content, "items[*].id")
|
|
assert result == [1, 2, 3]
|
|
|
|
def test_filter_expression(self, query_engine):
|
|
content = '{"items": [{"name": "a", "value": 10}, {"name": "b", "value": 20}]}'
|
|
result = query_engine.query(content, "items[?value > `15`].name")
|
|
assert result == ["b"]
|
|
|
|
def test_multi_select(self, query_engine):
|
|
content = '{"server": {"host": "localhost", "port": 8080}}'
|
|
result = query_engine.query(content, "{host: server.host, port: server.port}")
|
|
assert result["host"] == "localhost"
|
|
assert result["port"] == 8080
|
|
|
|
def test_query_nonexistent_key(self, query_engine):
|
|
content = '{"name": "test"}'
|
|
result = query_engine.query(content, "nonexistent")
|
|
assert result is None
|
|
|
|
def test_query_yaml_content(self, query_engine):
|
|
content = 'name: test\nserver:\n host: localhost\n port: 8080\n'
|
|
result = query_engine.query(content, "server.host")
|
|
assert result == "localhost"
|
|
|
|
def test_query_toml_content(self, query_engine):
|
|
content = 'name = "test"\n[server]\nhost = "localhost"\nport = 8080\n'
|
|
result = query_engine.query(content, "server.host")
|
|
assert result == "localhost"
|
|
|
|
def test_query_with_format(self, query_engine):
|
|
content = '{"name": "test"}'
|
|
result = query_engine.query(content, "name", format="json")
|
|
assert result == "test"
|
|
|
|
def test_invalid_jmespath_expression(self, query_engine):
|
|
content = '{"name": "test"}'
|
|
with pytest.raises(QueryError):
|
|
query_engine.query(content, "invalid``expression")
|
|
|
|
def test_query_json_output(self, query_engine):
|
|
content = '{"name": "test", "value": 123}'
|
|
result = query_engine.query_json(content, "name")
|
|
assert '"test"' in result
|
|
|
|
def test_complex_filter(self, query_engine):
|
|
content = '''
|
|
{
|
|
"services": [
|
|
{"name": "web", "port": 80, "enabled": true},
|
|
{"name": "api", "port": 8080, "enabled": true},
|
|
{"name": "db", "port": 5432, "enabled": false}
|
|
]
|
|
}
|
|
'''
|
|
result = query_engine.query(content, "services[?enabled == `true`].name")
|
|
assert result == ["web", "api"]
|
|
|
|
def test_query_empty_result(self, query_engine):
|
|
content = '{"items": [{"name": "a"}]}'
|
|
result = query_engine.query(content, "items[?value > `100`]")
|
|
assert result == []
|
|
|
|
def test_preserve_type_in_result(self, query_engine):
|
|
content = '{"int": 123, "float": 45.67, "bool": true, "string": "test"}'
|
|
result = query_engine.query(content, "int")
|
|
assert result == 123
|
|
result = query_engine.query(content, "float")
|
|
assert result == 45.67
|
|
result = query_engine.query(content, "bool")
|
|
assert result is True
|
|
result = query_engine.query(content, "string")
|
|
assert result == "test"
|
|
|
|
def test_nested_array_query(self, query_engine):
|
|
content = '''
|
|
{
|
|
"org": {
|
|
"teams": [
|
|
{"name": "engineering", "members": ["alice", "bob"]},
|
|
{"name": "design", "members": ["charlie"]}
|
|
]
|
|
}
|
|
}
|
|
'''
|
|
result = query_engine.query(content, "org.teams[*].members")
|
|
assert result == [["alice", "bob"], ["charlie"]]
|