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_can_set_properties_on_node_in_same_batch():
graph_db = neo4j.GraphDatabaseService()
batch = neo4j.WriteBatch(graph_db)
alice = batch.create({})
batch.set_properties(alice, {"name": "Alice", "age": 34})
results = batch.submit()
alice = results[batch.find(alice)]
assert alice["name"] == "Alice"
assert alice["age"] == 34
def test_can_delete_property_on_node_in_same_batch():
graph_db = neo4j.GraphDatabaseService()
batch = neo4j.WriteBatch(graph_db)
alice = batch.create({"name": "Alice", "age": 34})
batch.delete_property(alice, "age")
results = batch.submit()
alice = results[batch.find(alice)]
assert alice["name"] == "Alice"
assert alice["age"] is None
def test_can_create_path_with_existing_nodes():
graph_db = neo4j.GraphDatabaseService()
alice, bob = graph_db.create({"name": "Alice"}, {"name": "Bob"})
batch = neo4j.WriteBatch(graph_db)
batch.create_path(alice, "KNOWS", bob)
results = batch.submit()
path = results[0]
assert len(path) == 1
assert path.nodes[0] == alice
assert path.relationships[0].type == "KNOWS"
assert path.nodes[1] == bob
def test_can_add_labels_to_node_in_same_batch():
graph_db = neo4j.GraphDatabaseService()
batch = neo4j.WriteBatch(graph_db)
a = batch.create({"name": "Alice"})
batch.add_labels(a, "human", "female")
results = batch.submit()
alice = results[batch.find(a)]
assert alice.get_labels() == {"human", "female"}
def test_can_set_property_on_preexisting_node():
graph_db = neo4j.GraphDatabaseService()
alice, = graph_db.create({"name": "Alice"})
batch = neo4j.WriteBatch(graph_db)
batch.set_property(alice, "age", 34)
batch.run()
assert alice["age"] == 34
def test_can_set_labels_on_preexisting_node():
graph_db = neo4j.GraphDatabaseService()
if not graph_db.supports_node_labels:
return
alice, = graph_db.create({"name": "Alice"})
alice.add_labels("human", "female")
batch = neo4j.WriteBatch(graph_db)
batch.set_labels(alice, "mystery", "badger")
batch.run()
assert alice.get_labels() == {"mystery", "badger"}
def test_can_remove_labels_from_preexisting_node():
graph_db = neo4j.GraphDatabaseService()
alice, = graph_db.create({"name": "Alice"})
alice.add_labels("human", "female")
batch = neo4j.WriteBatch(graph_db)
batch.remove_label(alice, "human")
batch.run()
assert alice.get_labels() == {"female"}
def test_can_delete_property_on_preexisting_node():
graph_db = neo4j.GraphDatabaseService()
alice, = graph_db.create({"name": "Alice", "age": 34})
batch = neo4j.WriteBatch(graph_db)
batch.delete_property(alice, "age")
batch.run()
assert alice["name"] == "Alice"
assert alice["age"] is None
def destroy(self):
""" Removes all nodes, relationships and indexes in the store. This
object will no longer be usable after calling this method.
Construct a new Manager to re-initialise the database for kaiso.
WARNING: This will destroy everything in your Neo4j database.
"""
self._conn.clear()
# NB. we assume all indexes are from constraints (only use-case for
# kaiso) if any aren't, this will not work
batch = neo4j.WriteBatch(self._conn)
for label in self._conn.node_labels:
for key in self._conn.schema.get_indexed_property_keys(label):
batch.append_cypher(
"""
DROP CONSTRAINT ON (type:{type_id})
ASSERT type.{attr_name} IS UNIQUE
""".format(
type_id=label,
attr_name=key,
)
)
batch.run()
def clear(self):
""" Clear all nodes and relationships from the graph.
.. warning::
This method will permanently remove **all** nodes and relationships
from the graph and cannot be undone.
"""
batch = WriteBatch(self)
batch.append_cypher("START r=rel(*) DELETE r")
batch.append_cypher("START n=node(*) DELETE n")
batch.run()