Test refactoring

The testing script is getting too big and not very well organized. In
this commit a new module `integration_tests` is introduced that contains
parts of the original testing script split into multiple files. The
content should be the same, the only difference is that now you can run
the tests by invoking `python3 -m test`.

test/1-create-base.json

@@ -1,23 +0,0 @@
-{
-  "name": "base",
-  "stages": [
-    {
-      "name": "org.osbuild.dnf",
-      "options": {
-        "releasever": "30",
-        "repos": {
-          "fedora": {
-            "name": "Fedora",
-            "metalink": "https://mirrors.fedoraproject.org/metalink?repo=fedora-$releasever&arch=$basearch",
-            "gpgkey": "file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch"
-          }
-        },
-        "packages": [
-          "@Core",
-          "selinux-policy-targeted",
-          "grub2-pc"
-        ]
-      }
-    }
-  ]
-}

test/2-configure-web-server.json

@@ -1,36 +0,0 @@
-{
-  "name": "base-qcow2",
-  "base": "1f663f817473ffa5b01241b17adbd71bc734962313f5d4eef230073c0ac5884e",
-  "stages": [
-    {
-      "name": "org.osbuild.script",
-      "options": {
-        "script": "echo root | passwd --stdin root; echo 'SELINUX=disabled' > /etc/selinux/config;"
-      }
-    },
-    {
-      "name": "org.osbuild.script",
-      "options": {
-        "script": "mkdir -p /var/web; echo 'hello, world!' > /var/web/index; echo -e \"[Unit]\\nDescription=Testing web server\\nAfter=network.target\\n\\n[Service]\\nType=simple\\nExecStart=python3 -m http.server 8888\\nWorkingDirectory=/var/web/\\n\\n[Install]\\nWantedBy=multi-user.target\" > /etc/systemd/system/web-server.service;"
-      }
-    },
-    {
-      "name": "org.osbuild.systemd",
-      "options": {
-        "enabled_services": [
-          "NetworkManager",
-          "web-server"
-        ],
-        "disabled_services": [
-          "firewalld"
-        ]
-      }
-    },
-    {
-      "name": "org.osbuild.grub2",
-      "options": {
-        "root_fs_uuid": "76a22bf4-f153-4541-b6c7-0332c0dfaeac"
-      }
-    }
-  ]
-}

test/3-compose-qcow2.json

@@ -1,11 +0,0 @@
-{
-  "name": "base-qcow2",
-  "base": "552b5555bdf64c5e19bf4ca8a709da37fb3046678643a8f8499297b6dd95c7e7",
-  "assembler": {
-    "name": "org.osbuild.qcow2",
-    "options": {
-      "filename": "base.qcow2",
-      "root_fs_uuid": "76a22bf4-f153-4541-b6c7-0332c0dfaeac"
-    }
-  }
-}

test/__main__.py

@@ -0,0 +1,81 @@
+import argparse
+import logging
+import subprocess
+import os
+
+from test.integration_tests.test_case import IntegrationTestCase, IntegrationTestType
+from test.integration_tests.config import *
+
+logging.basicConfig(level=logging.getLevelName(os.environ.get("TESTS_LOGLEVEL", "INFO")))
+
+
+def test_web_server_with_curl():
+    cmd = ["curl", "-s", "http://127.0.0.1:8888/index"]
+    logging.info(f"Running curl: {cmd}")
+    curl = subprocess.run(cmd, capture_output=True)
+    logging.info(f"Curl returned: code={curl.returncode}, stdout={curl.stdout.decode()}, stderr={curl.stderr.decode()}")
+    assert curl.returncode == 0
+    assert curl.stdout.decode("utf-8").strip() == "hello, world!"
+
+
+def test_timezone(extract_dir):
+    link = os.readlink(f"{extract_dir}/etc/localtime")
+    assert "Europe/Prague" in link
+
+
+def test_firewall(extract_dir):
+    with open(f"{extract_dir}/etc/firewalld/zones/public.xml") as f:
+        content = f.read()
+        assert 'service name="http"' in content
+        assert 'service name="ftp"' in content
+        assert 'service name="telnet"' not in content
+        assert 'port port="53" protocol="tcp"' in content
+        assert 'port port="88" protocol="udp"' in content
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='Run integration tests')
+    parser.add_argument('--list', dest='list', action='store_true', help='list test cases')
+    parser.add_argument('--case', dest='specific_case', metavar='TEST_CASE', help='run single test case')
+    args = parser.parse_args()
+
+    logging.info(f"Using {OBJECTS} for objects storage.")
+    logging.info(f"Using {OUTPUT_DIR} for output images storage.")
+    logging.info(f"Using {OSBUILD} for building images.")
+
+    web_server = IntegrationTestCase(
+        name="web-server",
+        pipeline="web-server.json",
+        output_image="web-server.qcow2",
+        test_cases=[test_web_server_with_curl],
+        type=IntegrationTestType.BOOT_WITH_QEMU
+    )
+    timezone = IntegrationTestCase(
+        name="timezone",
+        pipeline="timezone.json",
+        output_image="timezone.tar.xz",
+        test_cases=[test_timezone],
+        type=IntegrationTestType.EXTRACT
+    )
+    firewall = IntegrationTestCase(
+        name="firewall",
+        pipeline="firewall.json",
+        output_image="firewall.tar.xz",
+        test_cases=[test_firewall],
+        type=IntegrationTestType.EXTRACT
+    )
+
+    cases = [web_server, timezone, firewall]
+
+    if args.list:
+        print("Available test cases:")
+        for case in cases:
+            print(f" - {case.name}")
+    else:
+        if not args.specific_case:
+            for case in cases:
+                case.run()
+        else:
+            for case in cases:
+                if case.name == args.specific_case:
+                    case.run()

test/integration_tests/__init__.py

@@ -0,0 +1,14 @@
+from .config import *
+
+
+def evaluate_test(test, name=None):
+    try:
+        test()
+        print(f"{RESET}{BOLD}{name or test.__name__}: Success{RESET}")
+    except AssertionError as e:
+        print(f"{RESET}{BOLD}{name or test.__name__}: {RESET}{RED}Fail{RESET}")
+        print(e)
+
+
+def rel_path(fname: str) -> str:
+    return os.path.join(os.path.dirname(os.path.dirname(__file__)), fname)

test/integration_tests/build.py

@@ -0,0 +1,25 @@
+import logging
+import subprocess
+import sys
+
+from .config import *
+
+
+def run_osbuild(pipeline: str, check=True):
+    cmd = OSBUILD + ["--store", OBJECTS, "-o", OUTPUT_DIR, pipeline]
+    logging.info(f"Running osbuild: {cmd}")
+    osbuild = subprocess.run(cmd, capture_output=True)
+    if osbuild.returncode != 0:
+        logging.error(f"{RED}osbuild failed!{RESET}")
+        print(f"{BOLD}STDERR{RESET}")
+        print(osbuild.stderr.decode())
+        print(f"{BOLD}STDOUT{RESET}")
+        print(osbuild.stdout.decode())
+        if check:
+            sys.exit(1)
+
+    return osbuild.returncode
+
+
+def build_testing_image(pipeline_full_path):
+    run_osbuild(pipeline_full_path)

test/integration_tests/config.py

@@ -0,0 +1,11 @@
+import tempfile
+import os
+
+
+EXPECTED_TIME_TO_BOOT = 60  # seconds
+RESET = "\033[0m"
+BOLD = "\033[1m"
+RED = "\033[31m"
+OBJECTS = os.environ.get("OBJECTS", tempfile.mkdtemp(prefix="osbuild-"))
+OUTPUT_DIR = os.environ.get("OUTPUT_DIR", tempfile.mkdtemp(prefix="osbuild-"))
+OSBUILD = os.environ.get("OSBUILD", "osbuild").split(' ')

test/integration_tests/run.py

@@ -0,0 +1,32 @@
+import contextlib
+import logging
+import subprocess
+import time
+
+from .config import *
+
+
+@contextlib.contextmanager
+def boot_image(file_name: str):
+    acceleration = ["-accel", "kvm:hvf:tcg"]
+    network = ["-net", "nic,model=rtl8139", "-net", "user,hostfwd=tcp::8888-:8888"]
+    cmd = ["qemu-system-x86_64", "-nographic", "-m", "1024", "-snapshot"] + \
+          acceleration + [f"{OUTPUT_DIR}/{file_name}"] + network
+    logging.info(f"Booting image: {cmd}")
+    vm = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    try:
+        time.sleep(EXPECTED_TIME_TO_BOOT)
+        yield None
+    finally:
+        vm.kill()
+
+
+@contextlib.contextmanager
+def extract_image(file_name: str):
+    extract_dir = tempfile.mkdtemp(prefix="osbuild-")
+    subprocess.run(["tar", "xf", f"{OUTPUT_DIR}/{file_name}"], cwd=extract_dir, check=True)
+    try:
+        yield extract_dir
+    finally:
+        # Clean up?
+        pass

test/integration_tests/test_case.py

@@ -0,0 +1,38 @@
+from dataclasses import dataclass
+from enum import Enum
+from typing import List, Callable, Any
+
+from . import evaluate_test, rel_path
+from .build import run_osbuild
+from .run import boot_image, extract_image
+
+
+class IntegrationTestType(Enum):
+    EXTRACT=0
+    BOOT_WITH_QEMU=1
+
+
+@dataclass
+class IntegrationTestCase:
+    name: str
+    pipeline: str
+    output_image: str
+    test_cases: List[Callable[[Any], None]]
+    type: IntegrationTestType
+
+    def run(self):
+        run_osbuild(rel_path(f"pipelines/{self.pipeline}"))
+        if self.type == IntegrationTestType.BOOT_WITH_QEMU:
+            self.boot_and_run()
+        else:
+            self.extract_and_run()
+
+    def boot_and_run(self):
+        with boot_image(self.output_image):
+            for test in self.test_cases:
+                evaluate_test(test)
+
+    def extract_and_run(self):
+        with extract_image(self.output_image) as fstree:
+            for test in self.test_cases:
+                evaluate_test(lambda: test(fstree), name=test.__name__)

test/pipelines/firewall.json

@@ -27,7 +27,7 @@
   "assembler": {
     "name": "org.osbuild.tar",
     "options": {
-      "filename": "firewall-output.tar.xz",
+      "filename": "firewall.tar.xz",
       "compression": "xz"
     }
   }

test/pipelines/timezone.json

@@ -25,7 +25,7 @@
   "assembler": {
     "name": "org.osbuild.tar",
     "options": {
-      "filename": "timezone-output.tar.xz",
+      "filename": "timezone.tar.xz",
       "compression": "xz"
     }
   }

test/pipelines/web-server.json

@@ -53,7 +53,7 @@
   "assembler": {
     "name": "org.osbuild.qcow2",
     "options": {
-      "filename": "base.qcow2",
+      "filename": "web-server.qcow2",
       "root_fs_uuid": "76a22bf4-f153-4541-b6c7-0332c0dfaeac"
     }
   }

test/run-tests.py

@@ -1,122 +0,0 @@
-import contextlib
-import logging
-import os
-import subprocess
-import sys
-import tempfile
-import time
-
-EXPECTED_TIME_TO_BOOT = 60  # seconds
-RESET = "\033[0m"
-BOLD = "\033[1m"
-RED = "\033[31m"
-OBJECTS = os.environ.get("OBJECTS", tempfile.mkdtemp(prefix="osbuild-"))
-OUTPUT_DIR = os.environ.get("OUTPUT_DIR", tempfile.mkdtemp(prefix="osbuild-"))
-OSBUILD = os.environ.get("OSBUILD", "osbuild").split(' ')
-IMAGE_PATH = os.environ.get("IMAGE_PATH", OUTPUT_DIR + "/base.qcow2")
-
-
-logging.basicConfig(level=logging.getLevelName(os.environ.get("TESTS_LOGLEVEL", "INFO")))
-
-
-def run_osbuild(pipeline: str, check=True):
-    cmd = OSBUILD + ["--store", OBJECTS, "-o", OUTPUT_DIR, pipeline]
-    logging.info(f"Running osbuild: {cmd}")
-    osbuild = subprocess.run(cmd, capture_output=True)
-    if osbuild.returncode != 0:
-        logging.error(f"{RED}osbuild failed!{RESET}")
-        print(f"{BOLD}STDERR{RESET}")
-        print(osbuild.stderr.decode())
-        print(f"{BOLD}STDOUT{RESET}")
-        print(osbuild.stdout.decode())
-        if check:
-            sys.exit(1)
-
-    return osbuild.returncode
-
-
-def rel_path(fname: str) -> str:
-    return os.path.join(os.path.dirname(__file__), fname)
-
-
-def build_web_server_image():
-    run_osbuild(rel_path("4-all.json"))
-
-
-@contextlib.contextmanager
-def boot_image(path: str):
-    acceleration = ["-accel", "kvm:hvf:tcg"]
-    network = ["-net", "nic,model=rtl8139", "-net", "user,hostfwd=tcp::8888-:8888"]
-    cmd = ["qemu-system-x86_64", "-nographic", "-m", "1024", "-snapshot"] + acceleration + [path] + network
-    logging.info(f"Booting image: {cmd}")
-    vm = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    try:
-        time.sleep(EXPECTED_TIME_TO_BOOT)
-        yield None
-    finally:
-        vm.kill()
-
-
-def test_web_server():
-    cmd = ["curl", "-s", "http://127.0.0.1:8888/index"]
-    logging.info(f"Running curl: {cmd}")
-    curl = subprocess.run(cmd, capture_output=True)
-    logging.info(f"Curl returned: code={curl.returncode}, stdout={curl.stdout.decode()}, stderr={curl.stderr.decode()}")
-    assert curl.returncode == 0
-    assert curl.stdout.decode("utf-8").strip() == "hello, world!"
-
-
-def build_timezone_image():
-    run_osbuild(rel_path("timezone-test.json"))
-
-
-def build_firewall_image():
-    run_osbuild(rel_path("firewall-test.json"))
-
-
-def extract_to_tempdir(image_file):
-    extract_dir = tempfile.mkdtemp(prefix="osbuild-")
-    subprocess.run(["tar", "xf", OUTPUT_DIR + image_file], cwd=extract_dir, check=True)
-    return extract_dir
-
-
-def test_timezone():
-    extract_dir = extract_to_tempdir("timezone-output.tar.xz")
-    ls = subprocess.run(["ls", "-l", "etc/localtime"], cwd=extract_dir, check=True, stdout=subprocess.PIPE)
-    ls_output = ls.stdout.decode("utf-8")
-    assert "Europe/Prague" in ls_output
-
-
-def test_firewall():
-    extract_dir = extract_to_tempdir("firewall-output.tar.xz")
-    cat = subprocess.run(["cat", "etc/firewalld/zones/public.xml"], cwd=extract_dir, check=True, stdout=subprocess.PIPE)
-    cat_output = cat.stdout.decode("utf-8")
-    assert 'service name="http"' in cat_output
-    assert 'service name="ftp"' in cat_output
-    assert 'service name="telnet"' not in cat_output
-    assert 'port port="53" protocol="tcp"' in cat_output
-    assert 'port port="88" protocol="udp"' in cat_output
-
-
-def evaluate_test(test):
-    try:
-        test()
-        print(f"{RESET}{BOLD}{test.__name__}: Success{RESET}")
-    except AssertionError as e:
-        print(f"{RESET}{BOLD}{test.__name__}: {RESET}{RED}Fail{RESET}")
-        print(e)
-
-
-if __name__ == '__main__':
-    logging.info("Running tests")
-    build_web_server_image()
-    tests = [test_web_server]
-    with boot_image(IMAGE_PATH):
-        for test in tests:
-            evaluate_test(test)
-
-    build_timezone_image()
-    evaluate_test(test_timezone)
-
-    build_firewall_image()
-    evaluate_test(test_firewall)

test/variables

@@ -1,6 +0,0 @@
-export OSBUILD=../osbuild
-export BASE_INPUT=/osbuild/workdir
-export BASE_OUTPUT=/osbuild/base-tree
-export WEB_OUTPUT=/osbuild/web-tree
-export QCOW2_OUTPUT=/osbuild/output
-export IMAGE_PATH=/osbuild/output/base.qcow2