Add behave tests

behave/features/steps/common.py

@@ -0,0 +1,257 @@
+import errno
+import os
+import shutil
+import subprocess
+
+import behave
+
+
+def makedirs(path):
+    try:
+        os.makedirs(path)
+    except OSError as e:
+        if e.errno != errno.EEXIST:
+            raise
+
+
+def run(cmd, shell=True, cwd=None, env=None):
+    """
+    Run a command.
+    Return exitcode, stdout, stderr
+    """
+
+    proc = subprocess.Popen(
+        cmd,
+        shell=shell,
+        cwd=cwd,
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        encoding="utf-8",
+        universal_newlines=True,
+        errors="surrogateescape",
+    )
+
+    stdout, stderr = proc.communicate()
+    return proc.returncode, stdout, stderr
+
+
+def check_exit_code(context):
+    # check if the previous command finished successfully
+    # or if the exit code was tested in a scenario
+    if not getattr(context, "cmd", None):
+        return
+
+    if context.cmd_exitcode_checked:
+        return
+
+    the_exit_code_is(context, 0)
+
+
+def run_in_context(context, cmd, can_fail=False, **run_args):
+    check_exit_code(context)
+
+    context.cmd = cmd
+
+    if hasattr(context.scenario, "working_dir") and 'cwd' not in run_args:
+        run_args['cwd'] = context.scenario.working_dir
+
+    if hasattr(context.scenario, "PATH"):
+        env = os.environ.copy()
+        path = context.scenario.PATH
+        env["PATH"] = path.replace("$PATH", env["PATH"])
+        run_args["env"] = env
+
+    if context.config.userdata.get("DEBUG", False):
+        print(f"DEBUG: command: {cmd}")
+
+    context.cmd_exitcode, context.cmd_stdout, context.cmd_stderr = run(cmd, **run_args)
+    context.cmd_exitcode_checked = False
+
+    if context.config.userdata.get("DEBUG", False):
+        print(f"DEBUG: exit code: {context.cmd_exitcode}")
+        print(f"DEBUG: stdout: {context.cmd_stdout}")
+        print(f"DEBUG: stderr: {context.cmd_stderr}")
+
+    if not can_fail and context.cmd_exitcode != 0:
+        raise AssertionError('Running command "%s" failed: %s' % (cmd, context.cmd_exitcode))
+
+
+@behave.step("stdout contains \"{text}\"")
+def step_impl(context, text):
+    if re.search(text.format(context=context), context.cmd_stdout):
+        return
+    raise AssertionError("Stdout doesn't contain: %s" % text)
+
+
+@behave.step("stderr contains \"{text}\"")
+def step_impl(context, text):
+    if re.search(text.format(context=context), context.cmd_stderr):
+        return
+    raise AssertionError("Stderr doesn't contain: %s" % text)
+
+
+@behave.step("stdout is")
+def step_impl(context):
+    expected = context.text.format(context=context).rstrip().split('\n')
+    found = context.cmd_stdout.rstrip().split('\n')
+
+    if found == expected:
+        return
+
+    expected_str = "\n".join(expected)
+    found_str = "\n".join(found)
+    raise AssertionError(f"Stdout is not:\n{expected_str}\n\nActual stdout:\n{found_str}")
+
+
+@behave.step('I set working directory to "{path}"')
+def step_impl(context, path):
+    path = path.format(context=context)
+    context.scenario.working_dir = path
+
+
+@behave.step('I set PATH to "{path}"')
+def step_impl(context, path):
+    path = path.format(context=context)
+    context.scenario.PATH = path
+
+
+@behave.step('I copy file "{source}" to "{destination}"')
+def step_impl(context, source, destination):
+    # substitutions
+    source = source.format(context=context)
+    destination = destination.format(context=context)
+
+    # if destination is a directory, append the source filename
+    if destination.endswith("/"):
+        destination = os.path.join(destination, os.path.basename(source))
+
+    # copy file without attributes
+    makedirs(os.path.dirname(destination))
+    shutil.copyfile(source, destination)
+
+
+@behave.step('file "{path}" exists')
+def step_impl(context, path):
+    path = path.format(context=context)
+    if not os.path.isfile(path):
+        raise AssertionError(f"File doesn't exist: {path}")
+
+
+@behave.step('file "{path}" does not exist')
+def step_impl(context, path):
+    path = path.format(context=context)
+    if os.path.isfile(path):
+        raise AssertionError(f"File exists: {path}")
+
+
+@behave.step('file "{one}" is identical to "{two}"')
+def step_impl(context, one, two):
+    one = one.format(context=context)
+    two = two.format(context=context)
+    data_one = open(one, "r").read()
+    data_two = open(two, "r").read()
+    if data_one != data_two:
+        raise AssertionError(f"Files differ: {one} != {two}")
+
+
+@behave.step('directory "{path}" exists')
+def step_impl(context, path):
+    path = path.format(context=context)
+    if not os.path.isdir(path):
+        raise AssertionError(f"Directory doesn't exist: {path}")
+
+
+@behave.step('directory "{path}" does not exist')
+def step_impl(context, path):
+    path = path.format(context=context)
+    if os.path.isdir(path):
+        raise AssertionError(f"Directory exists: {path}")
+
+
+@behave.step('I create file "{path}" with perms "{mode}"')
+def step_impl(context, path, mode):
+    path = path.format(context=context)
+    mode = int(mode, 8)
+    content = context.text.format(context=context).rstrip()
+
+    makedirs(os.path.dirname(path))
+    open(path, "w").write(content)
+    os.chmod(path, mode)
+
+
+@behave.step("the exit code is {exitcode}")
+def the_exit_code_is(context, exitcode):
+    if context.cmd_exitcode != int(exitcode):
+        raise AssertionError(f"Command has exited with code {context.cmd_exitcode}: {context.cmd}")
+    context.cmd_exitcode_checked = True
+
+
+@behave.step('directory listing of "{path}" is')
+def step_impl(context, path):
+    path = path.format(context=context)
+    expected = context.text.format(context=context).rstrip().split('\n')
+    expected = [i for i in expected if i.strip()]
+    found = os.listdir(path)
+
+    expected = set(expected)
+    found = set(found)
+
+    if found == expected:
+        return
+
+    extra = sorted(set(found) - set(expected))
+    missing = sorted(set(expected) - set(found))
+
+    msg = []
+    if extra:
+        msg.append("Unexpected files found on disk:")
+        for fn in extra:
+            msg.append(f"  {fn}")
+    if missing:
+        msg.append("Files missing on disk:")
+        for fn in missing:
+            msg.append(f"  {fn}")
+
+    msg = "\n".join(msg)
+
+    raise AssertionError(f"Directory listing does not match:\n{msg}")
+
+
+@behave.step('directory tree in "{path}" is')
+def step_impl(context, path):
+    path = path.format(context=context)
+    path = os.path.abspath(path)
+    expected = context.text.format(context=context).rstrip().split('\n')
+    expected = [i for i in expected if i.strip()]
+
+    found = []
+    for root, dirs, files in os.walk(path):
+        for fn in files:
+            file_abspath = os.path.join(root, fn)
+            file_relpath = file_abspath[len(path) + 1:]
+            found.append(file_relpath)
+#    found = os.listdir(path)
+
+    expected = set(expected)
+    found = set(found)
+
+    if found == expected:
+        return
+
+    extra = sorted(set(found) - set(expected))
+    missing = sorted(set(expected) - set(found))
+
+    msg = []
+    if extra:
+        msg.append("Unexpected files found on disk:")
+        for fn in extra:
+            msg.append(f"  {fn}")
+    if missing:
+        msg.append("Files missing on disk:")
+        for fn in missing:
+            msg.append(f"  {fn}")
+
+    msg = "\n".join(msg)
+
+    raise AssertionError(f"Directory listing does not match:\n{msg}")

behave/features/steps/kanku.py

@@ -0,0 +1,76 @@
+#!/usr/bin/python3
+
+
+import os
+import re
+import subprocess
+
+import behave
+from ruamel.yaml import YAML
+
+
+class Kanku:
+    def __init__(self, context, kankufile):
+        self.kankufile = kankufile
+        self.kankudir = os.path.dirname(self.kankufile)
+        self.domain_name = self._get_domain_name()
+        self.ip = self._get_ip()
+
+    def _get_domain_name(self):
+        """
+        Get domain name directly from KankuFile yaml
+        """
+        yaml = YAML(typ='safe')
+        doc = yaml.load(open(self.kankufile, "r"))
+        return doc["domain_name"]
+
+    def _run_kanku(self, args):
+        cmd = ["kanku"] + args
+        env = os.environ.copy()
+        env["KANKU_CONFIG"] = self.kankufile
+        proc = subprocess.Popen(
+            cmd,
+            cwd=self.kankudir,
+            env=env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            encoding="utf-8"
+        )
+        return proc
+
+    def _get_ip(self):
+        """
+        Get IP from calling `kanku ip`
+        """
+        proc = self._run_kanku(["ip"])
+        stdout, stderr = proc.communicate()
+        match = re.search(r"IP Address: ([\d\.]+)", stderr)
+        return match.group(1)
+
+    def create_snapshot(self):
+        # unmount /tmp/kanku so we are able to create a snapshot of the VM
+        self.run_command("umount /tmp/kanku")
+        proc = self._run_kanku(["snapshot", "--create", "--name", "current"])
+        proc.communicate()
+
+    def revert_to_snapshot(self):
+        proc = self._run_kanku(["snapshot", "--revert", "--name", "current"])
+        proc.communicate()
+
+    def delete_snapshot(self):
+        proc = self._run_kanku(["snapshot", "--remove", "--name", "current"])
+        proc.communicate()
+
+    def run_command(self, ssh_cmd, user="root"):
+        proc = self._run_kanku(["ssh", "-u", user, "--execute", ssh_cmd])
+        proc.wait()
+
+
+@behave.step("I create VM snapshot")
+def func(context, args):
+    context.kanku.create_snapshot(sudo=True)
+
+
+@behave.step("I revert to VM snapshot")
+def func(context, args):
+    context.kanku.revert_to_snapshot(sudo=True)

behave/features/steps/osc.py

@@ -0,0 +1,75 @@
+import os
+import shutil
+import tempfile
+import time
+
+import behave
+
+from steps.common import run_in_context
+
+
+class Osc:
+    def __init__(self, context):
+        self.temp = tempfile.mkdtemp(prefix="osc_behave_")
+
+        if not hasattr(context, "kanku"):
+            raise RuntimeError("context doesn't have kanku object set")
+
+        self.oscrc = os.path.join(self.temp, "oscrc")
+        with open(self.oscrc, "w") as f:
+            f.write("[general]\n")
+            f.write("\n")
+            f.write(f"[https://{context.kanku.ip}]\n")
+            f.write("user=Admin\n")
+            f.write("pass=opensuse\n")
+            f.write("credentials_mgr_class=osc.credentials.PlaintextConfigFileCredentialsManager\n")
+            f.write("sslcertck=0\n")
+            f.write("trusted_prj=openSUSE.org:openSUSE:Tumbleweed\n")
+
+    def __del__(self):
+        try:
+            shutil.rmtree(self.temp)
+        except Exception:
+            pass
+
+    def get_cmd(self, context):
+        osc_cmd = context.config.userdata.get("osc", "osc")
+        cmd = [osc_cmd]
+        cmd += ["--config", self.oscrc]
+        cmd += ["-A", f"https://{context.kanku.ip}"]
+        return cmd
+
+
+@behave.step("I execute osc with args \"{args}\"")
+def step_impl(context, args):
+    args = args.format(context=context)
+    cmd = context.osc.get_cmd(context) + [args]
+    cmd = " ".join(cmd)
+    run_in_context(context, cmd, can_fail=True)
+
+
+@behave.step('I wait for osc results for "{project}" "{package}"')
+def step_impl(context, project, package):
+    args = f"results {project} {package} --csv --format='%(code)s,%(dirty)s'"
+    cmd = context.osc.get_cmd(context) + [args]
+    cmd = " ".join(cmd)
+
+    while True:
+        # wait for a moment before checking the status even for the first time
+        # for some reason, packages appear to be "broken" for a while after they get commited
+        time.sleep(5)
+
+        run_in_context(context, cmd, can_fail=True)
+        results = []
+        for line in context.cmd_stdout.splitlines():
+            code, dirty = line.split(",")
+            dirty = dirty.lower() == "true"
+            results.append((code, dirty))
+
+        if all((code == "succeeded" and not dirty for code, dirty in results)):
+            # all builds have succeeded and all dirty flags are false
+            break
+
+        if any((code in ("unresolvable", "failed", "broken", "blocked", "locked", "excluded") and not dirty for code, dirty in results)):
+            # failed build with dirty flag false
+            raise AssertionError("Package build failed:\n" + context.cmd_stdout)