@contextmanager
def workspace():
    """
    Provides a workspace folder within the project root and changes into it.
    The workspace folder can be used to perform tests.
    """
    with tempfile.TemporaryDirectory(dir=get_project_root(), prefix="tmp-pytest-") as workspace_dir:
        chown_to_user(workspace_dir)
        current_cwd = os.getcwd()
        os.chdir(workspace_dir)

        try:
            yield workspace_dir
        finally:
            os.chdir(current_cwd) def get_random_string(length):
    return ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(length)) def get_project_root():
    return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) def suppress_chown_during_debuild(monkeypatch):
    def fakechown(*_):
        pass

    if get_user() == 'root':  # debuild case
        monkeypatch.setattr(shutil, 'chown', fakechown) def run(popenargs, sudo=False, input=None, timeout=None, check=True, universal_newlines=True,
        stdout=_ADAPTIVE, log_threshold=logging.DEBUG,
        **kwargs):
    """
    Small wrapper around subprocess.run().
    """

    assert type(popenargs) is list

    subprocess_stdout = stdout

    if subprocess_stdout == _ADAPTIVE:
        if logging.getLogger().isEnabledFor(log_threshold):
            subprocess_stdout = None
        else:
            subprocess_stdout = subprocess.PIPE

    all_args = list()
    if not sudo and os.getuid() == 0:
        current_user = get_user()
        if current_user != 'root':
            # drop privileges
            all_args.extend(['sudo', '-u', current_user])
    elif sudo and os.getuid() != 0:
        all_args.append('sudo')

    all_args.extend(popenargs)

    logging.log(log_threshold, "Running command: {0}".format(all_args))

    result = mockablerun.run_mockable(all_args, input=input, timeout=timeout, check=check,
                                      universal_newlines=universal_newlines,
                                      stdout=subprocess_stdout, **kwargs)

    if (logging.getLogger().isEnabledFor(log_threshold) and
            subprocess_stdout is subprocess.PIPE):
        logging.log(log_threshold, result.stdout)

    return result def get_debian_architecture():
    cmd = ['dpkg', '--print-architecture']
    return run(cmd, stdout=subprocess.PIPE).stdout.strip('\n') @require('lxc', lxd_install_hint, LxdVersion.check)
def get_server_image_compression_algorithm():
    cmd = [lxc_exec(), 'config', 'get', 'images.compression_algorithm']
    algorithm = run(cmd, stdout=subprocess.PIPE).stdout.strip('\n')
    if not algorithm:
        return 'gzip'
    else:
        return algorithm def lxc_exec():
    return Executables.get('lxc') def get_file_extension_from_image_compression_algorithm(algorithm):
    mapping = {
        'bzip2': '.tar.bz2',
        'gzip': '.tar.gz',
        'lzma': '.tar.lzma',
        'xz': '.tar.xz',
        'none': '.tar',
    }

    extension = mapping.get(algorithm, None)
    if not extension:
        raise FatalError(('''Unhandled lxc image compression algorithm '{}'.'''
                          ).format(algorithm))

    return extension class Create(Image):

    def __init__(self):
        super().__init__()
        self.section = 'postprocessing_commands'

    @classmethod
    def advertise(cls, subparsers):
        help_text = "create a re-distributable image"
        description_text = "Create a re-distributable image."
        parser = subparsers.add_parser(cls._get_short_command_name(),
                                       help=help_text,
                                       description=description_text)
        cls._offer_options(parser, introspection=True, clean=True)
        cls._require_config_file(parser)

    def run_cli(self, cli_args):
        self._dispatch(*self._unpack_cli_args(cli_args), run_method=self._get_run_method(cli_args))

    def dry_run(self, config_file):
        return self._dispatch(config_file, run_method=self._dry_run)

    def _dry_run(self):
        plugins = {}
        if self._input_artifact() is not None:
            plugins.update(Export().dry_run(self.config.get_base_config_file()))
        command_runner = CommandRunner(self.config, self.section, self._input_artifact())
        plugins.update(command_runner.get_plugin_report())
        return plugins

    def run(self, config_file):
        return self._dispatch(config_file, run_method=self._run)

    def _run(self):
        command_runner = CommandRunner(self.config, self.section, self._input_artifact())

        if command_runner.require_root():
            self._require_sudo()

        if self._input_artifact() is not None:
            Export().run(self.config.get_base_config_file())
        else:
            logging.info("Creating new image without bootstrapping other artifacts.")

        print("Going to post process image - be patient.")

        result = command_runner.run()

        if result:
            print_success(("Completed the image creation post processing commands.\n"
                           "The following artifacts are now available:\n- {}".format('\n- '.join(result))))
        return result

    def clean_recursive(self, config_file, depth):
        self.clean_depth = depth
        self._dispatch(config_file, run_method=self._clean)

    def clean(self, config_file):
        self._dispatch(config_file, run_method=self._clean)

    def _clean(self):
        command_runner = CommandRunner(self.config, self.section, self._input_artifact())
        if command_runner.require_root_for_clean():
            self._require_sudo()
        command_runner.clean()

        if self.clean_depth > 0 and self._input_artifact() is not None:
            Export().clean_recursive(self.config.get_base_config_file(), self.clean_depth - 1)

    def _dispatch(self, config_file, run_method):
        with command_context({'edi_create_distributable_image': True}):
            self._setup_parser(config_file)
            return run_method()

    def _input_artifact(self):
        if self.config.has_bootstrap_node():
            return Export().result(self.config.get_base_config_file())
        else:
            return None def get_artifact_dir():
    return os.path.join(get_workdir(), 'artifacts')
import pytest
import os
import edi
import subprocess
from tests.libtesting.contextmanagers.workspace import workspace
from tests.libtesting.helpers import get_random_string, get_project_root, suppress_chown_during_debuild
from edi.lib.shellhelpers import run, get_debian_architecture
from edi.lib.lxchelpers import (get_server_image_compression_algorithm, lxc_exec,
                                get_file_extension_from_image_compression_algorithm)
from edi.commands.imagecommands.create import Create
from edi.lib.helpers import get_artifact_dir
# -*- coding: utf-8 -*-
# Copyright (C) 2017 Matthias Luescher
#
# Authors:
#  Matthias Luescher
#
# This file is part of edi.
#
# edi is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# edi is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with edi.  If not, see <http://www.gnu.org/licenses/>.




@pytest.mark.requires_lxc
@pytest.mark.requires_ansible
@pytest.mark.requires_debootstrap
@pytest.mark.requires_sudo
def test_create_buster_image(capsys):
    print(os.getcwd())
    with workspace():
        edi_exec = os.path.join(get_project_root(), 'bin', 'edi')
        project_name = 'pytest-{}'.format(get_random_string(6))
        config_command = [edi_exec, 'config', 'init', project_name,
                          'debian-buster-{}'.format(get_debian_architecture())]
        run(config_command)  # run as non root

        parser = edi._setup_command_line_interface()
        cli_args = parser.parse_args(['image', 'create', '{}-develop.yml'.format(project_name)])

        Create().run_cli(cli_args)
        out, err = capsys.readouterr()
        print(out)
        assert not err

        lxc_compression_algo = get_server_image_compression_algorithm()
        lxc_export_extension = get_file_extension_from_image_compression_algorithm(lxc_compression_algo)

        images = [
            os.path.join(get_artifact_dir(), '{}-develop_edicommand_image_bootstrap_di.tar.gz'.format(project_name)),

            os.path.join(get_artifact_dir(), '{}-develop_edicommand_lxc_prepare_di.tar.gz'.format(project_name)),