Test cases for RDFLib/rdflib-sqlalchemy#42 and RDFLib/rdflib-sqlalchemy#39

42-and-39-tests.py

import logging
import rdflib
from rdflib.term import URIRef
from rdflib import RDF, plugin
from rdflib.graph import Graph
from rdflib.store import Store
from movementmetadata import MovementMetadata
import tempfile
import os
from os.path import join as p
import shutil

# This is, apparently, necessary -- don't care why
try:
    from rdflib_sqlalchemy import registerplugins
    from sqlalchemy import event
except ImportError:
    raise Exception('The rdflib-sqlalchemy package is not installed.')
registerplugins()


def test_type_add(graph):
    graph.add((URIRef('http://example.org#type-add'), RDF.type, URIRef('http://example.org/cra')))
    graph.add((URIRef('http://example.org#type-add'), RDF.type, URIRef('http://example.org/cra')))


def test_type_addn(graph):
    graph.addN([
        (URIRef('http://example.org#type-addn'), RDF.type, URIRef('http://example.org/cra'), graph),
        (URIRef('http://example.org#type-addn'), RDF.type, URIRef('http://example.org/cra'), graph)
    ])


def test_add(graph):
    graph.add((URIRef('http://example.org#add'), URIRef('http://example.org/blah'), URIRef('http://example.org/cra')))
    graph.add((URIRef('http://example.org#add'), URIRef('http://example.org/blah'), URIRef('http://example.org/cra')))


def test_addn(graph):
    graph.addN([
        (URIRef('http://example.org#addn'), URIRef('http://example.org/blah'), URIRef('http://example.org/cra'), graph),
        (URIRef('http://example.org#addn'), URIRef('http://example.org/blah'), URIRef('http://example.org/cra'), graph)
    ])


def test_namespace_change_prefix_binding(graph):
    graph.namespace_manager.bind('change_binding', URIRef('http://example.org/change-binding-1#'),
            replace=True)
    graph.namespace_manager.bind('change_binding', URIRef('http://example.org/change-binding-2#'),
            replace=True)
    assert ('change_binding',
            URIRef('http://example.org/change-binding-2#')) in list(graph.namespace_manager.namespaces())


def test_namespace_rebind_prefix(graph):
    graph.namespace_manager.bind('rebind', URIRef('http://example.org/rebind#'))
    graph.namespace_manager.bind('rebind', URIRef('http://example.org/rebind#'))


def graph():
    store = plugin.get("SQLAlchemy", Store)(
            identifier=URIRef("http://example.org/rdflib-sqlalchemy-duplicate-insert-test"))
    testdir = tempfile.mkdtemp(prefix=__name__ + '.')
    try:
        os.unlink(p(testdir, 'rdflib-sqlitedb'))
    except OSError:
        pass
    g = Graph(store)
    g.open('sqlite:///' + p(testdir, 'rdflib-sqlitedb'), create=True)
    yield g
    g.close()
    shutil.rmtree(testdir)


if __name__ == '__main__':
    # test runner
    import sys
    import io
    import os
    my_vars = dict(**vars())
    tests = []
    for k, v in my_vars.items():
        if k.startswith('test_'):
            tests.append([k, v])

    for name, test_func in tests:
        old_stdout = sys.stdout
        sys.stdout = io.StringIO()
        old_stderr = sys.stderr
        sys.stderr = io.StringIO()
        gg = graph()
        try:
            g = next(gg)
            test_func(g)
        except Exception as e:
            stderr = sys.stderr.getvalue()
            stdout = sys.stdout.getvalue()
            sys.stderr = old_stderr
            sys.stdout = old_stdout
            print("STDERR")
            print(stderr)
            print("STDOUT")
            print(stdout)
            print('Test "{}" failed'.format(name), e)
        finally:
            next(gg, None)
            sys.stderr = old_stderr
            sys.stdout = old_stdout