""" Part of the bsfs test suite. A copy of the license is provided with the project. Author: Matthias Baumgartner, 2022 """ # imports import rdflib import unittest # bsie imports from bsfs import schema as _schema from bsfs.namespace import ns from bsfs.triple_store.sparql import SparqlStore from bsfs.utils import errors, URI # objects to test from bsfs.graph.nodes import Nodes ## code ## class TestNodes(unittest.TestCase): def setUp(self): # initialize backend self.backend = SparqlStore() self.backend.schema = _schema.Schema.from_string(''' prefix rdfs: prefix xsd: prefix bsfs: prefix bsm: prefix bse: prefix bst: bsfs:Entity rdfs:subClassOf bsfs:Node . bsfs:Tag rdfs:subClassOf bsfs:Node . bsfs:User rdfs:subClassOf bsfs:Node . xsd:string rdfs:subClassOf bsfs:Literal . xsd:integer rdfs:subClassOf bsfs:Literal . # predicates mandated by Nodes bsm:t_created rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Node ; rdfs:range xsd:integer ; bsfs:unique "true"^^xsd:boolean . # additionally defined predicates bse:comment rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:string ; bsfs:unique "false"^^xsd:boolean . bse:filesize rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:integer ; bsfs:unique "true"^^xsd:boolean . bse:tag rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range bsfs:Tag ; bsfs:unique "false"^^xsd:boolean . bse:author rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range bsfs:User ; bsfs:unique "true"^^xsd:boolean . bst:representative rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Tag ; rdfs:range bsfs:Entity ; bsfs:unique "true"^^xsd:boolean . ''') # Nodes constructor args self.user = URI('http://example.com/me') # set args self.tag_type = self.backend.schema.node(ns.bsfs.Tag) self.ent_type = self.backend.schema.node(ns.bsfs.Entity) self.user_type = self.backend.schema.node(ns.bsfs.User) self.p_filesize = self.backend.schema.predicate(ns.bse.filesize) self.p_author = self.backend.schema.predicate(ns.bse.author) self.p_tag = self.backend.schema.predicate(ns.bse.tag) self.p_representative = self.backend.schema.predicate(URI('http://bsfs.ai/schema/Tag#representative')) self.t_created = self.backend.schema.predicate(ns.bsm.t_created) self.ent_ids = { URI('http://example.com/me/entity#1234'), URI('http://example.com/me/entity#4321'), } self.tag_ids = { URI('http://example.com/me/tag#1234'), URI('http://example.com/me/tag#4321'), } def test_str(self): # str baseline nodes = Nodes(self.backend, self.user, self.ent_type, self.ent_ids) self.assertEqual(str(nodes), f'Nodes({self.ent_type}, {self.ent_ids})') self.assertEqual(repr(nodes), f'Nodes({self.backend}, {self.user}, {self.ent_type}, {self.ent_ids})') # str respects node_type nodes = Nodes(self.backend, self.user, self.tag_type, self.tag_ids) self.assertEqual(str(nodes), f'Nodes({self.tag_type}, {self.tag_ids})') self.assertEqual(repr(nodes), f'Nodes({self.backend}, {self.user}, {self.tag_type}, {self.tag_ids})') # str respects guids nodes = Nodes(self.backend, self.user, self.ent_type, {URI('http://example.com/me/entity#foo')}) self.assertEqual(str(nodes), f'Nodes({self.ent_type}, {{\'http://example.com/me/entity#foo\'}})') self.assertEqual(repr(nodes), f'Nodes({self.backend}, {self.user}, {self.ent_type}, {{\'http://example.com/me/entity#foo\'}})') # repr respects backend class Foo(SparqlStore): pass backend = Foo.Open() backend.schema = self.backend.schema nodes = Nodes(backend, self.user, self.ent_type, self.ent_ids) self.assertEqual(repr(nodes), f'Nodes({backend}, {self.user}, {self.ent_type}, {self.ent_ids})') # repr respects user nodes = Nodes(self.backend, URI('http://example.com/you'), self.ent_type, self.ent_ids) self.assertEqual(repr(nodes), f'Nodes({self.backend}, http://example.com/you, {self.ent_type}, {self.ent_ids})') def test_equality(self): nodes = Nodes(self.backend, self.user, self.ent_type, self.ent_ids) # instance is equal to itself self.assertEqual(nodes, nodes) self.assertEqual(hash(nodes), hash(nodes)) # instance is equal to a clone self.assertEqual(nodes, Nodes(self.backend, self.user, self.ent_type, self.ent_ids)) self.assertEqual(Nodes(self.backend, self.user, self.ent_type, self.ent_ids), nodes) self.assertEqual(hash(nodes), hash(Nodes(self.backend, self.user, self.ent_type, self.ent_ids))) # equality respects backend backend = SparqlStore.Open() backend.schema = self.backend.schema self.assertNotEqual(nodes, Nodes(backend, self.user, self.ent_type, self.ent_ids)) self.assertNotEqual(hash(nodes), hash(Nodes(backend, self.user, self.ent_type, self.ent_ids))) # equality respects user self.assertNotEqual(nodes, Nodes(self.backend, URI('http://example.com/you'), self.ent_type, self.ent_ids)) self.assertNotEqual(hash(nodes), hash(Nodes(self.backend, URI('http://example.com/you'), self.ent_type, self.ent_ids))) # equality respects node_type self.assertNotEqual(nodes, Nodes(self.backend, self.user, self.tag_type, self.ent_ids)) self.assertNotEqual(hash(nodes), hash(Nodes(self.backend, self.user, self.tag_type, self.ent_ids))) # equality respects guids self.assertNotEqual(nodes, Nodes(self.backend, self.user, self.ent_type, self.tag_ids)) self.assertNotEqual(hash(nodes), hash(Nodes(self.backend, self.user, self.ent_type, self.tag_ids))) def test_properties(self): # node_type self.assertEqual(self.ent_type, Nodes( self.backend, self.user, self.ent_type, self.ent_ids).node_type) self.assertEqual(self.tag_type, Nodes( self.backend, self.user, self.tag_type, self.tag_ids).node_type) # guids self.assertSetEqual(self.ent_ids, set(Nodes( self.backend, self.user, self.ent_type, self.ent_ids).guids)) self.assertSetEqual(self.tag_ids, set(Nodes( self.backend, self.user, self.tag_type, self.tag_ids).guids)) def test__ensure_nodes(self): nodes = Nodes(self.backend, self.user, self.ent_type, self.ent_ids) # missing nodes are created self.assertSetEqual(self.ent_ids, nodes._ensure_nodes(self.ent_type, self.ent_ids)) # get creation time from backend manually time_triples = list(self.backend._graph.objects(rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri))) t_ent_created = float(time_triples[0]) if len(time_triples) > 0 else 0.0 # check triples self.assertSetEqual(set(self.backend._graph), { # entity definitions (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), # bookkeeping (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), }) # existing nodes remain unchanged self.assertSetEqual(self.ent_ids, nodes._ensure_nodes(self.ent_type, self.ent_ids)) self.assertSetEqual(set(self.backend._graph), { # entity definitions (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), # bookkeeping (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), }) # type and guids don't need to match the node instance's members self.assertSetEqual(self.tag_ids, nodes._ensure_nodes(self.tag_type, self.tag_ids)) # get creation time from backend manually time_triples = list(self.backend._graph.objects(rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.URIRef(self.t_created.uri))) t_tag_created = float(time_triples[0]) if len(time_triples) > 0 else 0.0 # check triples self.assertSetEqual(set(self.backend._graph), { # previous triples (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), # new triples (rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), (rdflib.URIRef('http://example.com/me/tag#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), (rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_tag_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/tag#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_tag_created, datatype=rdflib.XSD.integer)), }) def test___set(self): # setup nodes = Nodes(self.backend, self.user, self.ent_type, {URI('http://example.com/me/entity#1234'), URI('http://example.com/me/entity#4321')}) self.assertSetEqual(set(self.backend._graph), set()) set_ = nodes._Nodes__set # node_type must match predicate's domain self.assertRaises(errors.ConsistencyError, set_, self.p_representative.uri, self.ent_ids) # cannot set protected predicates self.assertRaises(errors.PermissionDeniedError, set_, self.t_created.uri, 1234) # set literal value set_(self.p_filesize.uri, 1234) # get creation time from backend manually time_triples = list(self.backend._graph.objects(rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri))) t_ent_created = float(time_triples[0]) if len(time_triples) > 0 else 0.0 # verify triples self.assertSetEqual(set(self.backend._graph), { # entity definitions (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), # bookkeeping (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), # literals (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), }) # set node value tags = Nodes(self.backend, self.user, self.tag_type, {URI('http://example.com/me/tag#1234'), URI('http://example.com/me/tag#4321')}) set_(self.p_tag.uri, tags) # get creation time from backend manually time_triples = list(self.backend._graph.objects(rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.URIRef(self.t_created.uri))) t_tag_created = float(time_triples[0]) if len(time_triples) > 0 else 0.0 # verify triples self.assertSetEqual(set(self.backend._graph), { # previous values (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_ent_created, datatype=rdflib.XSD.integer)), # tag definitions (rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), (rdflib.URIRef('http://example.com/me/tag#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), # tag bookkeeping (rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_tag_created, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/tag#4321'), rdflib.URIRef(self.t_created.uri), rdflib.Literal(t_tag_created, datatype=rdflib.XSD.integer)), # entity -> tag links (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#1234')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#1234')), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#4321')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#4321')), }) # value must be a nodes instance self.assertRaises(TypeError, set_, self.p_tag.uri, 'foobar') self.assertRaises(TypeError, set_, self.p_tag.uri, self.tag_ids) self.assertRaises(TypeError, set_, self.p_tag.uri, URI('http://example.com/me/tag#1234')) # value's node_type must match the predicate's range self.assertRaises(errors.ConsistencyError, set_, self.p_tag.uri, Nodes(self.backend, self.user, self.ent_type, self.ent_ids)) def test_set(self): self.assertSetEqual(set(self.backend._graph), set()) nodes = Nodes(self.backend, self.user, self.ent_type, self.ent_ids) # can set literal values self.assertEqual(nodes, nodes.set(self.p_filesize.uri, 1234)) self.assertTrue(set(self.backend._graph).issuperset({ # nodes exist (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), # links exist (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), })) # can set node values self.assertEqual(nodes, nodes.set(self.p_tag.uri, Nodes(self.backend, self.user, self.tag_type, self.tag_ids))) self.assertTrue(set(self.backend._graph).issuperset({ # nodes exist (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), (rdflib.URIRef('http://example.com/me/tag#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), # links exist (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#1234')), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#4321')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#1234')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#4321')), })) # cannot set protected predicate curr = set(self.backend._graph) self.assertRaises(errors.PermissionDeniedError, nodes.set, self.t_created.uri, 12345) self.assertSetEqual(curr, set(self.backend._graph)) # predicate.domain must match node_type self.assertRaises(errors.ConsistencyError, nodes.set, self.p_representative.uri, nodes) self.assertSetEqual(curr, set(self.backend._graph)) # value's node_type must match predicate's range self.assertRaises(errors.ConsistencyError, nodes.set, self.p_tag.uri, nodes) self.assertSetEqual(curr, set(self.backend._graph)) # value type must match predicate's range type self.assertRaises(TypeError, nodes.set, self.p_tag.uri, 'invalid') self.assertSetEqual(curr, set(self.backend._graph)) # cannot assing multiple values to unique predicate self.assertRaises(ValueError, nodes.set, self.p_author.uri, Nodes(self.backend, self.user, self.user_type, {URI('http://example.com/me/user#1234'), URI('http://example.com/me/user#4321')})) self.assertSetEqual(curr, set(self.backend._graph)) def test_set_from_iterable(self): self.assertSetEqual(set(self.backend._graph), set()) nodes = Nodes(self.backend, self.user, self.ent_type, self.ent_ids) # can set literal and node values simultaneously self.assertEqual(nodes, nodes.set_from_iterable({ self.p_filesize.uri: 1234, self.p_tag.uri: Nodes(self.backend, self.user, self.tag_type, self.tag_ids), }.items())) self.assertTrue(set(self.backend._graph).issuperset({ # nodes exist (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Entity')), (rdflib.URIRef('http://example.com/me/tag#1234'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), (rdflib.URIRef('http://example.com/me/tag#4321'), rdflib.RDF.type, rdflib.URIRef('http://bsfs.ai/schema/Tag')), # links exist (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_filesize.uri), rdflib.Literal(1234, datatype=rdflib.XSD.integer)), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#1234')), (rdflib.URIRef('http://example.com/me/entity#1234'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#4321')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#1234')), (rdflib.URIRef('http://example.com/me/entity#4321'), rdflib.URIRef(self.p_tag.uri), rdflib.URIRef('http://example.com/me/tag#4321')), })) # cannot set protected predicate curr = set(self.backend._graph) self.assertRaises(errors.PermissionDeniedError, nodes.set_from_iterable, ((self.p_filesize.uri, 1234), (self.t_created.uri, 12345))) self.assertSetEqual(curr, set(self.backend._graph)) # predicate.domain must match node_type self.assertRaises(errors.ConsistencyError, nodes.set_from_iterable, ((self.p_filesize.uri, 1234), (self.p_representative.uri, nodes))) self.assertSetEqual(curr, set(self.backend._graph)) # value's node_type must match predicate's range self.assertRaises(errors.ConsistencyError, nodes.set_from_iterable, ((self.p_filesize.uri, 1234), (self.p_tag.uri, nodes))) self.assertSetEqual(curr, set(self.backend._graph)) # value type must match predicate's range type self.assertRaises(TypeError, nodes.set_from_iterable, ((self.p_filesize.uri, 1234), (self.p_tag.uri, 'invalid'))) self.assertSetEqual(curr, set(self.backend._graph)) # cannot assing multiple values to unique predicate self.assertRaises(ValueError, nodes.set_from_iterable, ((self.p_filesize.uri, 1234), (self.p_author.uri, Nodes(self.backend, self.user, self.user_type, {URI('http://example.com/me/user#1234'), URI('http://example.com/me/user#4321')})))) self.assertSetEqual(curr, set(self.backend._graph)) ## main ## if __name__ == '__main__': unittest.main() ## EOF ##