path: root/test/test_dataset/test_dataset.py

# -*- 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)