Initial upload: config-converter-cli v1.0.0

tests/test_query.py

@@ -0,0 +1,118 @@
+"""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"]]