debian-forge/test/osbuildtest.py

import difflib

import json
import os
import pprint
import shutil
import subprocess
import tempfile
import unittest

import osbuild


class TestCase(unittest.TestCase):
    """A TestCase to test running the osbuild program.

    Each test case can use `self.run_osbuild()` to run osbuild. A temporary
    store is used, which can be accessed through `self.store`. The store is
    persistent for whole test class due to performance concerns.

    To speed up local development, OSBUILD_TEST_STORE can be set to an existing
    store. Note that this might make tests dependant of each other. Do not use
    it for actual testing.
    """

    def setUp(self):
        self.store = os.getenv("OSBUILD_TEST_STORE")
        if not self.store:
            self.store = tempfile.mkdtemp(prefix="osbuild-test-", dir="/var/tmp")

    def tearDown(self):
        if not os.getenv("OSBUILD_TEST_STORE"):
            shutil.rmtree(self.store)

    def run_osbuild(self, pipeline, input_data=None):
        osbuild_cmd = ["python3", "-m", "osbuild", "--json", "--store", self.store, "--libdir", ".", pipeline]

        build_env = os.getenv("OSBUILD_TEST_BUILD_ENV", None)
        if build_env:
            osbuild_cmd.append("--build-env")
            osbuild_cmd.append(build_env)

        # Create a checkpoint at the last stage, i.e.
        # commit the final tree to the store, so that
        # tests can use it to compare against
        if input_data:
            manifest = json.loads(input_data)
        else:
            with open(pipeline, "r") as f:
                manifest = json.load(f)

        parsed = osbuild.load(manifest.get("pipeline", {}), manifest.get("sources", {}))
        if parsed.tree_id:
            osbuild_cmd.append("--checkpoint")
            osbuild_cmd.append(parsed.tree_id)

        stdin = subprocess.PIPE if input_data else None

        p = subprocess.Popen(osbuild_cmd, encoding="utf-8", stdin=stdin, stdout=subprocess.PIPE)
        try:
            output, _ = p.communicate(input_data)
            if p.returncode != 0:
                print(output)
            self.assertEqual(p.returncode, 0)
        except KeyboardInterrupt:
            # explicitly wait again to let osbuild clean up
            p.wait()
            raise

        result = json.loads(output)
        return result.get("tree_id"), result.get("output_id")

    # pylint: disable=no-self-use
    def run_tree_diff(self, tree1, tree2):
        tree_diff_cmd = ["./tree-diff", tree1, tree2]

        r = subprocess.run(tree_diff_cmd, encoding="utf-8", stdout=subprocess.PIPE, check=True)

        return json.loads(r.stdout)

    def get_path_to_store(self, tree_id):
        return f"{self.store}/refs/{tree_id}"

    def assertTreeDiffsEqual(self, tree_diff1, tree_diff2):
        """
        Asserts two tree diffs for equality.

        Before assertion, the two trees are sorted, therefore order of files
        doesn't matter.

        There's a special rule for asserting differences where we don't
        know the exact before/after value. This is useful for example if
        the content of file is dependant on current datetime. You can use this
        feature by putting null value in difference you don't care about.

        Example:
            "/etc/shadow": {content: ["sha256:xxx", null]}

            In this case the after content of /etc/shadow doesn't matter.
            The only thing that matters is the before content and that
            the content modification happened.
        """
        tree_diff1 = _sorted_tree(tree_diff1)
        tree_diff2 = _sorted_tree(tree_diff2)

        def raise_assertion(msg):
            raise TreeAssertionError(msg, tree_diff1, tree_diff2)

        self.assertEqual(tree_diff1['added_files'], tree_diff2['added_files'])
        self.assertEqual(tree_diff1['deleted_files'], tree_diff2['deleted_files'])

        if len(tree_diff1['differences']) != len(tree_diff2['differences']):
            raise_assertion('length of differences different')

        for (file1, differences1), (file2, differences2) in \
                zip(tree_diff1['differences'].items(), tree_diff2['differences'].items()):

            if file1 != file2:
                raise_assertion(f"filename different: {file1}, {file2}")

            if len(differences1) != len(differences2):
                raise_assertion("length of file differences different")

            for (difference1_kind, difference1_values), (difference2_kind, difference2_values) in \
                    zip(differences1.items(), differences2.items()):
                if difference1_kind != difference2_kind:
                    raise_assertion(f"different difference kinds: {difference1_kind}, {difference2_kind}")

                if difference1_values[0] is not None \
                        and difference2_values[0] is not None \
                        and difference1_values[0] != difference2_values[0]:
                    raise_assertion(f"before values are different: {difference1_values[0]}, {difference2_values[0]}")

                if difference1_values[1] is not None \
                        and difference2_values[1] is not None \
                        and difference1_values[1] != difference2_values[1]:
                    raise_assertion(f"after values are different: {difference1_values[1]}, {difference2_values[1]}")


def _sorted_tree(tree):
    sorted_tree = json.loads(json.dumps(tree, sort_keys=True))
    sorted_tree["added_files"] = sorted(sorted_tree["added_files"])
    sorted_tree["deleted_files"] = sorted(sorted_tree["deleted_files"])

    return sorted_tree


class TreeAssertionError(AssertionError):
    def __init__(self, msg, tree_diff1, tree_diff2):
        diff = ('\n'.join(difflib.ndiff(
            pprint.pformat(tree_diff1).splitlines(),
            pprint.pformat(tree_diff2).splitlines())))
        super().__init__(f"{msg}\n\n{diff}")