83 lines
2.1 KiB
Python
83 lines
2.1 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import logging
|
|
|
|
# TODO: auto import
|
|
# TODO: better logging (scenario visibility)
|
|
# TODO: result storage
|
|
from scenario import no_000001, no_000002
|
|
from neo4j.v1 import GraphDatabase, basic_auth, types, Node
|
|
|
|
# scenarios = [no_000001.scenario_1, no_000002.scenario_2]
|
|
scenarios = [no_000002.scenario_2]
|
|
|
|
# logging init
|
|
log = logging.getLogger(__name__)
|
|
|
|
# initialize driver and create session
|
|
session = GraphDatabase.driver("bolt://localhost",
|
|
auth=basic_auth("neo4j", "1234"),
|
|
encrypted=0).session()
|
|
|
|
|
|
def check(condition, scenario_no, message):
|
|
'''
|
|
Checks condition if condition is false the whole test will be stopped and
|
|
scenario_no + message will be printed out.
|
|
|
|
:param condition: bool
|
|
:param scenario_no: int
|
|
:param message: assert message
|
|
:returns: None
|
|
'''
|
|
if not condition:
|
|
log.error("Error in scenario: %s" % scenario_no)
|
|
assert condition, message
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
for scenario in scenarios:
|
|
|
|
# iterate through all queries and execute them agains the database
|
|
for query, result in scenario.queries:
|
|
|
|
# extract count and properties from test entries
|
|
count, expected = result
|
|
records = [record for record in session.run(query)]
|
|
|
|
log.info("%s" % records)
|
|
|
|
# check count
|
|
check(len(records) == count, scenario.no,
|
|
"Number of results for %s isn't good;" \
|
|
" expected: %s, got %s" % (query, count, len(records)))
|
|
|
|
# in case that the result is single
|
|
if count == 1:
|
|
|
|
# extract properties from record
|
|
record = records[0]
|
|
record_name, = record
|
|
data = record[record_name]
|
|
|
|
if type(data) is Node:
|
|
received = {k: v for (k, v) in data.items()}
|
|
else:
|
|
received = data
|
|
|
|
# get expected elements
|
|
expected = expected[0]
|
|
|
|
# check properties
|
|
check(expected == received, scenario.no,
|
|
"Received data for %s isn't OK; expected: %s, " \
|
|
"got %s" % (query, expected, received))
|
|
|
|
elif count > 1:
|
|
# TODO: implement the full test
|
|
pass
|