test: implement tests for construct_ungrouped_pagionation_query

tests/tests_adapter/test_adapter_ungrouped_pagionation.py

@@ -0,0 +1,60 @@
+"""Basic tests for rdfproxy.SPARQLModelAdapter pagination with ungrouped models."""
+
+from collections.abc import Iterator
+from itertools import permutations
+from string import Template
+
+import pytest
+
+from pydantic import BaseModel
+from rdfproxy import Page, SPARQLModelAdapter
+
+
+def _generate_queries() -> Iterator[str]:
+    _values = [
+        "('z' 'a' 'foo')",
+        "('y' 'b' UNDEF)",
+        "('x' UNDEF UNDEF)",
+    ]
+
+    values_permutations = permutations(_values, r=3)
+    query_template = Template(
+        "select ?parent ?child ?name where { values (?parent ?child ?name) { $values }}"
+    )
+
+    for values in values_permutations:
+        values = " ".join(values)
+        yield query_template.substitute(values=values)
+
+
+class Model(BaseModel):
+    parent: str
+    child: str | None = None
+    name: str | None = None
+
+
+@pytest.mark.remote
+@pytest.mark.parametrize("query", _generate_queries())
+def test_ungrouped_pagination(query):
+    expected = Page[Model](
+        items=[
+            {"parent": "x", "child": None, "name": None},
+            {"parent": "y", "child": "b", "name": None},
+            {"parent": "z", "child": "a", "name": "foo"},
+        ],
+        page=1,
+        size=100,
+        total=3,
+        pages=1,
+    )
+
+    print("INFO: ", query)
+    print()
+
+    adapter = SPARQLModelAdapter(
+        target="https://graphdb.r11.eu/repositories/RELEVEN",
+        query=query,
+        model=Model,
+    )
+
+    assert adapter.query(page=1, size=100) == expected

tests/unit/test_construct_ungrouped_pagination_query.py

@@ -0,0 +1,55 @@
+"""Unit tests for rdfproxy.utils.sparql_utils.construct_ungrouped_pagination_query."""
+
+from typing import NamedTuple
+
+import pytest
+
+from rdfproxy.utils.sparql_utils import construct_ungrouped_pagination_query
+
+
+class UngroupedPaginationQueryParameter(NamedTuple):
+    query: str
+    expected: str
+    order_by: str | None = None
+
+
+ungrouped_pagination_query_parameters = [
+    UngroupedPaginationQueryParameter(
+        query="select ?s ?p ?o where {?s ?p ?o .}",
+        expected="select ?s ?p ?o where {?s ?p ?o .} order by ?s limit 1 offset 2",
+    ),
+    UngroupedPaginationQueryParameter(
+        query="select ?s ?p ?o where {?s ?p ?o .}",
+        expected="select ?s ?p ?o where {?s ?p ?o .} order by ?test limit 1 offset 2",
+        order_by="test",
+    ),
+    UngroupedPaginationQueryParameter(
+        query="select * where {?s ?p ?o .}",
+        expected="select * where {?s ?p ?o .} order by ?s limit 1 offset 2",
+    ),
+    UngroupedPaginationQueryParameter(
+        query="select * where {?s ?p ?o .}",
+        expected="select * where {?s ?p ?o .} order by ?test limit 1 offset 2",
+        order_by="test",
+    ),
+    UngroupedPaginationQueryParameter(
+        query="select ?s ?p ?o where {values (?s ?p ?o) {(1 2 3)}}",
+        expected="select ?s ?p ?o where {values (?s ?p ?o) {(1 2 3)}} order by ?test limit 1 offset 2",
+        order_by="test",
+    ),
+    UngroupedPaginationQueryParameter(
+        query="select * where {values (?s ?p ?o) {(1 2 3)}}",
+        expected="select * where {values (?s ?p ?o) {(1 2 3)}} order by ?test limit 1 offset 2",
+        order_by="test",
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    ["query", "expected", "order_by"], ungrouped_pagination_query_parameters
+)
+def test_basic_construct_ungrouped_pagination_query_default(query, expected, order_by):
+    constructed_query = construct_ungrouped_pagination_query(
+        query, 1, 2, order_by=order_by
+    )
+    assert constructed_query == expected

tests/unit/test_get_query_projection.py

@@ -0,0 +1,110 @@
+"""Unit tests for rdfproxy.utils.sparql_utils.get_query_projection."""
+
+from typing import NamedTuple
+
+import pytest
+
+from rdfproxy.utils.sparql_utils import get_query_projection
+
+
+class QueryProjectionParameter(NamedTuple):
+    query: str
+    exptected_projection: list[str]
+
+
+queries = []
+
+
+query_projection_parameters = [
+    QueryProjectionParameter(
+        query="select ?s ?p ?o where {?s ?p ?o .}",
+        exptected_projection=["s", "p", "o"],
+    ),
+    QueryProjectionParameter(
+        query="select ?s ?o where {?s ?p ?o .}",
+        exptected_projection=["s", "o"],
+    ),
+    QueryProjectionParameter(
+        query="select * where {?s ?p ?o .}",
+        exptected_projection=[
+            "s",
+            "p",
+            "o",
+        ],
+    ),
+    QueryProjectionParameter(
+        query="select * where {?s ?p ?o ; <urn:predicate> ?o2 .}",
+        exptected_projection=[
+            "s",
+            "p",
+            "o",
+            "o2",
+        ],
+    ),
+    QueryProjectionParameter(
+        query="select * where {?s1 <urn:predicate> ?o1 ; ?p ?o .}",
+        exptected_projection=[
+            "s1",
+            "o1",
+            "p",
+            "o",
+        ],
+    ),
+    QueryProjectionParameter(
+        query="select ?o ?s ?p where {?s ?p ?o ; <urn:predicate> ?o2 .}",
+        exptected_projection=[
+            "o",
+            "s",
+            "p",
+        ],
+    ),
+    QueryProjectionParameter(
+        query="select ?p ?o ?s where {?s ?p ?o .}",
+        exptected_projection=[
+            "p",
+            "o",
+            "s",
+        ],
+    ),
+    QueryProjectionParameter(
+        query="select ?s where {?s <urn:prediate> <urn:object> .}",
+        exptected_projection=["s"],
+    ),
+    QueryProjectionParameter(
+        query="select ?p where {<urn:subject> ?p <urn:object> .}",
+        exptected_projection=["p"],
+    ),
+    QueryProjectionParameter(
+        query="select ?o where {<urn:subject> <urn:predicate> ?o .}",
+        exptected_projection=["o"],
+    ),
+    QueryProjectionParameter(
+        query="select ?s where {?s ?p ?o . {?s2 ?p2 ?o2 .}}",
+        exptected_projection=["s"],
+    ),
+    QueryProjectionParameter(
+        query="select ?s2 where {?s ?p ?o . {?s2 ?p2 ?o2 .}}",
+        exptected_projection=["s2"],
+    ),
+    QueryProjectionParameter(
+        query="select ?s ?p ?o where { values (?s ?p ?o) { (1 2 3) } }",
+        exptected_projection=[
+            "s",
+            "p",
+            "o",
+        ],
+    ),
+    QueryProjectionParameter(
+        query="select * where { values (?s ?p ?o) { (1 2 3) } }",
+        exptected_projection=[
+            "s",
+            "p",
+            "o",
+        ],
+    ),
+]
+
+
+@pytest.mark.parametrize(["query", "expected_projection"], query_projection_parameters)
+def test_basic_get_query_projection(query, expected_projection):
+    assert list(map(str, get_query_projection(query))) == expected_projection