run-database-tests

#!/usr/bin/env python3

import argparse
import json
import os
import re
import shutil
import subprocess
import sys

parser = argparse.ArgumentParser()
parser.add_argument("working_dir", type=str,
                    help="Working directory of tests")
parser.add_argument("-c","--config_file", type=str,
                    help="Test configuration file (default: testsuite.json)",
                    default="testsuite.json")
parser.add_argument("-d", "--database", type=str,choices=["all","mysql8","pgsql","sqlite3"],
                    help="Run tests for specific database (all, mysql8, pgsql, sqlite3).  \"all\" selects all databases listed in the testsuite configuration file. Default: \"pgsql\"",
                    default="pgsql")
parser.add_argument("-l", "--local", action="store_true",
                    help="Test local database configuration (<working_dir>/config) if present; otherwise, the database(s) specified by the --database parameter are tested.",
                    default=False)
parser.add_argument("-v", "--verbose", action="store_true",
                    help="Print verbose debugging messages",
                    default=False)
args = parser.parse_args()

def main():
    working_dir = args.working_dir
    config_file = args.config_file
    verbose = args.verbose

    links_dir = os.getcwd()
    links_exec = os.path.join(links_dir,"links")

    def debug(s):
        if verbose:
            print(s)

    def links_cmd(config,file):
        setdebug = "--set=debug=true" if verbose else ""
        debug("%s --config=%s %s %s" % (links_exec,config,setdebug,file))
        if verbose:
            completed_process = subprocess.run([links_exec, "--config=%s" % config, setdebug, file])
        else:
            completed_process = subprocess.run([links_exec, "--config=%s" % config, setdebug, file],
                                               stdout=subprocess.PIPE,stderr=subprocess.PIPE)
        return completed_process.returncode

    matcher = re.compile("database_args=([^:]*):([^:]*):([^:]*):([^\n:]*)")
    def parse_db_config(config):
        debug("Parsing %s" % config)
        with open(config) as fp:
            for line in fp:
                m = matcher.match(line)
                if m != None:
                    return ( os.path.expandvars(m.group(1))
                           , os.path.expandvars(m.group(2))
                           , os.path.expandvars(m.group(3))
                           , os.path.expandvars(m.group(4)) )
        return (None,None,None,None)

    driver_matcher = re.compile("database_driver=([^\n]*)")
    def parse_db_driver(config):
        debug("Parsing %s" % config)
        with open(config) as fp:
            for line in fp:
                m = driver_matcher.match(line)
                if m != None:
                    if m.group(1) == "postgresql":
                        return ("pgsql")
                    else:
                        return (m.group(1))
        return None

    def prepare(db,db_config,setup):
        (host, port, user, passwd) = db_config

        # if a specialized setup file exists for this database, use it
        sql_file = "%s.%s" % (setup,db)
        # otherwise, use the plain sql one
        if not(os.path.exists("%s.%s" % (setup,db))):
            sql_file = "%s.sql" % setup

        debug("setting up %s with %s" % (db,sql_file))
        if db == 'pgsql':
            cmnd = "PGPASSWORD=%s psql -h %s -p %s %s -v ON_ERROR_STOP=1 -q -d links -f %s" % (passwd, host, port, user, sql_file)
        elif db == 'mysql8':
            cmnd = "mysql links -h %s -P %s -u %s -p%s  < %s" % (host, port, user, passwd, sql_file)
        elif db == 'sqlite3':
            cmnd = "sqlite3 links < %s" % sql_file
        else:
            raise SystemExit("Unsupported database %s" % db)
        debug(cmnd)
        returncode = os.system(cmnd)
        if returncode != 0:
            raise SystemExit("Database setup failed: %s for database %s" % (sql_file,db))

    def run(db,test):
        debug("running %s %s" % (db,test))
        db_config = "config." + db
        test_file = test + ".links"
        return_code = links_cmd(db_config,test_file)
        if return_code != 0:
            print("!FAILURE: Test %s on database %s" % (test, db))
            return 1
        else:
            print(" SUCCESS: Test %s on database %s" % (test, db))
            return 0


    def run_tests(db,params):
        failures = 0
        print("%s: Setting up %s" % (working_dir,db))
        for setup in config['setups']:
            prepare(db,params,setup)
        print("%s: Testing %s" % (working_dir,db))
        for test in config['tests']:
            if run(db,test) > 0:
                failures = failures + 1
        print("%s: %d failures for %s" % (working_dir, failures, db))
        return failures > 0

    os.chdir(working_dir)

    with open(config_file) as fp:
        config = json.load(fp)

    debug(config)

    if args.database == "all":
        db_configs = [(db, parse_db_config("config.%s" % db)) for db in config['databases']]
    else:
        db_configs = [(args.database, parse_db_config("config.%s" % args.database))]

    # but override and use local configuration if present
    if args.local and os.path.exists("config"):
        print("Local configuration file found; ignoring --database parameter")
        # parse the config file to work out what database we are using
        db_configs = [(parse_db_driver("config"), parse_db_config("config"))]
    else:
        print("Local configuration file not found; testing databases specified by --database parameter")


    failed = False
    for (db, params) in db_configs:
        failed = failed or run_tests(db, params)

    if failed:
        exit(1)
    else:
        exit(0)

if __name__ == '__main__':
    main()