Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from __future__ import print_function
import warnings
import os
from neomodel import config, db, clear_neo4j_database, change_neo4j_password
from neo4j.v1 import CypherError
warnings.simplefilter('default')
config.DATABASE_URL = os.environ.get('NEO4J_BOLT_URL', 'bolt://neo4j:neo4j@localhost:7687')
config.AUTO_INSTALL_LABELS = True
try:
clear_neo4j_database(db)
except CypherError as ce:
# handle instance without password being changed
if 'The credentials you provided were valid, but must be changed before you can use this instance' in str(ce):
change_neo4j_password(db, 'test')
db.set_connection('bolt://neo4j:test@localhost:7687')
print("New database with no password set, setting password to 'test'")
print("Please 'export NEO4J_BOLT_URL=bolt://neo4j:test@localhost:7687' for subsequent test runs")
else:
raise ce
def test_cypher_syntax_error():
jim = User2(email='jim1@test.com').save()
try:
jim.cypher("MATCH a WHERE id(a)={self} RETURN xx")
except CypherError as e:
assert hasattr(e, 'message')
assert hasattr(e, 'code')
else:
assert False, "CypherError not raised."
def run(self, source):
source = source.strip()
if not source:
return
try:
if source.startswith("/"):
try:
self.run_command(source)
except TypeError:
self.run_source(source)
else:
self.run_source(source)
except CypherError as error:
if error.classification == "ClientError":
pass
elif error.classification == "DatabaseError":
pass
elif error.classification == "TransientError":
pass
else:
pass
click.secho("{}: {}".format(error.title, error.message), err=True)
except TransactionError:
click.secho("Transaction error", err=True, fg=self.err_colour)
except ServiceUnavailable:
raise
except Exception as error:
click.secho("{}: {}".format(error.__class__.__name__, str(error)), err=True, fg=self.err_colour)
# Try local_password.
try:
driver = GraphDatabase.driver(uri, auth=basic_auth("neo4j", local_password))
return driver
except:
pass
# Try default_password.
# Password handling from https://github.com/robinedwards/django-neomodel
driver = GraphDatabase.driver(uri, auth=basic_auth("neo4j", default_password))
with driver.session() as neo4j_session:
try:
result = neo4j_session.run("MATCH (a:Person) WHERE a.name = {name} RETURN a", {"name": "Crick"})
except CypherError as ce:
if 'The credentials you provided were valid, but must be changed before you can use this instance' in str(ce):
neo4j_session.run("CALL dbms.changePassword({password})", {'password': local_password})
print("New database with no password set, setting password to '", local_password, "'.")
neo4j_session.close()
else:
raise ce
return driver