lakesuperior_legacy/tests/0_data_structures/test_graph.py

import pdb
import pytest

from shutil import rmtree

from rdflib import Graph, Namespace, URIRef

from lakesuperior.model.rdf.graph import Graph
from lakesuperior.store.ldp_rs.lmdb_store import LmdbStore


@pytest.fixture(scope='class')
def store():
    """
    Test LMDB store.

    This store has a different life cycle than the one used for tests in higher
    levels of the stack and is not bootstrapped (i.e. starts completely empty).
    """
    env_path = '/tmp/test_lmdbstore'
    # Remove previous test DBs
    rmtree(env_path, ignore_errors=True)
    store = LmdbStore(env_path)
    yield store
    store.close()
    store.destroy()


@pytest.fixture(scope='class')
def trp():
    return (
        (URIRef('urn:s:0'), URIRef('urn:p:0'), URIRef('urn:o:0')),
        # Exact same as [0].
        (URIRef('urn:s:0'), URIRef('urn:p:0'), URIRef('urn:o:0')),
        # NOTE: s and o are in reversed order.
        (URIRef('urn:o:0'), URIRef('urn:p:0'), URIRef('urn:s:0')),
        (URIRef('urn:s:0'), URIRef('urn:p:1'), URIRef('urn:o:0')),
        (URIRef('urn:s:0'), URIRef('urn:p:1'), URIRef('urn:o:1')),
        (URIRef('urn:s:1'), URIRef('urn:p:1'), URIRef('urn:o:1')),
        (URIRef('urn:s:1'), URIRef('urn:p:2'), URIRef('urn:o:2')),
    )

@pytest.mark.usefixtures('trp')
@pytest.mark.usefixtures('store')
class TestGraphInit:
    """
    Test initialization of graphs with different base data sets.
    """
    def test_empty(self, store):
        """
        Test creation of an empty graph.
        """
        with store.txn_ctx():
            gr = Graph(store)

        # len() should not need a DB transaction open.
        assert len(gr) == 0


    def test_init_triples(self, trp, store):
        """
        Test creation using a Python set.
        """
        with store.txn_ctx(True):
            pdb.set_trace()
            gr = Graph(store, data=set(trp))

            assert len(gr) == 6

            for t in trp:
                assert t in gr


@pytest.mark.usefixtures('trp')
class TestGraphLookup:
    """
    Test triple lookup.

    TODO
    """

    @pytest.mark.skip(reason='TODO')
    def test_lookup_pattern(self, trp):
        """
        Test lookup by basic pattern.
        """
        pass


@pytest.mark.usefixtures('trp')
class TestGraphOps:
    """
    Test various graph operations.
    """
    def test_len(self, trp):
        """
        Test the length of a graph with and without duplicates.
        """
        gr = Graph()
        assert len(gr) == 0

        gr.add((trp[0],))
        assert len(gr) == 1

        gr.add((trp[1],)) # Same values
        assert len(gr) == 1

        gr.add((trp[2],))
        assert len(gr) == 2

        gr.add(trp)
        assert len(gr) == 6


    def test_dup(self, trp):
        """
        Test operations with duplicate triples.
        """
        gr = Graph()
        #import pdb; pdb.set_trace()

        gr.add((trp[0],))
        assert trp[1] in gr
        assert trp[2] not in gr


    def test_remove(self, trp):
        """
        Test adding and removing triples.
        """
        gr = Graph()

        gr.add(trp)
        gr.remove(trp[0])
        assert len(gr) == 5
        assert trp[0] not in gr
        assert trp[1] not in gr

        # This is the duplicate triple.
        gr.remove(trp[1])
        assert len(gr) == 5

        # This is the triple in reverse order.
        gr.remove(trp[2])
        assert len(gr) == 4

        gr.remove(trp[4])
        assert len(gr) == 3


    def test_union(self, trp):
        """
        Test graph union.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr3 = gr1 | gr2

        assert len(gr3) == 5
        assert trp[0] in gr3
        assert trp[4] in gr3


    def test_ip_union(self, trp):
        """
        Test graph in-place union.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr1 |= gr2

        assert len(gr1) == 5
        assert trp[0] in gr1
        assert trp[4] in gr1


    def test_addition(self, trp):
        """
        Test graph addition.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr3 = gr1 + gr2

        assert len(gr3) == 5
        assert trp[0] in gr3
        assert trp[4] in gr3


    def test_ip_addition(self, trp):
        """
        Test graph in-place addition.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr1 += gr2

        assert len(gr1) == 5
        assert trp[0] in gr1
        assert trp[4] in gr1


    def test_subtraction(self, trp):
        """
        Test graph addition.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr3 = gr1 - gr2

        assert len(gr3) == 1
        assert trp[0] in gr3
        assert trp[1] in gr3
        assert trp[2] not in gr3
        assert trp[3] not in gr3
        assert trp[4] not in gr3

        gr3 = gr2 - gr1

        assert len(gr3) == 2
        assert trp[0] not in gr3
        assert trp[1] not in gr3
        assert trp[2] not in gr3
        assert trp[3] not in gr3
        assert trp[4] in gr3
        assert trp[5] in gr3


    def test_ip_subtraction(self, trp):
        """
        Test graph in-place addition.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr1 -= gr2

        assert len(gr1) == 1
        assert trp[0] in gr1
        assert trp[1] in gr1
        assert trp[2] not in gr1
        assert trp[3] not in gr1
        assert trp[4] not in gr1



    def test_intersect(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr3 = gr1 & gr2

        assert len(gr3) == 2
        assert trp[2] in gr3
        assert trp[3] in gr3
        assert trp[0] not in gr3
        assert trp[5] not in gr3


    def test_ip_intersect(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr1 &= gr2

        assert len(gr1) == 2
        assert trp[2] in gr1
        assert trp[3] in gr1
        assert trp[0] not in gr1
        assert trp[5] not in gr1


    def test_xor(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr3 = gr1 ^ gr2

        assert len(gr3) == 3
        assert trp[2] not in gr3
        assert trp[3] not in gr3
        assert trp[0] in gr3
        assert trp[5] in gr3


    def test_ip_xor(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph()
        gr2 = Graph()

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr1 ^= gr2

        assert len(gr1) == 3
        assert trp[2] not in gr1
        assert trp[3] not in gr1
        assert trp[0] in gr1
        assert trp[5] in gr1



@pytest.mark.usefixtures('trp')
class TestNamedGraphOps:
    """
    Test various operations on a named graph.
    """
    def test_len(self, trp):
        """
        Test the length of a graph with and without duplicates.
        """
        imr = Graph(uri='http://example.edu/imr01')
        assert len(imr) == 0

        imr.add((trp[0],))
        assert len(imr) == 1

        imr.add((trp[1],)) # Same values
        assert len(imr) == 1

        imr.add((trp[2],))
        assert len(imr) == 2

        imr.add(trp)
        assert len(imr) == 6


    def test_dup(self, trp):
        """
        Test operations with duplicate triples.
        """
        imr = Graph(uri='http://example.edu/imr01')
        #import pdb; pdb.set_trace()

        imr.add((trp[0],))
        assert trp[1] in imr
        assert trp[2] not in imr


    def test_remove(self, trp):
        """
        Test adding and removing triples.
        """
        imr = Graph(uri='http://example.edu/imr01')

        imr.add(trp)
        imr.remove(trp[0])
        assert len(imr) == 5
        assert trp[0] not in imr
        assert trp[1] not in imr

        # This is the duplicate triple.
        imr.remove(trp[1])
        assert len(imr) == 5

        # This is the triple in reverse order.
        imr.remove(trp[2])
        assert len(imr) == 4

        imr.remove(trp[4])
        assert len(imr) == 3


    def test_union(self, trp):
        """
        Test graph union.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr3 = gr1 | gr2

        assert len(gr3) == 5
        assert trp[0] in gr3
        assert trp[4] in gr3

        assert gr3.uri == URIRef('http://example.edu/imr01')


    def test_ip_union(self, trp):
        """
        Test graph in-place union.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr1 |= gr2

        assert len(gr1) == 5
        assert trp[0] in gr1
        assert trp[4] in gr1

        assert gr1.uri == URIRef('http://example.edu/imr01')


    def test_addition(self, trp):
        """
        Test graph addition.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr3 = gr1 + gr2

        assert len(gr3) == 5
        assert trp[0] in gr3
        assert trp[4] in gr3

        assert gr3.uri == URIRef('http://example.edu/imr01')


    def test_ip_addition(self, trp):
        """
        Test graph in-place addition.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr1 += gr2

        assert len(gr1) == 5
        assert trp[0] in gr1
        assert trp[4] in gr1

        assert gr1.uri == URIRef('http://example.edu/imr01')


    def test_subtraction(self, trp):
        """
        Test graph addition.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr3 = gr1 - gr2

        assert len(gr3) == 1
        assert trp[0] in gr3
        assert trp[1] in gr3
        assert trp[2] not in gr3
        assert trp[3] not in gr3
        assert trp[4] not in gr3

        assert gr3.uri == URIRef('http://example.edu/imr01')

        gr3 = gr2 - gr1

        assert len(gr3) == 2
        assert trp[0] not in gr3
        assert trp[1] not in gr3
        assert trp[2] not in gr3
        assert trp[3] not in gr3
        assert trp[4] in gr3
        assert trp[5] in gr3

        assert gr3.uri == URIRef('http://example.edu/imr02')


    def test_ip_subtraction(self, trp):
        """
        Test graph in-place addition.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr1 -= gr2

        assert len(gr1) == 1
        assert trp[0] in gr1
        assert trp[1] in gr1
        assert trp[2] not in gr1
        assert trp[3] not in gr1
        assert trp[4] not in gr1

        assert gr1.uri == URIRef('http://example.edu/imr01')



    def test_intersect(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr3 = gr1 & gr2

        assert len(gr3) == 2
        assert trp[2] in gr3
        assert trp[3] in gr3
        assert trp[0] not in gr3
        assert trp[5] not in gr3

        assert gr3.uri == URIRef('http://example.edu/imr01')


    def test_ip_intersect(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr1 &= gr2

        assert len(gr1) == 2
        assert trp[2] in gr1
        assert trp[3] in gr1
        assert trp[0] not in gr1
        assert trp[5] not in gr1

        assert gr1.uri == URIRef('http://example.edu/imr01')


    def test_xor(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr3 = gr1 ^ gr2

        assert len(gr3) == 3
        assert trp[2] not in gr3
        assert trp[3] not in gr3
        assert trp[0] in gr3
        assert trp[5] in gr3

        assert gr3.uri == URIRef('http://example.edu/imr01')


    def test_ip_xor(self, trp):
        """
        Test graph intersextion.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph(uri='http://example.edu/imr02')

        gr1.add(trp[0:4])
        gr2.add(trp[2:6])

        gr1 ^= gr2

        assert len(gr1) == 3
        assert trp[2] not in gr1
        assert trp[3] not in gr1
        assert trp[0] in gr1
        assert trp[5] in gr1

        assert gr1.uri == URIRef('http://example.edu/imr01')


@pytest.mark.usefixtures('trp')
class TestHybridOps:
    """
    Test operations between IMR and graph.
    """


    def test_union(self, trp):
        """
        Test hybrid IMR + graph union.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph()

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr3 = gr1 | gr2

        assert len(gr3) == 5
        assert trp[0] in gr3
        assert trp[4] in gr3

        assert isinstance(gr3, Graph)
        assert gr3.uri == URIRef('http://example.edu/imr01')

        gr4 = gr2 | gr1

        assert isinstance(gr4, Graph)

        assert gr3 == gr4


    def test_ip_union_imr(self, trp):
        """
        Test IMR + graph in-place union.
        """
        gr1 = Graph(uri='http://example.edu/imr01')
        gr2 = Graph()

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr1 |= gr2

        assert len(gr1) == 5
        assert trp[0] in gr1
        assert trp[4] in gr1

        assert gr1.uri == URIRef('http://example.edu/imr01')


    def test_ip_union_gr(self, trp):
        """
        Test graph + IMR in-place union.
        """
        gr1 = Graph()
        gr2 = Graph(uri='http://example.edu/imr01')

        gr1.add(trp[0:3])
        gr2.add(trp[2:6])

        gr1 |= gr2

        assert len(gr1) == 5
        assert trp[0] in gr1
        assert trp[4] in gr1

        assert isinstance(gr1, Graph)