class FatalError(Error):
    """Exception raised for fatal errors needing corrective actions from user.

    Attributes:
        message -- explanation of the error
    """

    def __init__(self, message):
        self.message = message @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 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 def lxc_exec():
    return Executables.get('lxc') def get_lxd_version():
    if not Executables.has('lxd'):
        return '0.0.0'

    cmd = [Executables.get('lxd'), '--version']
    result = run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    if not result.stdout:
        # recent lxd versions print the version to stderr!
        return result.stderr.strip('\n')
    else:
        return result.stdout.strip('\n') class LxdVersion:
    """
    Make sure that the lxc/lxd version is >= 3.0.0.
    """
    _check_done = False
    _required_minimal_version = '3.0.0'

    def __init__(self, clear_cache=False):
        if clear_cache:
            LxdVersion._check_done = False

    @staticmethod
    def check():
        if LxdVersion._check_done:
            return

        if Version(get_stripped_version(get_lxd_version())) < Version(LxdVersion._required_minimal_version):
            raise FatalError(('The current lxd installation ({}) does not meet the minimal requirements (>={}).\n'
                              'Please update your lxd installation using snaps or xenial-backports!'
                              ).format(get_lxd_version(), LxdVersion._required_minimal_version))
        else:
            LxdVersion._check_done = True @require('lxc', lxd_install_hint, LxdVersion.check)
def is_bridge_available(bridge_name):
    cmd = [lxc_exec(), "network", "list", "--format=json"]
    result = run(cmd, stdout=subprocess.PIPE)

    try:
        parsed_result = yaml.safe_load(result.stdout)
        bridge_names = [item.get("name") for item in parsed_result]
        if bridge_name in bridge_names:
            return True
        else:
            return False
    except yaml.YAMLError as exc:
        raise FatalError("Unable to parse lxc output ({}).".format(exc)) @require('lxc', lxd_install_hint, LxdVersion.check)
def create_bridge(bridge_name):
    cmd = [lxc_exec(), "network", "create", bridge_name]
    run(cmd) _ADAPTIVE = -42
def run(popenargs, sudo=False, input=None, timeout=None, check=True, universal_newlines=True,
        stdout=_ADAPTIVE, log_threshold=logging.DEBUG,
        **kwargs):
def get_chroot_cmd(rootfs):
def get_environment_variable(name, default=None):
def get_current_display():
def get_user_home_directory(username):
def get_debian_architecture():
def safely_remove_artifacts_folder(abs_folder_path, sudo=False):
def mount_aware_tempdir(parent_dir, log_warning=False):
def gpg_agent(directory):
    def __init__(self, clear_cache=False):
    def has(executable):
    def get(cls, executable):
def require(executable, installation_command=None, version_check=None):
    def require_decorator(func):
        def func_wrapper(*args, **kwargs):
class Executables: def get_command(popenargs):
    def get_real_command(cmd, expected_position):
        if cmd[expected_position] == 'env':
            return cmd[expected_position + 2]
        else:
            return cmd[expected_position]

    command = popenargs[0]
    if command[0] == 'sudo':
        if command[1] == '-u':
            return get_real_command(command, 3)
        else:
            return get_real_command(command, 1)
    else:
        return get_real_command(command, 0) def get_sub_command(popenargs):
    main_command = get_command(popenargs)
    return get_command_parameter(popenargs, main_command) @contextmanager
def mocked_executable(executable, mock=str(os.path.join(os.sep, 'bin', 'true'))):
    """
    Mocks away an executable that gets fetched using the Executables class.
    :param executable: The executable to be mocked.
    :param mock: The replacement for the mocked executable.
    :return: The mock.
    """
    Executables._cache[executable] = mock

    try:
        yield mock
    finally:
        del Executables._cache[executable] @contextmanager
def mocked_lxd_version_check():
    LxdVersion._check_done = True
    try:
        yield
    finally:
        LxdVersion._check_done = False
import subprocess
import pytest
from subprocess import CalledProcessError
from contextlib import contextmanager
from edi.lib.helpers import FatalError
from edi.lib.lxchelpers import (get_server_image_compression_algorithm,
                                get_file_extension_from_image_compression_algorithm, lxc_exec,
                                get_lxd_version, LxdVersion, is_bridge_available, create_bridge)
from edi.lib.shellhelpers import mockablerun, run
from tests.libtesting.helpers import get_command, get_sub_command
from tests.libtesting.contextmanagers.mocked_executable import mocked_executable, mocked_lxd_version_check
# -*- 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
def test_get_server_image_compression():
    result = get_server_image_compression_algorithm()
    assert result in ['bzip2', 'gzip', 'lzma', 'xz', 'none']


def test_get_server_image_compression_bzip2(monkeypatch):
    with mocked_executable('lxc', '/here/is/no/lxc'):
        with mocked_lxd_version_check():
            def fake_lxc_config_command(*popenargs, **kwargs):
                if get_command(popenargs).endswith('lxc') and get_sub_command(popenargs) == 'config':
                    return subprocess.CompletedProcess("fakerun", 0, stdout='bzip2')
                else:
                    return subprocess.run(*popenargs, **kwargs)

            monkeypatch.setattr(mockablerun, 'run_mockable', fake_lxc_config_command)
            result = get_server_image_compression_algorithm()
            assert result == 'bzip2'


@pytest.mark.parametrize("algorithm, expected_extension", [
    ("none", ".tar"),
    ("bzip2", ".tar.bz2"),
    ("gzip", ".tar.gz"),
])
def test_get_file_extension_from_image_compression_algorithm(algorithm, expected_extension):
    extension = get_file_extension_from_image_compression_algorithm(algorithm)
    assert extension == expected_extension


def test_get_file_extension_from_image_compression_algorithm_failure():
    with pytest.raises(FatalError) as e:
        get_file_extension_from_image_compression_algorithm('42')

    assert 'compression algorithm' in str(e)
    assert '42' in str(e)


@pytest.mark.requires_lxc
def test_get_lxd_version():
    version = get_lxd_version()
    assert '.' in version


@contextmanager
def clear_lxd_version_check_cache():
    try:
        LxdVersion(clear_cache=True)
        yield
    finally:
        LxdVersion(clear_cache=True)


@pytest.mark.requires_lxc
def test_lxd_version_check():
    with clear_lxd_version_check_cache():
        check_method = LxdVersion.check
        check_method()


def patch_lxd_get_version(monkeypatch, fake_version):
    def fake_lxd_version_command(*popenargs, **kwargs):
        if get_command(popenargs).endswith('lxd') and get_sub_command(popenargs) == '--version':
            return subprocess.CompletedProcess("fakerun", 0,
                                               stdout='{}\n'.format(fake_version))
        else:
            return subprocess.run(*popenargs, **kwargs)

    monkeypatch.setattr(mockablerun, 'run_mockable', fake_lxd_version_command)


def test_invalid_version(monkeypatch):
    patch_lxd_get_version(monkeypatch, '2.2.0')
    with mocked_executable('lxd', '/here/is/no/lxd'):
        with clear_lxd_version_check_cache():
            check_method = LxdVersion.check

            with pytest.raises(FatalError) as error:
                check_method()

            assert '2.2.0' in error.value.message
            assert '>=3.0.0' in error.value.message
            assert 'xenial-backports' in error.value.message


def test_valid_version(monkeypatch):
    patch_lxd_get_version(monkeypatch, '3.0.0+bingo')
    with mocked_executable('lxd', '/here/is/no/lxd'):
        with clear_lxd_version_check_cache():
            LxdVersion.check()


@pytest.mark.requires_lxc
def test_is_bridge_available():
    bridge_name = "edibrtesttest42"

    assert not is_bridge_available(bridge_name)