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_install_labels_db_property():
class SomeNode(StructuredNode):
id_ = UniqueIdProperty(db_property='id')
stdout = StringIO()
install_labels(SomeNode, quiet=False, stdout=stdout)
assert 'id_' in stdout.getvalue()
# make sure that the id_ constraint doesn't exist
with pytest.raises(DatabaseError) as exc_info:
db.cypher_query(
'DROP CONSTRAINT on (n:SomeNode) ASSERT n.id_ IS UNIQUE')
assert 'No such constraint' in exc_info.exconly()
# make sure the id constraint exists and can be removed
db.cypher_query('DROP CONSTRAINT on (n:SomeNode) ASSERT n.id IS UNIQUE')
def test_install_labels_db_property():
class SomeNode(StructuredNode):
id_ = UniqueIdProperty(db_property='id')
stdout = StringIO()
install_labels(SomeNode, quiet=False, stdout=stdout)
assert 'id_' in stdout.getvalue()
# make sure that the id_ constraint doesn't exist
with pytest.raises(DatabaseError) as exc_info:
db.cypher_query(
'DROP CONSTRAINT on (n:SomeNode) ASSERT n.id_ IS UNIQUE')
assert 'No such constraint' in exc_info.exconly()
# make sure the id constraint exists and can be removed
db.cypher_query('DROP CONSTRAINT on (n:SomeNode) ASSERT n.id IS UNIQUE')
def test_install_all():
install_labels(AbstractNode)
# run install all labels
install_all_labels()
assert True
# remove constraint for above test
db.cypher_query("DROP CONSTRAINT on (n:NoConstraintsSetup) ASSERT n.name IS UNIQUE")
def test_install_all():
# run install all labels
install_all_labels()
assert True
# remove constraint for above test
db.cypher_query("DROP CONSTRAINT on (n:NoConstraintsSetup) ASSERT n.name IS UNIQUE")
def _execute(self):
query = self.build_query()
results, _ = db.cypher_query(query, self._query_params)
if results:
return [self._ast['result_class'].inflate(n[0]) for n in results]
return []
def _count(self):
self._ast['return'] = 'count({})'.format(self._ast['return'])
# drop order_by, results in an invalid query
self._ast.pop('order_by', None)
query = self.build_query()
results, _ = db.cypher_query(query, self._query_params)
return int(results[0][0])