Migrate some tests to pytest

Preparatory work for dataset-rework, eases review

2 files changed, 195 insertions, 200 deletions

diff --git a/test/test_dataset/test_dataset.py b/test/test_dataset/test_dataset.py
index c08162de..a50f023a 100644
--- a/test/test_dataset/test_dataset.py
+++ b/test/test_dataset/test_dataset.py
@@ -1,14 +1,14 @@
-import sys
+# -*- coding: utf-8 -*-
 import os
-from typing import Optional
-import unittest
-
-from tempfile import mkdtemp, mkstemp
 import shutil
+import tempfile
+from test.data import context1, likes, pizza, tarek
 
 import pytest
-from rdflib import Dataset, URIRef, plugin
-from rdflib.graph import DATASET_DEFAULT_GRAPH_ID
+
+from rdflib import FOAF, XSD, BNode, Literal, URIRef, logger, plugin
+from rdflib.graph import DATASET_DEFAULT_GRAPH_ID, Dataset, Graph
+from rdflib.plugins.stores.sparqlstore import SPARQLUpdateStore
 
 # Will also run SPARQLUpdateStore tests against local SPARQL1.1 endpoint if
 # available. This assumes SPARQL1.1 query/update endpoints running locally at
@@ -25,215 +25,210 @@ from rdflib.graph import DATASET_DEFAULT_GRAPH_ID
 HOST = "http://localhost:3030"
 DB = "/db/"
 
+pluginstores = []
+
+for s in plugin.plugins(None, plugin.Store):
+    if s.name in ("default", "Memory", "Auditable", "Concurrent", "SPARQLStore"):
+        continue  # these are tested by default
+
+    if not s.getClass().graph_aware:
+        continue
 
-class DatasetTestCase(unittest.TestCase):
-    store = "default"
-    skip_reason = None
-    slow = True
-    tmppath = None
+    if s.name == "SPARQLUpdateStore":
+        from urllib.request import urlopen
 
-    def setUp(self):
-        if self.skip_reason is not None:
-            self.skipTest(skip_reason)
         try:
-            self.graph = Dataset(store=self.store)
-        except ImportError:
-            pytest.skip("Dependencies for store '%s' not available!" % self.store)
-        if self.store == "SQLite":
-            _, self.tmppath = mkstemp(prefix="test", dir="/tmp", suffix=".sqlite")
-        elif self.store == "SPARQLUpdateStore":
-            root = HOST + DB
-            self.graph.open((root + "sparql", root + "update"))
-        else:
-            self.tmppath = mkdtemp()
-
-        if self.store != "SPARQLUpdateStore":
-            self.graph.open(self.tmppath, create=True)
-        self.michel = URIRef("urn:example:michel")
-        self.tarek = URIRef("urn:example:tarek")
-        self.bob = URIRef("urn:example:bob")
-        self.likes = URIRef("urn:example:likes")
-        self.hates = URIRef("urn:example:hates")
-        self.pizza = URIRef("urn:example:pizza")
-        self.cheese = URIRef("urn:cheese")
-
-        # Use regular URIs because SPARQL endpoints like Fuseki alter short names
-        self.c1 = URIRef("urn:example:context-1")
-        self.c2 = URIRef("urn:example:context-2")
-
-        # delete the graph for each test!
-        self.graph.remove((None, None, None))
-        for c in self.graph.contexts():
-            c.remove((None, None, None))
-            assert len(c) == 0
-            self.graph.remove_graph(c)
-
-    def tearDown(self):
-        self.graph.close()
-        if self.store == "SPARQLUpdateStore":
+            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
-        else:
-            if os.path.isdir(self.tmppath):
-                shutil.rmtree(self.tmppath)
+    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:
-                os.remove(self.tmppath)
+                raise Exception(e)
+
+    yield store, graph
 
-    def testGraphAware(self):
+    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
 
-        if not self.graph.store.graph_aware:
-            return
 
-        g = self.graph
-        g1 = g.graph(self.c1)
+def test_graph_aware(get_dataset):
 
-        # Some SPARQL endpoint backends (e.g. TDB) do not consider
-        # empty named graphs
-        if self.store != "SPARQLUpdateStore":
-            # added graph exists
-            self.assertEqual(
-                set(x.identifier for x in self.graph.contexts()),
-                set([self.c1, DATASET_DEFAULT_GRAPH_ID]),
-            )
+    store, dataset = get_dataset
 
-        # added graph is empty
-        self.assertEqual(len(g1), 0)
+    if not dataset.store.graph_aware:
+        return
 
-        g1.add((self.tarek, self.likes, self.pizza))
+    g1 = dataset.graph(context1)
 
-        # added graph still exists
-        self.assertEqual(
-            set(x.identifier for x in self.graph.contexts()),
-            set([self.c1, DATASET_DEFAULT_GRAPH_ID]),
+    # 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 contains one triple
-        self.assertEqual(len(g1), 1)
+    # added graph is empty
+    assert len(g1) == 0
 
-        g1.remove((self.tarek, self.likes, self.pizza))
+    g1.add((tarek, likes, pizza))
 
-        # added graph is empty
-        self.assertEqual(len(g1), 0)
+    # 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
 
-        # Some SPARQL endpoint backends (e.g. TDB) do not consider
-        # empty named graphs
-        if self.store != "SPARQLUpdateStore":
-            # graph still exists, although empty
-            self.assertEqual(
-                set(x.identifier for x in self.graph.contexts()),
-                set([self.c1, DATASET_DEFAULT_GRAPH_ID]),
-            )
+    g1.remove((tarek, likes, pizza))
 
-        g.remove_graph(self.c1)
+    # added graph is empty
+    assert len(g1) == 0
 
-        # graph is gone
-        self.assertEqual(
-            set(x.identifier for x in self.graph.contexts()),
-            set([DATASET_DEFAULT_GRAPH_ID]),
+    # 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]
         )
 
-    def testDefaultGraph(self):
-        # Something the default graph is read-only (e.g. TDB in union mode)
-        if self.store == "SPARQLUpdateStore":
-            print(
-                "Please make sure updating the default graph "
-                "is supported by your SPARQL endpoint"
-            )
-
-        self.graph.add((self.tarek, self.likes, self.pizza))
-        self.assertEqual(len(self.graph), 1)
-        # only default exists
-        self.assertEqual(
-            set(x.identifier for x in self.graph.contexts()),
-            set([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"
         )
 
-        # removing default graph removes triples but not actual graph
-        self.graph.remove_graph(DATASET_DEFAULT_GRAPH_ID)
+    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])
 
-        self.assertEqual(len(self.graph), 0)
-        # default still exists
-        self.assertEqual(
-            set(x.identifier for x in self.graph.contexts()),
-            set([DATASET_DEFAULT_GRAPH_ID]),
+
+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"
         )
 
-    def testNotUnion(self):
-        # Union depends on the SPARQL endpoint configuration
-        if self.store == "SPARQLUpdateStore":
-            print(
-                "Please make sure your SPARQL endpoint has not configured "
-                "its default graph as the union of the named graphs"
-            )
-        g1 = self.graph.graph(self.c1)
-        g1.add((self.tarek, self.likes, self.pizza))
-
-        self.assertEqual(list(self.graph.objects(self.tarek, None)), [])
-        self.assertEqual(list(g1.objects(self.tarek, None)), [self.pizza])
-
-    def testIter(self):
-        """PR 1382: adds __iter__ to Dataset"""
-        d = 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.add_graph(URIRef("https://example.com/g1"))
-        d.add((uri_a, uri_b, uri_c, URIRef("https://example.com/g1")))
-        d.add(
-            (uri_a, uri_b, uri_c, URIRef("https://example.com/g1"))
-        )  # pointless addition: duplicates above
-
-        d.add_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/g1")))  # new, uri_d
-
-        # 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
-
-        self.assertEqual(i_new, i_trad)  # both should be 3
-
-
-# dynamically create classes for each registered Store
-
-pluginname = None
-if __name__ == "__main__":
-    if len(sys.argv) > 1:
-        pluginname = sys.argv[1]
-
-tests = 0
-
-for s in plugin.plugins(pluginname, plugin.Store):
-    skip_reason: Optional[str] = None
-    if s.name in ("default", "Memory", "Auditable", "Concurrent", "SPARQLStore"):
-        continue  # these are tested by default
+    subgraph1 = dataset.graph(context1)
+    subgraph1.add((tarek, likes, pizza))
 
-    if not s.getClass().graph_aware:
-        continue
+    assert list(dataset.objects(tarek, None)) == []
+    assert list(subgraph1.objects(tarek, None)) == [pizza]
 
-    if s.name == "SPARQLUpdateStore":
-        from urllib.request import urlopen
 
-        try:
-            assert len(urlopen(HOST).read()) > 0
-        except BaseException:
-            skip_reason = "No SPARQL endpoint for %s (tests skipped)\n" % s.name
-            sys.stderr.write(skip_reason)
-
-    locals()["t%d" % tests] = type(
-        "%sContextTestCase" % s.name,
-        (DatasetTestCase,),
-        {"store": s.name, "skip_reason": skip_reason},
-    )
-    tests += 1
+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
 
-if __name__ == "__main__":
-    unittest.main()
+    assert i_new == i_trad  # both should be 3
diff --git a/test/test_dataset/test_dataset_generators.py b/test/test_dataset/test_dataset_generators.py
index c6951336..03416ff8 100644
--- a/test/test_dataset/test_dataset_generators.py
+++ b/test/test_dataset/test_dataset_generators.py
@@ -1,18 +1,18 @@
 import os
-from rdflib import Dataset, URIRef
-
-timblcardn3 = open(
-    os.path.join(os.path.dirname(__file__), "..", "consistent_test_data", "timbl-card.n3")
-).read()
+from test.data import (
+    CONSISTENT_DATA_DIR,
+    bob,
+    cheese,
+    hates,
+    likes,
+    michel,
+    pizza,
+    tarek,
+)
 
+from rdflib import Dataset, URIRef
 
-michel = URIRef("urn:example:michel")
-tarek = URIRef("urn:example:tarek")
-bob = URIRef("urn:example:bob")
-likes = URIRef("urn:example:likes")
-hates = URIRef("urn:example:hates")
-pizza = URIRef("urn:example:pizza")
-cheese = URIRef("urn:example:cheese")
+timblcardn3 = open(os.path.join(CONSISTENT_DATA_DIR, "timbl-card.n3")).read()
 
 
 def add_stuff(graph):