To detect logic bugs in graph database engines by mutating graph query patterns. ICSE'24.
#!/usr/bin/env python3

import os
import time
import json
import threading
import configparser
import datetime
from schema_scanner import *
from query_generator import *
from query_mutator import *

class Testing:
    def init_testing_configs(self):
        config = configparser.ConfigParser()
        config.read('graphgenie.ini')
        self.threads = []
        self.prev_results = []
        self.bug_rules_eval = [0, 0, 0]
        self.variant = int(config['testing_configs']['variant'])
        self.multi_threading = int(config['testing_configs']['multi_threading'])
        self.perf_issue = int(config['testing_configs']['perf_issue'])
        self.threshold = float(config['testing_configs']['threshold'])
        self.variant_threshold = float(config['testing_configs']['variant_threshold'])
        self.minimum_test_ms = float(config['testing_configs']['minimum_test_ms'])
        self.ip = config['default']['ip']
        self.port = int(config['default']['port'])
        self.username = config['default']['username']
        self.password = config['default']['password']
        self.logpath = config['testing_configs']['logpath']
        self.bug_logpath = config['testing_configs']['bug_logpath']
        self.exception_logpath = config['testing_configs']['exception_logpath']
        self.min_save_log_size = int(config['testing_configs']['min_save_log_size'])
        self.max_testing_query_num = int(config['testing_configs']['max_testing_query_num'])
        self.testing_times = int(config['testing_configs']['testing_times'])
        self.statistics = int(config['testing_configs']['statistics'])
        self.logging_stop = int(config['testing_configs']['logging_stop'])

    def init_log(self):
        if os.path.exists(self.logpath) and os.path.getsize(self.logpath)>self.min_save_log_size:
            os.system("mv {} ./logs/testing.log-{}".format(self.logpath, str(time.time())))
        if os.path.exists(self.bug_logpath):
            os.system("mv {} ./logs/buggy.log-{}".format(self.bug_logpath, str(time.time())))
        if os.path.exists(self.exception_logpath):
            os.system("mv {} ./logs/exception.log-{}".format(self.exception_logpath, str(time.time())))

    # TODO: multi-thread sync

    def log(self, text):
        f = open(self.logpath, "a+")
        f.write(text)
        f.close()

    def bug_log(self, text):
        f = open(self.bug_logpath, "a+")
        f.write(text)
        f.close()

    def except_log(self, text):
        if "imeout" in text:
            return
        f = open(self.exception_logpath, "a+")
        f.write(text)
        f.close()

    def print_testing_results(self):
        self.log("number of tested base queries: {}\n".format(self.executed_base_query_num))
        self.log("number of detected inconsistent pair: {}\n".format(self.detected_bug_num))

    def time_checking(self, base_time, testing_time):
        diff = max(base_time, testing_time)/min(base_time, testing_time)
        if diff>self.threshold and testing_time>50.0 and testing_time<base_time:
            self.log("[***** Potential Performance Bug: DIFF={:.2f}times *****]\n".format(diff))
            self.bug_rules_eval[0] += self.current_rules_eval[0]
            self.bug_rules_eval[1] += self.current_rules_eval[1]
            self.bug_rules_eval[2] += self.current_rules_eval[2]
            return 1
        return 0

    def variant_time_checking(self, base_time, light_time):
        diff = max(light_time, base_time)/min(light_time, base_time)
        if diff>self.variant_threshold and light_time>base_time:
            self.log("[***** Potential Performance Bug: DIFF={:.2f}times *****]\n".format(diff))
            return 1
        return 0

    def result_checking(self, base_result, test_result, base_query, test_query):
        buggy = type(base_result)!=type(test_result) or base_result!=test_result
        if buggy and "count" in test_query:
            if base_result==None or test_result==None:
                self.log("[***** None Check *****]\n")
            else:
                # remove reduplicate bugs in one testing case
                if test_result in self.prev_results:
                    return
                self.detected_bug_num += 1
                self.bug_rules_eval[0] += self.current_rules_eval[0]
                self.bug_rules_eval[1] += self.current_rules_eval[1]
                self.bug_rules_eval[2] += self.current_rules_eval[2]
                self.log("[***** No.{} Potential Logic Bug: {} {} *****]\n".format(self.detected_bug_num, base_result, test_result))
                self.bug_log("{}\n[***** No.{} Potential Logic Bug:\n\tbase_query={}\n\ttest_query={}\n\tbase_result={}\ttest_result={}\n*****]\n".format(datetime.datetime.now(), self.detected_bug_num, base_query, test_query, base_result, test_result))
                self.prev_results.append(test_result)

    def restricted_result_checking(self, base_result, restricted_result, query):
        if base_result!=None and restricted_result!=None and base_result<restricted_result:
            self.detected_bug_num += 1
            self.log("[***** Potential Restricted Logic Bug: {} {} *****]\n".format(base_result, restricted_result))
            self.bug_rules_eval[0] += self.current_rules_eval[0]
            self.bug_rules_eval[1] += self.current_rules_eval[1]
            self.bug_rules_eval[2] += self.current_rules_eval[2]

    def performance_verification(self, base_query, testing_query, version):
        base_dbhits = self.execute_ret_dbhits(version, base_query)
        testing_dbhits = self.execute_ret_dbhits(version, testing_query)

    # function for multi-threading
    # Note: db may have sync issue => result inconsistency
    def eq_testing(self, base_result, base_time, base_query, eq_query):
        eq_query_result, eq_query_time = self.execute_ret_result_time(eq_query, log_str="[Equivalent]")
        if base_result!=None or eq_query_result!=None:
            self.result_checking(base_result, eq_query_result, base_query, eq_query)
        if self.perf_issue==1 and base_time>self.minimum_test_ms:
            self.time_checking(base_time, eq_query_time)

    # only for testing single version single instance, no index testing
    def testing(self, random_cypher_generator, cypher_query_mutator):
        for i in range(self.testing_times):
            print("{} round testing".format(i+1))
            self.current_rules_eval = [0, 0, 0]
            self.executed_allquery = 0
            self.executed_query_num = 0
            self.executed_base_query_num = 0
            self.detected_bug_num = 0
            random_cypher_generator.init()
            while True:
                if self.executed_base_query_num % self.logging_stop == 0:
                    self.log("{} executed. Rules evaluation statitics: {}\n".format(self.executed_base_query_num, str(self.bug_rules_eval)))
                self.prev_results.clear()
                base_query = random_cypher_generator.random_query_generator()
                self.print_testing_results()
                self.executed_base_query_num += 1
                self.log("=================================\n")
                self.log("No.{} Base Query: {} \n".format(self.executed_base_query_num, base_query))
                base_query_result = -1
                base_query_time = -1
                base_query_result, base_query_time = self.execute_ret_result_time(query=base_query, log_str="[BASE QUERY]")
                equivalent_queries, equivalent_rules_eval = cypher_query_mutator.generate_equivalent_queries(base_query)
                if self.variant:
                    restricted_queries, restricted_rules_eval = cypher_query_mutator.generate_restricted_queries(base_query)
                for i in range(len(equivalent_queries)):
                    eq_query = equivalent_queries[i]
                    self.current_rules_eval = equivalent_rules_eval[i]
                    # Use multi threads
                    if self.multi_threading:
                        if base_query_time>0:
                            new_thread = threading.Thread(target=self.eq_testing, args=(base_query_result, base_query_time, base_query, eq_query))
                            new_thread.start()
                            self.threads.append(new_thread)
                    # No multi threads
                    else:
                        if base_query_time>0:
                            eq_query_result, eq_query_time = self.execute_ret_result_time(query=eq_query, log_str="[Equivalent]")
                            self.result_checking(base_query_result, eq_query_result, base_query, eq_query)
                            if self.perf_issue==1 and base_query_time>self.minimum_test_ms:
                                self.time_checking(base_query_time, eq_query_time)
                if self.multi_threading:
                    # wait for all threads finishing
                    for t in self.threads:
                        t.join()
                    self.threads.clear()

                # variant: testing variant queries
                # multi-threading is not used here
                if self.variant:
                    for i in range(len(restricted_queries)):
                        each_query = restricted_queries[i]
                        self.current_rules_eval = restricted_rules_eval[i]
                        if base_query_time>0:
                            query_result, query_time = self.execute_ret_result_time(query=each_query, log_str="[Restricted]")
                            self.restricted_result_checking(base_query_result, query_result, each_query)

class Neo4jTesting(Testing):
    # Neo4j Config
    def __init__(self):
        self.init_testing_configs()
        self.start_time = time.time()
        if os.path.exists(self.logpath):
            self.init_log()
        self.log("***** Testing Neo4j *****\n")
        self.log("\tmulti_threading={}\n".format(self.multi_threading))
        self.log("\tvariant={}\n".format(self.variant))
        self.bolt_uri = "bolt://{}:{}".format(self.ip, self.port)
        self.driver = GraphDatabase.driver(self.bolt_uri, auth=(self.username, self.password))

    # log/return result+time
    def execute_ret_result_time(self, query, log_str):
        driver = self.driver
        query_result = None
        query_time = -1
        with driver.session() as session:
            try:
                # if possible, clear cache first
                clear_query = "CALL db.clearQueryCaches();"
                session.execute_write(self._new_execute, clear_query)
                query_result, query_time = session.execute_write(self._new_execute, query)
            except Exception as e:
                self.except_log('\nQuery:{}\nInfo:{}\n'.format(query, str(e)))
                return None, -1
            self.executed_allquery += 1
            self.log("No.{} {} Query=\"{}\"\n\tQuery Result={}\n\tQuery Time={}\n".format(self.executed_allquery, log_str, query, query_result, query_time))
        self.executed_query_num += 1
        return query_result, query_time

    @staticmethod
    def _new_execute(tx, query):
        ## clear cache first
        clear_query = "CALL db.clearQueryCaches();"
        tx.run(clear_query)
        query_result = -1
        query_time = -1
        query_execute = tx.run(query)
        query_data = query_execute.data()
        if len(query_data)==0:
            query_result = None
        else:
            query_result = (list(query_data[0].values())[0])
        query_time = query_execute.consume().result_available_after + query_execute.consume().result_consumed_after
        return query_result, query_time

class RedisGraphTesting(Testing):
    # Redis Config
    graph = "db"
    def __init__(self):
        self.init_testing_configs()
        self.start_time = time.time()
        if os.path.exists(self.logpath):
            self.init_log()
        self.log("***** Testing RedisGraph *****\n")
        self.log("\tmulti_threading={}\n".format(self.multi_threading))
        self.log("\tvariant={}\n".format(self.variant))
        r = redis.Redis(host=self.ip, port=self.port)
        self.driver = Graph(self.graph, r)

    def execute_ret_result_time(self, query, log_str):
        driver = self.driver
        query_result = None
        query_time = -1
        try:
            result = self.driver.query(query)
            first_result = result.result_set
            if len(first_result)==0:
                query_result = None
            else:
                query_result = first_result[0][0]
            query_time = result.run_time_ms
        except Exception as e:
            self.except_log('\nQuery:{}\nInfo:{}\n'.format(query, str(e)))
            return None, -1
        self.log("{} Query=\"{}\"\n\tQuery Result={}\n\tQuery Time={}\n".format(log_str, query, query_result, query_time))
        self.executed_query_num += 1
        return query_result, query_time

class AgensGraphTesting(Testing):
    def __init__(self):
        self.init_testing_configs()
        self.start_time = time.time()
        if os.path.exists(self.logpath):
            self.init_log()
        self.log("***** Testing AgensGraph *****")
        self.connstr = "host={} port={} user={} password={}".format(
            self.ip,
            self.port,
            self.username,
            self.password
        )
        self.init_testing_configs()
        self.conn = psycopg2.connect(self.connstr)
        self.cur = self.conn.cursor()
        self.cur.execute("SET graph_path = t")

    def execute_ret_result_time(self, query, log_str):
        query_result = None
        query_time = -1
        query = query.replace('--', '-[]-')
        try:
            start_time = time.time()
            self.cur.execute(query)
            query_result = self.cur.fetchone()
            if query_result and len(query_result)>0:
                query_result = query_result[0]
            query_time = (time.time()-start_time)*1000
            self.cur.fetchall()
        except Exception as e:
            self.conn = psycopg2.connect(self.connstr)
            self.cur = self.conn.cursor()
            self.cur.execute("SET graph_path = t")
            self.except_log('\nQuery:{}\nInfo:{}\n'.format(query, str(e)))
            return None, -1
        self.executed_query_num += 1
        self.log("{} Query=\"{}\"\n\tQuery Result={}\n\tQuery Time={}\n".format(log_str, query, query_result, query_time))
        return query_result, query_time

if __name__ == "__main__":
    print("=====GraphGenie=====")
    config = configparser.ConfigParser()
    config.read('graphgenie.ini')
    graphdb = config['default']['graphdb']
    language = config['default']['language']
    ip = config['default']['ip']
    port = int(config['default']['port'])
    username = config['default']['username']
    password = config['default']['password']

    if graphdb=="neo4j":
        from neo4j import GraphDatabase
        test = Neo4jTesting()
    elif graphdb=="redisgraph":
        import redis
        from redisgraph import Node, Edge, Graph, Path
        test = RedisGraphTesting()
    elif graphdb=="agensgraph":
        import psycopg2
        test = AgensGraphTesting()
    else:
        pass

    if language=="cypher" and graphdb=="neo4j":
        schema_scanner = Neo4jSchemaScanner(ip, port, username, password)
        node_labels, edge_labels, node_properties, connectivity_matrix = schema_scanner.scan()
    else:
        # for other graph-dbs, please manually specify labels below
        node_labels = ["u3zY", "U", "sLnkyz1k", "CFn2"]
        edge_labels = ['EgysWPD', 'phUVN3WK1', 'kxA']
        # better to have but
        # it is ok for leave empty for node properties and connectivity matrix
        node_properties = []
        connectivity_matrix = []

    random_cypher_generator = RandomCypherGenerator(node_labels, edge_labels, node_properties, connectivity_matrix)
    cypher_query_mutator = CypherQueryMutator(node_labels, edge_labels, node_properties, connectivity_matrix)
    test.testing(random_cypher_generator, cypher_query_mutator)

main.py
YuanchengJiang/GraphGenie
#!/usr/bin/env python3
import string
import re
import configparser
from random import randint, choice

# this is a lightweight cypher query generator

class RandomCypherGenerator():
    # in GraphGenie, the high-level idea is to mutated the graph query pattern
    # that is, the _path, rather than _predicate mutated by many existing works
    cypher_query_pattern = "{_match} {_path} {_predicate} {_return} {_other}"

    # Note: we generate cypher queries in an incremental way
    # we start from a specific number (node_number) which limits the count of nodes
    # self._node_num = int(config['testing_configs']['_node_num'])
    # You can start from two nodes so it only generates (x)-[y]-(z)
    # path vectors store the previously tested graph query patterns
    # if we test duplicated patterns for many times (recorded in `stuck`)
    # we would increase the node number, now the condition is
    # self.stuck==2*self._node_num*self._node_num
    _path_vectors = []
    _last_vector_length = 0
    stuck = 0

    # cypher query elements
    _match = ""
    _path = ""
    _predicate = ""
    _return = ""
    _other = ""

    def __init__(self, node_labels, edge_labels, node_properties, connectivity_matrix):
        config = configparser.ConfigParser()
        config.read('graphgenie.ini')
        self.graphdb = config['default']['graphdb']
        self.language = config['default']['language']
        self._node_num = int(config['testing_configs']['_node_num'])
        self.min_node_num = int(config['query_generation_args']['min_node_num'])
        self.max_node_num = int(config['query_generation_args']['max_node_num'])
        self.variable_pathlen_rate = float(config['query_generation_args']['variable_pathlen_rate'])
        self.node_symbol_rate = float(config['query_generation_args']['node_symbol_rate'])
        self.edge_symbol_rate = float(config['query_generation_args']['edge_symbol_rate'])
        self.node_label_rate = float(config['query_generation_args']['node_label_rate'])
        self.edge_label_rate = float(config['query_generation_args']['edge_label_rate'])
        self.multi_node_label_rate = float(config['query_generation_args']['multi_node_label_rate'])
        self.multi_edge_label_rate = float(config['query_generation_args']['multi_edge_label_rate'])
        self.cyclic_rate = float(config['query_generation_args']['cyclic_rate'])
        self.random_symbol_len = int(config['query_generation_args']['random_symbol_len'])
        self.cyclic_symbol = config['query_generation_args']['cyclic_symbol']
        self.multi_node_labels = int(config['query_generation_args']['multi_node_labels'])
        self.multi_edge_labels = int(config['query_generation_args']['multi_edge_labels'])
        self.node_labels = node_labels
        self.edge_labels = edge_labels
        self.node_properties = node_properties
        self.connectivity_matrix = connectivity_matrix
        pass

    # call before each run of test
    def init(self):
        config = configparser.ConfigParser()
        config.read('graphgenie.ini')
        self._node_num = int(config['testing_configs']['_node_num'])
        self._path_vectors = []
        self._last_vector_length = 0
        self.stuck = 0
        self.init_query()

    # note: call before each generation
    def init_query(self):
        self._match = ""
        self._path = ""
        self._predicate = ""
        self._return = ""
        self._condition = ""
        self.symbols = []
        self.node_symbols = []
        self.edge_symbols = []
        self.name_label_dict = {}
        self.nodes_num = 0

    # this is a random choice api for a given rate
    # e.g., if given_rate = 0.3, then 30% returns true and 70% return false
    def random_choice(self, given_rate):
        if randint(1,100)<=given_rate*100:
            return True
        else:
            return False

    # this is a random symbol generator
    # note: we only consider lowercase letters in ascii
    def random_symbol(self):
        return ''.join(choice(string.ascii_lowercase) for _ in range(self.random_symbol_len))

    # _match indicates the query is a graph-matching query rather than add/update/delete queries
    # cypher: `match` or `optional match` clause
    def match_generator(self):
        match_candidates = ["MATCH", "OPTIONAL MATCH"]
        self._match = choice(match_candidates)

    def random_node_multi_labels(self):
        label_num = len(self.node_labels)
        random_num = randint(2, label_num)
        node_label = choice(self.node_labels)
        for i in range(random_num-1):
            node_label += "|{}".format(choice(self.node_labels))
        return node_label

    def random_edge_types(self):
        type_num = len(self.edge_labels)
        random_num = randint(2, type_num)
        edge_type = choice(self.edge_labels)
        for i in range(random_num-1):
            edge_type += "|{}".format(choice(self.edge_labels))
        return edge_type

    # given the previous node, use connectivity matrix to find connectable node labels
    def connectable_node_labels(self, prev_node_label, prev_node_direction):
        if self.graphdb!="neo4j":
            return self.node_labels
        if prev_node_label==None or prev_node_label=="" or "%" in self.node_labels:
            return self.node_labels
        possible_node_labels = []
        for each_prev_node_label in prev_node_label.split('|'):
            prev_node_index = self.node_labels.index(each_prev_node_label)
            for each_label in self.node_labels:
                each_label_index = self.node_labels.index(each_label)
                if prev_node_direction==">":
                    if self.connectivity_matrix[prev_node_index][each_label_index]!=0:
                        possible_node_labels.append(each_label)
                elif prev_node_direction=="<":
                    if self.connectivity_matrix[each_label_index][prev_node_index]!=0:
                        possible_node_labels.append(each_label)
                else:
                    if self.connectivity_matrix[prev_node_index][each_label_index]!=0 or self.connectivity_matrix[each_label_index][prev_node_index]!=0:
                        possible_node_labels.append(each_label)
        return possible_node_labels

    # to generate random path unit
    def random_path_unit(self, prev_node_label=None, prev_node_direction=None):
        connectable_node_labels = self.connectable_node_labels(prev_node_label, prev_node_direction)
        path_unit_candidates = [
            "({node_sym})-[{edge_sym}]-",
            "({node_sym})<-[{edge_sym}]-",
            "({node_sym})-[{edge_sym}]->"
            ]
        random_unit = choice(path_unit_candidates)
        random_node_sym = self.random_symbol() if self.random_choice(self.node_symbol_rate) else ""
        random_edge_sym = self.random_symbol() if self.random_choice(self.edge_symbol_rate) else ""
        # determine whether we need node label
        if self.random_choice(self.node_label_rate) and random_node_sym!="" and len(connectable_node_labels)!=0:
            node_labels = ""
            if self.graphdb=="neo4j" and self.multi_node_labels:
                node_labels = self.random_node_multi_labels() if self.random_choice(self.multi_node_label_rate) else choice(connectable_node_labels)
            else:
                node_labels = choice(connectable_node_labels)
            random_node_sym = "{}:{}".format(random_node_sym, node_labels)
        # determine whether we need edge label
        if self.random_choice(self.edge_label_rate) and random_edge_sym!="":
            # do not support multiple edge labels
            random_edge_sym = "{}:{}".format(random_edge_sym, choice(self.edge_labels))
        return random_unit.format(node_sym=random_node_sym, edge_sym=random_edge_sym)

    def cypher_get_unit_direction(self, path_unit):
        if "<" in path_unit:
            return "<"
        elif ">" in path_unit:
            return ">"
        else:
            return "-"

    def parse_path_unit_node_label(self, path_unit):
        the_node = path_unit.split('-')[0]
        if ':' not in the_node:
            return ""
        else:
            # extract Person from (a:Person)
            return the_node.split(':')[1].split(')')[0]

    # it is important to generate diverse graph query patterns
    def path_generator(self):
        nodes_num = self._node_num
        path = ""
        prev_node_label = ""
        prev_node_direction = "-"
        # given the number of nodes, we generate each node unit
        # it takes previous node label and edge information and checks the connectivity
        # connectivity: before testing, we have parsed the target dataset to pre-analyze
        # the connectivity among different types of nodes
        # note: if supporting update/insert clauses later, we need incremental updates to
        # the connectivity matrix
        for i in range(nodes_num):
            new_path_unit = self.random_path_unit(prev_node_label, prev_node_direction)
            # update the previous node label and edge direction after generation
            prev_node_label = self.parse_path_unit_node_label(new_path_unit)
            prev_node_direction = self.cypher_get_unit_direction(new_path_unit)
            if self.random_choice(self.variable_pathlen_rate):
                variable_length_expressions = ["*..1]", "*0..1]", "*0..0]", "*1..1]"]
                new_path_unit = new_path_unit.replace(']', choice(variable_length_expressions))
            path += new_path_unit
        # to strip the tail edge
        path = ")".join(path.split(")")[:-1])+")"
        # to generate cyclic path
        if self.random_choice(self.cyclic_rate):
            cyclic_str = "{cyc_sym}{node_label}".format(
                cyc_sym = self.cyclic_symbol,
                node_label= ":"+choice(self.node_labels) if self.random_choice(self.node_label_rate) else ""
            )
            path = ("({cyc})-{path}-({cyc})".format(cyc=cyclic_str, path="-".join(path.split("-")[1:-1])))
        self._path = path
        self.path_parser()

    def path_parser(self):
        self.nodes_num = self._path.count('-')/2 + 1
        symbol_list = re.findall("[(,\[][a-z]{{{sym_len}}}".format(sym_len=self.random_symbol_len), self._path)
        symbol_list_with_labels = re.findall("[(,\[][a-z]{{{sym_len}}}[:]?[\w]+".format(sym_len=self.random_symbol_len), self._path)
        for each_symbol in symbol_list:
            symbol_name = each_symbol[1:]
            self.symbols.append(symbol_name)
            if each_symbol[0]=="(":
                self.node_symbols.append(symbol_name)
            elif each_symbol[0]=="[":
                self.edge_symbols.append(symbol_name)
        for each_symbol in symbol_list_with_labels:
            symbol_name = each_symbol[1:].split(':')[0]
            symbol_label = each_symbol[1:].split(':')[1]
            self.symbols.append(symbol_name)
            if each_symbol[0]=="(":
                self.node_symbols.append(symbol_name)
            elif each_symbol[0]=="[":
                self.edge_symbols.append(symbol_name)
            self.name_label_dict.update({symbol_name: symbol_label})
        # to record the path, for incrementally generation
        # record node labels only, save as vector.
        all_nodes = re.findall("\([A-Za-z0-9:]*\)", self._path)
        path_vector = []
        for each_node in all_nodes:
            if ':' not in each_node:
                path_vector.append(0)
            else:
                node_label = each_node.split(':')[1].split(')')[0]
                path_vector.append(self.node_labels.index(node_label)+1)

        # the code below is for incremental base query generation
        # we encode the graph pattern into vectors
        # too many deduplicated queries would increase the node number of graph pattern

        if path_vector not in self._path_vectors:
            print("node num: {} tested vectors:{}".format(self._node_num, self._last_vector_length+1))
            self._path_vectors.append(path_vector)
        if self._last_vector_length == len(self._path_vectors):
            self.stuck += 1
        else:
            self.stuck = 0
        self._last_vector_length = len(self._path_vectors)
        if self.stuck==2*self._node_num*self._node_num:
            self.stuck = 0
            self._node_num += 1
            self._last_vector_length = 0
            self._path_vectors.clear()

    # TODO: add more predicate
    def predicate_generator(self):
        pattern = "WHERE {}"
        predicate = "{} IS NOT NULL AND True".format(choice(self.node_symbols)) if len(self.node_symbols)>0 else "True"
        self._predicate = pattern.format(predicate)

    # note: we focus on testing `count`
    def return_generator(self):
        _return = "{return_keyword} {return_staff}"
        return_keywords = ["RETURN", "RETURN DISTINCT"]
        # TODO: to support count(DISTINCT ), max(), min()
        test_returns = ["count({})"]
        return_staff = choice(test_returns).format(choice(self.symbols)) if len(self.symbols)>0 else "count(1)"
        self._return = _return.format(
            return_keyword = choice(return_keywords),
            return_staff = return_staff
        )

    # for count() testing, we do not really need other clauses
    def other_generator(self):
        _other = "{order_by} {skip} {limit}"
        order_by_keywords = ["", "ORDER BY -1+1", "ORDER BY NULL"]
        skip_keywords = ["", "SKIP 1", "SKIP 0", "SKIP 0", "SKIP 0", "SKIP 0", "SKIP 0", "SKIP 0",]
        limit_keywords = ["", "LIMIT 0", "LIMIT 1", "LIMIT 1", "LIMIT 2", "LIMIT 3", "LIMIT 5", "LIMIT 1"]
        self._other = _other.format(
            order_by = choice(order_by_keywords),
            skip = choice(skip_keywords),
            limit = choice(limit_keywords)
        )

    def random_query_generator(self):
        self.init_query()
        self.match_generator()
        self.path_generator()
        self.predicate_generator()
        self.return_generator()
        self.other_generator()
        query = self.cypher_query_pattern.format(
            _match = self._match,
            _path = self._path,
            _predicate = self._predicate,
            _return = self._return,
            _other = self._other
        )
        query = re.sub(' +', ' ', query).strip(' ')
        query = re.sub('[*]+', '*', query).strip(' ')
        return query

query_generator.py
YuanchengJiang/GraphGenie
#!/usr/bin/env python3
import re
import string
import configparser
from random import choice, randint

class CypherQueryMutator:
    cypher_query_pattern = "{_match} {_path} {_predicate} {_return} {_other}"

    def __init__(self, node_labels, edge_labels, node_properties, connectivity_matrix):
        config = configparser.ConfigParser()
        config.read('graphgenie.ini')
        self.graphdb = config['default']['graphdb']
        self.language = config['default']['language']
        self.random_symbol_len = int(config['query_generation_args']['random_symbol_len'])
        self.graph_pattern_mutation = int(config['testing_strategy']['graph_pattern_mutation'])
        self.mutated_query_num = int(config['testing_configs']['mutated_query_num'])
        self.cyclic_symbol = config['query_generation_args']['cyclic_symbol']
        self.node_labels = node_labels
        self.edge_labels = edge_labels
        self.node_properties = node_properties
        self.connectivity_matrix = connectivity_matrix
        pass

    def cypher_query_parser(self, query):
        _match = "OPTIONAL MATCH" if "OPTIONAL" in query else "MATCH"
        _path = query.split("MATCH ")[1].split(' ')[0].strip(' ')
        if "WHERE " not in query:
            _predicate = ""
        else:
            _predicate = "WHERE " + query.split("WHERE")[1].split("RETURN")[0].strip(' ')
        return_symbol = "RETURN DISTINCT " if "RETURN DISTINCT" in query else "RETURN "
        _return = return_symbol + query.split(return_symbol)[1].split(')')[0]+')'
        _other = ')'.join(query.split(return_symbol)[1].split(')')[1:])
        return _match, _path, _predicate, _return, _other

    def path_parser(self, path):
        symbols = []
        node_symbols = []
        edge_symbols = []
        symbol_list = re.findall("[(,\[][a-z]{{{sym_len}}}".format(sym_len=self.random_symbol_len), path)
        for each_symbol in symbol_list:
            symbol_name = each_symbol[1:]
            symbols.append(symbol_name)
            if each_symbol[0]=="(":
                node_symbols.append(symbol_name)
            elif each_symbol[0]=="[":
                edge_symbols.append(symbol_name)
        return symbols, node_symbols, edge_symbols

    def init_for_each_base_query(self, base_query):
        # meta data for base query
        self.base_query = base_query
        self.base_match = ""
        self.base_path = ""
        self.base_predicate = ""
        self.base_return = ""
        self.base_other = ""
        self.base_symbols = []
        self.base_node_symbols = []
        self.base_edge_symbols = []

        self.mutated_match = ""
        self.mutated_path = ""
        self.mutated_predicate = ""
        self.mutated_return = ""
        self.mutated_other = ""
        self.mutated_symbols = []
        self.mutated_node_symbols = []
        self.mutated_edge_symbols = []

        # we split rules into three classes: Non-GQT, Property-GQT, and Structure-GQT
        self.Non_GQT = [1,0,0]
        self.Property_GQT = [0,1,0]
        self.Structure_GQT = [0,0,1]
        self.equivalent_queries = []
        self.restricted_queries = []

        # eval list (a list of triple) records the rule using the tuple: [x,y,z]
        # x=1 indicates it includes Non-GQT transformation
        # y=1 indicates it includes Property-GQT transformation
        # z=1 indicates it includes Structure-GQT transformation
        self.equivalent_queries_eval = []
        self.restricted_queries_eval = []

    def strip_spaces(self, query):
        return re.sub(" +", " ", query)

    # This is a random choice wrapper for a given rate
    # E.g., if given_rate = 0.3, then 30% returns True and 70% return False
    def random_choice(self, given_rate):
        if randint(1,100)<=given_rate*100:
            return True
        else:
            return False

    def random_symbol(self):
        # Symbol length for self.random_symbol()
        return ''.join(choice(string.ascii_lowercase) for _ in range(self.random_symbol_len))

    def query_parser(self, base_query):
        self.init_for_each_base_query(base_query)
        self.base_match, self.base_path, self.base_predicate, self.base_return, self.base_other = self.cypher_query_parser(base_query)
        self.base_symbols, self.base_node_symbols, self.base_edge_symbols = self.path_parser(self.base_path)
        self.mutated_match, self.mutated_path, self.mutated_predicate, self.mutated_return, self.mutated_other = self.cypher_query_parser(base_query)
        self.mutated_symbols, self.mutated_node_symbols, self.mutated_edge_symbols = self.path_parser(self.mutated_path)

    def generate_restricted_queries(self, base_query):
        self.query_parser(base_query)
        if "count(" in self.base_return:
            self.generate_restricted_add_edge_direction()
            self.generate_restricted_add_edge_label()
            self.generate_restricted_add_node_label()
            self.generate_restricted_add_node()
        return self.restricted_queries, self.restricted_queries_eval

    def generate_restricted_add_node_label(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if "()" not in self.base_path:
                return
            new_path = self.base_path.replace("()", "({}:{})".format(self.random_symbol(), choice(self.node_labels)), 1)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = self.base_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.restricted_queries.append(mutated_query)
        self.restricted_queries_eval.append(self.Property_GQT)

    def generate_restricted_add_edge_label(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if "-[]-" not in self.base_path:
                return
            new_path = self.base_path.replace("-[]-", "-[:{}]-".format(choice(self.edge_labels)), 1)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = self.base_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        if self.language=="gremlin":
            mutated_query = self.strip_dots(mutated_query)
        mutated_query = self.strip_spaces(mutated_query)
        self.restricted_queries.append(mutated_query)
        self.restricted_queries_eval.append(self.Property_GQT)

    def generate_restricted_add_edge_direction(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if ")-[]-(" not in self.base_path:
                return
            new_path = self.base_path.replace(")-[]-(", choice([")-[]->(", ")<-[]-("]), 1)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = self.base_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.restricted_queries.append(mutated_query)
        self.restricted_queries_eval.append(self.Non_GQT)

    def generate_restricted_add_node(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if "count(1)" in self.base_return:
                return
            new_path_pattern = ["{new_node}-[]-{path}", "{new_node}-[]->{path}", "{new_node}<-[]-{path}"]
            new_node_label = choice(["", ":{}".format(choice(self.node_labels))])
            new_node = "({}{})".format(self.random_symbol(), new_node_label)
            new_path = choice(new_path_pattern).format(new_node=new_node, path=self.base_path)
            new_return = self.base_return.replace("(", "(DISTINCT ", 1)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = self.base_predicate,
                _return = new_return,
                _other = self.base_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.restricted_queries.append(mutated_query)
        self.restricted_queries_eval.append(self.Non_GQT)

    def generate_equivalent_queries(self, base_query):
        # for mutation rules, we have two modes
        # if you call the function without argument
        # then it is single mutatoin
        # if you call the function with argument 1
        # then it is iterative mutation based on previous mutated query
        self.query_parser(base_query)
        # the query number for iterative mutation times
        recursive_mutated_query_num = 5
        # TODO: still have many mutation rules that I do not have time to implement
        # TODO: SplitNodeLabels - node with multiple labels can be splitted in to predicate
        # TODO: SplitEdgeLabels - edge with multiple labels can be splitted in to predicate
        # TODO: SplitDirection - edge without directions can be splitted into
        #   left direction plus right direction
        # note: all rules should be also applicable the other way round.
        if self.language=="cypher":
            self.generate_equivalent_switch_match()
            self.generate_equivalent_splitting_path()
            self.generate_equivalent_symmetrical_queries()
            self.generate_equivalent_unfold_cyclic()
            self.generate_equivalent_count_id()
            self.generate_equivalent_count_other_symbol()
            self.generate_equivalent_move_label_predicate()
            self.generate_equivalent_predicate_intersect()
            self.generate_equivalent_queries_adding_redundant_predicate_a()
            self.generate_equivalent_queries_adding_redundant_predicate_b()
            self.generate_equivalent_queries_adding_redundant_predicate_c()
            self.generate_equivalent_rename_symbols_up()
            self.generate_equivalent_rename_symbols_down()
            if self.graphdb=="neo4j":
                self.generate_equivalent_adding_call_wrapper()
            else:
                self.generate_equivalent_count_star()
            if self.language=="cypher" and self.graphdb!="agensgraph":
                # for iterative mutations, not all rules are applicable
                recursive_mutation_rules = [
                    "generate_equivalent_switch_match",
                    "generate_equivalent_count_other_symbol",
                    "generate_equivalent_queries_adding_redundant_predicate_a",
                    "generate_equivalent_queries_adding_redundant_predicate_b",
                    "generate_equivalent_queries_adding_redundant_predicate_c",
                    "generate_equivalent_symmetrical_queries",
                    "generate_equivalent_move_label_predicate",
                    "generate_equivalent_predicate_intersect"
                ]
                while True:
                    getattr(self, choice(recursive_mutation_rules))(1)
                    if len(self.equivalent_queries)==self.mutated_query_num:
                        break

            """
            # for Evaluation #
            # Structure-GQT
            self.generate_equivalent_splitting_path()
            self.generate_equivalent_symmetrical_queries()
            self.generate_equivalent_unfold_cyclic()
            # Property-GQT
            self.generate_equivalent_count_id()
            self.generate_equivalent_count_other_symbol()
            self.generate_equivalent_move_label_predicate()
            # Non
            self.generate_equivalent_predicate_intersect()
            self.generate_equivalent_queries_adding_redundant_predicate_b()
            self.generate_equivalent_rename_symbols_up()
            """
        return self.equivalent_queries, self.equivalent_queries_eval

    def generate_equivalent_count_other_symbol(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if len(self.base_node_symbols)<2:
                return
            if "min" in self.base_return or "max" in self.base_return:
                return
            return_sym = self.base_return.split('(')[1].split(' ')[-1].split(')')[0]
            new_return = self.base_return.replace(return_sym, "id({})".format(choice(self.base_node_symbols)), 1)
            # return_list = self.base_return.split(' ')
            # if "count(DISTINCT" in self.base_return:
            #     return_list[-1] = "count(DISTINCT {})".format(choice(self.base_node_symbols))
            # else:
            #     return_list[-1] = "count({})".format(choice(self.base_node_symbols))
            # new_return = ' '.join(return_list)
            mutated_query = self.cypher_query_pattern.format(
                _match = self.base_match,
                _path = self.base_path,
                _predicate = self.base_predicate,
                _return = new_return,
                _other = self.base_other
            )
            mutated_query = self.strip_spaces(mutated_query)
        else:
            if len(self.mutated_node_symbols)<2:
                return
            if "min" in self.mutated_return or "max" in self.mutated_return:
                return
            return_list = self.mutated_return.split(' ')
            if "count(DISTINCT" in self.mutated_return:
                return_list[-1] = "count(DISTINCT {})".format(choice(self.mutated_node_symbols))
            else:
                return_list[-1] = "count({})".format(choice(self.mutated_node_symbols))
            self.mutated_return = ' '.join(return_list)
            mutated_query = self.cypher_query_pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
            mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Property_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([last_eval[0], 1, last_eval[-1]])

    # This mutation is not recursive
    def generate_equivalent_count_star(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            new_return = "RETURN count(*)"
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = self.base_path,
                _predicate = self.base_predicate,
                _return = new_return,
                _other = self.base_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Property_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([last_eval[0], 1, last_eval[-1]])

    # This mutation is not recursive
    def generate_equivalent_count_id(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if "count(1)" in self.base_return or "max" in self.base_return or "min" in self.base_return:
            return
        return_sym = self.base_return.split('(')[1].split(' ')[-1].split(')')[0]
        new_return = self.base_return.replace(return_sym, "id({})".format(return_sym), 1)
        # return_list = self.base_return.split(' ')
        # return_list[-1] = return_list[-1].replace(")", "))").replace("(", "(id(")
        # new_return = ' '.join(return_list)
        pattern = self.cypher_query_pattern
        mutated_query = pattern.format(
            _match = self.base_match,
            _path = self.base_path,
            _predicate = self.base_predicate,
            _return = new_return,
            _other = self.base_other
        )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        self.equivalent_queries_eval.append(self.Property_GQT)

    def generate_equivalent_switch_match(self, mode=0):
        if mode==0:
            mutated_query = self.cypher_query_pattern.format(
                _match = "OPTIONAL MATCH" if self.base_match=="MATCH" else "MATCH",
                _path = self.base_path,
                _predicate = self.base_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            self.mutated_match = "OPTIONAL MATCH" if self.mutated_match=="MATCH" else "MATCH"
            mutated_query = self.cypher_query_pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Non_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([1, last_eval[1], last_eval[2]])

    def generate_equivalent_predicate_intersect(self, mode=0):
        if mode==0:
            mutated_query = self.base_query.replace('AND', 'WITH * WHERE')
        else:
            self.mutated_predicate = self.mutated_predicate.replace('AND', 'WITH * WHERE')
            mutated_query = self.cypher_query_pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Non_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([1, last_eval[1], last_eval[2]])

    def generate_equivalent_queries_adding_redundant_predicate_a(self, mode=0):
        if mode==0:
            if "WHERE " in self.base_predicate:
                new_predicate = self.base_predicate + " AND True"
            else:
                new_predicate = "WHERE True"
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = self.base_path,
                _predicate = new_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            if "WHERE " in self.mutated_predicate:
                self.mutated_predicate = self.mutated_predicate + " AND True"
            else:
                self.mutated_predicate = "WHERE True"
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Non_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([1, last_eval[1], last_eval[2]])

    def generate_equivalent_unfold_cyclic(self, mode=0):
        if self.cyclic_symbol in self.base_path:
            new_path = self.base_path.replace(self.cyclic_symbol, "start", 1).replace(self.cyclic_symbol, "end", 1)
            if "WHERE " in self.base_predicate:
                new_predicate = self.base_predicate + " AND start=end"
            else:
                new_predicate = "WHERE start=end"
            new_predicate = new_predicate.replace(self.cyclic_symbol, "start")
            new_return = self.base_return.replace(self.cyclic_symbol, "start")
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = new_predicate,
                _return = new_return,
                _other = self.base_other
            )
            self.equivalent_queries.append(mutated_query)
            self.equivalent_queries_eval.append(self.Structure_GQT)


    def generate_equivalent_queries_adding_redundant_predicate_b(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if len(self.base_symbols)==0:
                return
            if "WHERE " in self.base_predicate:
                new_predicate = self.base_predicate + " AND id({})>=0.0".format(choice(self.base_symbols))
            else:
                new_predicate = "WHERE id({})>=0.0".format(choice(self.base_symbols))
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = self.base_path,
                _predicate = new_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            if len(self.mutated_symbols)==0:
                return
            if "WHERE " in self.mutated_predicate:
                self.mutated_predicate = self.mutated_predicate + " AND id({})>=0.0".format(choice(self.mutated_symbols))
            else:
                self.mutated_predicate = "WHERE id({})>=0.0".format(choice(self.mutated_symbols))
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Non_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([1, last_eval[1], last_eval[2]])

    def generate_equivalent_queries_adding_redundant_predicate_c(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            if len(self.base_symbols)==0:
                return
            if "WHERE " in self.base_predicate:
                new_predicate = self.base_predicate + " AND {sym}={sym}".format(sym=choice(self.base_symbols))
            else:
                new_predicate = "WHERE {sym}={sym}".format(sym=choice(self.base_symbols))
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = self.base_path,
                _predicate = new_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            if len(self.mutated_symbols)==0:
                return
            if "WHERE " in self.mutated_predicate:
                self.mutated_predicate = self.mutated_predicate + " AND {sym}={sym}".format(sym=choice(self.mutated_symbols))
            else:
                self.mutated_predicate = "WHERE {sym}={sym}".format(sym=choice(self.mutated_symbols))
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Non_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([1, last_eval[1], last_eval[2]])


    def generate_equivalent_symmetrical_queries(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            path = self.base_path.split('-')
            path.reverse()
            for i in range(len(path)):
                if path[i]!='':
                    if path[i][0]=='>' and path[i][-1]!='<':
                        path[i] = path[i][1:] + '<'
                    elif path[i][0]!='>' and path[i][-1]=='<':
                        path[i] = '>' + path[i][:-1]
            new_path = '-'.join(path)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = self.base_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            path = self.mutated_path.split('-')
            path.reverse()
            for i in range(len(path)):
                if path[i]!='':
                    if path[i][0]=='>' and path[i][-1]!='<':
                        path[i] = path[i][1:] + '<'
                    elif path[i][0]!='>' and path[i][-1]=='<':
                        path[i] = '>' + path[i][:-1]
            self.mutated_path = '-'.join(path)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Structure_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([last_eval[0], last_eval[1], 1])

    # # Mutation Rules:
    # def generate_equivalent_redundant_match_with(self):
    def generate_equivalent_adding_call_wrapper(self):
        return_alias = "deadbeef"
        new_return = "{} AS {}".format(self.base_return, return_alias)
        mutated_query = self.cypher_query_pattern.format(
                _match = self.base_match,
                _path = self.base_path,
                _predicate = self.base_predicate,
                _return = new_return,
                _other = self.base_other
            )
        mutated_query = "CALL {{ {} }} RETURN {}".format(mutated_query, return_alias)
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        self.equivalent_queries_eval.append(self.Non_GQT)

    # # Mutation Rules:
    def generate_equivalent_splitting_path(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            path = self.base_path
            split_sym_test = self.base_return.split('(')[1]
            if ' ' in split_sym_test:
                split_sym = split_sym_test.split(' ')[1].split(')')[0]
            else:
                split_sym = split_sym_test.split(')')[0]
            if split_sym not in self.base_node_symbols or split_sym==self.cyclic_symbol:
                return
            split_pattern = "{sym}{node_label}),({sym}"
            after_split = path.split(split_sym)
            node_label = after_split[1].split(')')[0] if after_split[1][0]==":" else ""
            new_split = split_pattern.format(sym=split_sym, node_label=node_label)
            new_path = after_split[0] + new_split + after_split[1]
            mutated_query = self.cypher_query_pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = self.base_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            path = self.mutated_path
            split_sym = self.mutated_return.split('(')[1].split(')')[0]
            if split_sym not in self.mutated_node_symbols or split_sym==self.cyclic_symbol:
                return
            split_pattern = "{sym}{node_label}),({sym}"
            after_split = path.split(split_sym)
            node_label = after_split[1].split(')')[0] if after_split[1][0]==":" else ""
            new_split = split_pattern.format(sym=split_sym, node_label=node_label)
            self.mutated_path = after_split[0] + new_split + after_split[1]
            mutated_query = self.cypher_query_pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Structure_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([last_eval[0], last_eval[1], 1])


    #Before: match (a:Movie)--(b) return a;
    #After: match (a)--(b) where a:Movie return a;
    def generate_equivalent_move_label_predicate(self, mode=0):
        if self.graph_pattern_mutation==0:
            return
        if mode==0:
            path = self.base_path
            node_label_list = re.findall("[(,\[][a-z]{{{}}}:[\w]*[),\]]".format(self.random_symbol_len), path)
            if len(node_label_list)==0:
                return
            target_label = choice(node_label_list)
            target = target_label.split(':')[0][1:]
            if self.cyclic_symbol in target:
                return
            label = target_label.split(':')[1][:-1]
            new_path = path.split(target_label)[0] + target_label.split(':')[0] + (')' if target_label[0]=='(' else ']') + path.split(target_label)[1]

            predicate = self.base_predicate
            if "WHERE " not in predicate:
                new_predicate = "WHERE {}:{}".format(target, label)
            else:
                new_predicate = "{} AND {}:{}".format(predicate, target, label)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.base_match,
                _path = new_path,
                _predicate = new_predicate,
                _return = self.base_return,
                _other = self.base_other
            )
        else:
            path = self.mutated_path
            node_label_list = re.findall("[(,\[][a-z]{{{}}}:[\w]*[),\]]".format(self.random_symbol_len), path)
            if len(node_label_list)==0:
                return
            target_label = choice(node_label_list)
            target = target_label.split(':')[0][1:]
            if self.cyclic_symbol in target:
                return
            label = target_label.split(':')[1][:-1]
            self.mutated_path = path.split(target_label)[0] + target_label.split(':')[0] + (')' if target_label[0]=='(' else ']') + path.split(target_label)[1]
            predicate = self.mutated_predicate
            if "WHERE " not in predicate:
                self.mutated_predicate = "WHERE {}:{}".format(target, label)
            else:
                self.mutated_predicate = "{} AND {}:{}".format(predicate, target, label)
            pattern = self.cypher_query_pattern
            mutated_query = pattern.format(
                _match = self.mutated_match,
                _path = self.mutated_path,
                _predicate = self.mutated_predicate,
                _return = self.mutated_return,
                _other = self.mutated_other
            )
        mutated_query = self.strip_spaces(mutated_query)
        self.equivalent_queries.append(mutated_query)
        if mode==0:
            self.equivalent_queries_eval.append(self.Property_GQT)
        else:
            last_eval = self.equivalent_queries_eval[-1]
            self.equivalent_queries_eval.append([last_eval[0], 1, last_eval[2]])

    def random_symbol_up(self):
        # Symbol length for self.random_symbol()
        return ''.join(choice(string.ascii_lowercase) for _ in range(32))

    # not combinable
    def generate_equivalent_rename_symbols_down(self):
        base_query = self.base_query
        count = 0
        targets = set(re.findall("[(,\[][a-z]{"+str(self.random_symbol_len)+"}", base_query))
        for each_sym in targets:
            base_query = re.sub(each_sym[1:], "s{}".format(count), base_query)
            count = count+1
        equivalent_query = base_query
        self.equivalent_queries.append(equivalent_query)
        self.equivalent_queries_eval.append(self.Non_GQT)

    def generate_equivalent_rename_symbols_up(self):
        base_query = self.base_query
        targets = set(re.findall("[(,\[][a-z]{"+str(self.random_symbol_len)+"}", base_query))
        for each_sym in targets:
            base_query = re.sub(each_sym[1:], self.random_symbol_up(), base_query)
        equivalent_query = base_query
        self.equivalent_queries.append(equivalent_query)
        self.equivalent_queries_eval.append(self.Non_GQT)
query_mutator.py
YuanchengJiang/GraphGenie
#!/usr/bin/env python3

import os
import time

from neo4j import GraphDatabase

# currently we only support neo4j

class SchemaScanner:
    node_count = 0
    edge_count = 0
    node_labels = []
    edge_labels = []
    node_properties = {}
    connectivity_matrix = []

    def __init__(self, ip, port, username, password):
        self.ip = ip
        self.port = port
        self.username = username
        self.password = password
        pass

    def scan(self):
        pass

    def print_schema_info(self):
        print("Node count: {}".format(self.node_count))
        print("Edge count: {}".format(self.edge_count))
        print("Node labels: {}".format(str(self.node_labels)))
        print("Edge labels: {}".format(str(self.edge_labels)))

    def print_connectivity(self):
        for i in self.node_labels:
            print(i[0], end=" ")
        print()
        for x in self.connectivity_matrix:
            for y in x:
                print(y, end=" ")
            print()

class Neo4jSchemaScanner(SchemaScanner):
    def neo4j_init(self):
        url = "bolt://{}:{}".format(self.ip, self.port)
        username = self.username
        password = self.password
        self.driver = GraphDatabase.driver(url, auth=(username, password))
        
    def scan(self):
        self.neo4j_init()
        get_node_count_query = "MATCH (n) RETURN count(n)"
        self.node_count = (list(self.execute_query(get_node_count_query)[0].values())[0])
        get_edge_count_query = "MATCH ()-[n]-() RETURN count(n)"
        self.edge_count = (list(self.execute_query(get_edge_count_query)[0].values())[0])
        get_node_labels_query = """
            MATCH (n)
            WITH DISTINCT labels(n) AS labels
            UNWIND labels AS label
            RETURN DISTINCT label
            ORDER BY label;
        """
        query_result = self.execute_query(get_node_labels_query)
        for i in query_result:
            self.node_labels.append(i['label'])
        get_edge_labels_query = """
            MATCH ()-[n]-()
            WITH DISTINCT type(n) AS labels
            UNWIND labels AS label
            RETURN DISTINCT label
            ORDER BY label;
        """
        query_result = self.execute_query(get_edge_labels_query)
        for i in query_result:
            self.edge_labels.append(i['label'])
        self.print_schema_info()
        self.scan_properties()
        self.scan_connectivity()
        return self.node_labels, self.edge_labels, self.node_properties, self.connectivity_matrix

    def scan_properties(self):
        get_properties_query = """
            MATCH (n:{})
            WITH DISTINCT(keys(n)) as key_sets
            UNWIND(key_sets) as keys
            RETURN DISTINCT(keys) as key;
        """
        for each_node_label in self.node_labels:
            properties = []
            query_result = self.execute_query(get_properties_query.format(each_node_label))
            for i in query_result:
                properties.append(i['key'])
            self.node_properties[each_node_label] = properties

    def scan_connectivity(self):
        for each_node_label in self.node_labels:
            matrix_row = []
            for each_test_node_label in self.node_labels:
                if each_node_label==each_test_node_label:
                    matrix_row.append(0)
                else:
                    test_query = "MATCH (a:{})-->(b:{}) RETURN count(a)".format(
                        each_node_label,
                        each_test_node_label
                    )
                    query_result = (list(self.execute_query(test_query)[0].values())[0])
                    matrix_row.append(0 if query_result==0 else 1)
            self.connectivity_matrix.append(matrix_row)
        self.print_connectivity()

    def execute_query(self, query):
        with self.driver.session() as session:
            query_result = session.execute_write(self._new_execute, query)
            return query_result

    @staticmethod
    def _new_execute(tx, query):
        query_result = -1
        query_execute = tx.run(query)
        query_data = query_execute.data()
        return query_data

schema_scanner.py
YuanchengJiang/GraphGenie