From e5f3cc91215799897b02a940ab8e06ef082ae2b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philippe=20H=C3=B6ij?= Date: Sat, 15 Nov 2025 08:14:29 +0100 Subject: [PATCH] Improve coverage further --- terminusdb_client/tests/test_main.py | 0 .../tests/test_woql_idgen_random.py | 0 .../tests/test_woql_query_builders.py | 421 +++++++++++ .../tests/test_woql_query_cleaners.py | 659 ++++++++++++++++++ .../tests/test_woql_query_utils.py | 381 ++++++++++ terminusdb_client/tests/woql_test_helpers.py | 160 +++++ test_detection.py | 0 test_random_key.py | 0 8 files changed, 1621 insertions(+) create mode 100644 terminusdb_client/tests/test_main.py create mode 100644 terminusdb_client/tests/test_woql_idgen_random.py create mode 100644 terminusdb_client/tests/test_woql_query_builders.py create mode 100644 terminusdb_client/tests/test_woql_query_cleaners.py create mode 100644 terminusdb_client/tests/test_woql_query_utils.py create mode 100644 terminusdb_client/tests/woql_test_helpers.py create mode 100644 test_detection.py create mode 100644 test_random_key.py diff --git a/terminusdb_client/tests/test_main.py b/terminusdb_client/tests/test_main.py new file mode 100644 index 00000000..e69de29b diff --git a/terminusdb_client/tests/test_woql_idgen_random.py b/terminusdb_client/tests/test_woql_idgen_random.py new file mode 100644 index 00000000..e69de29b diff --git a/terminusdb_client/tests/test_woql_query_builders.py b/terminusdb_client/tests/test_woql_query_builders.py new file mode 100644 index 00000000..2df6301f --- /dev/null +++ b/terminusdb_client/tests/test_woql_query_builders.py @@ -0,0 +1,421 @@ +"""Tests for WOQL query builder methods using standardized helpers.""" +from terminusdb_client.woqlquery.woql_query import WOQLQuery, Var +from terminusdb_client.tests.woql_test_helpers import WOQLTestHelpers as H + + +# Basic Query Builders + +def test_triple_basic(): + """Test triple() creates proper Triple structure.""" + query = WOQLQuery().triple("v:Subject", "rdf:type", "v:Object") + + H.assert_triple_structure(query) + H.assert_is_variable(query._query["subject"], "Subject") + H.assert_is_variable(query._query["object"], "Object") + + +def test_triple_with_strings(): + """Test triple() with plain string arguments.""" + query = WOQLQuery().triple("doc:Person1", "rdfs:label", "Alice") + + H.assert_triple_structure(query) + assert query._query["subject"]["node"] == "doc:Person1" + assert query._query["predicate"]["node"] == "rdfs:label" + + +def test_triple_with_var_objects(): + """Test triple() with Var objects.""" + s = Var("Subject") + p = Var("Pred") + o = Var("Obj") + + query = WOQLQuery().triple(s, p, o) + + H.assert_triple_structure(query) + H.assert_is_variable(query._query["subject"], "Subject") + H.assert_is_variable(query._query["predicate"], "Pred") + H.assert_is_variable(query._query["object"], "Obj") + + +def test_triple_with_opt(): + """Test triple() with opt=True wraps in Optional.""" + query = WOQLQuery().triple("v:S", "v:P", "v:O", opt=True) + + H.assert_optional_structure(query) + # The triple is nested inside the optional + inner = query._query["query"] + assert inner["@type"] == "Triple" + + +def test_select_basic(): + """Test select() creates proper Select structure.""" + query = WOQLQuery().select("v:X", "v:Y") + + H.assert_select_structure(query) + assert query._query["variables"] == ["X", "Y"] + + +def test_select_with_strings_and_subquery(): + """Test select() with string variables (strings don't have to_dict).""" + query = WOQLQuery().select("v:X", "v:Y", "v:Z") + + H.assert_select_structure(query) + assert query._query["variables"] == ["X", "Y", "Z"] + + +def test_select_with_subquery(): + """Test select() with embedded subquery.""" + subquery = WOQLQuery().triple("v:X", "rdf:type", "owl:Class") + + query = WOQLQuery().select("v:X", subquery) + + H.assert_select_structure(query) + assert query._query["variables"] == ["X"] + assert query._query["query"]["@type"] == "Triple" + + +def test_select_empty(): + """Test select() with no arguments.""" + query = WOQLQuery().select() + + H.assert_select_structure(query) + assert query._query["variables"] == [] + + +def test_distinct_basic(): + """Test distinct() creates proper Distinct structure.""" + query = WOQLQuery().distinct("v:X") + + H.assert_query_type(query, "Distinct") + H.assert_has_key(query, "variables") + assert query._query["variables"] == ["X"] + + +def test_distinct_with_subquery(): + """Test distinct() with embedded subquery.""" + subquery = WOQLQuery().triple("v:X", "v:P", "v:Y") + + query = WOQLQuery().distinct("v:X", "v:Y", subquery) + + H.assert_query_type(query, "Distinct") + assert query._query["variables"] == ["X", "Y"] + assert query._query["query"]["@type"] == "Triple" + + +def test_woql_and_basic(): + """Test woql_and() creates proper And structure.""" + q1 = WOQLQuery().triple("v:X", "rdf:type", "v:T") + q2 = WOQLQuery().triple("v:X", "rdfs:label", "v:L") + + query = WOQLQuery().woql_and(q1, q2) + + H.assert_and_structure(query, expected_count=2) + assert query._query["and"][0]["@type"] == "Triple" + assert query._query["and"][1]["@type"] == "Triple" + + +def test_woql_and_flattens_nested_and(): + """Test woql_and() flattens nested And queries.""" + q1 = WOQLQuery().triple("v:A", "v:B", "v:C") + q2 = WOQLQuery().triple("v:D", "v:E", "v:F") + nested = WOQLQuery().woql_and(q1, q2) + + q3 = WOQLQuery().triple("v:G", "v:H", "v:I") + query = WOQLQuery().woql_and(nested, q3) + + H.assert_and_structure(query, expected_count=3) + + +def test_woql_and_with_existing_cursor(): + """Test woql_and() with existing query in cursor.""" + query = WOQLQuery().triple("v:X", "v:Y", "v:Z") + query.woql_and(WOQLQuery().triple("v:A", "v:B", "v:C")) + + H.assert_and_structure(query, expected_count=2) + + +def test_woql_or_basic(): + """Test woql_or() creates proper Or structure.""" + q1 = WOQLQuery().triple("v:X", "rdf:type", "Person") + q2 = WOQLQuery().triple("v:X", "rdf:type", "Company") + + query = WOQLQuery().woql_or(q1, q2) + + H.assert_or_structure(query, expected_count=2) + assert query._query["or"][0]["@type"] == "Triple" + assert query._query["or"][1]["@type"] == "Triple" + + +def test_woql_or_multiple(): + """Test woql_or() with multiple queries.""" + q1 = WOQLQuery().triple("v:X", "v:P1", "v:O1") + q2 = WOQLQuery().triple("v:X", "v:P2", "v:O2") + q3 = WOQLQuery().triple("v:X", "v:P3", "v:O3") + + query = WOQLQuery().woql_or(q1, q2, q3) + + H.assert_or_structure(query, expected_count=3) + + +def test_woql_not(): + """Test not operator creates proper Not structure.""" + inner = WOQLQuery().triple("v:X", "v:P", "v:O") + + query = ~inner + + H.assert_not_structure(query) + assert query._query["query"]["@type"] == "Triple" + + +def test_using_basic(): + """Test using() creates proper Using structure.""" + query = WOQLQuery().using("myCollection") + + H.assert_query_type(query, "Using") + H.assert_key_value(query, "collection", "myCollection") + + +def test_using_with_subquery(): + """Test using() with subquery.""" + subquery = WOQLQuery().triple("v:X", "v:Y", "v:Z") + + query = WOQLQuery().using("myCollection", subquery) + + H.assert_query_type(query, "Using") + assert query._query["collection"] == "myCollection" + assert query._query["query"]["@type"] == "Triple" + + +def test_using_invalid_collection(): + """Test using() raises ValueError for invalid collection.""" + try: + WOQLQuery().using(123) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "Collection ID" in str(e) + + +def test_comment_basic(): + """Test comment() creates proper Comment structure.""" + query = WOQLQuery().comment("This is a comment") + + H.assert_query_type(query, "Comment") + H.assert_has_key(query, "comment") + assert query._query["comment"]["@type"] == "xsd:string" + + +def test_comment_with_subquery(): + """Test comment() with subquery.""" + subquery = WOQLQuery().triple("v:X", "v:Y", "v:Z") + + query = WOQLQuery().comment("Test comment", subquery) + + H.assert_query_type(query, "Comment") + assert query._query["query"]["@type"] == "Triple" + + +def test_woql_from_basic(): + """Test woql_from() creates proper From structure.""" + query = WOQLQuery().woql_from("instance/main") + + H.assert_query_type(query, "From") + H.assert_key_value(query, "graph", "instance/main") + + +def test_woql_from_with_subquery(): + """Test woql_from() with subquery.""" + subquery = WOQLQuery().triple("v:X", "v:Y", "v:Z") + + query = WOQLQuery().woql_from("instance/main", subquery) + + H.assert_query_type(query, "From") + assert query._query["graph"] == "instance/main" + assert query._query["query"]["@type"] == "Triple" + + +def test_woql_from_invalid_graph(): + """Test woql_from() raises ValueError for invalid graph.""" + try: + WOQLQuery().woql_from(123) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "Graph Filter Expression" in str(e) + + +def test_into_basic(): + """Test into() creates proper Into structure.""" + subquery = WOQLQuery().triple("v:X", "v:Y", "v:Z") + + query = WOQLQuery().into("instance/main", subquery) + + H.assert_query_type(query, "Into") + H.assert_key_value(query, "graph", "instance/main") + + +def test_into_invalid_graph(): + """Test into() raises ValueError for invalid graph.""" + subquery = WOQLQuery().triple("v:X", "v:Y", "v:Z") + + try: + WOQLQuery().into(None, subquery) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "Graph Filter Expression" in str(e) + + +def test_execute_without_commit(): + """Test execute() calls client.query without commit_msg.""" + query = WOQLQuery().triple("v:X", "v:Y", "v:Z") + client = H.create_mock_client() + + result = query.execute(client) + + client.query.assert_called_once_with(query) + assert result["@type"] == "api:WoqlResponse" + + +def test_execute_with_commit(): + """Test execute() calls client.query with commit_msg.""" + query = WOQLQuery().triple("v:X", "v:Y", "v:Z") + client = H.create_mock_client() + + result = query.execute(client, commit_msg="Test commit") + + client.query.assert_called_once_with(query, "Test commit") + assert result["@type"] == "api:WoqlResponse" + + +# Chaining Tests + +def test_chaining_triple_and_triple(): + """Test chaining multiple triple() calls.""" + query = WOQLQuery().triple("v:X", "rdf:type", "Person") + query.triple("v:X", "rdfs:label", "v:Name") + + H.assert_and_structure(query, expected_count=2) + assert query._query["and"][0]["@type"] == "Triple" + assert query._query["and"][1]["@type"] == "Triple" + + +def test_chaining_select_with_triple(): + """Test chaining select() with triple().""" + inner_query = WOQLQuery().triple("v:X", "rdf:type", "Person") + query = WOQLQuery().select("v:X", inner_query) + + H.assert_select_structure(query) + assert query._query["query"]["@type"] == "Triple" + + +def test_operators_return_self(): + """Test that query methods return self for chaining.""" + query = WOQLQuery() + + result1 = query.triple("v:X", "v:Y", "v:Z") + assert result1 is query + + result2 = query.select("v:X") + # select wraps, so cursor changes but returns self + assert result2 is query + + +# Edge Cases + +def test_select_with_empty_list(): + """Test select() explicitly with empty list.""" + query = WOQLQuery().select() + + H.assert_select_structure(query) + assert query._query["variables"] == [] + + +def test_woql_and_empty(): + """Test woql_and() with no arguments.""" + query = WOQLQuery().woql_and() + + H.assert_and_structure(query, expected_count=0) + + +def test_woql_or_empty(): + """Test woql_or() with no arguments.""" + query = WOQLQuery().woql_or() + + H.assert_or_structure(query, expected_count=0) + + +def test_triple_with_integer_object(): + """Test triple() with integer as object.""" + query = WOQLQuery().triple("v:X", "schema:age", 42) + + H.assert_triple_structure(query) + obj = query._query["object"] + assert obj["@type"] == "Value" + assert obj["data"]["@type"] == "xsd:integer" + assert obj["data"]["@value"] == 42 + + +def test_triple_with_string_object(): + """Test triple() with string literal as object.""" + query = WOQLQuery().triple("v:X", "rdfs:label", "Alice") + + H.assert_triple_structure(query) + obj = query._query["object"] + # String without v: prefix becomes a node + assert "node" in obj or obj.get("@type") == "Value" + + +# Args Introspection Tests + +def test_using_args_introspection(): + """Test using() returns args list when called with 'args'.""" + result = WOQLQuery().using("args") + + assert result == ["collection", "query"] + + +def test_comment_args_introspection(): + """Test comment() returns args list when called with 'args'.""" + result = WOQLQuery().comment("args") + + assert result == ["comment", "query"] + + +def test_select_args_introspection(): + """Test select() returns args list when called with 'args'.""" + result = WOQLQuery().select("args") + + assert result == ["variables", "query"] + + +def test_distinct_args_introspection(): + """Test distinct() returns args list when called with 'args'.""" + result = WOQLQuery().distinct("args") + + assert result == ["variables", "query"] + + +def test_woql_and_args_introspection(): + """Test woql_and() returns args list when called with 'args'.""" + result = WOQLQuery().woql_and("args") + + assert result == ["and"] + + +def test_woql_or_args_introspection(): + """Test woql_or() returns args list when called with 'args'.""" + result = WOQLQuery().woql_or("args") + + assert result == ["or"] + + +def test_woql_from_args_introspection(): + """Test woql_from() returns args list when called with 'args'.""" + result = WOQLQuery().woql_from("args") + + assert result == ["graph", "query"] + + +def test_into_args_introspection(): + """Test into() returns args list when called with 'args'.""" + result = WOQLQuery().into("args", None) + + assert result == ["graph", "query"] diff --git a/terminusdb_client/tests/test_woql_query_cleaners.py b/terminusdb_client/tests/test_woql_query_cleaners.py new file mode 100644 index 00000000..64466aef --- /dev/null +++ b/terminusdb_client/tests/test_woql_query_cleaners.py @@ -0,0 +1,659 @@ +"""Tests for WOQLQuery cleaning and expansion methods.""" +import datetime +from terminusdb_client.woqlquery.woql_query import WOQLQuery, Var, Doc + + +def test_clean_subject_with_string(): + """Test _clean_subject with plain string.""" + query = WOQLQuery() + + result = query._clean_subject("testSubject") + + assert result["@type"] == "NodeValue" + assert result["node"] == "testSubject" + + +def test_clean_subject_with_colon(): + """Test _clean_subject with URI string containing colon.""" + query = WOQLQuery() + + result = query._clean_subject("doc:Person") + + assert result["@type"] == "NodeValue" + assert result["node"] == "doc:Person" + + +def test_clean_subject_with_vocab(): + """Test _clean_subject uses vocab mapping.""" + query = WOQLQuery() + + result = query._clean_subject("type") + + # "type" maps to "rdf:type" in SHORT_NAME_MAPPING + assert result["@type"] == "NodeValue" + assert result["node"] == "rdf:type" + + +def test_clean_subject_with_var(): + """Test _clean_subject with Var object.""" + query = WOQLQuery() + var = Var("Subject") + + result = query._clean_subject(var) + + assert result["@type"] == "NodeValue" + assert result["variable"] == "Subject" + + +def test_clean_subject_with_dict(): + """Test _clean_subject returns dict as-is.""" + query = WOQLQuery() + input_dict = {"@type": "NodeValue", "node": "test"} + + result = query._clean_subject(input_dict) + + assert result == input_dict + + +def test_clean_subject_with_invalid_type(): + """Test _clean_subject raises ValueError for invalid type.""" + query = WOQLQuery() + + try: + query._clean_subject(123) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "Subject must be a URI string" in str(e) + + +def test_clean_predicate_with_string(): + """Test _clean_predicate with plain string.""" + query = WOQLQuery() + + result = query._clean_predicate("testPredicate") + + assert result["@type"] == "NodeValue" + assert result["node"] == "testPredicate" + + +def test_clean_predicate_with_colon(): + """Test _clean_predicate with URI containing colon.""" + query = WOQLQuery() + + result = query._clean_predicate("rdf:type") + + assert result["@type"] == "NodeValue" + assert result["node"] == "rdf:type" + + +def test_clean_predicate_with_vocab(): + """Test _clean_predicate uses vocab mapping.""" + query = WOQLQuery() + + result = query._clean_predicate("label") + + # "label" maps to "rdfs:label" + assert result["@type"] == "NodeValue" + assert result["node"] == "rdfs:label" + + +def test_clean_predicate_with_var(): + """Test _clean_predicate with Var object.""" + query = WOQLQuery() + var = Var("Predicate") + + result = query._clean_predicate(var) + + assert result["@type"] == "NodeValue" + assert result["variable"] == "Predicate" + + +def test_clean_predicate_with_dict(): + """Test _clean_predicate returns dict as-is.""" + query = WOQLQuery() + input_dict = {"@type": "NodeValue", "node": "test"} + + result = query._clean_predicate(input_dict) + + assert result == input_dict + + +def test_clean_predicate_with_invalid_type(): + """Test _clean_predicate raises ValueError for invalid type.""" + query = WOQLQuery() + + try: + query._clean_predicate(123) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "Predicate must be a URI string" in str(e) + + +def test_clean_path_predicate_with_colon(): + """Test _clean_path_predicate with URI containing colon.""" + query = WOQLQuery() + + result = query._clean_path_predicate("doc:parent") + + assert result == "doc:parent" + + +def test_clean_path_predicate_with_vocab(): + """Test _clean_path_predicate uses vocab mapping.""" + query = WOQLQuery() + + result = query._clean_path_predicate("type") + + assert result == "rdf:type" + + +def test_clean_path_predicate_with_plain_string(): + """Test _clean_path_predicate with plain string.""" + query = WOQLQuery() + + result = query._clean_path_predicate("customPredicate") + + assert result == "customPredicate" + + +def test_clean_path_predicate_with_none(): + """Test _clean_path_predicate returns False for None.""" + query = WOQLQuery() + + result = query._clean_path_predicate(None) + + assert result is False + + +def test_clean_object_with_string(): + """Test _clean_object with plain string.""" + query = WOQLQuery() + + result = query._clean_object("testObject") + + assert result["@type"] == "Value" + assert result["node"] == "testObject" + + +def test_clean_object_with_v_prefix(): + """Test _clean_object with v: prefixed variable.""" + query = WOQLQuery() + + result = query._clean_object("v:MyVar") + + assert result["@type"] == "Value" + assert result["variable"] == "MyVar" + + +def test_clean_object_with_var(): + """Test _clean_object with Var object.""" + query = WOQLQuery() + var = Var("Object") + + result = query._clean_object(var) + + assert result["@type"] == "Value" + assert result["variable"] == "Object" + + +def test_clean_object_with_doc(): + """Test _clean_object with Doc object.""" + query = WOQLQuery() + doc = Doc({"name": "Alice"}) + + result = query._clean_object(doc) + + # Should return the encoded version + assert "@type" in result + + +def test_clean_object_with_int(): + """Test _clean_object with integer.""" + query = WOQLQuery() + + result = query._clean_object(42) + + assert result["@type"] == "Value" + assert result["data"]["@type"] == "xsd:integer" + assert result["data"]["@value"] == 42 + + +def test_clean_object_with_float(): + """Test _clean_object with float.""" + query = WOQLQuery() + + result = query._clean_object(3.14) + + assert result["@type"] == "Value" + assert result["data"]["@type"] == "xsd:decimal" + assert result["data"]["@value"] == 3.14 + + +def test_clean_object_with_bool(): + """Test _clean_object with boolean.""" + query = WOQLQuery() + + result = query._clean_object(True) + + assert result["@type"] == "Value" + assert result["data"]["@type"] == "xsd:boolean" + assert result["data"]["@value"] is True + + +def test_clean_object_with_date(): + """Test _clean_object with datetime.""" + query = WOQLQuery() + dt = datetime.datetime(2025, 1, 1, 12, 0, 0) + + result = query._clean_object(dt) + + assert result["@type"] == "Value" + assert result["data"]["@type"] == "xsd:dateTime" + assert result["data"]["@value"] == "2025-01-01T12:00:00" + + +def test_clean_object_with_dict_value(): + """Test _clean_object with dict containing @value.""" + query = WOQLQuery() + input_dict = {"@type": "xsd:string", "@value": "test"} + + result = query._clean_object(input_dict) + + assert result["@type"] == "Value" + assert result["data"] == input_dict + + +def test_clean_object_with_plain_dict(): + """Test _clean_object with dict without @value.""" + query = WOQLQuery() + input_dict = {"@type": "Custom", "node": "test"} + + result = query._clean_object(input_dict) + + assert result == input_dict + + +def test_clean_object_with_list(): + """Test _clean_object with list.""" + query = WOQLQuery() + + result = query._clean_object([1, "test", True]) + + assert isinstance(result, list) + assert len(result) == 3 + + +def test_clean_object_with_custom_target(): + """Test _clean_object with custom target type.""" + query = WOQLQuery() + + result = query._clean_object(42, "xsd:long") + + assert result["data"]["@type"] == "xsd:long" + + +def test_clean_data_value_with_string(): + """Test _clean_data_value with string.""" + query = WOQLQuery() + + result = query._clean_data_value("test") + + assert result["@type"] == "DataValue" + assert result["data"]["@type"] == "xsd:string" + assert result["data"]["@value"] == "test" + + +def test_clean_data_value_with_v_prefix(): + """Test _clean_data_value with v: prefixed variable.""" + query = WOQLQuery() + + result = query._clean_data_value("v:MyVar") + + assert result["@type"] == "DataValue" + assert result["variable"] == "MyVar" + + +def test_clean_data_value_with_var(): + """Test _clean_data_value with Var object.""" + query = WOQLQuery() + var = Var("Data") + + result = query._clean_data_value(var) + + assert result["@type"] == "DataValue" + assert result["variable"] == "Data" + + +def test_clean_data_value_with_int(): + """Test _clean_data_value with integer.""" + query = WOQLQuery() + + result = query._clean_data_value(100) + + assert result["@type"] == "DataValue" + assert result["data"]["@type"] == "xsd:integer" + assert result["data"]["@value"] == 100 + + +def test_clean_data_value_with_float(): + """Test _clean_data_value with float.""" + query = WOQLQuery() + + result = query._clean_data_value(2.5) + + assert result["@type"] == "DataValue" + assert result["data"]["@type"] == "xsd:decimal" + assert result["data"]["@value"] == 2.5 + + +def test_clean_data_value_with_bool(): + """Test _clean_data_value with boolean.""" + query = WOQLQuery() + + result = query._clean_data_value(False) + + assert result["@type"] == "DataValue" + assert result["data"]["@type"] == "xsd:boolean" + assert result["data"]["@value"] is False + + +def test_clean_data_value_with_date(): + """Test _clean_data_value with datetime.""" + query = WOQLQuery() + dt = datetime.date(2025, 1, 1) + + result = query._clean_data_value(dt) + + assert result["@type"] == "DataValue" + assert result["data"]["@type"] == "xsd:dateTime" + assert "2025-01-01" in result["data"]["@value"] + + +def test_clean_data_value_with_dict_value(): + """Test _clean_data_value with dict containing @value.""" + query = WOQLQuery() + input_dict = {"@type": "xsd:decimal", "@value": 3.14} + + result = query._clean_data_value(input_dict) + + assert result["@type"] == "DataValue" + assert result["data"] == input_dict + + +def test_clean_data_value_with_plain_dict(): + """Test _clean_data_value with dict without @value.""" + query = WOQLQuery() + input_dict = {"@type": "Custom"} + + result = query._clean_data_value(input_dict) + + assert result == input_dict + + +def test_clean_arithmetic_value_with_string(): + """Test _clean_arithmetic_value with string.""" + query = WOQLQuery() + + result = query._clean_arithmetic_value("42") + + assert result["@type"] == "ArithmeticValue" + assert result["data"]["@value"] == "42" + + +def test_clean_arithmetic_value_with_v_prefix(): + """Test _clean_arithmetic_value with v: prefixed variable.""" + query = WOQLQuery() + + result = query._clean_arithmetic_value("v:Count") + + assert result["@type"] == "ArithmeticValue" + assert result["variable"] == "Count" + + +def test_clean_arithmetic_value_with_int(): + """Test _clean_arithmetic_value with integer.""" + query = WOQLQuery() + + result = query._clean_arithmetic_value(50) + + assert result["@type"] == "ArithmeticValue" + assert result["data"]["@type"] == "xsd:integer" + assert result["data"]["@value"] == 50 + + +def test_clean_arithmetic_value_with_float(): + """Test _clean_arithmetic_value with float.""" + query = WOQLQuery() + + result = query._clean_arithmetic_value(7.5) + + assert result["@type"] == "ArithmeticValue" + assert result["data"]["@type"] == "xsd:decimal" + assert result["data"]["@value"] == 7.5 + + +def test_clean_arithmetic_value_with_bool(): + """Test _clean_arithmetic_value with boolean.""" + query = WOQLQuery() + + result = query._clean_arithmetic_value(True) + + assert result["@type"] == "ArithmeticValue" + assert result["data"]["@type"] == "xsd:boolean" + + +def test_clean_arithmetic_value_with_date(): + """Test _clean_arithmetic_value with datetime.""" + query = WOQLQuery() + dt = datetime.date(2025, 6, 15) + + result = query._clean_arithmetic_value(dt) + + assert result["@type"] == "ArithmeticValue" + assert result["data"]["@type"] == "xsd:dateTime" + + +def test_clean_arithmetic_value_with_dict_value(): + """Test _clean_arithmetic_value with dict containing @value.""" + query = WOQLQuery() + input_dict = {"@type": "xsd:integer", "@value": 999} + + result = query._clean_arithmetic_value(input_dict) + + assert result["@type"] == "ArithmeticValue" + assert result["data"] == input_dict + + +def test_clean_arithmetic_value_with_plain_dict(): + """Test _clean_arithmetic_value with dict without @value.""" + query = WOQLQuery() + input_dict = {"@type": "Custom"} + + result = query._clean_arithmetic_value(input_dict) + + assert result == input_dict + + +def test_clean_node_value_with_string(): + """Test _clean_node_value with plain string.""" + query = WOQLQuery() + + result = query._clean_node_value("nodeId") + + assert result["@type"] == "NodeValue" + assert result["node"] == "nodeId" + + +def test_clean_node_value_with_v_prefix(): + """Test _clean_node_value with v: prefixed variable.""" + query = WOQLQuery() + + result = query._clean_node_value("v:Node") + + assert result["@type"] == "NodeValue" + assert result["variable"] == "Node" + + +def test_clean_node_value_with_var(): + """Test _clean_node_value with Var object.""" + query = WOQLQuery() + var = Var("MyNode") + + result = query._clean_node_value(var) + + assert result["@type"] == "NodeValue" + assert result["variable"] == "MyNode" + + +def test_clean_node_value_with_dict(): + """Test _clean_node_value with dict.""" + query = WOQLQuery() + input_dict = {"@type": "Custom", "node": "test"} + + result = query._clean_node_value(input_dict) + + assert result == input_dict + + +def test_clean_node_value_with_other_type(): + """Test _clean_node_value with non-string, non-Var, non-dict.""" + query = WOQLQuery() + + result = query._clean_node_value(12345) + + assert result["@type"] == "NodeValue" + assert result["node"] == 12345 + + +def test_clean_graph(): + """Test _clean_graph returns graph as-is.""" + query = WOQLQuery() + + result = query._clean_graph("instance/main") + + assert result == "instance/main" + + +def test_expand_variable_with_v_prefix(): + """Test _expand_variable with v: prefix.""" + query = WOQLQuery() + + result = query._expand_variable("v:MyVar", "Value") + + assert result["@type"] == "Value" + assert result["variable"] == "MyVar" + + +def test_expand_variable_with_always_true(): + """Test _expand_variable with always=True.""" + query = WOQLQuery() + + result = query._expand_variable("MyVar", "Value", always=True) + + assert result["@type"] == "Value" + assert result["variable"] == "MyVar" + + +def test_expand_variable_without_v_prefix(): + """Test _expand_variable without v: prefix and always=False.""" + query = WOQLQuery() + + result = query._expand_variable("nodeId", "NodeValue") + + assert result["@type"] == "NodeValue" + assert result["node"] == "nodeId" + + +def test_expand_variable_with_var_object(): + """Test _expand_variable with Var object.""" + query = WOQLQuery() + var = Var("X") + + result = query._expand_variable(var, "DataValue") + + assert result["@type"] == "DataValue" + assert result["variable"] == "X" + + +def test_expand_value_variable(): + """Test _expand_value_variable helper.""" + query = WOQLQuery() + + result = query._expand_value_variable("v:Val") + + assert result["@type"] == "Value" + assert result["variable"] == "Val" + + +def test_expand_node_variable(): + """Test _expand_node_variable helper.""" + query = WOQLQuery() + + result = query._expand_node_variable("v:Node") + + assert result["@type"] == "NodeValue" + assert result["variable"] == "Node" + + +def test_expand_data_variable(): + """Test _expand_data_variable helper.""" + query = WOQLQuery() + + result = query._expand_data_variable("v:Data") + + assert result["@type"] == "DataValue" + assert result["variable"] == "Data" + + +def test_expand_arithmetic_variable(): + """Test _expand_arithmetic_variable helper.""" + query = WOQLQuery() + + result = query._expand_arithmetic_variable("v:Arith") + + assert result["@type"] == "ArithmeticValue" + assert result["variable"] == "Arith" + + +def test_to_json(): + """Test to_json serializes query.""" + query = WOQLQuery({"@type": "Triple"}) + + result = query.to_json() + + assert isinstance(result, str) + assert "@type" in result + assert "Triple" in result + + +def test_from_json(): + """Test from_json deserializes query.""" + query = WOQLQuery() + json_str = '{"@type": "Triple", "subject": "v:X"}' + + result = query.from_json(json_str) + + assert result is query + assert query._query["@type"] == "Triple" + + +def test_json_without_input(): + """Test _json returns JSON string when no input.""" + query = WOQLQuery({"@type": "And", "and": []}) + + result = query._json() + + assert isinstance(result, str) + # The query is serialized to JSON + assert "{" in result and "}" in result + + +def test_json_with_input(): + """Test _json sets query when input provided.""" + query = WOQLQuery() + json_str = '{"@type": "Or"}' + + result = query._json(json_str) + + assert result is query + assert query._query["@type"] == "Or" diff --git a/terminusdb_client/tests/test_woql_query_utils.py b/terminusdb_client/tests/test_woql_query_utils.py new file mode 100644 index 00000000..d4f1c66e --- /dev/null +++ b/terminusdb_client/tests/test_woql_query_utils.py @@ -0,0 +1,381 @@ +"""Tests for WOQLQuery utility methods.""" +from terminusdb_client.woqlquery.woql_query import WOQLQuery + + +def test_to_dict(): + """Test to_dict returns copy of query.""" + query = WOQLQuery({"@type": "Triple", "subject": "v:X"}) + + result = query.to_dict() + + assert result["@type"] == "Triple" + assert result["subject"] == "v:X" + # Should be a copy, not the same object + assert result is not query._query + + +def test_from_dict(): + """Test from_dict sets query from dict.""" + query = WOQLQuery() + dictdata = {"@type": "And", "and": []} + + result = query.from_dict(dictdata) + + assert result is query # Returns self + assert query._query["@type"] == "And" + + +def test_find_last_subject_with_subject(): + """Test _find_last_subject finds subject in simple query.""" + query = WOQLQuery() + json_obj = {"@type": "Triple", "subject": "v:X", "predicate": "v:P"} + + result = query._find_last_subject(json_obj) + + assert result == json_obj + + +def test_find_last_subject_in_and(): + """Test _find_last_subject finds subject in And clause.""" + query = WOQLQuery() + triple = {"@type": "Triple", "subject": "v:Y"} + json_obj = { + "@type": "And", + "and": [ + {"@type": "Other"}, + triple + ] + } + + result = query._find_last_subject(json_obj) + + assert result == triple + + +def test_find_last_subject_in_or(): + """Test _find_last_subject finds subject in Or clause.""" + query = WOQLQuery() + triple = {"@type": "Triple", "subject": "v:Z"} + json_obj = { + "@type": "Or", + "or": [ + {"@type": "Other"}, + triple + ] + } + + result = query._find_last_subject(json_obj) + + assert result == triple + + +def test_find_last_subject_in_nested_query(): + """Test _find_last_subject finds subject in nested query.""" + query = WOQLQuery() + triple = {"@type": "Triple", "subject": "v:A"} + json_obj = { + "@type": "Select", + "query": triple + } + + result = query._find_last_subject(json_obj) + + assert result == triple + + +def test_find_last_subject_not_found(): + """Test _find_last_subject returns False when no subject found.""" + query = WOQLQuery() + json_obj = {"@type": "Other", "data": "test"} + + result = query._find_last_subject(json_obj) + + assert result is False + + +def test_find_last_subject_reverse_order(): + """Test _find_last_subject searches in reverse order.""" + query = WOQLQuery() + first = {"@type": "Triple", "subject": "v:First"} + last = {"@type": "Triple", "subject": "v:Last"} + json_obj = { + "@type": "And", + "and": [first, last] + } + + result = query._find_last_subject(json_obj) + + # Should find the last one + assert result == last + + +def test_find_last_property_with_object_property(): + """Test _find_last_property finds owl:ObjectProperty.""" + query = WOQLQuery() + json_obj = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdf:type", + "object": "owl:ObjectProperty" + } + + result = query._find_last_property(json_obj) + + assert result == json_obj + + +def test_find_last_property_with_datatype_property(): + """Test _find_last_property finds owl:DatatypeProperty.""" + query = WOQLQuery() + json_obj = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdf:type", + "object": "owl:DatatypeProperty" + } + + result = query._find_last_property(json_obj) + + assert result == json_obj + + +def test_find_last_property_with_domain(): + """Test _find_last_property finds rdfs:domain predicate.""" + query = WOQLQuery() + json_obj = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdfs:domain", + "object": "v:Y" + } + + result = query._find_last_property(json_obj) + + assert result == json_obj + + +def test_find_last_property_with_range(): + """Test _find_last_property finds rdfs:range predicate.""" + query = WOQLQuery() + json_obj = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdfs:range", + "object": "xsd:string" + } + + result = query._find_last_property(json_obj) + + assert result == json_obj + + +def test_find_last_property_in_and(): + """Test _find_last_property finds property in And clause.""" + query = WOQLQuery() + prop_triple = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdfs:domain", + "object": "v:Y" + } + json_obj = { + "@type": "And", + "and": [ + {"@type": "Other"}, + prop_triple + ] + } + + result = query._find_last_property(json_obj) + + assert result == prop_triple + + +def test_find_last_property_in_or(): + """Test _find_last_property finds property in Or clause.""" + query = WOQLQuery() + prop_triple = { + "@type": "Triple", + "subject": "v:X", + "object": "owl:ObjectProperty" + } + json_obj = { + "@type": "Or", + "or": [ + {"@type": "Other"}, + prop_triple + ] + } + + result = query._find_last_property(json_obj) + + assert result == prop_triple + + +def test_find_last_property_in_nested_query(): + """Test _find_last_property finds property in nested query.""" + query = WOQLQuery() + prop_triple = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdfs:range", + "object": "v:Y" + } + json_obj = { + "@type": "Select", + "query": prop_triple + } + + result = query._find_last_property(json_obj) + + assert result == prop_triple + + +def test_find_last_property_not_found(): + """Test _find_last_property returns False when no property found.""" + query = WOQLQuery() + json_obj = { + "@type": "Triple", + "subject": "v:X", + "predicate": "someOther:predicate", + "object": "v:Y" + } + + result = query._find_last_property(json_obj) + + assert result is False + + +def test_is_property_triple_object_property(): + """Test _is_property_triple with owl:ObjectProperty.""" + query = WOQLQuery() + + result = query._is_property_triple("rdf:type", "owl:ObjectProperty") + + assert result is True + + +def test_is_property_triple_datatype_property(): + """Test _is_property_triple with owl:DatatypeProperty.""" + query = WOQLQuery() + + result = query._is_property_triple("rdf:type", "owl:DatatypeProperty") + + assert result is True + + +def test_is_property_triple_domain(): + """Test _is_property_triple with rdfs:domain predicate.""" + query = WOQLQuery() + + result = query._is_property_triple("rdfs:domain", "v:Something") + + assert result is True + + +def test_is_property_triple_range(): + """Test _is_property_triple with rdfs:range predicate.""" + query = WOQLQuery() + + result = query._is_property_triple("rdfs:range", "xsd:string") + + assert result is True + + +def test_is_property_triple_false(): + """Test _is_property_triple returns False for non-property triple.""" + query = WOQLQuery() + + result = query._is_property_triple("rdf:type", "owl:Class") + + assert result is False + + +def test_is_property_triple_with_dict_pred(): + """Test _is_property_triple with dict predicate.""" + query = WOQLQuery() + pred = {"@type": "NodeValue", "node": "rdfs:domain"} + + result = query._is_property_triple(pred, "v:X") + + assert result is True + + +def test_is_property_triple_with_dict_obj(): + """Test _is_property_triple with dict object.""" + query = WOQLQuery() + obj = {"@type": "NodeValue", "node": "owl:ObjectProperty"} + + result = query._is_property_triple("rdf:type", obj) + + assert result is True + + +def test_is_property_triple_with_both_dicts(): + """Test _is_property_triple with both dict pred and obj.""" + query = WOQLQuery() + pred = {"@type": "NodeValue", "node": "rdfs:range"} + obj = {"@type": "NodeValue", "node": "xsd:string"} + + result = query._is_property_triple(pred, obj) + + assert result is True + + +def test_is_property_triple_with_dict_no_node(): + """Test _is_property_triple with dict without node.""" + query = WOQLQuery() + pred = {"@type": "NodeValue", "variable": "P"} + obj = {"@type": "NodeValue", "variable": "O"} + + result = query._is_property_triple(pred, obj) + + # Neither has 'node', so extracts None + assert result is False + + +def test_find_last_subject_deeply_nested(): + """Test _find_last_subject with deep nesting.""" + query = WOQLQuery() + triple = {"@type": "Triple", "subject": "v:Deep"} + json_obj = { + "@type": "And", + "and": [ + { + "@type": "Or", + "or": [ + {"@type": "Select", "query": triple} + ] + } + ] + } + + result = query._find_last_subject(json_obj) + + assert result == triple + + +def test_find_last_property_deeply_nested(): + """Test _find_last_property with deep nesting.""" + query = WOQLQuery() + triple = { + "@type": "Triple", + "subject": "v:X", + "predicate": "rdfs:domain", + "object": "v:Y" + } + json_obj = { + "@type": "And", + "and": [ + { + "@type": "Or", + "or": [ + {"@type": "Select", "query": triple} + ] + } + ] + } + + result = query._find_last_property(json_obj) + + assert result == triple diff --git a/terminusdb_client/tests/woql_test_helpers.py b/terminusdb_client/tests/woql_test_helpers.py new file mode 100644 index 00000000..5f996413 --- /dev/null +++ b/terminusdb_client/tests/woql_test_helpers.py @@ -0,0 +1,160 @@ +"""Helper utilities for testing WOQL query generation. + +This module provides standardized mocks and assertion helpers to verify +that WOQLQuery methods generate correct JSON-LD structures. +""" +from unittest.mock import Mock + + +class WOQLTestHelpers: + """Helper class for systematic WOQL testing.""" + + @staticmethod + def create_mock_client(): + """Create a mock client for testing execute() method.""" + client = Mock() + client.query = Mock(return_value={"@type": "api:WoqlResponse"}) + return client + + @staticmethod + def assert_query_type(query, expected_type): + """Assert that query has the expected @type.""" + assert query._query.get("@type") == expected_type, \ + f"Expected @type={expected_type}, got {query._query.get('@type')}" + + @staticmethod + def assert_has_key(query, key): + """Assert that query has a specific key.""" + assert key in query._query, \ + f"Expected key '{key}' in query, got keys: {list(query._query.keys())}" + + @staticmethod + def assert_key_value(query, key, expected_value): + """Assert that query has a key with expected value.""" + actual = query._query.get(key) + assert actual == expected_value, \ + f"Expected {key}={expected_value}, got {actual}" + + @staticmethod + def assert_is_variable(obj, expected_name=None): + """Assert that an object is a variable structure.""" + assert isinstance(obj, dict), f"Expected dict, got {type(obj)}" + assert obj.get("@type") in ["Value", "NodeValue", "DataValue", "ArithmeticValue"], \ + f"Expected variable type, got {obj.get('@type')}" + assert "variable" in obj, f"Expected 'variable' key in {obj}" + if expected_name: + assert obj["variable"] == expected_name, \ + f"Expected variable name '{expected_name}', got '{obj['variable']}'" + + @staticmethod + def assert_is_node(obj, expected_node=None): + """Assert that an object is a node structure.""" + assert isinstance(obj, dict), f"Expected dict, got {type(obj)}" + assert "node" in obj or obj.get("@type") in ["Value", "NodeValue"], \ + f"Expected node structure, got {obj}" + if expected_node and "node" in obj: + assert obj["node"] == expected_node, \ + f"Expected node '{expected_node}', got '{obj['node']}'" + + @staticmethod + def assert_is_data_value(obj, expected_type=None, expected_value=None): + """Assert that an object is a typed data value.""" + assert isinstance(obj, dict), f"Expected dict, got {type(obj)}" + assert "data" in obj, f"Expected 'data' key in {obj}" + data = obj["data"] + assert "@type" in data, f"Expected '@type' in data: {data}" + assert "@value" in data, f"Expected '@value' in data: {data}" + if expected_type: + assert data["@type"] == expected_type, \ + f"Expected type '{expected_type}', got '{data['@type']}'" + if expected_value is not None: + assert data["@value"] == expected_value, \ + f"Expected value '{expected_value}', got '{data['@value']}'" + + @staticmethod + def get_query_dict(query): + """Get the internal query dictionary.""" + return query._query + + @staticmethod + def assert_triple_structure(query, check_subject=True, check_predicate=True, check_object=True): + """Assert that query has proper triple structure.""" + WOQLTestHelpers.assert_query_type(query, "Triple") + if check_subject: + WOQLTestHelpers.assert_has_key(query, "subject") + if check_predicate: + WOQLTestHelpers.assert_has_key(query, "predicate") + if check_object: + WOQLTestHelpers.assert_has_key(query, "object") + + @staticmethod + def assert_quad_structure(query): + """Assert that query has proper quad structure.""" + WOQLTestHelpers.assert_query_type(query, "AddQuad") + WOQLTestHelpers.assert_has_key(query, "subject") + WOQLTestHelpers.assert_has_key(query, "predicate") + WOQLTestHelpers.assert_has_key(query, "object") + WOQLTestHelpers.assert_has_key(query, "graph") + + @staticmethod + def assert_and_structure(query, expected_count=None): + """Assert that query has proper And structure.""" + WOQLTestHelpers.assert_query_type(query, "And") + WOQLTestHelpers.assert_has_key(query, "and") + and_list = query._query["and"] + assert isinstance(and_list, list), f"Expected 'and' to be list, got {type(and_list)}" + if expected_count is not None: + assert len(and_list) == expected_count, \ + f"Expected {expected_count} items in 'and', got {len(and_list)}" + + @staticmethod + def assert_or_structure(query, expected_count=None): + """Assert that query has proper Or structure.""" + WOQLTestHelpers.assert_query_type(query, "Or") + WOQLTestHelpers.assert_has_key(query, "or") + or_list = query._query["or"] + assert isinstance(or_list, list), f"Expected 'or' to be list, got {type(or_list)}" + if expected_count is not None: + assert len(or_list) == expected_count, \ + f"Expected {expected_count} items in 'or', got {len(or_list)}" + + @staticmethod + def assert_select_structure(query): + """Assert that query has proper Select structure.""" + WOQLTestHelpers.assert_query_type(query, "Select") + WOQLTestHelpers.assert_has_key(query, "variables") + WOQLTestHelpers.assert_has_key(query, "query") + + @staticmethod + def assert_not_structure(query): + """Assert that query has proper Not structure.""" + WOQLTestHelpers.assert_query_type(query, "Not") + WOQLTestHelpers.assert_has_key(query, "query") + + @staticmethod + def assert_optional_structure(query): + """Assert that query has proper Optional structure.""" + WOQLTestHelpers.assert_query_type(query, "Optional") + WOQLTestHelpers.assert_has_key(query, "query") + + @staticmethod + def print_query_structure(query, indent=0): + """Print query structure for debugging (helper method).""" + q = query._query + print(" " * indent + f"Query type: {q.get('@type', 'Unknown')}") + for key, value in q.items(): + if key != "@type": + if isinstance(value, dict): + print(" " * indent + f"{key}:") + if "@type" in value: + print(" " * (indent + 2) + f"@type: {value['@type']}") + elif isinstance(value, list): + print(" " * indent + f"{key}: [{len(value)} items]") + else: + print(" " * indent + f"{key}: {value}") + + +# Convenience function for quick access +def helpers(): + """Quick access to WOQLTestHelpers.""" + return WOQLTestHelpers diff --git a/test_detection.py b/test_detection.py new file mode 100644 index 00000000..e69de29b diff --git a/test_random_key.py b/test_random_key.py new file mode 100644 index 00000000..e69de29b