path: root/benchmarking/test-suite/testsweet.py

#!/bin/python3
# -*- coding: utf-8 -*-

import datetime
import os
import subprocess
import sys
import tarfile
import warnings
from abc import abstractmethod
from urllib.request import urlretrieve

CACHEDIR = "installCache"
STATSDIR = "stats"
VERSION  = "0.1.0"
DATE_FORMAT = "%Y-%m-%d_%H-%M-%S"

# Glances is not included in the benchmarking suite as it can be installed with pip
# using requirements.txt
SOURCES = {
    "7z": {
        "default_iterations": 5,
        "description": "Built-in 7zip benchmark",
        "linux_bin": "7z",
        "windows_bin": "7z",
        "linux_url": "https://www.7-zip.org/a/7z2301-linux-x64.tar.xz",
        "windows_url": "https://www.7-zip.org/a/7z2301-x64.exe",
        "linux_flags": "b",
        "windows_flags": "b",
    },
    "geekbench": {
        "default_iterations": 1,
        "description": "Geekbench 6 CPU benchmark",
        "linux_bin": "geekbench_x86_64",
        "windows_bin": "geekbench_x86_64.exe",
        "linux_url": "https://cdn.geekbench.com/Geekbench-6.2.1-Linux.tar.gz",
        "windows_url": "https://cdn.geekbench.com/Geekbench-6.2.1-WindowsSetup.exe",
        "linux_flags": "--cpu",
        "windows_flags": "--cpu",
    },
    "baseline": {
        "default_iterations": 60,
        "description": "Monitors the system without running any tests",
        "linux_bin":     None,
        "windows_bin":   None,
        "linux_url":     None,
        "windows_url":   None,
        "linux_flags":   None,
        "windows_flags": None,
    },
}

PROCEDURE_ORDER = [
    "version",
    "list-all",
    "list-installed",
    "help",
    "force-install",
    "install",
    "test"
]


def get_os(operating_system=os.name):
    """Returns the OS object for the current system"""
    if operating_system == "posix":
        return Linux()
    if operating_system == "nt":
        return Windows()
    raise NotImplementedError(f"OS not supported: {operating_system}")

class OS:
    """Abstract class for OS"""

    def __init__(self, name):
        """Initializes the OS object"""
        if name is not None:
            self.name = name
            self.packages = list(SOURCES.keys())
            self.cache_dir = CACHEDIR
            self.stats_dir = STATSDIR
            # Create the cache and stats directories if they don't exist
            for directory in [self.cache_dir, self.stats_dir]:
                if not os.path.exists(directory):
                    os.makedirs(directory)

    @abstractmethod
    def is_installed(self, package):
        """Determines if the package is installed on the system"""

    @abstractmethod
    def install(self, package, force=False):
        """Downloads and installs the package on the system"""

    @abstractmethod
    def run(self, package, iterations, monitor=False):
        """Runs the package on the system
        If monitor is True, the system will be monitored with glances
        during the test and the results will be saved to a CSV file.
        """


class Linux(OS):
    """Linux Operating System"""

    def __init__(self):
        super().__init__("linux")

    def is_installed(self, package):
        # Export every extracted directory to the path (if not already there)
        for directory in os.listdir(self.cache_dir):
            if directory not in os.environ["PATH"]:
                os.environ["PATH"] += f":{self.cache_dir}/{directory}"
        return True

    def install(self, package, force=False):
        print(f"Installing {package}...")
        if force is False:
            if self.is_installed(package):
                print(f"{package} is already installed")
                return
        # Download the tarball to the cachedir
        tarball = f"{CACHEDIR}/{package}.tar.gz"
        urlretrieve(SOURCES[package][self.name+"_url"], tarball)
        # Extract the tarball to the cachedir (in a subdirectory)
        with tarfile.open(tarball) as tar:
            tar.extractall(CACHEDIR)
        print("Done")

    def run(self, package, iterations, monitor=False):
        print(f"Running test {package}...")
        # Start glances in the background
        if monitor:
            # Gather the date and time (YYYY-MM-DD_HH-MM-SS)
            date = datetime.datetime.now().strftime(DATE_FORMAT)
            os.system("glances "+
                      "--quiet "+
                      "--disable-webui "+
                      "--time 1 "+
                      "--export csv "+
                      f"--export-csv-file {self.cache_dir}/{date}_{package}.csv &")
            # Run the package
            for _ in range(iterations):
                os.system(f"{SOURCES[package][self.name+'_bin']} "+
                          f"{SOURCES[package][self.name+'_flags']}")
            # Kill glances
            os.system("killall glances")





class Windows(OS):
    """Windows Operating System"""

    def __init__(self):
        super().__init__("windows")

    def is_installed(self, package):
        # Check if the package is installed on the system with where
        return os.system(f"where {package}") == 0

    def install(self, package, force=False):
        # Cd into the cachedir
        os.chdir(self.cache_dir)

        print(f"Installing {package}...")
        if not force:
            if self.is_installed(package):
                print(f"{package} is already installed")
                return
        else:
            # Get the current directory (to return to it later)
            root_dir = os.getcwd()
            # Cd into the cachedir
            os.chdir(self.cache_dir)
            # Remove the package if it is already installed
            if self.is_installed(package):
                os.system(f"{package} /uninstall")
            # Download the exe to the cachedir
            subprocess.run(["powershell",
                            "Invoke-WebRequest",
                            SOURCES[package][self.name+"_url"]])
            # Execute the installer
            os.system(f"{package} /S")
            # Return to the previous directory
            os.chdir(root_dir)
    def run(self, package, iterations, monitor=False):
        if monitor:
            # Gather the date and time (YYYY-MM-DD_HH-MM-SS)
            date = datetime.datetime.now().strftime(DATE_FORMAT)
            subprocess.Popen(["glances",
                "--quiet",
                "--disable-webui",
                "--time", "1",
                "--export", "csv",
                f"--export-csv-file={self.stats_dir}/{date}_{package}.csv"])
        for _ in range(iterations):
            subprocess.run([f"{package}",
                f"{SOURCES[package][self.name+'_flags']}"])
        if monitor:
            subprocess.run(["taskkill", "/IM", "glances.exe", "/F"])



def parse_args(flag_index, flags, args=sys.argv):
    """Parses every argument given to a flag
    Arguments can be space or comma separated (or both)
    So every argument between the selected flag and the next
    get parsed and returned as a list. When two flags are
    adjacent, the list is empty.
    args: list of all arguments (flags AND arguments)
    flags: list of tuples (index, value) of all flags in args
    flag_index: index of the selected flag in args
    """
    # Offset the flag index by 1 to get the first argument
    flag_index += 1
    arguments = []
    last_arg = 0
    try:
        # Get the index of the next flag
        last_arg = min(i for i, v in flags if i > flag_index)
    except ValueError:
        # If there is no next flag, get the index of the last argument
        last_arg = len(args)-1
    for i in range(flag_index + 1, last_arg+1):
        arguments.extend([v.strip() for v in args[i].split(",")])
    return arguments

def build_procedure(args=sys.argv, host=get_os()):
    """To make sure every command is executed in the right order,
    this function builds a dictionary containing an organized and
    easily parsable list of instructions.

    "procedure": {
        "help": [],      # If this key is present, print the help message
        "version": [],   # If this key is present, print the version
        "force-install": [], # List of packages to force install
        "install": [],       # List of packages to install
        "test": [(test, iterations),...], # List of tests to run
        "list": [],  # "all" available or "installed" packages
    }
    """
    # dictionary storing the procedure for the runtime
    procedure = {}
    # Gather the arguments as a list
    args = sys.argv[1:]
    # Gather the flags as a list of tuples (index, value)
    flags =  [(i, v) for i, v in enumerate(args) if v.startswith("-")]

    # If 'o|output' is in the flags, add 'output': [] to the procedure
    if any(v in ("-o", "--output") for i, v in flags):
        # Get the index of the flag (first one)
        flag_index = [i for i, v in flags if v in ("-o", "--output")][0]
        procedure["output"] = parse_args(flag_index, flags)
        if len(procedure["output"]) > 0:
            host.stats_dir = procedure["output"][0]
            # Warn that only the first output directory will be used if more than one is given
            if len(procedure["output"]) > 1:
                warnings.warn(
                    "Only one output directory can be specified yet "+
                    f"{len(procedure['output'])} were given.\n"+
                    f"Using {procedure['output'][0]} as the output directory.")

    # If 'c|cache' is in the flags, add 'cache': [] to the procedure
    if any(v in ("-c", "--cache") for i, v in flags):
        # Get the index of the flag
        flag_index = [i for i, v in flags if v in ("-c", "--cache")][0]
        procedure["cache"] = parse_args(flag_index, flags)
        if len(procedure["cache"]) > 0:
            host.cache_dir = procedure["cache"][0]
            # Warn that only the first cache directory will be used if more than one is given
            if len(procedure["cache"]) > 1:
                warnings.warn(
                    "Only one cache directory can be specified yet "+
                    f"{len(procedure['cache'])} were given.\n"+
                    f"Using {procedure['cache'][0]} as the cache directory.")

    # If no flags are given (or 'a'|'all' is in the flags),
    # Override the flags to install and run all tests
    if len(flags) == 0 or any(v in ("-a", "--all") for i, v in flags):
        args = ["-i", "all", "-t", "all"]

    # If 'h|help' is in the flags, add 'help': [] to the procedure
    if any(v in ("-h", "--help") for i, v in flags):
        procedure["help"] = []

    # If 'v|version' is in the flags, add 'version': [] to the procedure
    if any(v in ("-v", "--version") for i, v in flags):
        procedure["version"] = [VERSION]

    # If 'i|install' is in the flags, add 'install': [] to the procedure
    if any(v in ("-i", "--install") for i, v in flags):
        # Get the index of the flag
        flag_index = [i for i, v in flags if v in ("-i", "--install")][0]
        procedure["install"] = parse_args(flag_index, flags)
        if (procedure["install"] == []) or (procedure["install"] == ["all"]):
            procedure["install"] = list(host.packages)

    # If 'f|force-install' is in the flags, add 'force-install': [] to the procedure
    if any(v in ("-f", "--force-install") for i, v in flags):
        # Get the index of the flag
        flag_index = [i for i, v in flags if v in ("-f", "--force-install")][0]
        procedure["force-install"] = parse_args(flag_index, flags)
        if procedure["force-install"] == []:
            procedure["force-install"] = list(host.packages)

    # If 't|test' is in the flags, add 'test': [(test,iterations),...] to the procedure
    if any(v in ("-t", "--test") for i, v in flags):
        # Get the index of the flag
        flag_index = [i for i, v in flags if v in ("-t", "--test")][0]
        tests = []
        arguments = parse_args(flag_index, flags)
        # Parse the arguments to get the test name and the number of iterations
        for test in arguments:
            if "(" in test:
                test, iterations = test.split("(")
                iterations = int(iterations[:-1])
                tests.append((test, iterations))
            # Use the default number of iterations if none were given
            else:
                tests.append((test, SOURCES[test]["default_iterations"]))
        # Add all the tests to the procedure if none were given
        if not arguments:
            for software in SOURCES:
                tests.append((software, SOURCES[software]["default_iterations"]))
        procedure["test"] = tests

    # If 'l|list' is in the flags, add 'list': [] to the procedure
    if any(v in ("-l", "--list") for i, v in flags):
        # Get the index of the flag
        flag_index = [i for i, v in flags if v in ("-l", "--list")][0]
        procedure["list"] = parse_args(flag_index, flags)
        if (procedure["list"] == []) or (procedure["list"] == ["all"]):
            procedure["list"] = ["all"]
        if "installed" in procedure["list"]:
            procedure["list-installed"] = []

    return procedure


def main():
    """
    Benchmarking Suite:
    This program runs a sweet of tests to measure the performance
    of a system and installs the required tools if necessary.

    Usage:
    python3 testsweet.py [options] [test ...]

    Options:
    -h, --help           Show this screen
    -v, --version        Show version
    -i, --install        Install the required tools
    -f, --force-install  Force installation of packages
    -o, --output         Output directory
    -c, --cache          Cache directory
    -t, --test           Run the specified test(s)
    -l, --list           List available tests
    -a, --all            Install and run all tests

    Install (-i):
    This option installs the required tools for the tests.
    Arguments can be given comma or space separated. Only
    arguments directly following this option are considered.
    (ex: `-i 7z,geekbench -t 7z` will install both 7z and
    geekbench but only run the 7z test) Available arguments:
        * 7z
        * geekbench
        * all (equivalent to `-i 7z,geekbench`)

    Force Install (-f):
    This option forces the installation of the specified packages
    even if they are already installed. See the Install option for
    more details on it's usage.

    Output (-o):
    This option specifies the output directory for the test results.
    The default directory is `stats` in the current working directory.
    (ex: `-o /home/user/stats`). CSV labelling is done automatically
    using the current date and time and the test name.
    (ex: `2021-01-01_00-00-00_7z.csv`).

    Test (-t):
    This option runs the specified test(s). Arguments can be given
    comma or space separated. Each argument can be given a number of
    iterations in parentesis after the test name. The available tests
    are listed below.
        * baseline: Monitors the system without running any tests
            (Default: 60 iterations of 1 second)
        * 7z: Runs the built-in 7z benchmark
            (Default: 5 iteration of the test)
        * geekbench:
            (Default: 1 iteration of the test)

    List (-l):
    This option lists the available tests and their descriptions.
    Available arguments:
        * all: List all available tests (default)
        * installed: List only installed tests

    All (-a):
    This option installs and runs all available tests.
    See the Install and Test options for more details on their usage.
    This is equivalent to `-i all -t all`.
    Please note that this option will override any other options given
    (except for the Output and Cache options) and use soft installation
    and the default number of iterations.
    """

    # Gather and set the Operating System
    host = get_os()

    # Build the procedure
    procedure = build_procedure(host=host)

    for step in PROCEDURE_ORDER:
        if step in procedure:
            if step == "help":
                print(main.__doc__)
            elif step == "version":
                print(f"Version: {VERSION}")
            elif step == "force-install":
                for package in procedure["force-install"]:
                    host.install(package, force=True)
            elif step == "install":
                for package in procedure["install"]:
                    host.install(package)
            elif step == "test":
                for test, iterations in procedure["test"]:
                    print(f"Running {test} test...")
                    monitor = True
                    if not host.is_installed(test):
                        warnings.warn(f"{test} is not installed. Skipping test.")
                        continue
                    if not host.is_installed("glances"):
                        warnings.warn("glances is not installed. "+
                                      " Monitoring will be disabled.\n"+
                                      "Consider installing it with pip install requirements.txt."+
                                      "On windows, also use requirements_win.txt.")
                        monitor = False
                    host.run(test, iterations, monitor)
            elif step == "list-installed":
                print("Installed tests:")
                print(f"\tbaseline: {SOURCES['baseline']['description']}")
                for package in host.packages:
                    if host.is_installed(package):
                        print(f"\t{package}: {SOURCES[package]['description']}")
            elif step == "list-all":
                print("Available tests:")
                print(f"\tbaseline: {SOURCES['baseline']['description']}")
                for package in SOURCES:
                    print(f"\t{package}: {SOURCES[package]['description']}")


if __name__ == "__main__":
    main()