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_create_nodes_and_relationship_4(self):
self.graph.delete_all()
with self.graph.begin() as tx:
a = Node()
b = Node()
c = Node()
ab = Relationship(a, "TO", b)
bc = Relationship(b, "TO", c)
ca = Relationship(c, "TO", a)
tx.create(ab | bc | ca)
self.assertEqual(a.graph, self.graph)
self.assertIsNotNone(a.identity)
self.assertEqual(b.graph, self.graph)
self.assertIsNotNone(b.identity)
self.assertEqual(c.graph, self.graph)
self.assertIsNotNone(c.identity)
self.assertEqual(ab.graph, self.graph)
self.assertIsNotNone(ab.identity)
assert ab.start_node == a
assert ab.end_node == b
self.assertEqual(bc.graph, self.graph)
def test_can_create_and_delete_relationship(self):
ab = Relationship(Node(), "KNOWS", Node())
self.graph.create(ab)
assert isinstance(ab, Relationship)
self.assertEqual(ab.graph, self.graph)
self.assertIsNotNone(ab.identity)
assert self.graph.exists(ab)
self.graph.delete(ab | ab.start_node | ab.end_node)
assert not self.graph.exists(ab)
def test_can_delete_relationship_by_separating(self):
a = Node()
b = Node()
r = Relationship(a, "TO", b)
self.graph.create(r)
assert self.graph.exists(r)
with self.graph.begin() as tx:
tx.separate(r)
assert not self.graph.exists(r)
assert self.graph.exists(a)
assert self.graph.exists(b)
from __future__ import absolute_import
from unittest import TestCase
from neo4j.exceptions import ConstraintError, CypherSyntaxError
from py2neo.data import Node, Relationship, Path, Record
from py2neo.database import Database, Graph, TransactionFinished
from py2neo.internal.json import JSONHydrator
from py2neo.testing import IntegrationTestCase
alice = Node("Person", "Employee", name="Alice", age=33)
bob = Node("Person")
carol = Node("Person")
dave = Node("Person")
alice_knows_bob = Relationship(alice, "KNOWS", bob, since=1999)
alice_likes_carol = Relationship(alice, "LIKES", carol)
carol_dislikes_bob = Relationship(carol, "DISLIKES", bob)
carol_married_to_dave = Relationship(carol, "MARRIED_TO", dave)
dave_works_for_dave = Relationship(dave, "WORKS_FOR", dave)
record_keys = ["employee_id", "Person"]
record_a = Record(zip(record_keys, [1001, alice]))
record_b = Record(zip(record_keys, [1002, bob]))
record_c = Record(zip(record_keys, [1003, carol]))
record_d = Record(zip(record_keys, [1004, dave]))
def setUp(self):
self.graph.delete_all()
self.alice = Node(name="Alice")
self.bob = Node(name="Bob")
self.carol = Node(name="Carol")
s = (Relationship(self.alice, "LOVES", self.bob) |
Relationship(self.bob, "LOVES", self.alice) |
Relationship(self.alice, "KNOWS", self.bob) |
Relationship(self.bob, "KNOWS", self.alice) |
Relationship(self.bob, "KNOWS", self.carol) |
Relationship(self.carol, "KNOWS", self.bob))
self.graph.create(s)
def test_node_cache_is_thread_local(self):
from threading import Thread
node = Node()
self.graph.create(node)
assert node.identity in self.graph.node_cache
other_cache_keys = []
def check_cache():
other_cache_keys.extend(self.graph.node_cache.keys())
thread = Thread(target=check_cache)
thread.start()
thread.join()
assert node.identity in self.graph.node_cache
assert node.identity not in other_cache_keys
def test_custom_conditions_with_parameters(self):
found = list(self.matcher.match("Person").where(("_.name = {1}", {"1": "Keanu Reeves"})))
assert len(found) == 1
first = found[0]
assert isinstance(first, Node)
assert first["name"] == "Keanu Reeves"
assert first["born"] == 1964
def test_can_create_nodes_and_relationship_2(self):
self.graph.delete_all()
with self.graph.begin() as tx:
a = Node("Person", name="Alice")
b = Node("Person", name="Bob")
tx.create(a)
tx.create(b)
r = Relationship(a, "KNOWS", b, since=1999)
tx.create(r)
self.assertEqual(a.graph, self.graph)
self.assertIsNotNone(a.identity)
self.assertEqual(b.graph, self.graph)
self.assertIsNotNone(b.identity)
self.assertEqual(r.graph, self.graph)
self.assertIsNotNone(r.identity)
assert r.start_node == a
assert r.end_node == b
self.assertEqual(len(self.graph.nodes), 2)
self.assertEqual(len(self.graph.relationships), 1)
def setUp(self):
TO = Relationship.type("TO")
self.graph.delete_all()
a = self.a = Node()
b = self.b = Node()
c = self.c = Node()
d = self.d = Node()
self.r = [TO(a, b), TO(b, a), TO(b, c), TO(b, b), TO(c, d), TO(a, d)]
self.graph.create(reduce(or_, self.r))
def test_can_remove_label(self):
node = Node("Person", "Employee", name="Alice")
node.remove_label("Employee")
assert set(node.labels) == {"Person"}