# -*- coding: utf-8 -*- import os import shutil import tempfile from test.data import context1, likes, pizza, tarek import pytest from rdflib import URIRef, plugin from rdflib.graph import DATASET_DEFAULT_GRAPH_ID, Dataset, Graph, Namespace # Will also run SPARQLUpdateStore tests against local SPARQL1.1 endpoint if # available. This assumes SPARQL1.1 query/update endpoints running locally at # http://localhost:3030/db/ # # Testing SPARQLUpdateStore Dataset behavior needs a different endpoint behavior # than our ConjunctiveGraph tests in test_sparqlupdatestore.py! # # For the tests here to run, you can for example start fuseki with: # ./fuseki-server --mem --update /db # THIS WILL DELETE ALL DATA IN THE /db dataset HOST = "http://localhost:3030" DB = "/db/" pluginstores = [] for s in plugin.plugins(None, plugin.Store): if s.name in ("Memory", "Auditable", "Concurrent", "SPARQLStore"): continue # these are tested by default if not s.getClass().graph_aware: continue if s.name == "SPARQLUpdateStore": from urllib.request import urlopen try: assert len(urlopen(HOST).read()) > 0 except Exception: continue pluginstores.append(s.name) @pytest.fixture( scope="function", params=pluginstores, ) def get_dataset(request): store = request.param try: dataset = Dataset(store=store) except ImportError: pytest.skip("Dependencies for store '%s' not available!" % store) graph = Dataset(store=store) if not graph.store.graph_aware: return if store in ["SQLiteLSM", "LevelDB"]: path = os.path.join(tempfile.gettempdir(), f"test_{store.lower()}") try: shutil.rmtree(path) except Exception: pass elif store == "SPARQLUpdateStore": root = HOST + DB path = root + "sparql", root + "update" else: path = tempfile.mkdtemp() graph.open(path, create=True if store != "SPARQLUpdateStore" else False) if store == "SPARQLUpdateStore": try: graph.store.update("CLEAR ALL") except Exception as e: if "SPARQLStore does not support BNodes! " in str(e): pass else: raise Exception(e) yield store, graph if store == "SPARQLUpdateStore": try: graph.store.update("CLEAR ALL") except Exception as e: if "SPARQLStore does not support BNodes! " in str(e): pass else: raise Exception(e) graph.close() else: graph.close() graph.destroy(path) if os.path.isdir(path): shutil.rmtree(path) else: try: os.remove(path) except: pass def test_graph_aware(get_dataset): store, dataset = get_dataset if not dataset.store.graph_aware: return g1 = dataset.graph(context1) # Some SPARQL endpoint backends (e.g. TDB) do not consider # empty named graphs if store != "SPARQLUpdateStore": # added graph exists assert set(x.identifier for x in dataset.contexts()) == set( [context1, DATASET_DEFAULT_GRAPH_ID] ) # added graph is empty assert len(g1) == 0 g1.add((tarek, likes, pizza)) # added graph still exists assert set(x.identifier for x in dataset.contexts()) == set( [context1, DATASET_DEFAULT_GRAPH_ID] ) # added graph contains one triple assert len(g1) == 1 g1.remove((tarek, likes, pizza)) # added graph is empty assert len(g1) == 0 # Some SPARQL endpoint backends (e.g. TDB) do not consider # empty named graphs if store != "SPARQLUpdateStore": # graph still exists, although empty assert set(x.identifier for x in dataset.contexts()) == set( [context1, DATASET_DEFAULT_GRAPH_ID] ) dataset.remove_graph(context1) # graph is gone assert set(x.identifier for x in dataset.contexts()) == set( [DATASET_DEFAULT_GRAPH_ID] ) def test_default_graph(get_dataset): # Something the default graph is read-only (e.g. TDB in union mode) store, dataset = get_dataset if store == "SPARQLUpdateStore": print( "Please make sure updating the default graph " "is supported by your SPARQL endpoint" ) dataset.add((tarek, likes, pizza)) assert len(dataset) == 1 # only default exists assert list(dataset.contexts()) == [dataset.default_context] # removing default graph removes triples but not actual graph dataset.remove_graph(DATASET_DEFAULT_GRAPH_ID) assert len(dataset) == 0 # default still exists assert set(dataset.contexts()) == set([dataset.default_context]) def test_not_union(get_dataset): store, dataset = get_dataset # Union depends on the SPARQL endpoint configuration if store == "SPARQLUpdateStore": print( "Please make sure your SPARQL endpoint has not configured " "its default graph as the union of the named graphs" ) subgraph1 = dataset.graph(context1) subgraph1.add((tarek, likes, pizza)) assert list(dataset.objects(tarek, None)) == [] assert list(subgraph1.objects(tarek, None)) == [pizza] def test_iter(get_dataset): store, d = get_dataset """PR 1382: adds __iter__ to Dataset""" uri_a = URIRef("https://example.com/a") uri_b = URIRef("https://example.com/b") uri_c = URIRef("https://example.com/c") uri_d = URIRef("https://example.com/d") d.graph(URIRef("https://example.com/subgraph1")) d.add((uri_a, uri_b, uri_c, URIRef("https://example.com/subgraph1"))) d.add( (uri_a, uri_b, uri_c, URIRef("https://example.com/subgraph1")) ) # pointless addition: duplicates above d.graph(URIRef("https://example.com/g2")) d.add((uri_a, uri_b, uri_c, URIRef("https://example.com/g2"))) d.add((uri_a, uri_b, uri_d, URIRef("https://example.com/subgraph1"))) # traditional iterator i_trad = 0 for t in d.quads((None, None, None)): i_trad += 1 # new Dataset.__iter__ iterator i_new = 0 for t in d: i_new += 1 assert i_new == i_trad # both should be 3 EGSCHEMA = Namespace("example:") def test_subgraph_without_identifier() -> None: """ Graphs with no identifies assigned are identified by Skolem IRIs with a prefix that is bound to `genid`. TODO: This is somewhat questionable and arbitrary behaviour and should be reviewed at some point. """ dataset = Dataset() nman = dataset.namespace_manager genid_prefix = URIRef("https://rdflib.github.io/.well-known/genid/rdflib/") namespaces = set(nman.namespaces()) assert ( next((namespace for namespace in namespaces if namespace[0] == "genid"), None) is None ) subgraph: Graph = dataset.graph() subgraph.add((EGSCHEMA["subject"], EGSCHEMA["predicate"], EGSCHEMA["object"])) namespaces = set(nman.namespaces()) assert next( (namespace for namespace in namespaces if namespace[0] == "genid"), None ) == ("genid", genid_prefix) assert f"{subgraph.identifier}".startswith(genid_prefix)