python - How to create pytest session level neo4j instance with csv loading mechanism? -


what best way create 1 neo4j session level fixture during pytest , in neo4j, database must loaded nodes csv , relations csv files in session , test can execute in full database (not blank database ).

i have tried below approach, if have better solution , please answer ,

default configurations http port 7474 , bolt 7687 , suppose 1 neo4j running http port 7474 , can not run neo4j instance 7474 ,we have change port in neo4j.conf can start instance , resolve this

sock = socket.socket(socket.af_inet, socket.sock_stream) sock.bind(("", 0)) sock.listen(1) port = sock.getsockname()[1] 

it give free port , need change in neo4j.conf in runtime lets see pytest fixture

@pytest.fixture(scope="session") def neo4j_graph_instance(csv_node_file, csv_relation_file):     instancesdirectory = os.path.join(path_test_dir, "neo4j_instance_acceptance") if not os.path.exists(instancesdirectory):     os.makedirs(instancesdirectory)  archive_file = "neo4j-community-3.2.1-unix.tar.gz" filename = instancesdirectory + "/" + archive_file  if (os.path.isfile(filename)) == true:     try: tarfile.open(filename, "r:gz") archive:             archive.extractall(instancesdirectory)             print ("successfully extracted")      except ioerror:         print "file not there" else:     try:         uri = "http://dist.neo4j.org/neo4j-community-3.2.1-unix.tar.gz"         urlretrieve(uri, filename)         try:             tarfile.open(filename, "r:gz") archive:                 archive.extractall(instancesdirectory)         except ioerror:             print "file not there"     except ioerror:         print "could not connect internet download file"  neo4j_home = os.path.join(instancesdirectory, "neo4j-community-3.2.1") neo4j_import_dir = os.path.join(neo4j_home, "import") neo4j_inst = neo4jinstance(home=neo4j_home)  neo4j_inst.set_config("dbms.connector.http.listen_address", ":%s" % get_open_port()) neo4j_inst.set_config("dbms.connector.bolt.listen_address", ":%s" % get_open_port()) neo4j_inst.set_config("dbms.security.auth_enabled", "false") neo4j_inst.set_config("dbms.connector.https.enabled", "false")  # db loading mechanism # # rajib: getting csv files fixture , copying them in neo4j import dir  in run time. # getting bolt uri active instance  , , executing command 'load_db_script'. # @ end returning 1 graph instance loaded csv files.   neo4j_inst.start_neo4j_instance() time.sleep(15) #:todo need avoid sleep , instead check socket connection desired host , port print "copying csv files neo4j import direcotry"  copy_script = "cp %s %s %s" % (     csv_node_file, csv_relation_file,  neo4j_import_dir) call(copy_script, shell=true) print "successfully copied neo4j import directory " neo4j_inst_bolt_uri = "bolt://localhost:%d" % neo4j_inst.get_bolt_port() load_cypher_file = os.path.join(dms_path, "load_csv.cypher") load_db_script = "cat %s | %s -a %s --format verbose --debug" % (     load_cypher_file, neo4j_inst.cypher_shell_script(), neo4j_inst_bolt_uri) call(load_db_script, shell=true) neo4j_inst_http_port = neo4j_inst.get_http_port()  yield graph(host='localhost', http_port=neo4j_inst_http_port, bolt=false)  print "running teardown methods"  neo4j_inst.stop_neo4j_instance() neo4j_inst.delete_store() 

so every time give fixture neo4j instance port assigned in run time loaded database.

there can better approach , let me know if have done better .


Comments