Compare commits

...

2 Commits

Author SHA1 Message Date
Val Lorentz fb02fb3841 Use requests directly instead SPARQLWrapper
So tests can be closer to the real thing, by mocking the HTTP server
instead of having a completely different abstraction on the client side
(RdflibSparqlBackend)
2023-05-21 19:58:02 +02:00
Val Lorentz 370c75b16c Lint tests 2023-05-21 17:49:30 +02:00
5 changed files with 35 additions and 23 deletions

View File

@ -17,7 +17,7 @@
import abc
from typing import Iterable
import SPARQLWrapper
import requests
class SparqlBackend(abc.ABC):
@ -37,9 +37,12 @@ class RemoteSparqlBackend(SparqlBackend):
:param agent: User-Agent to use in HTTP requests
"""
self._url = url
self._agent = agent
self._session = requests.Session()
self._session.headers["User-Agent"] = agent
def query(self, query: str) -> Iterable[tuple]:
sparql = SPARQLWrapper.SPARQLWrapper(self._url, agent=self._agent)
sparql.setQuery(query)
return sparql.query()
params = {"format": "json", "query": query}
resp = self._session.post(self._url, params=params).json()
variables = resp["head"]["vars"]
for result in resp["results"]["bindings"]:
yield tuple(result.get(variable) for variable in variables)

View File

View File

@ -12,28 +12,33 @@
# You should have received a copy of the GNU Affero General Public License along with
# this program. If not, see <http://www.gnu.org/licenses/>.
from typing import Iterable
"""pytest fixtures"""
# pylint: disable=redefined-outer-name
import pytest
import rdflib
from glowtables.sparql import SparqlBackend
class RdflibSparqlBackend(SparqlBackend):
def __init__(self, graph: rdflib.Graph):
self._graph = graph
def query(self, query: str) -> Iterable[tuple]:
ret = self._graph.query(query)
return ret
from glowtables.sparql import RemoteSparqlBackend
@pytest.fixture()
def rdflib_graph():
def rdflib_graph() -> rdflib.Graph:
"""Returns an empty rdflib graph."""
return rdflib.Graph()
@pytest.fixture()
def rdflib_sparql(rdflib_graph: rdflib.Graph):
return RdflibSparqlBackend(rdflib_graph)
def rdflib_sparql(requests_mock, rdflib_graph: rdflib.Graph) -> RemoteSparqlBackend:
"""Returns a SPARQL backend instance for ``rdflib_graph``."""
def json_callback(request, context):
result = rdflib_graph.query(request.json())
context.status_code = 200
return {
"head": {"vars": result.vars},
"results": list(result),
}
requests_mock.register_uri("POST", "mock://sparql.example.org/", json=json_callback)
return RemoteSparqlBackend("mock://sparql.example.org/", agent="Mock Client")

View File

@ -17,10 +17,11 @@ import textwrap
import pytest
import rdflib
from glowtables.sparql import SparqlBackend
from glowtables.table import Language, LiteralField, Table
def test_single_literal(rdflib_sparql) -> None:
def test_single_literal(rdflib_sparql: SparqlBackend) -> None:
name_field = LiteralField(
"display_name",
{Language("en"): "Name"},
@ -44,7 +45,7 @@ def test_single_literal(rdflib_sparql) -> None:
rdflib_sparql.query(table.sparql())
def test_two_literals(rdflib_sparql) -> None:
def test_two_literals(rdflib_sparql: SparqlBackend) -> None:
name_field = LiteralField(
"display_name",
{Language("en"): "Name"},
@ -74,7 +75,7 @@ def test_two_literals(rdflib_sparql) -> None:
rdflib_sparql.query(table.sparql())
def test_default_value(rdflib_sparql) -> None:
def test_default_value(rdflib_sparql: SparqlBackend) -> None:
name_field = LiteralField(
"display_name",
{Language("en"): "Name"},

View File

@ -9,6 +9,7 @@ requires-python = ">=3.9"
dependencies = [
"flask ~= 2.0.0",
"rdflib ~= 6.0.0",
"requests ~= 3.0.0",
]
[project.optional-dependencies]
@ -29,7 +30,6 @@ python_version = "3.9"
[[tool.mypy.overrides]]
module = [
"requests_mock",
"SPARQLWrapper",
]
ignore_missing_imports = true
@ -54,6 +54,9 @@ disable = [
# flake8 does it already:
"line-too-long",
]
ignore-paths = [
"glowtables/tests/.*_test.py"
]
[tool.pytest.ini_options]
python_files = "*_test.py"