Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_base(self):
prop = model.get_qname('Thing:name')
node = Node(registry.entity, 'banana')
stmt = Statement(node, prop, "Theodore Böln")
assert stmt.subject == node
value = stmt.to_tuple()
other = stmt.from_tuple(model, value)
assert other == stmt, (stmt, other)
assert hash(other) == hash(stmt)
assert repr(other) == repr(stmt)
def test_graph(self):
g = DiGraph()
proxy = model.get_proxy(ENTITY)
node = proxy.node
self.assertEqual(str(node), node.id)
for stmt in proxy.statements:
stmt.to_digraph(g)
self.assertEqual(g.number_of_edges(), 8)
self.assertEqual(g.number_of_nodes(), 9)
self.assertIn(node.id, g.nodes)
prop = model.get_qname('Thing:name')
stmt = Statement(Node(registry.name, 'Bob'), prop, proxy.id,
inverted=True)
stmt.to_digraph(g)
self.assertEqual(g.number_of_edges(), 9)
stmt = Statement(node, prop, 'Blub', weight=0)
stmt.to_digraph(g)
self.assertEqual(g.number_of_edges(), 9)
prop = model.get_qname('Thing:summary')
stmt = Statement(node, prop, 'This is a text')
stmt.to_digraph(g)
self.assertEqual(g.number_of_edges(), 9)
node = Node(registry.entity, "passpoat")
prop = model.get_qname("Passport:holder")
adj = list(graph.get_adjacent(node, prop))
assert len(adj) == 1, adj
assert adj[0].source_prop == prop, adj[0].source_prop
assert adj[0].target_prop == prop.reverse, adj[0].target_prop
node = Node(registry.entity, "jodie")
prop = model.get_qname("Person:familyPerson")
adj = list(graph.get_adjacent(node, prop))
assert len(adj) == 1, adj
assert adj[0].source_prop == prop, adj[0].source_prop
node = Node(registry.entity, "ralph")
prop = model.get_qname("Person:familyRelative")
adj2 = list(graph.get_adjacent(node, prop))
assert len(adj2) == 1, adj2
assert adj2[0].target_prop == prop, adj2[0].target_prop
assert adj[0] == adj2[0], (adj[0], adj2[0])
assert adj[0].id in repr(adj[0]), repr(adj[0])
graph.add(model.get_proxy(ENTITY, cleaned=False))
graph.add(model.get_proxy(ENTITY2, cleaned=False))
graph.add(model.get_proxy(REL, cleaned=False))
graph.add(model.get_proxy(PASS, cleaned=False))
node = Node(registry.entity, "jodie")
adj = list(graph.get_adjacent(node))
assert len(adj) == 3, adj
node = Node(registry.entity, "ralph")
adj = list(graph.get_adjacent(node))
assert len(adj) == 7, adj
node = Node(registry.entity, "passpoat")
adj = list(graph.get_adjacent(node))
assert len(adj) == 2, adj
node = Node(registry.entity, "passpoat")
prop = model.get_qname("Passport:holder")
adj = list(graph.get_adjacent(node, prop))
assert len(adj) == 1, adj
assert adj[0].source_prop == prop, adj[0].source_prop
assert adj[0].target_prop == prop.reverse, adj[0].target_prop
node = Node(registry.entity, "jodie")
prop = model.get_qname("Person:familyPerson")
adj = list(graph.get_adjacent(node, prop))
assert len(adj) == 1, adj
assert adj[0].source_prop == prop, adj[0].source_prop
node = Node(registry.entity, "ralph")
prop = model.get_qname("Person:familyRelative")
adj2 = list(graph.get_adjacent(node, prop))
assert len(adj2) == 1, adj2
assert adj2[0].target_prop == prop, adj2[0].target_prop
def test_invert(self):
prop = model.get_qname('Thing:name')
node = Node(registry.entity, 'banana')
stmt = Statement(node, prop, "Theodore")
assert not stmt.inverted
inv = stmt.invert()
assert inv.inverted
assert inv.rdf() is None
banana = Node(registry.entity, 'banana')
peach = Node(registry.entity, 'peach')
prop = model.get_qname('Thing:sameAs')
stmt = Statement(banana, prop, peach.value)
inv = stmt.invert()
assert inv.subject == peach
assert inv.value_node == banana
assert inv.prop == stmt.prop
def test_base_functions(self):
proxy = EntityProxy.from_dict(model, ENTITY)
assert "test" in repr(proxy), repr(proxy)
assert hash(proxy) == hash(proxy.id)
assert proxy.get("name") == ["Ralph Tester"]
assert proxy.first("name") == "Ralph Tester"
prop = model.get_qname("LegalEntity:name")
assert proxy.get(prop) == ["Ralph Tester"]
assert proxy.caption == "Ralph Tester"
assert str(proxy) == "Ralph Tester"
name = "Ralph the Great"
proxy.add("name", name)
assert len(proxy.get("name")) == 2
proxy.add("name", None)
assert len(proxy.get("name")) == 2
proxy.add("name", "")
assert len(proxy.get("name")) == 2
proxy.add("name", [""])
assert len(proxy.get("name")) == 2
proxy.add("name", {"name": "banana"})
assert len(proxy.get("name")) == 3, proxy.get("name")
assert name in proxy.get("name")
def unpack(cls, ref, packed):
qname, qualifier, value = packed.split('>', 2)
prop = model.get_qname(qname)
# TODO: parse qualifier
return cls(ref, prop, value)
def iter_value_entities(type_, value):
value = stringify(value)
if type_.group is None or value is None:
return
key = cache.object_key(type(type_), value)
degree_key = cache.object_key(type(type_), value, 'deg1')
degree = cache.get(degree_key)
if degree is not None:
for item in cache.kv.sscan_iter(key):
qname, entity_id = item.decode('utf-8').split('@', 1)
prop = model.get_qname(qname)
yield entity_id, prop
else:
degree = 0
pipe = cache.kv.pipeline()
for entity_id, prop in _iter_value_entities(type_, value):
yield entity_id, prop
item = '@'.join((prop.qname, entity_id))
pipe.sadd(key, item)
degree += 1
pipe.set(degree_key, degree, ex=cache.EXPIRE)
pipe.execute()
facets = []
for prop in model.properties:
if prop.type != registry.entity:
continue
if not schema.is_a(prop.range):
continue
index = entities_read_index(prop.schema)
field = 'properties.%s' % prop.name
value = entity.get('id')
facets.append((index, prop.qname, group, field, value))
res = _filters_faceted_query(authz, facets)
for (qname, total) in res.items():
if total > 0:
yield (model.get_qname(qname), total)