updates_to_run.py

def run_tests_in_venv(test="", socket_dir=""):
    """Runs tests in the virtual environment set by venv_dir.

    Arguments:
    test: Name of the test to run (default: run all tests)
    socket_dir: Use running VPP's socket files
                (default: run a new VPP instance )
    """
    script = os.path.join(test_dir, "scripts", "run.sh")
    p = Popen(
        [script,
        f"--venv-dir={venv_dir}",
        f"--vpp-ws-dir={ws_root}",
        f"--use-running-vpp={socket_dir}",
        f"--filter={test}"],
        stdout=PIPE,
        stderr=STDOUT,
        cwd=ws_root,
    )
    show_progress(p.stdout)


if __name__ == "__main__":
    # Build a Virtual Environment for running tests on host & QEMU
    #(TODO): Create a single config object by merging the below args with existing
    # config.py
    parser = argparse.ArgumentParser(
        description="Run VPP Unit Tests", formatter_class=argparse.RawTextHelpFormatter
    )
    parser.add_argument(
        "--vm",
        dest="vm",
        required=False,
        action="store_true",
        help="Run Test Inside a QEMU VM",
    )
    parser.add_argument(
        "-d",
        "--debug",
        dest="debug",
        required=False,
        default=True,
        action="store_true",
        help="Run Tests on Debug Build",
    )
    parser.add_argument(
        "-r",
        "--release",
        dest="release",
        required=False,
        default=False,
        action="store_true",
        help="Run Tests on release Build",
    )
    parser.add_argument(
        "--test",
        dest="test_name",
        required=False,
        action="store",
        default="",
        help="Test Name or Test filter",
    )
    parser.add_argument(
        "--vm-kernel-image",
        dest="kernel_image",
        required=False,
        action="store",
        default="",
        help="Kernel Image Selection to Boot",
    )
    parser.add_argument(
        "--vm-cpu-list",
        dest="vm_cpu_list",
        required=False,
        action="store",
        default="5-8",
        help="Set CPU Affinity\n"
        "E.g. 5-7,10 will schedule on processors "
        "#5, #6, #7 and #10. (Default: 5-8)",
    )
    parser.add_argument(
        "--vm-mem",
        dest="vm_mem",
        required=False,
        action="store",
        default="2",
        help="Guest Memory in Gibibytes\n" "E.g. 4 (Default: 2)",
    )
    parser.add_argument(
        "--use-running-vpp",
        dest="socket_dir",
        required=False,
        action="store",
        default="",
        help="Run tests against a running VPP.\n"
        "Specify the path to running VPP's Socket Directory.\n"
        "Path can be relative or absolute. Default is cwd.\n"
        "The directory must contain the two socket files - "
        "named api.sock & stats.sock"
    )
    args = parser.parse_args()
    vm_tests = False
    # Enable VM tests
    if args.vm and args.test_name:
        test_data_dir = "/tmp/vpp-vm-tests"
        set_logging(test_data_dir, args.test_name)
        vm_tests = True
    elif args.vm and not args.test_name:
        print("Error: The --test argument must be set for running VM tests")
        sys.exit(1)
    build_venv()
    # Build VPP release or debug binaries
    debug = False if args.release else True
    build_vpp(debug, args.release)
    set_environ()
    # Run tests against an existing VPP 
    if os.path.isdir(args.socket_dir):
        print("Tests will be run against a running VPP..")
        run_tests_in_venv(args.test_name, args.socket_dir)
    # Run tests against a new instance of VPP
    elif not vm_tests:
        print("Tests will be run by spawning a new VPP instance..")
        run_tests_in_venv(args.test_name)
    # Run tests inside a VM
    elif vm_tests:
        print("Running VPP unit test(s):{0} inside a QEMU VM".format(args.test_name))
        # Check Available CPUs & Usable Memory
        cpus = expand_mix_string(args.vm_cpu_list)
        num_cpus, usable_cpus = (len(cpus.split(",")), len(os.sched_getaffinity(0)))
        if num_cpus > usable_cpus:
            print(f"Error:# of CPUs:{num_cpus} > Avail CPUs:{usable_cpus}")
            sys.exit(1)
        avail_mem = int(os.popen("free -t -g").readlines()[-1].split()[-1])
        if int(args.vm_mem) > avail_mem:
            print(f"Error: Mem Size:{args.vm_mem}G > Avail Mem:{avail_mem}G")
            sys.exit(1)
        vm_test_runner(
            args.test_name, args.kernel_image, test_data_dir, cpus, f"{args.vm_mem}G"
        )