import pytest from os import path from shutil import rmtree from rdflib import Graph, Namespace, URIRef from rdflib.graph import DATASET_DEFAULT_GRAPH_ID as RDFLIB_DEFAULT_GRAPH_URI from rdflib.namespace import RDF, RDFS from lakesuperior.store.ldp_rs.lmdb_store import LmdbStore from lakesuperior.model.graph.graph import Imr @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() # Leave store around for post-test inspection. #store.destroy() @pytest.fixture def bogus_trp(): """ 1000 bogus triples. With 2048-sized pages, even 2-bound indices with 5-byte values such as sp:o should yield more than one page (2048 bytes // 5 = 409 values). """ return { ( URIRef('urn:test_mp:s1'), URIRef(f'urn:test_mp:p{i // 10}'), URIRef(f'urn:test_mp:o{i}')) for i in range(1000) } def _clean(res): return {r[0] for r in res} @pytest.mark.usefixtures('store') class TestStoreInit: ''' Tests for intializing and shutting down store and transactions. ''' def test_open_close(self): ''' Test opening, closing and destroying a store. ''' env_path = '/tmp/test_lmdbstore_init' tmpstore = LmdbStore(env_path) assert tmpstore.is_open tmpstore.close() assert not tmpstore.is_open tmpstore.destroy() assert not path.exists(env_path) assert not path.exists(env_path + '-lock') def test_txn(self, store): ''' Test opening and closing the main transaction. ''' store.begin(True) assert store.is_txn_open store.commit() assert not store.is_txn_open store.begin(True) store.abort() assert not store.is_txn_open def test_ctx_mgr(self, store): ''' Test enclosing a transaction in a context. ''' with store.txn_ctx(): assert store.is_txn_open assert not store.is_txn_rw assert not store.is_txn_open with store.txn_ctx(True): assert store.is_txn_open assert store.is_txn_rw assert not store.is_txn_open assert not store.is_txn_rw try: with store.txn_ctx(): raise RuntimeError() except RuntimeError: assert not store.is_txn_open def test_rollback(self, store): ''' Test rolling back a transaction. ''' try: with store.txn_ctx(True): store.add(( URIRef('urn:nogo:s'), URIRef('urn:nogo:p'), URIRef('urn:nogo:o'))) raise RuntimeError() # This should roll back the transaction. except RuntimeError: pass with store.txn_ctx(): res = set(store.triples((None, None, None))) assert len(res) == 0 @pytest.mark.usefixtures('store') class TestBasicOps: ''' High-level tests for basic operations. ''' def test_create_triple(self, store): ''' Test creation of a single triple. ''' trp = ( URIRef('urn:test:s'), URIRef('urn:test:p'), URIRef('urn:test:o')) with store.txn_ctx(True): store.add(trp) with store.txn_ctx(): res1 = set(store.triples((None, None, None))) res2 = set(store.triples(trp)) assert len(res1) == 1 assert len(res2) == 1 clean_res1 = _clean(res1) clean_res2 = _clean(res2) assert trp in clean_res1 & clean_res2 def test_triple_match_1bound(self, store): ''' Test triple patterns matching one bound term. ''' with store.txn_ctx(): res1 = set(store.triples((URIRef('urn:test:s'), None, None))) res2 = set(store.triples((None, URIRef('urn:test:p'), None))) res3 = set(store.triples((None, None, URIRef('urn:test:o')))) assert _clean(res1) == {( URIRef('urn:test:s'), URIRef('urn:test:p'), URIRef('urn:test:o'))} assert _clean(res2) == _clean(res1) assert _clean(res3) == _clean(res2) def test_triple_match_2bound(self, store): ''' Test triple patterns matching two bound terms. ''' with store.txn_ctx(): res1 = set(store.triples( (URIRef('urn:test:s'), URIRef('urn:test:p'), None))) res2 = set(store.triples( (URIRef('urn:test:s'), None, URIRef('urn:test:o')))) res3 = set(store.triples( (None, URIRef('urn:test:p'), URIRef('urn:test:o')))) assert _clean(res1) == {( URIRef('urn:test:s'), URIRef('urn:test:p'), URIRef('urn:test:o'))} assert _clean(res2) == _clean(res1) assert _clean(res3) == _clean(res2) def test_triple_match_3bound(self, store): ''' Test triple patterns matching 3 bound terms (exact match). ''' with store.txn_ctx(): pattern = ( URIRef('urn:test:s'), URIRef('urn:test:p'), URIRef('urn:test:o')) res1 = set(store.triples(pattern)) assert _clean(res1) == {pattern} def test_triple_no_match_1bound(self, store): ''' Test empty matches with 1 bound term. ''' with store.txn_ctx(True): store.add(( URIRef('urn:test:s'), URIRef('urn:test:p2'), URIRef('urn:test:o2'))) store.add(( URIRef('urn:test:s3'), URIRef('urn:test:p3'), URIRef('urn:test:o3'))) res1 = set(store.triples((None, None, None))) assert len(res1) == 3 res1 = set(store.triples((URIRef('urn:test:s2'), None, None))) res2 = set(store.triples((None, URIRef('urn:test:p4'), None))) res3 = set(store.triples((None, None, URIRef('urn:test:o4')))) assert len(res1) == len(res2) == len(res3) == 0 def test_triple_no_match_2bound(self, store): ''' Test empty matches with 2 bound terms. ''' with store.txn_ctx(True): res1 = set(store.triples( (URIRef('urn:test:s2'), URIRef('urn:test:p'), None))) res2 = set(store.triples( (URIRef('urn:test:s3'), None, URIRef('urn:test:o')))) res3 = set(store.triples( (None, URIRef('urn:test:p3'), URIRef('urn:test:o2')))) assert len(res1) == len(res2) == len(res3) == 0 def test_triple_no_match_3bound(self, store): ''' Test empty matches with 3 bound terms. ''' with store.txn_ctx(True): res1 = set(store.triples(( URIRef('urn:test:s2'), URIRef('urn:test:p3'), URIRef('urn:test:o2')))) assert len(res1) == 0 def test_remove(self, store): ''' Test removing one or more triples. ''' with store.txn_ctx(True): store.remove((URIRef('urn:test:s3'), URIRef('urn:test:p3'), URIRef('urn:test:o3'))) with store.txn_ctx(): res1 = set(store.triples((None, None, None))) assert len(res1) == 2 with store.txn_ctx(True): store.remove((URIRef('urn:test:s'), None, None)) res2 = set(store.triples((None, None, None))) assert len(res2) == 0 @pytest.mark.usefixtures('store', 'bogus_trp') class TestEntryCount: ''' Count number of entries in databases. ''' @pytest.fixture def indices(self): """ List of index labels. """ return [ 's:po', 'p:so', 'o:sp', 'po:s', 'so:p', 'sp:o', 'spo:c', 'c:spo', ] def test_init(self, store, bogus_trp): """ Insert initial data and verify store length. """ with store.txn_ctx(True): for trp in bogus_trp: store.add(trp) with store.txn_ctx(): assert len(store) == 1000 def test_entries_full(self, store, indices): """ Count entries for the full data set. """ with store.txn_ctx(): stat = store.stats() for idxlabel in indices: assert stat['db_stats'][idxlabel]['ms_entries'] == 1000 # 1 subject, 100 predicates, 1000 objects, 1 context assert stat['db_stats']['t:st']['ms_entries'] == 1102 assert stat['db_stats']['th:t']['ms_entries'] == 1102 def test_entries_partial(self, store, indices): """ Delete some triples and verify numbers. """ with store.txn_ctx(True): store.remove((URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p0'), None)) with store.txn_ctx(): stat = store.stats() assert stat['db_stats']['t:st']['ms_entries'] == 1102 assert stat['db_stats']['th:t']['ms_entries'] == 1102 def test_entries_empty(self, store, indices): """ Delete all remainng triples and verify that indices are empty. """ with store.txn_ctx(True): store.remove((None, None, None)) with store.txn_ctx(): stat = store.stats() for idxlabel in indices: assert stat['db_stats'][idxlabel]['ms_entries'] == 0 assert stat['db_stats']['t:st']['ms_entries'] == 1102 assert stat['db_stats']['th:t']['ms_entries'] == 1102 @pytest.mark.usefixtures('store', 'bogus_trp') class TestMultiPageLookup: ''' Tests looking up results retrieved in multiple pages. ''' def test_init(self, store, bogus_trp): """ Insert initial data and verify store length. """ with store.txn_ctx(True): for trp in bogus_trp: store.add(trp) with store.txn_ctx(): assert len(store) == 1000 def test_mp_lookup_0bound(self, store): """ Lookup all triples in a multi-page setup. """ with store.txn_ctx(True): res = store.triples((None, None, None)) res_set = set(res) assert len(res_set) == 1000 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p99'), URIRef('urn:test_mp:o999')) in _clean(res_set) def test_mp_lookup_1bound(self, store): """ Lookup 1-bound triples in a multi-page setup. """ with store.txn_ctx(True): res1 = store.triples((URIRef('urn:test_mp:s1'), None, None)) res_set1 = set(res1) res2 = store.triples((None, URIRef('urn:test_mp:p69'), None)) res_set2 = set(res2) res3 = store.triples((None, None, URIRef('urn:test_mp:o699'))) res_set3 = set(res3) assert len(res_set1) == 1000 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p99'), URIRef('urn:test_mp:o999')) in _clean(res_set1) assert len(res_set2) == 10 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p69'), URIRef('urn:test_mp:o699')) in _clean(res_set2) assert len(res_set3) == 1 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p69'), URIRef('urn:test_mp:o699')) in _clean(res_set3) def test_mp_lookup_2bound(self, store): """ Lookup 2-bound triples in a multi-page setup. """ with store.txn_ctx(True): res1 = store.triples((URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p96'), None)) res_set1 = set(res1) res2 = store.triples((URIRef('urn:test_mp:s1'), None, URIRef('urn:test_mp:o966'))) res_set2 = set(res2) res3 = store.triples((None, URIRef('urn:test_mp:p96'), URIRef('urn:test_mp:o966'))) res_set3 = set(res3) assert len(res_set1) == 10 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p96'), URIRef('urn:test_mp:o966')) in _clean(res_set1) assert len(res_set2) == 1 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p96'), URIRef('urn:test_mp:o966')) in _clean(res_set2) assert len(res_set3) == 1 assert (URIRef('urn:test_mp:s1'), URIRef('urn:test_mp:p96'), URIRef('urn:test_mp:o966')) in _clean(res_set3) @pytest.mark.usefixtures('store') class TestRemoveMulti: ''' Tests for proper removal of multiple combinations of triple and context. ''' @pytest.fixture def data(self): return { 'spo1': ( URIRef('urn:test:s1'), URIRef('urn:test:p1'), URIRef('urn:test:o1')), 'spo2': ( URIRef('urn:test:s1'), URIRef('urn:test:p1'), URIRef('urn:test:o2')), 'spo3': ( URIRef('urn:test:s1'), URIRef('urn:test:p1'), URIRef('urn:test:o3')), 'c1': URIRef('urn:test:c1'), 'c2': URIRef('urn:test:c2'), 'c3': URIRef('urn:test:c3'), } def test_init(self, store, data): """ Initialize the store with test data. """ with store.txn_ctx(True): store.add(data['spo1'], data['c1']) store.add(data['spo2'], data['c1']) store.add(data['spo3'], data['c1']) store.add(data['spo1'], data['c2']) store.add(data['spo2'], data['c2']) store.add(data['spo3'], data['c2']) store.add(data['spo1'], data['c3']) store.add(data['spo2'], data['c3']) store.add(data['spo3'], data['c3']) with store.txn_ctx(): assert len(store) == 9 assert len(set(store.triples((None, None, None)))) == 3 assert len(set(store.triples((None, None, None), data['c1']))) == 3 assert len(set(store.triples((None, None, None), data['c2']))) == 3 assert len(set(store.triples((None, None, None), data['c3']))) == 3 def test_remove_1ctx(self, store, data): """ Test removing all triples from a context. """ with store.txn_ctx(True): store.remove((None, None, None), data['c1']) with store.txn_ctx(): assert len(store) == 6 assert len(set(store.triples((None, None, None)))) == 3 assert len(set(store.triples((None, None, None), data['c1']))) == 0 assert len(set(store.triples((None, None, None), data['c2']))) == 3 assert len(set(store.triples((None, None, None), data['c3']))) == 3 def test_remove_1subj(self, store, data): """ Test removing one subject from all contexts. """ with store.txn_ctx(True): store.remove((data['spo1'][0], None, None)) assert len(store) == 0 @pytest.mark.usefixtures('store') class TestCleanup: ''' Tests for proper cleanup on resource deletion. ''' @pytest.fixture def data(self): return { 'spo1': ( URIRef('urn:test:s1'), URIRef('urn:test:p1'), URIRef('urn:test:o1')), 'spo2': ( URIRef('urn:test:s2'), URIRef('urn:test:p2'), URIRef('urn:test:o2')), 'c1': URIRef('urn:test:c1'), 'c2': URIRef('urn:test:c2'), } def _is_empty(self, store): stats = store.stats()['db_stats'] for dblabel in ('spo:c', 'c:spo', 's:po', 'p:so', 'o:sp',): if stats[dblabel]['ms_entries'] > 0: return False return True def test_cleanup_spo(self, store, data): with store.txn_ctx(True): store.add(data['spo1']) with store.txn_ctx(): assert not self._is_empty(store) with store.txn_ctx(True): store.remove(data['spo1']) with store.txn_ctx(): assert self._is_empty(store) def test_cleanup_spoc1(self, store, data): with store.txn_ctx(True): store.add(data['spo1'], data['c1']) with store.txn_ctx(): assert not self._is_empty(store) with store.txn_ctx(True): store.remove(data['spo1'], data['c1']) with store.txn_ctx(): assert self._is_empty(store) def test_cleanup_spoc2(self, store, data): with store.txn_ctx(True): store.add(data['spo1'], data['c1']) with store.txn_ctx(True): store.remove((None, None, None), data['c1']) with store.txn_ctx(): assert self._is_empty(store) def test_cleanup_spoc3(self, store, data): with store.txn_ctx(True): store.add(data['spo1'], data['c1']) store.add(data['spo2'], data['c1']) with store.txn_ctx(True): store.remove((data['spo1'][0], None, None), data['c1']) with store.txn_ctx(): assert not self._is_empty(store) with store.txn_ctx(True): store.remove((data['spo2'][0], None, None), data['c1']) with store.txn_ctx(): assert self._is_empty(store) def test_cleanup_spoc4(self, store, data): with store.txn_ctx(True): store.add(data['spo1'], data['c1']) store.add(data['spo2'], data['c1']) store.add(data['spo2'], data['c2']) with store.txn_ctx(True): store.remove((None, None, None)) with store.txn_ctx(): assert self._is_empty(store) @pytest.mark.usefixtures('store') class TestBindings: ''' Tests for namespace bindings. ''' @pytest.fixture def bindings(self): return ( ('ns1', Namespace('http://test.org/ns#')), ('ns2', Namespace('http://my_org.net/ns#')), ('ns3', Namespace('urn:test:')), ('ns4', Namespace('info:myinst/graph#')), ) def test_ns(self, store, bindings): ''' Test namespace bindings. ''' with store.txn_ctx(True): for b in bindings: store.bind(*b) nslist = list(store.namespaces()) assert len(nslist) == len(bindings) for i in range(len(bindings)): assert nslist[i] == bindings[i] def test_ns2pfx(self, store, bindings): ''' Test namespace to prefix conversion. ''' with store.txn_ctx(True): for b in bindings: pfx, ns = b assert store.namespace(pfx) == ns def test_pfx2ns(self, store, bindings): ''' Test namespace to prefix conversion. ''' with store.txn_ctx(True): for b in bindings: pfx, ns = b assert store.prefix(ns) == pfx @pytest.mark.usefixtures('store') class TestContext: ''' Tests for context handling. ''' def test_add_empty_graph(self, store): ''' Test creating an empty and a non-empty graph. ''' gr_uri = URIRef('urn:bogus:graph#a') with store.txn_ctx(True): store.add_graph(gr_uri) assert gr_uri in {gr.uri for gr in store.contexts()} def test_add_graph_with_triple(self, store): ''' Test creating an empty and a non-empty graph. ''' trp = (URIRef('urn:test:s123'), URIRef('urn:test:p123'), URIRef('urn:test:o123')) ctx_uri = URIRef('urn:bogus:graph#b') with store.txn_ctx(True): store.add(trp, ctx_uri) with store.txn_ctx(): assert ctx_uri in {gr.uri for gr in store.contexts(trp)} def test_empty_context(self, store): ''' Test creating and deleting empty contexts. ''' gr_uri = URIRef('urn:bogus:empty#a') with store.txn_ctx(True): store.add_graph(gr_uri) assert gr_uri in {gr.uri for gr in store.contexts()} with store.txn_ctx(True): store.remove_graph(gr_uri) assert gr_uri not in {gr.uri for gr in store.contexts()} def test_context_ro_txn(self, store): ''' Test creating a context within a read-only transaction. ''' gr_uri = URIRef('urn:bogus:empty#b') with store.txn_ctx(): store.add_graph(gr_uri) a = 5 #bogus stuff for debugger # The lookup must happen in a separate transaction. The first # transaction opens a separate (non-child) R/W transaction while it is # already isolated so even after the RW txn is committed, the RO one # doesn't know anything about the changes. # If the RW transaction could be nested inside the RO one that would # allow a lookup in the same transaction, but this does not seem to be # possible. with store.txn_ctx(): assert gr_uri in {gr.uri for gr in store.contexts()} with store.txn_ctx(True): store.remove_graph(gr_uri) assert gr_uri not in {gr.uri for gr in store.contexts()} def test_add_trp_to_ctx(self, store): ''' Test adding triples to a graph. ''' gr_uri = URIRef('urn:bogus:graph#a') # From previous test gr2_uri = URIRef('urn:bogus:graph#z') # Never created before trp1 = (URIRef('urn:s:1'), URIRef('urn:p:1'), URIRef('urn:o:1')) trp2 = (URIRef('urn:s:2'), URIRef('urn:p:2'), URIRef('urn:o:2')) trp3 = (URIRef('urn:s:3'), URIRef('urn:p:3'), URIRef('urn:o:3')) trp4 = (URIRef('urn:s:4'), URIRef('urn:p:4'), URIRef('urn:o:4')) with store.txn_ctx(True): store.add(trp1, gr_uri) store.add(trp2, gr_uri) store.add(trp2, gr_uri) # Duplicate; dropped. store.add(trp2, None) # Goes to the default graph. store.add(trp3, gr2_uri) store.add(trp3, gr_uri) store.add(trp4) # Goes to the default graph. # Quick size checks. with store.txn_ctx(): assert len(set(store.triples((None, None, None)))) == 5 assert len(set(store.triples((None, None, None), RDFLIB_DEFAULT_GRAPH_URI))) == 2 assert len(set(store.triples((None, None, None), gr_uri))) == 3 assert len(set(store.triples((None, None, None), gr2_uri))) == 1 assert gr2_uri in {gr.uri for gr in store.contexts()} assert trp1 in _clean(store.triples((None, None, None))) assert trp1 not in _clean(store.triples((None, None, None), RDFLIB_DEFAULT_GRAPH_URI)) assert trp2 in _clean(store.triples((None, None, None), gr_uri)) assert trp2 in _clean(store.triples((None, None, None))) assert trp3 in _clean(store.triples((None, None, None), gr2_uri)) assert trp3 not in _clean(store.triples((None, None, None), RDFLIB_DEFAULT_GRAPH_URI)) # Verify that contexts are in the right place. with store.txn_ctx(): # trp3 is in both graphs. res_no_ctx = store.triples(trp3) res_ctx = store.triples(trp3, gr2_uri) for res in res_no_ctx: assert Imr(uri=gr_uri) in res[1] assert Imr(uri=gr2_uri) in res[1] for res in res_ctx: assert Imr(uri=gr_uri) in res[1] assert Imr(uri=gr2_uri) in res[1] def test_delete_from_ctx(self, store): ''' Delete triples from a named graph and from the default graph. ''' gr_uri = URIRef('urn:bogus:graph#a') gr2_uri = URIRef('urn:bogus:graph#b') with store.txn_ctx(True): store.remove((None, None, None), gr2_uri) assert len(set(store.triples((None, None, None), gr2_uri))) == 0 assert len(set(store.triples((None, None, None), gr_uri))) == 3 with store.txn_ctx(True): store.remove((URIRef('urn:s:1'), None, None)) assert len(set(store.triples( (URIRef('urn:s:1'), None, None), gr_uri))) == 0 assert len(set(store.triples((None, None, None), gr_uri))) == 2 assert len(set(store.triples((None, None, None)))) == 3 # This should result in no change because the graph does not exist. with store.txn_ctx(True): store.remove((None, None, None), URIRef('urn:phony:graph#xyz')) store.remove( (URIRef('urn:s:1'), None, None), URIRef('urn:phony:graph#xyz')) assert len(set(store.triples((None, None, None), gr_uri))) == 2 assert len(set(store.triples((None, None, None)))) == 3 with store.txn_ctx(True): store.remove((URIRef('urn:s:4'), None, None), RDFLIB_DEFAULT_GRAPH_URI) assert len(set(store.triples((None, None, None)))) == 2 with store.txn_ctx(True): store.remove((None, None, None)) assert len(set(store.triples((None, None, None)))) == 0 assert len(set(store.triples((None, None, None), gr_uri))) == 0 assert len(store) == 0 def test_remove_shared_ctx(self, store): """ Remove a context that shares triples with another one. """ trp1 = ( URIRef('urn:bogus:shared_s:1'), URIRef('urn:bogus:shared_p:1'), URIRef('urn:bogus:shared_o:1')) trp2 = ( URIRef('urn:bogus:shared_s:2'), URIRef('urn:bogus:shared_p:2'), URIRef('urn:bogus:shared_o:2')) trp3 = ( URIRef('urn:bogus:shared_s:3'), URIRef('urn:bogus:shared_p:3'), URIRef('urn:bogus:shared_o:3')) ctx1 = URIRef('urn:bogus:shared_graph#a') ctx2 = URIRef('urn:bogus:shared_graph#b') with store.txn_ctx(True): store.add(trp1, ctx1) store.add(trp2, ctx1) store.add(trp2, ctx2) store.add(trp3, ctx2) with store.txn_ctx(True): store.remove_graph(ctx1) with store.txn_ctx(): assert len(set(store.triples(trp1))) == 0 assert len(set(store.triples(trp2))) == 1 assert len(set(store.triples(trp3))) == 1 @pytest.mark.usefixtures('store') class TestTransactions: ''' Tests for transaction handling. ''' # @TODO Test concurrent reads and writes. pass #@pytest.mark.usefixtures('store') #class TestRdflib: # ''' # Test case adapted from # http://rdflib.readthedocs.io/en/stable/univrdfstore.html#interface-test-cases # ''' # # @pytest.fixture # def sample_gr(self): # from rdflib import Graph, plugin # from rdflib.store import Store # store = plugin.get('Lmdb', Store)('/tmp/rdflibtest') # return Graph(store).parse(''' # @prefix rdf: . # @prefix rdfs: . # @prefix : . # {:a :b :c; a :foo} => {:a :d :c} . # _:foo a rdfs:Class . # :a :d :c . # ''', format='n3') # # def test_basic(self, sample_gr): # from rdflib.namespace import RDF # with sample_gr.store.txn_ctx(): # implies = URIRef("http://www.w3.org/2000/10/swap/log#implies") # a = URIRef('http://test/a') # b = URIRef('http://test/b') # c = URIRef('http://test/c') # d = URIRef('http://test/d') # for s,p,o in g.triples((None,implies,None)): # formulaA = s # formulaB = o # # #contexts test # assert len(list(g.contexts()))==3 # # #contexts (with triple) test # assert len(list(g.contexts((a,d,c))))==2 # # #triples test cases # assert type(list(g.triples( # (None,RDF.type,RDFS.Class)))[0][0]) == BNode # assert len(list(g.triples((None,implies,None))))==1 # assert len(list(g.triples((None,RDF.type,None))))==3 # assert len(list(g.triples((None,RDF.type,None),formulaA)))==1 # assert len(list(g.triples((None,None,None),formulaA)))==2 # assert len(list(g.triples((None,None,None),formulaB)))==1 # assert len(list(g.triples((None,None,None))))==5 # assert len(list(g.triples( # (None,URIRef('http://test/d'),None),formulaB)))==1 # assert len(list(g.triples( # (None,URIRef('http://test/d'),None))))==1 # # #Remove test cases # g.remove((None,implies,None)) # assert len(list(g.triples((None,implies,None))))==0 # assert len(list(g.triples((None,None,None),formulaA)))==2 # assert len(list(g.triples((None,None,None),formulaB)))==1 # g.remove((None,b,None),formulaA) # assert len(list(g.triples((None,None,None),formulaA)))==1 # g.remove((None,RDF.type,None),formulaA) # assert len(list(g.triples((None,None,None),formulaA)))==0 # g.remove((None,RDF.type,RDFS.Class)) # # #remove_context tests # formulaBContext=Context(g,formulaB) # g.remove_context(formulaB) # assert len(list(g.triples((None,RDF.type,None))))==2 # assert len(g)==3 # assert len(formulaBContext)==0 # g.remove((None,None,None)) # assert len(g)==0