run_all_tests

#!/usr/bin/python

# Copyright 2013-present Barefoot Networks, Inc. 

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import subprocess
from subprocess import Popen
import os
import re
import sys
import time
import argparse
from functools import wraps
import errno
import signal

try:
    import termcolor as T
    _termcolor_available = True
except ImportError:
    _termcolor_available = False

parser = argparse.ArgumentParser(description='test arguments')
parser.add_argument('--jobs', '-j', type=int, required=False,
                    help='Number of jobs when running make')
parser.add_argument('--no-bmv2', action='store_true', default=False,
                    help='Disable BMv2 tests')
parser.add_argument('--no-bmv1', action='store_true', default=False,
                    help='Disable BMv1 tests')
parser.add_argument('--travis', action='store_true', default=False,
                    help='Skip some tests because of travis time limit')
parser.add_argument('--test-type', type=str, default='all',
                    help='Run specific test suite')


args = parser.parse_args()

_PD_TARGETS = [
    "basic_routing"
]

_THIS_DIR = os.path.dirname(os.path.realpath(__file__))

sw_binaries = ["behavioral-model"]
# bmv2_exes = set(["lt-simple_switch"])
bmv2_exes = set()

_TEST_RESULTS = []

def cleanup():
    for binary_path in sw_binaries:
        binary = os.path.basename(os.path.normpath(binary_path))
        cmd = ['sudo', 'killall', '-q', binary]
        print ' '.join(cmd)
        subprocess.call(cmd)

def bmv2_cleanup():
    for binary_path in bmv2_exes:
        binary = os.path.basename(os.path.normpath(binary_path))
        cmd = ['sudo', 'killall', '-q', binary]
        print ' '.join(cmd)
        subprocess.call(cmd)

def killall(name):
    cmd = ['sudo', 'killall', '-q', name]
    print ' '.join(cmd)
    subprocess.call(cmd)

def my_print(msg, color=None):
    if _termcolor_available:
        print T.colored(msg, color)
    else:
        print msg

def print_success(test_name):
    my_print(test_name + " : " + "PASSED", "green")

def print_failure(test_name):
    my_print(test_name + " : " + "FAILED", "red")

def one_test(target, test_script="./run_tests.py",
             test_subdir="ptf-tests", do_make_clean=True,
             make_tgt=None, make_args="", extra_args=""):
    cleanup()
    target_dir = os.path.join(_THIS_DIR, "targets", target)
    save_cwd = os.getcwd()
    os.chdir(target_dir)

    test_name = target
    if test_subdir != "":
        test_name += "--" + test_subdir.replace('/', '_')

    my_uniteresting_user_uid = int(os.environ['SUDO_UID'])

    os.seteuid(my_uniteresting_user_uid)

    if do_make_clean:
        cmd = ['make', 'clean']
        print ' '.join(cmd)
        subprocess.call(cmd)

    if make_tgt:
        cmd = ['make', make_tgt]
    else:
        cmd = ['make']
    if args.jobs is not None and args.jobs > 0:
        cmd.append("-j" + str(args.jobs))

    if make_args:
        cmd.append(make_args)

    print ' '.join(cmd)
    subprocess.call(cmd)

    os.seteuid(0)

    cmd = ['sudo', './behavioral-model']
    print ' '.join(cmd)
    logfile = "/tmp/" + test_name + ".log"
    logfd = open(logfile, 'w')
    bm_proc = Popen(cmd, stdout = logfd, stderr = subprocess.STDOUT)

    time.sleep(5)

    if target == "switch":
        test_dir = os.path.join("..", "..", "submodules", "switch",
                                "tests", test_subdir)
    else:
        test_dir = os.path.join("tests", test_subdir)
    cmd = ['sudo', test_script, '--test-dir', test_dir]
    if extra_args:
        cmd.append(extra_args)
    print ' '.join(cmd)
    returncode = subprocess.call(cmd)

    bm_proc.kill()

    os.chdir(save_cwd)

    if(returncode == 0):
        print_success(test_name)
    else:
        print_failure(test_name)
    _TEST_RESULTS.append( (test_name, returncode) )

class TimeoutError(Exception):
    pass

def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
    def decorator(func):
        def _handle_timeout(signum, frame):
            raise TimeoutError(error_message)

        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, _handle_timeout)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result

        return wraps(func)(wrapper)

    return decorator

def bmv2_one_test(target, test_script="./run_tests.py",
                  test_subdir="ptf-tests", do_make_clean=True,
                  make_tgt=None, run_bmv2="run_bm.sh", drivers_exe="drivers",
                  test_name=None):
    do_make_clean = False
    bmv2_cleanup()
    target_dir = os.path.join(_THIS_DIR, "targets", target, "bmv2")
    save_cwd = os.getcwd()
    os.chdir(target_dir)

    if test_name is None:
        test_name = "bmv2--" + target
        if test_subdir != "":
            test_name += "--" + test_subdir.replace('/', '_')

    user = os.environ['SUDO_USER']

    if do_make_clean:
        cmd = ["su", user, "-c", "make clean"]
        print ' '.join(cmd)
        subprocess.call(cmd)

    make_cmd = "make"
    if make_tgt:
        make_cmd += " " + make_tgt
    if args.jobs is not None and args.jobs > 0:
        make_cmd += " -j" + str(args.jobs)
    cmd = ["su", user, "-c", make_cmd]

    print ' '.join(cmd)
    subprocess.call(cmd)

    bmv2_exes.add(target + "_bmv2")
    cmd = ['bash', run_bmv2]
    print ' '.join(cmd)
    logfile = "/tmp/" + test_name + ".log"
    logfd = open(logfile, 'w')
    bm_proc = Popen(cmd, stdout = logfd, stderr = subprocess.STDOUT,
                    preexec_fn=os.setsid)

    time.sleep(5)

    bmv2_exes.add(drivers_exe)
    cmd = ['./' + drivers_exe]
    print ' '.join(cmd)
    logfile = "/tmp/" + test_name + "-drivers" + ".log"
    logfd = open(logfile, 'w')
    drivers_proc = Popen(cmd, stdout = logfd, stderr = subprocess.STDOUT,
                         preexec_fn=os.setsid)

    time.sleep(5)

    os.chdir("..")

    # TODO: Sometimes the test hangs but I haven't figured out why yet
    # I consider it is a success. I need to take care of this.
    @timeout(600)
    def run_that_test():
        if target == "switch":
            test_dir = os.path.join("..", "..", "submodules", "switch",
                                    "tests", test_subdir)
        else:
            test_dir = os.path.join("tests", test_subdir)
        cmd = [test_script, '--test-dir', test_dir]
        print ' '.join(cmd)
        return subprocess.call(cmd)

    try:
        returncode = run_that_test()
    except TimeoutError:
        my_print("test " + test_name + " timed out, counting it as success",
                 "red")
        returncode = 0

    # bm_proc.kill()
    # drivers_proc.kill()
    # kill() not enough, we need to kill all children as well
    os.killpg(os.getpgid(bm_proc.pid), signal.SIGTERM)
    os.killpg(os.getpgid(drivers_proc.pid), signal.SIGTERM)

    os.chdir(save_cwd)

    if(returncode == 0):
        print_success(test_name)
    else:
        print_failure(test_name)
    _TEST_RESULTS.append( (test_name, returncode) )

def pd_tests():
    for target in _PD_TARGETS:
        one_test(target)

    one_test("l2_switch")
    if args.travis: return
    one_test("switch", test_subdir=os.path.join("ptf-tests", "pd-tests"))

def bmv2_pd_tests():
    os.environ["BMV2_TEST"] = "1"
    bmv2_one_test("basic_routing")
    # Temporarily disabled (nanomsg socket is hanging)
    #
    # NN test
    # does not need to run as root, but
    # - easier to reuse code if I run it as root
    # - may get a conflict with some PTF output files, created as sudo earlier
    # bmv2_one_test("basic_routing", run_bmv2="run_bm_nn.sh",
    #               test_script="./run_tests_nn.py",
    #               do_make_clean=False, test_name="basic_routing-nn")
    bmv2_one_test("l2_switch", do_make_clean=False)
    bmv2_one_test("switch", test_subdir=os.path.join("ptf-tests", "pd-tests"),
                  do_make_clean=False)
    del os.environ["BMV2_TEST"]

def switchapi_test():
    if args.travis: return
    one_test("switch", test_subdir=os.path.join("ptf-tests", "api-tests"),
             do_make_clean=False, make_tgt="bm-switchapi")

def bmv2_switchapi_test():
    os.environ["BMV2_TEST"]="1"
    bmv2_one_test("switch", test_subdir=os.path.join("ptf-tests", "api-tests"),
                  do_make_clean=False, make_tgt="bm-switchapi",
                  drivers_exe="drivers-switchapi")
    del os.environ["BMV2_TEST"]

def switchsai_test():
    if args.travis: return
    one_test("switch", test_subdir=os.path.join("ptf-tests", "sai-tests"),
             do_make_clean=False, make_tgt="bm-switchsai")

def sai_p4_test():
    one_test("sai_p4", test_subdir=os.path.join("ptf-tests", "sai_thrift"),
             do_make_clean=True)

def bmv2_switchsai_test():
    os.environ["BMV2_TEST"] = "1"
    bmv2_one_test("switch", test_subdir=os.path.join("ptf-tests", "sai-tests"),
                  do_make_clean=False, make_tgt="bm-switchsai",
                  drivers_exe="drivers-switchsai")
    del os.environ["BMV2_TEST"]

def openflow_test():
    one_test("l2_switch",
             test_script="./run_of_tests.py", test_subdir="of-tests",
             make_tgt="bm-p4ofagent",
             make_args="PLUGIN_OPENFLOW=1")
    # I have to make clean here, otherwise it fails...
    # and takes too long for travis
    if args.travis: return
    one_test("switch", make_args="PLUGIN_OPENFLOW=1",
             test_script="./run_of_tests.py", test_subdir="of-tests",
             make_tgt="bm-p4ofagent")

def check_ifaces():
    ifconfig_out = subprocess.check_output(['ifconfig'])
    iface_list = re.findall(
        r'^(\S+)', ifconfig_out, re.S | re.M
    )
    ifaces = set(iface_list)
    NUM_VETH_PAIRS = 9
    for i in xrange(NUM_VETH_PAIRS * 2):
        iface_name = "veth" + str(i)
        if iface_name not in ifaces:
            my_print("missing veth interface: " + iface_name, "red")
            my_print("Did you remember to run ./tools/veth_setup.sh ?", "red")
            sys.exit(1)

    my_print("All required veth interfaces are present", "green")

    my_print("If some tests fail because extra packets are received, " +\
             "you may want to try disabling IPv6 with:\n" +\
             "sudo sysctl -w net.ipv6.conf.eth0.disable_ipv6=1",
             "red")

def ifaces_up_down():
    target_dir = os.path.join(_THIS_DIR, "targets", "switch", "bmv2")
    save_cwd = os.getcwd()
    os.chdir(target_dir)
    cmd = ['make', 'clean']
    print ' '.join(cmd)
    subprocess.call(cmd)
    os.chdir(save_cwd)

    cmd = ['sudo', './tools/veth_teardown.sh']
    print ' '.join(cmd)
    subprocess.call(cmd)

    cmd = ['sudo', './tools/veth_setup.sh']
    print ' '.join(cmd)
    subprocess.call(cmd)

    cmd = ['sudo', './tools/veth_disable_ipv6.sh']
    print ' '.join(cmd)
    subprocess.call(cmd)

def main():
    if(os.getuid() != 0):
        my_print("I cannot run as a mortal, I need to be root, rrrOOOOOOTTTTT",
                 "red")
        sys.exit(1)

    check_ifaces()

    if not args.no_bmv2:
        if args.test_type == 'pd' or args.test_type == 'all':
            bmv2_pd_tests()
        if args.test_type == 'api' or args.test_type == 'all':
            bmv2_switchapi_test()
        if args.test_type == 'sai' or args.test_type == 'all':
            bmv2_switchsai_test()
        if args.test_type == 'intf' or args.test_type == 'all':
            ifaces_up_down()

    if not args.no_bmv1:
        if args.test_type == 'pd' or args.test_type == 'all':
            pd_tests()
        if args.test_type == 'api' or args.test_type == 'all':
            switchapi_test()
        if args.test_type == 'sai' or args.test_type == 'all':
            switchsai_test()
        if args.test_type == 'oflow' or args.test_type == 'all':
            openflow_test()
        if args.test_type == 'saip4' or args.test_type == 'all':
            sai_p4_test()

    print "**************************************"

    passed_count, total_count = 0, 0

    for test_name, returncode in _TEST_RESULTS:
        total_count += 1
        if(returncode == 0):
            print_success(test_name)
            passed_count += 1
        else:
            logfile = "/tmp/" + test_name + ".log"
            print_failure(test_name)
            my_print("take a look at the BM log file: " + logfile, "red")

    print
    print
    print "**************************************"

    my_print("TESTS PASSED: " + str(passed_count) + "/" + str(total_count))

    cleanup()
    bmv2_cleanup()

    if passed_count == total_count:
        my_print("ALL TESTS PASSED!", "green")
        sys.exit(0)
    else:
        my_print(str(total_count - passed_count) + " TESTS FAILED!", "red")

        my_print("If some tests fail because extra packets are received, " +\
                 "you may want to try disabling IPv6 with:\n" +\
                 "sudo sysctl -w net.ipv6.conf.eth0.disable_ipv6=1",
                 "red")
        sys.exit(1)

if __name__ == '__main__':
    main()