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
Post a Comment