test_conda

#!/usr/bin/env python

from __future__ import print_function

import sys
import os
import argparse
import importlib
import subprocess as sb
import yaml
import json
import platform
import random

from GetImportBinLibs import getBins, getLibs, getImports

os.environ['PYTHON_VERSION'] = '%d.%d' % (sys.version_info.major, sys.version_info.minor)

############# python3 porting issues ##############
# http://python3porting.com/differences.html
try:
    input = raw_input
except NameError:
    pass

if sys.version_info < (3,):
    def b(x):
        return x
else:
    import codecs
    def b(x):
        return codecs.latin_1_encode(x)[0]
############### python 3 porting #####################

TESTDB = os.path.join(os.path.split(__file__)[0], "testdb.yaml")

description='''
driver to test a conda installation.
'''

epilog='''There are four tests to do:
basic - this is the recommended option to give. It reads the testdb yaml file (defaults to %s)
        and does all the python imports, runs the bins with -h, and loads the .so files, that are
        listed. Before running this, if
        you are testing an environment with new packages, do --dryrun  to see what the
        new imports are, then add them to the testdb yaml file.
pkgs - to do more than imports, give a command like --pkgs psana or --pkgs numpy,scipy. To
       test all pkgs, to --pkgs all

The update command first reads the testdb file, and queries you about all the new imports and bins
that it finds.
''' % TESTDB


parser = argparse.ArgumentParser(description=description,
                                 formatter_class=argparse.RawDescriptionHelpFormatter,
                                 epilog=epilog)

parser.add_argument('--pkglist', type=str, help="comma separated list of packages to test, or all", default='hdf5,openmpi,mpi4py,h5py,psana-conda,numpy')
parser.add_argument('--testdb', type=str, help="supply alternate yaml file for import/bin/lib tests. default=%s" % TESTDB, default=TESTDB)
parser.add_argument('--dryrun', action='store_true', help="dry run, do not execute tests", default=False)
parser.add_argument('--sep', action='store_true', help="separate python process for each import. Slower, but more robust. Default is False", default=False)
parser.add_argument('--default', action='store_true', help="do default testing (bins, imports, libs and default pkglist", default=False)
parser.add_argument('--soft', action='store_true', help="soft test, only test imports/bins/libs/pkgs that are in environment. Default fail is test not present.")
parser.add_argument('--interactive', action='store_true', help="interactive mode, stops after a test fails", default=False)
parser.add_argument('--bins', action='store_true', help="test the bins", default=False)
parser.add_argument('--libs', action='store_true', help="test the libs", default=False)
parser.add_argument('--imports', action='store_true', help="test the imports", default=False)
parser.add_argument('--pkgs', action='store_true', help='test packages', default=False)
parser.add_argument('--verbose', action='store_true', help="verbose testing", default=False)

def externalCommandTest(hdr, cmd, verbose, dryrun, interactive):
    if verbose or interactive or dryrun:
        print("------ %s: %s -------" % (hdr, cmd))
    if dryrun: return 'NOT-PRESENT'
    print(cmd)
    sys.stdout.flush()
    if verbose or interactive:
        retcode = os.system(cmd)
    else:
        p = sb.Popen(cmd, shell=True, stdout=sb.PIPE, stderr=sb.PIPE, universal_newlines=True)
        out, err = p.communicate()
        retcode = p.returncode
    retmsg = 'PASS'
    if retcode != 0: retmsg = 'FAIL'
    if interactive:
        raw_input("hit enter")
    if retcode == 0:
        return "PASS"
    return "FAIL"

def binTest(tst, verbose, dryrun, interactive):
    cmd = '%s -h' % tst
    return externalCommandTest("binTest", cmd, verbose, dryrun, interactive)

def importTest(tst, verbose, dryrun, interactive, sep=False):
    if sep:
        cmd = 'python -c "import %s"' % tst
        return externalCommandTest("importTest", cmd, verbose, dryrun, interactive)
        
    ret = 'NOT-PRESENT'
    if verbose or interactive or dryrun:
        print("----- import %s" % tst)
        sys.stdout.flush()
    if dryrun:
        return ret
    try:
        importlib.import_module(tst)
    except Exception as exp:
        print("FAIL import %s exception: %r" % (tst, exp))
        ret = 'FAIL'
    if interactive:
        raw_input("hit enter")
    return 'PASS'
    

def libTest(tst, verbose, dryrun, interactive):
    basedir=os.path.split(__file__)[0]
    loadlib = os.path.join(basedir, 'loadlib')
    assert os.path.exists(loadlib), "lib testing program not found: %s" % loadlib
    cmd = '%s %s' % (loadlib, tst)
    return externalCommandTest("libTest", cmd, verbose, dryrun, interactive)

def loadTests(testdb_fname, verbose):
    assert os.path.exists(testdb_fname), "testdb file: %s not found" % testdb_fname
    testdb = yaml.load(open(testdb_fname,'r'))
    for nm in ['imports', 'bins', 'libs']:
        assert nm in testdb, "testdb yaml file is improperly formed, %s is missing" % nm
        nmDict = testdb[nm]
        for fld in ['do','skip']:
            assert fld in nmDict, "testdb yaml file is improperly formed, %s is missing from the %s section" % \
                (fld, nm)
    if verbose:
        print("Loaded %s" % testdb_fname)
        sys.stdout.flush()
    return testdb

def partition_matches(gl, candlist):
    assert len(gl.split('*'))==2, "gl=%s does not have exactly one *" % gl
    before, after = gl.split('*')
    if after:
        after_match = [el for el in candlist if el.endswith(after)]
    else:
        after_match = [el for el in candlist]
    if before:
        matches = [el for el in after_match if el.startswith(before)]
    else:
        matches = [el for el in candlist]
    filtered_candlist = [el for el in candlist if el not in matches]
    return matches, filtered_candlist

def remove_tests(globs, normal, tests):
    normal = set(normal)
    tests = [tst for tst in tests if tst not in normal]
    for gl in globs:
        matches, tests = partition_matches(gl, tests)
    return tests

def identify_tests_todo(ttype, testdict, candidate_tests):
    # candidate_tests - what is available in the environment,
    # first throw out tests we don't want to do, if they are in this environment
    if ttype == 'lib': assert 'libxtcrunset.so' in candidate_tests, "fail 1"
    skip_list = [el for el in testdict['skip'] if el]
    skip_globs = [el for el in skip_list if el.find('*')>=0]
    skip_normal = set([el for el in skip_list if el.find('*')<0])
    candidate_tests = remove_tests(skip_globs, skip_normal, candidate_tests)
    if ttype == 'lib': assert 'libxtcrunset.so' in candidate_tests, "fail 2"

    # but together the tests that we want todo, keep track of tests we want to
    # do that are not in the environment
    tests_not_in_env = []
    todo = []
    for el in testdict['do']:
        if el.find('*')<0:
            if el in candidate_tests:
                todo.append(el)
                candidate_tests.remove(el)
            else:
                tests_not_in_env.append(el)
        else:
            glob_matches, candidate_tests = partition_matches(el, candidate_tests)
            # candidate_tests is now filtered, glob matches removed
            if len(glob_matches)==0:
                tests_not_in_env.append(el)
            else:
                todo.extend(glob_matches)

    tests_in_env_but_not_referenced_in_testdb = candidate_tests
    return todo, tests_not_in_env, tests_in_env_but_not_referenced_in_testdb 
    
def basicTest(testdb, bins, imports, libs, verbose, interactive, doBins, doImports, doLibs, dryrun, soft, sep):
    all_success = True
    for ttype, tlist in zip(['bins','imports','libs'],
                            [bins, imports, libs]):

        if ttype == 'bins' and (not doBins): continue
        if ttype == 'imports' and (not doImports): continue
        if ttype == 'libs' and (not doLibs): continue

        tests_todo, tests_not_in_env, tests_not_in_db  = identify_tests_todo(ttype, testdb[ttype], tlist)
        if not soft:
            assert 0==len(tests_not_in_env), "ERROR: for test='%s' this environment is missing the following tests: %s" % (ttype, '\n  '.join(tests_not_in_env))
            
        if len(tests_not_in_db)>0:
            print("%s: %d items not in testdb, use verbose flag to see them" % (ttype, len(tests_not_in_db)))
            if verbose:
                tests_not_in_db.sort()
                print("  %s" % '  \n'.join(tests_not_in_db))
                sys.stdout.flush()
        results = {'PASS':[], 'FAIL':[], 'NOT-PRESENT':[]}
        print("%s: about to execute %d tests" % (ttype, len(tests_todo)))
        sys.stdout.flush()
        for tst in tests_todo:
            if ttype == 'bins':
                results[binTest(tst, verbose, dryrun, interactive)].append(tst)
            elif ttype == 'imports':
                results[importTest(tst, verbose, dryrun, interactive, sep)].append(tst)
            elif ttype == 'libs':
                results[libTest(tst, verbose, dryrun, interactive)].append(tst)

        summary = "%s:" % ttype
        for ky,val in results.items():
            summary += " %s-%d" % (ky, len(val))
        print(summary)
        sys.stdout.flush()
        assert len(results['FAIL'])==0, "besicTest: %s has failures: %s" % (ttype, '  \n'.join(results['FAIL']))

#def batchTest(testdb, verbose):
#    pass

#def perfTest(testdb, verbose):
#    pass

def getAvailPkgs():
    cmd = 'conda list --json'
    p = sb.Popen(cmd, shell=True, stdout=sb.PIPE, stderr=sb.PIPE,  universal_newlines=True)
    stdout, stderr = p.communicate()
    assert p.returncode==0, "cmd: %s returncode !=0\n===stdout:\n%s\n===stderr:\n%s" % (cmd, stdout, stderr)
    pkg_dicts = json.loads(stdout)
    pkgs = set()
    for pkg_dict in pkg_dicts:
        pkg_name = pkg_dict['name']
        if pkg_name in pkgs:
            sys.stderr.write("WARNING: pkg %s appears more than once in the environment\n" % pkg_name)
        pkgs.add(pkg_dict['name'])
    return pkgs

def pkgsTest(testdb, pkgs, verbose, dryrun, soft, interactive):
    if len(pkgs)==1 and pkgs[0]=='all':
        pkgs = testdb['pkgs'].keys()
    else:
        # an explicit list of packages is given, don't do soft
        if soft:
            print("pkgsTest: explicit list of packages has been given, turning off soft for pkgsTest")
            sys.stdout.flush()
            soft = False
    pkgs = list(pkgs)
    pkgs.sort()
    avail_pkgs = set(getAvailPkgs())
    os.chdir(os.path.split(__file__)[0])  # testdb has commands like python pkgs/xx that only run relative to current directory
    for pkg in pkgs:
        # below, conda will not show up in the environment package list, even though it is available
        if (pkg != 'conda') and not (pkg in avail_pkgs):
            if not soft:
                assert pkg in avail_pkgs, "testPkgs: pkg=%s is not in avail_pkgs=%s" % (pkg, '\n  '.join(avail_pkgs))
            print("pkgTest: skipping %s" % pkg)
            sys.stdout.flush()
            continue
        print("Testing pkg=%s" % pkg)
        sys.stdout.flush()
        results = {'PASS':[], 'FAIL':[], 'NOT-PRESENT':[]}
        pkgTestDb = testdb['pkgs'][pkg]
        commands = pkgTestDb['commands']
        if 'skiphost' in pkgTestDb:
            skiphost = pkgTestDb['skiphost']
            node = platform.node()
            if node == skiphost:
                commands = []
                print("pkg=%s skiphost=%s match - not testing" % (pkg, node))
                sys.stdout.flush()
            else:
                print("pkg=%s skiphost=%s not a match (node=%s) testing" % (pkg, skiphost, node))
                sys.stdout.flush()
        for cmd in commands:
            results[externalCommandTest('pkg=%s' % pkg, cmd, True, dryrun, interactive)].append(cmd)

        summary = "%s:" % pkg
        for ky,val in results.items():
            summary += " %s-%d" % (ky, len(val))
        print(summary)
        sys.stdout.flush()
        assert len(results['FAIL'])==0, "pkgsTest: %s has failures: %s" % (pkg, '  \n'.join(results['FAIL']))

if __name__ == '__main__':
    args = parser.parse_args()
    if args.default:
        args.imports=True
        args.pkgs=True
        args.bins=True
        args.libs=True
    assert any([args.imports, args.pkgs, args.bins, args.libs]), "Must specify at least one of --default --bins --imports --libs --pkgs. give -h option for help"
    if args.dryrun: args.verbose=True

    testdb = loadTests(args.testdb, args.verbose)

    if args.imports or args.bins or args.libs:
        imports = getImports(args.verbose)
        bins = getBins(args.verbose)
        libs = getLibs(args.verbose)
        basicTest(testdb=testdb,
                  bins=bins,
                  imports=imports,
                  libs=libs,
                  verbose=args.verbose,
                  interactive=args.interactive,
                  doBins=args.bins,
                  doImports=args.imports,
                  doLibs=args.libs,
                  dryrun=args.dryrun,
                  soft=args.soft,
                  sep=args.sep)

    if args.pkgs:
        pkgsTest(testdb=testdb,
                 pkgs=args.pkglist.split(','),
                 verbose=args.verbose,
                 dryrun=args.dryrun,
                 soft=args.soft,
                 interactive=args.interactive)