heat-cfntools/heat_cfntools/tests/test_cfn_helper.py

#
# Copyright 2013 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import json
import os
import tempfile
from unittest import mock

import boto.cloudformation as cfn
import fixtures
import testtools
import testtools.matchers as ttm

from heat_cfntools.cfntools import cfn_helper


def popen_root_calls(calls, shell=False):
    kwargs = {'env': None, 'cwd': None, 'stderr': -1, 'stdout': -1,
              'shell': shell}
    return [
        mock.call(call, **kwargs)
        for call in calls
    ]


class FakePOpen(object):
    def __init__(self, stdout='', stderr='', returncode=0):
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr

    def communicate(self):
        return (self.stdout, self.stderr)

    def wait(self):
        pass


@mock.patch.object(cfn_helper.pwd, 'getpwnam')
@mock.patch.object(cfn_helper.os, 'seteuid')
@mock.patch.object(cfn_helper.os, 'geteuid')
class TestCommandRunner(testtools.TestCase):

    def test_command_runner(self, mock_geteuid, mock_seteuid, mock_getpwnam):
        def returns(*args, **kwargs):
            if args[0][0] == '/bin/command1':
                return FakePOpen('All good')
            elif args[0][0] == '/bin/command2':
                return FakePOpen('Doing something', 'error', -1)
            else:
                raise Exception('This should never happen')

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            cmd2 = cfn_helper.CommandRunner(['/bin/command2'])
            cmd1 = cfn_helper.CommandRunner(['/bin/command1'],
                                            nextcommand=cmd2)
            cmd1.run('root')
            self.assertEqual(
                'CommandRunner:\n\tcommand: [\'/bin/command1\']\n\tstdout: '
                'All good',
                str(cmd1))
            self.assertEqual(
                'CommandRunner:\n\tcommand: [\'/bin/command2\']\n\tstatus: '
                '-1\n\tstdout: Doing something\n\tstderr: error',
                str(cmd2))
            calls = popen_root_calls([['/bin/command1'], ['/bin/command2']])
            mock_popen.assert_has_calls(calls)

    def test_privileges_are_lowered_for_non_root_user(self, mock_geteuid,
                                                      mock_seteuid,
                                                      mock_getpwnam):
        pw_entry = mock.Mock()
        pw_entry.pw_uid = 1001
        mock_getpwnam.return_value = pw_entry
        mock_geteuid.return_value = 0
        calls = [mock.call(1001), mock.call(0)]
        with mock.patch('subprocess.Popen') as mock_popen:
            command = ['/bin/command', '--option=value', 'arg1', 'arg2']
            cmd = cfn_helper.CommandRunner(command)
            cmd.run(user='nonroot')
            self.assertTrue(mock_geteuid.called)
            mock_getpwnam.assert_called_once_with('nonroot')
            mock_seteuid.assert_has_calls(calls)
            self.assertTrue(mock_popen.called)

    def test_run_returns_when_cannot_set_privileges(self, mock_geteuid,
                                                    mock_seteuid,
                                                    mock_getpwnam):
        msg = '[Error 1] Permission Denied'
        mock_seteuid.side_effect = Exception(msg)
        with mock.patch('subprocess.Popen') as mock_popen:
            command = ['/bin/command2']
            cmd = cfn_helper.CommandRunner(command)
            cmd.run(user='nonroot')
            self.assertTrue(mock_getpwnam.called)
            self.assertTrue(mock_seteuid.called)
            self.assertFalse(mock_popen.called)
            self.assertEqual(126, cmd.status)
            self.assertEqual(msg, cmd.stderr)

    def test_privileges_are_restored_for_command_failure(self, mock_geteuid,
                                                         mock_seteuid,
                                                         mock_getpwnam):
        pw_entry = mock.Mock()
        pw_entry.pw_uid = 1001
        mock_getpwnam.return_value = pw_entry
        mock_geteuid.return_value = 0
        calls = [mock.call(1001), mock.call(0)]
        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = ValueError('Something wrong')
            command = ['/bin/command', '--option=value', 'arg1', 'arg2']
            cmd = cfn_helper.CommandRunner(command)
            self.assertRaises(ValueError, cmd.run, user='nonroot')
            self.assertTrue(mock_geteuid.called)
            mock_getpwnam.assert_called_once_with('nonroot')
            mock_seteuid.assert_has_calls(calls)
            self.assertTrue(mock_popen.called)


@mock.patch.object(cfn_helper, 'controlled_privileges')
class TestPackages(testtools.TestCase):

    def test_yum_install(self, mock_cp):

        def returns(*args, **kwargs):
            if args[0][0] == 'rpm' and args[0][1] == '-q':
                return FakePOpen(returncode=1)
            else:
                return FakePOpen(returncode=0)

        calls = [['which', 'yum']]
        for pack in ('httpd', 'wordpress', 'mysql-server'):
            calls.append(['rpm', '-q', pack])
            calls.append(['yum', '-y', '--showduplicates', 'list',
                          'available', pack])
        calls = popen_root_calls(calls)

        packages = {
            "yum": {
                "mysql-server": [],
                "httpd": [],
                "wordpress": []
            }
        }

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            cfn_helper.PackagesHandler(packages).apply_packages()
            mock_popen.assert_has_calls(calls, any_order=True)

    def test_dnf_install_yum_unavailable(self, mock_cp):

        def returns(*args, **kwargs):
            if ((args[0][0] == 'rpm' and args[0][1] == '-q')
                    or (args[0][0] == 'which' and args[0][1] == 'yum')):
                return FakePOpen(returncode=1)
            else:
                return FakePOpen(returncode=0)

        calls = [['which', 'yum']]
        for pack in ('httpd', 'wordpress', 'mysql-server'):
            calls.append(['rpm', '-q', pack])
            calls.append(['dnf', '-y', '--showduplicates', 'list',
                          'available', pack])
        calls = popen_root_calls(calls)

        packages = {
            "yum": {
                "mysql-server": [],
                "httpd": [],
                "wordpress": []
            }
        }

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            cfn_helper.PackagesHandler(packages).apply_packages()
            mock_popen.assert_has_calls(calls, any_order=True)

    def test_dnf_install(self, mock_cp):

        def returns(*args, **kwargs):
            if args[0][0] == 'rpm' and args[0][1] == '-q':
                return FakePOpen(returncode=1)
            else:
                return FakePOpen(returncode=0)

        calls = []
        for pack in ('httpd', 'wordpress', 'mysql-server'):
            calls.append(['rpm', '-q', pack])
            calls.append(['dnf', '-y', '--showduplicates', 'list',
                          'available', pack])
        calls = popen_root_calls(calls)

        packages = {
            "dnf": {
                "mysql-server": [],
                "httpd": [],
                "wordpress": []
            }
        }

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            cfn_helper.PackagesHandler(packages).apply_packages()
            mock_popen.assert_has_calls(calls, any_order=True)

    def test_zypper_install(self, mock_cp):

        def returns(*args, **kwargs):
            if args[0][0].startswith('rpm') and args[0][1].startswith('-q'):
                return FakePOpen(returncode=1)
            else:
                return FakePOpen(returncode=0)

        calls = []
        for pack in ('httpd', 'wordpress', 'mysql-server'):
            calls.append(['rpm', '-q', pack])
            calls.append(['zypper', '-n', '--no-refresh', 'search', pack])
        calls = popen_root_calls(calls)

        packages = {
            "zypper": {
                "mysql-server": [],
                "httpd": [],
                "wordpress": []
            }
        }

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            cfn_helper.PackagesHandler(packages).apply_packages()
            mock_popen.assert_has_calls(calls, any_order=True)

    def test_apt_install(self, mock_cp):
        packages = {
            "apt": {
                "mysql-server": [],
                "httpd": [],
                "wordpress": []
            }
        }

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.return_value = FakePOpen(returncode=0)
            cfn_helper.PackagesHandler(packages).apply_packages()
            self.assertTrue(mock_popen.called)


@mock.patch.object(cfn_helper, 'controlled_privileges')
class TestServicesHandler(testtools.TestCase):

    def test_services_handler_systemd(self, mock_cp):
        calls = []
        returns = []

        # apply_services
        calls.append(['/bin/systemctl', 'enable', 'httpd.service'])
        returns.append(FakePOpen())
        calls.append(['/bin/systemctl', 'status', 'httpd.service'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/bin/systemctl', 'start', 'httpd.service'])
        returns.append(FakePOpen())
        calls.append(['/bin/systemctl', 'enable', 'mysqld.service'])
        returns.append(FakePOpen())
        calls.append(['/bin/systemctl', 'status', 'mysqld.service'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/bin/systemctl', 'start', 'mysqld.service'])
        returns.append(FakePOpen())

        # monitor_services not running
        calls.append(['/bin/systemctl', 'status', 'httpd.service'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/bin/systemctl', 'start', 'httpd.service'])
        returns.append(FakePOpen())

        calls = popen_root_calls(calls)

        calls.extend(popen_root_calls(['/bin/services_restarted'], shell=True))
        returns.append(FakePOpen())

        calls.extend(popen_root_calls([['/bin/systemctl', 'status',
                                        'mysqld.service']]))
        returns.append(FakePOpen(returncode=-1))
        calls.extend(popen_root_calls([['/bin/systemctl', 'start',
                                        'mysqld.service']]))
        returns.append(FakePOpen())

        calls.extend(popen_root_calls(['/bin/services_restarted'], shell=True))
        returns.append(FakePOpen())

        # monitor_services running
        calls.extend(popen_root_calls([['/bin/systemctl', 'status',
                                        'httpd.service']]))
        returns.append(FakePOpen())
        calls.extend(popen_root_calls([['/bin/systemctl', 'status',
                                        'mysqld.service']]))
        returns.append(FakePOpen())

        services = {
            "systemd": {
                "mysqld": {"enabled": "true", "ensureRunning": "true"},
                "httpd": {"enabled": "true", "ensureRunning": "true"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.return_value = True
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.side_effect = returns

                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                # services not running
                sh.monitor_services()

                # services running
                sh.monitor_services()
                mock_popen.assert_has_calls(calls, any_order=True)
            mock_exists.assert_called_with('/bin/systemctl')

    def test_services_handler_systemd_disabled(self, mock_cp):
        calls = []

        # apply_services
        calls.append(['/bin/systemctl', 'disable', 'httpd.service'])
        calls.append(['/bin/systemctl', 'status', 'httpd.service'])
        calls.append(['/bin/systemctl', 'stop', 'httpd.service'])
        calls.append(['/bin/systemctl', 'disable', 'mysqld.service'])
        calls.append(['/bin/systemctl', 'status', 'mysqld.service'])
        calls.append(['/bin/systemctl', 'stop', 'mysqld.service'])
        calls = popen_root_calls(calls)

        services = {
            "systemd": {
                "mysqld": {"enabled": "false", "ensureRunning": "false"},
                "httpd": {"enabled": "false", "ensureRunning": "false"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]
        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.return_value = True
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.return_value = FakePOpen()
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                mock_popen.assert_has_calls(calls, any_order=True)
            mock_exists.assert_called_with('/bin/systemctl')

    def test_services_handler_sysv_service_chkconfig(self, mock_cp):

        def exists(*args, **kwargs):
            return args[0] != '/bin/systemctl'

        calls = []
        returns = []

        # apply_services
        calls.append(['/sbin/chkconfig', 'httpd', 'on'])
        returns.append(FakePOpen())
        calls.append(['/sbin/service', 'httpd', 'status'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/sbin/service', 'httpd', 'start'])
        returns.append(FakePOpen())

        # monitor_services not running
        calls.append(['/sbin/service', 'httpd', 'status'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/sbin/service', 'httpd', 'start'])
        returns.append(FakePOpen())

        calls = popen_root_calls(calls)

        calls.extend(popen_root_calls(['/bin/services_restarted'], shell=True))
        returns.append(FakePOpen())

        # monitor_services running
        calls.extend(popen_root_calls([['/sbin/service', 'httpd', 'status']]))
        returns.append(FakePOpen())

        services = {
            "sysvinit": {
                "httpd": {"enabled": "true", "ensureRunning": "true"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.side_effect = exists
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.side_effect = returns
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                # services not running
                sh.monitor_services()

                # services running
                sh.monitor_services()
                mock_popen.assert_has_calls(calls)
            mock_exists.assert_any_call('/bin/systemctl')
            mock_exists.assert_any_call('/sbin/service')
            mock_exists.assert_any_call('/sbin/chkconfig')

    def test_services_handler_sysv_disabled_service_chkconfig(self, mock_cp):
        def exists(*args, **kwargs):
            return args[0] != '/bin/systemctl'

        calls = []

        # apply_services
        calls.append(['/sbin/chkconfig', 'httpd', 'off'])
        calls.append(['/sbin/service', 'httpd', 'status'])
        calls.append(['/sbin/service', 'httpd', 'stop'])

        calls = popen_root_calls(calls)

        services = {
            "sysvinit": {
                "httpd": {"enabled": "false", "ensureRunning": "false"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.side_effect = exists
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.return_value = FakePOpen()
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                mock_popen.assert_has_calls(calls)
            mock_exists.assert_any_call('/bin/systemctl')
            mock_exists.assert_any_call('/sbin/service')
            mock_exists.assert_any_call('/sbin/chkconfig')

    def test_services_handler_sysv_systemctl(self, mock_cp):
        calls = []
        returns = []

        # apply_services
        calls.append(['/bin/systemctl', 'enable', 'httpd.service'])
        returns.append(FakePOpen())
        calls.append(['/bin/systemctl', 'status', 'httpd.service'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/bin/systemctl', 'start', 'httpd.service'])
        returns.append(FakePOpen())

        # monitor_services not running
        calls.append(['/bin/systemctl', 'status', 'httpd.service'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/bin/systemctl', 'start', 'httpd.service'])
        returns.append(FakePOpen())

        shell_calls = []
        shell_calls.append('/bin/services_restarted')
        returns.append(FakePOpen())

        calls = popen_root_calls(calls)
        calls.extend(popen_root_calls(shell_calls, shell=True))

        # monitor_services running
        calls.extend(popen_root_calls([['/bin/systemctl', 'status',
                                        'httpd.service']]))
        returns.append(FakePOpen())

        services = {
            "sysvinit": {
                "httpd": {"enabled": "true", "ensureRunning": "true"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.return_value = True
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.side_effect = returns
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                # services not running
                sh.monitor_services()

                # services running
                sh.monitor_services()
                mock_popen.assert_has_calls(calls)
            mock_exists.assert_called_with('/bin/systemctl')

    def test_services_handler_sysv_disabled_systemctl(self, mock_cp):
        calls = []

        # apply_services
        calls.append(['/bin/systemctl', 'disable', 'httpd.service'])
        calls.append(['/bin/systemctl', 'status', 'httpd.service'])
        calls.append(['/bin/systemctl', 'stop', 'httpd.service'])

        calls = popen_root_calls(calls)

        services = {
            "sysvinit": {
                "httpd": {"enabled": "false", "ensureRunning": "false"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.return_value = True
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.return_value = FakePOpen()
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                mock_popen.assert_has_calls(calls)
            mock_exists.assert_called_with('/bin/systemctl')

    def test_services_handler_sysv_service_updaterc(self, mock_cp):
        calls = []
        returns = []

        # apply_services
        calls.append(['/usr/sbin/update-rc.d', 'httpd', 'enable'])
        returns.append(FakePOpen())
        calls.append(['/usr/sbin/service', 'httpd', 'status'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/usr/sbin/service', 'httpd', 'start'])
        returns.append(FakePOpen())

        # monitor_services not running
        calls.append(['/usr/sbin/service', 'httpd', 'status'])
        returns.append(FakePOpen(returncode=-1))
        calls.append(['/usr/sbin/service', 'httpd', 'start'])
        returns.append(FakePOpen())

        shell_calls = []
        shell_calls.append('/bin/services_restarted')
        returns.append(FakePOpen())

        calls = popen_root_calls(calls)
        calls.extend(popen_root_calls(shell_calls, shell=True))

        # monitor_services running
        calls.extend(popen_root_calls([['/usr/sbin/service', 'httpd',
                                        'status']]))
        returns.append(FakePOpen())

        services = {
            "sysvinit": {
                "httpd": {"enabled": "true", "ensureRunning": "true"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.return_value = False
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.side_effect = returns
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                # services not running
                sh.monitor_services()

                # services running
                sh.monitor_services()
                mock_popen.assert_has_calls(calls)
            mock_exists.assert_any_call('/bin/systemctl')
            mock_exists.assert_any_call('/sbin/service')
            mock_exists.assert_any_call('/sbin/chkconfig')

    def test_services_handler_sysv_disabled_service_updaterc(self, mock_cp):
        calls = []
        returns = []

        # apply_services
        calls.append(['/usr/sbin/update-rc.d', 'httpd', 'disable'])
        returns.append(FakePOpen())
        calls.append(['/usr/sbin/service', 'httpd', 'status'])
        returns.append(FakePOpen())
        calls.append(['/usr/sbin/service', 'httpd', 'stop'])
        returns.append(FakePOpen())

        calls = popen_root_calls(calls)

        services = {
            "sysvinit": {
                "httpd": {"enabled": "false", "ensureRunning": "false"}
            }
        }
        hooks = [
            cfn_helper.Hook(
                'hook1',
                'service.restarted',
                'Resources.resource1.Metadata',
                'root',
                '/bin/services_restarted')
        ]

        with mock.patch('os.path.exists') as mock_exists:
            mock_exists.return_value = False
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.side_effect = returns
                sh = cfn_helper.ServicesHandler(services, 'resource1', hooks)
                sh.apply_services()
                mock_popen.assert_has_calls(calls)
            mock_exists.assert_any_call('/bin/systemctl')
            mock_exists.assert_any_call('/sbin/service')
            mock_exists.assert_any_call('/sbin/chkconfig')


class TestHupConfig(testtools.TestCase):

    def test_load_main_section(self):
        fcreds = tempfile.NamedTemporaryFile()
        fcreds.write('AWSAccessKeyId=foo\nAWSSecretKey=bar\n'.encode('UTF-8'))
        fcreds.flush()

        main_conf = tempfile.NamedTemporaryFile()
        main_conf.write(('''[main]
stack=teststack
credential-file=%s''' % fcreds.name).encode('UTF-8'))
        main_conf.flush()
        mainconfig = cfn_helper.HupConfig([open(main_conf.name)])
        self.assertEqual(
            '{stack: teststack, credential_file: %s, '
            'region: nova, interval:10}' % fcreds.name,
            str(mainconfig))
        main_conf.close()

        main_conf = tempfile.NamedTemporaryFile()
        main_conf.write(('''[main]
stack=teststack
region=region1
credential-file=%s-invalid
interval=120''' % fcreds.name).encode('UTF-8'))
        main_conf.flush()
        e = self.assertRaises(cfn_helper.InvalidCredentialsException,
                              cfn_helper.HupConfig,
                              [open(main_conf.name)])
        self.assertIn('invalid credentials file', str(e))
        fcreds.close()

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_hup_config(self, mock_cp):
        hooks_conf = tempfile.NamedTemporaryFile()

        def write_hook_conf(f, name, triggers, path, action):
            f.write((
                '[%s]\ntriggers=%s\npath=%s\naction=%s\nrunas=root\n\n' % (
                    name, triggers, path, action)).encode('UTF-8'))

        write_hook_conf(
            hooks_conf,
            'hook2',
            'service2.restarted',
            'Resources.resource2.Metadata',
            '/bin/hook2')
        write_hook_conf(
            hooks_conf,
            'hook1',
            'service1.restarted',
            'Resources.resource1.Metadata',
            '/bin/hook1')
        write_hook_conf(
            hooks_conf,
            'hook3',
            'service3.restarted',
            'Resources.resource3.Metadata',
            '/bin/hook3')
        write_hook_conf(
            hooks_conf,
            'cfn-http-restarted',
            'service.restarted',
            'Resources.resource.Metadata',
            '/bin/cfn-http-restarted')
        hooks_conf.flush()

        fcreds = tempfile.NamedTemporaryFile()
        fcreds.write('AWSAccessKeyId=foo\nAWSSecretKey=bar\n'.encode('UTF-8'))
        fcreds.flush()

        main_conf = tempfile.NamedTemporaryFile()
        main_conf.write(('''[main]
stack=teststack
credential-file=%s
region=region1
interval=120''' % fcreds.name).encode('UTF-8'))
        main_conf.flush()

        mainconfig = cfn_helper.HupConfig([
            open(main_conf.name),
            open(hooks_conf.name)])
        unique_resources = mainconfig.unique_resources_get()
        self.assertThat([
            'resource',
            'resource1',
            'resource2',
            'resource3',
        ], ttm.Equals(sorted(unique_resources)))

        hooks = sorted(mainconfig.hooks,
                       key=lambda hook: hook.resource_name_get())
        self.assertEqual(len(hooks), 4)
        self.assertEqual(
            '{cfn-http-restarted, service.restarted,'
            ' Resources.resource.Metadata, root, /bin/cfn-http-restarted}',
            str(hooks[0]))
        self.assertEqual(
            '{hook1, service1.restarted, Resources.resource1.Metadata,'
            ' root, /bin/hook1}', str(hooks[1]))
        self.assertEqual(
            '{hook2, service2.restarted, Resources.resource2.Metadata,'
            ' root, /bin/hook2}', str(hooks[2]))
        self.assertEqual(
            '{hook3, service3.restarted, Resources.resource3.Metadata,'
            ' root, /bin/hook3}', str(hooks[3]))

        calls = []
        calls.extend(popen_root_calls(['/bin/cfn-http-restarted'], shell=True))
        calls.extend(popen_root_calls(['/bin/hook1'], shell=True))
        calls.extend(popen_root_calls(['/bin/hook2'], shell=True))
        calls.extend(popen_root_calls(['/bin/hook3'], shell=True))

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.return_value = FakePOpen('All good')

            for hook in hooks:
                hook.event(hook.triggers, None, hook.resource_name_get())

            hooks_conf.close()
            fcreds.close()
            main_conf.close()
            mock_popen.assert_has_calls(calls)


class TestCfnHelper(testtools.TestCase):

    def _check_metadata_content(self, content, value):
        with tempfile.NamedTemporaryFile() as metadata_info:
            metadata_info.write(content.encode('UTF-8'))
            metadata_info.flush()
            port = cfn_helper.metadata_server_port(metadata_info.name)
            self.assertEqual(value, port)

    def test_metadata_server_port(self):
        self._check_metadata_content("http://172.20.42.42:8000\n", 8000)

    def test_metadata_server_port_https(self):
        self._check_metadata_content("https://abc.foo.bar:6969\n", 6969)

    def test_metadata_server_port_noport(self):
        self._check_metadata_content("http://172.20.42.42\n", None)

    def test_metadata_server_port_justip(self):
        self._check_metadata_content("172.20.42.42", None)

    def test_metadata_server_port_weird(self):
        self._check_metadata_content("::::", None)
        self._check_metadata_content("beforecolons:aftercolons", None)

    def test_metadata_server_port_emptyfile(self):
        self._check_metadata_content("\n", None)
        self._check_metadata_content("", None)

    def test_metadata_server_nofile(self):
        random_filename = self.getUniqueString()
        self.assertIsNone(cfn_helper.metadata_server_port(random_filename))

    def test_to_boolean(self):
        self.assertTrue(cfn_helper.to_boolean(True))
        self.assertTrue(cfn_helper.to_boolean('true'))
        self.assertTrue(cfn_helper.to_boolean('yes'))
        self.assertTrue(cfn_helper.to_boolean('1'))
        self.assertTrue(cfn_helper.to_boolean(1))

        self.assertFalse(cfn_helper.to_boolean(False))
        self.assertFalse(cfn_helper.to_boolean('false'))
        self.assertFalse(cfn_helper.to_boolean('no'))
        self.assertFalse(cfn_helper.to_boolean('0'))
        self.assertFalse(cfn_helper.to_boolean(0))
        self.assertFalse(cfn_helper.to_boolean(None))
        self.assertFalse(cfn_helper.to_boolean('fingle'))

    def test_parse_creds_file(self):
        def parse_creds_test(file_contents, creds_match):
            with tempfile.NamedTemporaryFile(mode='w') as fcreds:
                fcreds.write(file_contents)
                fcreds.flush()
                creds = cfn_helper.parse_creds_file(fcreds.name)
                self.assertThat(creds_match, ttm.Equals(creds))
        parse_creds_test(
            'AWSAccessKeyId=foo\nAWSSecretKey=bar\n',
            {'AWSAccessKeyId': 'foo', 'AWSSecretKey': 'bar'}
        )
        parse_creds_test(
            'AWSAccessKeyId =foo\nAWSSecretKey= bar\n',
            {'AWSAccessKeyId': 'foo', 'AWSSecretKey': 'bar'}
        )
        parse_creds_test(
            'AWSAccessKeyId    =    foo\nAWSSecretKey    =    bar\n',
            {'AWSAccessKeyId': 'foo', 'AWSSecretKey': 'bar'}
        )


class TestMetadataRetrieve(testtools.TestCase):

    def setUp(self):
        super(TestMetadataRetrieve, self).setUp()
        self.tdir = self.useFixture(fixtures.TempDir())
        self.last_file = os.path.join(self.tdir.path, 'last_metadata')

    def test_metadata_retrieve_files(self):

        md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
            "/tmp/foo": {"content": "bar"}}}}}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)

        with tempfile.NamedTemporaryFile(mode='w+') as default_file:
            default_file.write(md_str)
            default_file.flush()
            self.assertThat(default_file.name, ttm.FileContains(md_str))

            self.assertTrue(
                md.retrieve(default_path=default_file.name,
                            last_path=self.last_file))

            self.assertThat(self.last_file, ttm.FileContains(md_str))
            self.assertThat(md_data, ttm.Equals(md._metadata))

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(default_path=default_file.name,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))

    def test_metadata_retrieve_none(self):

        md = cfn_helper.Metadata('teststack', None)
        default_file = os.path.join(self.tdir.path, 'default_file')

        self.assertFalse(md.retrieve(default_path=default_file,
                                     last_path=self.last_file))
        self.assertIsNone(md._metadata)

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display()
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(), "")

    def test_metadata_retrieve_passed(self):

        md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
            "/tmp/foo": {"content": "bar"}}}}}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(meta_str=md_data,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertEqual(md_str, str(md))

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display()
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(),
                         "{\"AWS::CloudFormation::Init\": {\"config\": {"
                         "\"files\": {\"/tmp/foo\": {\"content\": \"bar\"}"
                         "}}}}\n")

    def test_metadata_retrieve_by_key_passed(self):

        md_data = {"foo": {"bar": {"fred.1": "abcd"}}}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(meta_str=md_data,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertEqual(md_str, str(md))

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display("foo")
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(),
                         "{\"bar\": {\"fred.1\": \"abcd\"}}\n")

    def test_metadata_retrieve_by_nested_key_passed(self):

        md_data = {"foo": {"bar": {"fred.1": "abcd"}}}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(meta_str=md_data,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertEqual(md_str, str(md))

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display("foo.bar.'fred.1'")
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(),
                         '"abcd"\n')

    def test_metadata_retrieve_key_none(self):

        md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
            "/tmp/foo": {"content": "bar"}}}}}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(meta_str=md_data,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertEqual(md_str, str(md))

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display("no_key")
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(), "")

    def test_metadata_retrieve_by_nested_key_none(self):

        md_data = {"foo": {"bar": {"fred.1": "abcd"}}}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(meta_str=md_data,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertEqual(md_str, str(md))

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display("foo.fred")
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(), "")

    def test_metadata_retrieve_by_nested_key_none_with_matching_string(self):

        md_data = {"foo": "bar"}
        md_str = json.dumps(md_data)

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(md.retrieve(meta_str=md_data,
                                    last_path=self.last_file))
        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertEqual(md_str, str(md))

        displayed = self.useFixture(fixtures.StringStream('stdout'))
        fake_stdout = displayed.stream
        self.useFixture(fixtures.MonkeyPatch('sys.stdout', fake_stdout))
        md.display("foo.bar")
        fake_stdout.flush()
        self.assertEqual(displayed.getDetails()['stdout'].as_text(), "")

    def test_metadata_creates_cache(self):
        temp_home = tempfile.mkdtemp()

        def cleanup_temp_home(thome):
            os.unlink(os.path.join(thome, 'cache', 'last_metadata'))
            os.rmdir(os.path.join(thome, 'cache'))
            os.rmdir(os.path.join(thome))

        self.addCleanup(cleanup_temp_home, temp_home)

        last_path = os.path.join(temp_home, 'cache', 'last_metadata')
        md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
            "/tmp/foo": {"content": "bar"}}}}}
        md_str = json.dumps(md_data)
        md = cfn_helper.Metadata('teststack', None)

        self.assertFalse(os.path.exists(last_path),
                         "last_metadata file already exists")
        self.assertTrue(md.retrieve(meta_str=md_str, last_path=last_path))
        self.assertTrue(os.path.exists(last_path),
                        "last_metadata file should exist")
        # Ensure created dirs and file have right perms
        self.assertTrue(os.stat(last_path).st_mode & 0o600 == 0o600)
        self.assertTrue(
            os.stat(os.path.dirname(last_path)).st_mode & 0o700 == 0o700)

    def test_is_valid_metadata(self):
        md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
            "/tmp/foo": {"content": "bar"}}}}}

        md = cfn_helper.Metadata('teststack', None)
        self.assertTrue(
            md.retrieve(meta_str=md_data, last_path=self.last_file))

        self.assertThat(md_data, ttm.Equals(md._metadata))
        self.assertTrue(md._is_valid_metadata())
        self.assertThat(
            md_data['AWS::CloudFormation::Init'], ttm.Equals(md._metadata))

    def test_remote_metadata(self):
        md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
            "/tmp/foo": {"content": "bar"}}}}}

        with mock.patch.object(
            cfn.CloudFormationConnection, 'describe_stack_resource'
        ) as mock_dsr:
            mock_dsr.return_value = {
                'DescribeStackResourceResponse': {
                    'DescribeStackResourceResult': {
                        'StackResourceDetail': {'Metadata': md_data}}}}
            md = cfn_helper.Metadata(
                'teststack',
                None,
                access_key='foo',
                secret_key='bar')
            self.assertTrue(md.retrieve(last_path=self.last_file))
            self.assertThat(md_data, ttm.Equals(md._metadata))

            with tempfile.NamedTemporaryFile(mode='w') as fcreds:
                fcreds.write('AWSAccessKeyId=foo\nAWSSecretKey=bar\n')
                fcreds.flush()
                md = cfn_helper.Metadata(
                    'teststack', None, credentials_file=fcreds.name)
                self.assertTrue(md.retrieve(last_path=self.last_file))
            self.assertThat(md_data, ttm.Equals(md._metadata))

    def test_nova_meta_with_cache(self):
        meta_in = {"uuid": "f9431d18-d971-434d-9044-5b38f5b4646f",
                   "availability_zone": "nova",
                   "hostname": "as-wikidatabase-4ykioj3lgi57.novalocal",
                   "launch_index": 0,
                   "meta": {},
                   "public_keys": {"heat_key": "ssh-rsa etc...\n"},
                   "name": "as-WikiDatabase-4ykioj3lgi57"}
        md_str = json.dumps(meta_in)

        md = cfn_helper.Metadata('teststack', None)
        with tempfile.NamedTemporaryFile(mode='w+') as default_file:
            default_file.write(md_str)
            default_file.flush()
            self.assertThat(default_file.name, ttm.FileContains(md_str))
            meta_out = md.get_nova_meta(cache_path=default_file.name)

            self.assertEqual(meta_in, meta_out)

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_nova_meta_curl(self, mock_cp):
        url = 'http://169.254.169.254/openstack/2012-08-10/meta_data.json'
        temp_home = tempfile.mkdtemp()
        cache_path = os.path.join(temp_home, 'meta_data.json')

        def cleanup_temp_home(thome):
            os.unlink(cache_path)
            os.rmdir(thome)

        self.addCleanup(cleanup_temp_home, temp_home)

        meta_in = {"uuid": "f9431d18-d971-434d-9044-5b38f5b4646f",
                   "availability_zone": "nova",
                   "hostname": "as-wikidatabase-4ykioj3lgi57.novalocal",
                   "launch_index": 0,
                   "meta": {"freddy": "is hungry"},
                   "public_keys": {"heat_key": "ssh-rsa etc...\n"},
                   "name": "as-WikiDatabase-4ykioj3lgi57"}
        md_str = json.dumps(meta_in)

        def write_cache_file(*params, **kwargs):
            with open(cache_path, 'w+') as cache_file:
                cache_file.write(md_str)
                cache_file.flush()
                self.assertThat(cache_file.name, ttm.FileContains(md_str))
            return FakePOpen('Downloaded', '', 0)

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = write_cache_file
            md = cfn_helper.Metadata('teststack', None)
            meta_out = md.get_nova_meta(cache_path=cache_path)
            self.assertEqual(meta_in, meta_out)
            mock_popen.assert_has_calls(
                popen_root_calls([['curl', '-o', cache_path, url]]))

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_nova_meta_curl_corrupt(self, mock_cp):
        url = 'http://169.254.169.254/openstack/2012-08-10/meta_data.json'
        temp_home = tempfile.mkdtemp()
        cache_path = os.path.join(temp_home, 'meta_data.json')

        def cleanup_temp_home(thome):
            os.unlink(cache_path)
            os.rmdir(thome)

        self.addCleanup(cleanup_temp_home, temp_home)

        md_str = "this { is not really json"

        def write_cache_file(*params, **kwargs):
            with open(cache_path, 'w+') as cache_file:
                cache_file.write(md_str)
                cache_file.flush()
                self.assertThat(cache_file.name, ttm.FileContains(md_str))
            return FakePOpen('Downloaded', '', 0)

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = write_cache_file
            md = cfn_helper.Metadata('teststack', None)
            meta_out = md.get_nova_meta(cache_path=cache_path)
            self.assertIsNone(meta_out)
            mock_popen.assert_has_calls(
                popen_root_calls([['curl', '-o', cache_path, url]]))

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_nova_meta_curl_failed(self, mock_cp):
        url = 'http://169.254.169.254/openstack/2012-08-10/meta_data.json'
        temp_home = tempfile.mkdtemp()
        cache_path = os.path.join(temp_home, 'meta_data.json')

        def cleanup_temp_home(thome):
            os.rmdir(thome)

        self.addCleanup(cleanup_temp_home, temp_home)

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.return_value = FakePOpen('Failed', '', 1)
            md = cfn_helper.Metadata('teststack', None)
            meta_out = md.get_nova_meta(cache_path=cache_path)
            self.assertIsNone(meta_out)
            mock_popen.assert_has_calls(
                popen_root_calls([['curl', '-o', cache_path, url]]))

    def test_get_tags(self):
        fake_tags = {'foo': 'fee',
                     'apple': 'red'}
        md_data = {"uuid": "f9431d18-d971-434d-9044-5b38f5b4646f",
                   "availability_zone": "nova",
                   "hostname": "as-wikidatabase-4ykioj3lgi57.novalocal",
                   "launch_index": 0,
                   "meta": fake_tags,
                   "public_keys": {"heat_key": "ssh-rsa etc...\n"},
                   "name": "as-WikiDatabase-4ykioj3lgi57"}
        tags_expect = fake_tags
        tags_expect['InstanceId'] = md_data['uuid']

        md = cfn_helper.Metadata('teststack', None)

        with mock.patch.object(md, 'get_nova_meta') as mock_method:
            mock_method.return_value = md_data
            tags = md.get_tags()
            mock_method.assert_called_once_with()

        self.assertEqual(tags_expect, tags)

    def test_get_instance_id(self):
        uuid = "f9431d18-d971-434d-9044-5b38f5b4646f"
        md_data = {"uuid": uuid,
                   "availability_zone": "nova",
                   "hostname": "as-wikidatabase-4ykioj3lgi57.novalocal",
                   "launch_index": 0,
                   "public_keys": {"heat_key": "ssh-rsa etc...\n"},
                   "name": "as-WikiDatabase-4ykioj3lgi57"}

        md = cfn_helper.Metadata('teststack', None)

        with mock.patch.object(md, 'get_nova_meta') as mock_method:
            mock_method.return_value = md_data
            self.assertEqual(md.get_instance_id(), uuid)
            mock_method.assert_called_once_with()


class TestCfnInit(testtools.TestCase):

    def setUp(self):
        super(TestCfnInit, self).setUp()
        self.tdir = self.useFixture(fixtures.TempDir())
        self.last_file = os.path.join(self.tdir.path, 'last_metadata')

    def test_cfn_init(self):

        with tempfile.NamedTemporaryFile(mode='w+') as foo_file:
            md_data = {"AWS::CloudFormation::Init": {"config": {"files": {
                foo_file.name: {"content": "bar"}}}}}

            md = cfn_helper.Metadata('teststack', None)
            self.assertTrue(
                md.retrieve(meta_str=md_data, last_path=self.last_file))
            md.cfn_init()
            self.assertThat(foo_file.name, ttm.FileContains('bar'))

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_cfn_init_with_ignore_errors_false(self, mock_cp):
        md_data = {"AWS::CloudFormation::Init": {"config": {"commands": {
            "00_foo": {"command": "/bin/command1",
                       "ignoreErrors": "false"}}}}}
        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.return_value = FakePOpen('Doing something', 'error', -1)
            md = cfn_helper.Metadata('teststack', None)
            self.assertTrue(
                md.retrieve(meta_str=md_data, last_path=self.last_file))
            self.assertRaises(cfn_helper.CommandsHandlerRunError, md.cfn_init)
            mock_popen.assert_has_calls(popen_root_calls(['/bin/command1'],
                                                         shell=True))

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_cfn_init_with_ignore_errors_true(self, mock_cp):
        calls = []
        returns = []
        calls.extend(popen_root_calls(['/bin/command1'], shell=True))
        returns.append(FakePOpen('Doing something', 'error', -1))
        calls.extend(popen_root_calls(['/bin/command2'], shell=True))
        returns.append(FakePOpen('All good'))

        md_data = {"AWS::CloudFormation::Init": {"config": {"commands": {
            "00_foo": {"command": "/bin/command1",
                       "ignoreErrors": "true"},
            "01_bar": {"command": "/bin/command2",
                       "ignoreErrors": "false"}
        }}}}

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            md = cfn_helper.Metadata('teststack', None)
            self.assertTrue(
                md.retrieve(meta_str=md_data, last_path=self.last_file))
            md.cfn_init()
            mock_popen.assert_has_calls(calls)

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_cfn_init_runs_list_commands_without_shell(self, mock_cp):
        calls = []
        returns = []
        # command supplied as list shouldn't run on shell
        calls.extend(popen_root_calls([['/bin/command1', 'arg']], shell=False))
        returns.append(FakePOpen('Doing something'))
        # command supplied as string should run on shell
        calls.extend(popen_root_calls(['/bin/command2'], shell=True))
        returns.append(FakePOpen('All good'))

        md_data = {"AWS::CloudFormation::Init": {"config": {"commands": {
            "00_foo": {"command": ["/bin/command1", "arg"]},
            "01_bar": {"command": "/bin/command2"}
        }}}}

        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.side_effect = returns
            md = cfn_helper.Metadata('teststack', None)
            self.assertTrue(
                md.retrieve(meta_str=md_data, last_path=self.last_file))
            md.cfn_init()
            mock_popen.assert_has_calls(calls)


class TestSourcesHandler(testtools.TestCase):
    def test_apply_sources_empty(self):
        sh = cfn_helper.SourcesHandler({})
        sh.apply_sources()

    def _test_apply_sources(self, url, end_file):
        dest = tempfile.mkdtemp()
        self.addCleanup(os.rmdir, dest)
        sources = {dest: url}
        td = os.path.dirname(end_file)
        er = "mkdir -p '%s'; cd '%s'; curl -s '%s' | gunzip | tar -xvf -"
        calls = popen_root_calls([er % (dest, dest, url)], shell=True)

        with mock.patch.object(tempfile, 'mkdtemp') as mock_mkdtemp:
            mock_mkdtemp.return_value = td
            with mock.patch('subprocess.Popen') as mock_popen:
                mock_popen.return_value = FakePOpen('Curl good')
                sh = cfn_helper.SourcesHandler(sources)
                sh.apply_sources()
                mock_popen.assert_has_calls(calls)
            mock_mkdtemp.assert_called_with()

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_apply_sources_github(self, mock_cp):
        url = "https://github.com/NoSuchProject/tarball/NoSuchTarball"
        dest = tempfile.mkdtemp()
        self.addCleanup(os.rmdir, dest)
        sources = {dest: url}
        er = "mkdir -p '%s'; cd '%s'; curl -s '%s' | gunzip | tar -xvf -"
        calls = popen_root_calls([er % (dest, dest, url)], shell=True)
        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.return_value = FakePOpen('Curl good')
            sh = cfn_helper.SourcesHandler(sources)
            sh.apply_sources()
            mock_popen.assert_has_calls(calls)

    @mock.patch.object(cfn_helper, 'controlled_privileges')
    def test_apply_sources_general(self, mock_cp):
        url = "https://website.no.existe/a/b/c/file.tar.gz"
        dest = tempfile.mkdtemp()
        self.addCleanup(os.rmdir, dest)
        sources = {dest: url}
        er = "mkdir -p '%s'; cd '%s'; curl -s '%s' | gunzip | tar -xvf -"
        calls = popen_root_calls([er % (dest, dest, url)], shell=True)
        with mock.patch('subprocess.Popen') as mock_popen:
            mock_popen.return_value = FakePOpen('Curl good')
            sh = cfn_helper.SourcesHandler(sources)
            sh.apply_sources()
            mock_popen.assert_has_calls(calls)

    def test_apply_source_cmd(self):
        sh = cfn_helper.SourcesHandler({})
        er = "mkdir -p '%s'; cd '%s'; curl -s '%s' | %s | tar -xvf -"
        dest = '/tmp'
        # test tgz
        url = 'http://www.example.com/a.tgz'
        cmd = sh._apply_source_cmd(dest, url)
        self.assertEqual(er % (dest, dest, url, "gunzip"), cmd)
        # test tar.gz
        url = 'http://www.example.com/a.tar.gz'
        cmd = sh._apply_source_cmd(dest, url)
        self.assertEqual(er % (dest, dest, url, "gunzip"), cmd)
        # test github - tarball 1
        url = 'https://github.com/openstack/heat-cfntools/tarball/master'
        cmd = sh._apply_source_cmd(dest, url)
        self.assertEqual(er % (dest, dest, url, "gunzip"), cmd)
        # test github - tarball 2
        url = 'https://github.com/openstack/heat-cfntools/tarball/master/'
        cmd = sh._apply_source_cmd(dest, url)
        self.assertEqual(er % (dest, dest, url, "gunzip"), cmd)
        # test tbz2
        url = 'http://www.example.com/a.tbz2'
        cmd = sh._apply_source_cmd(dest, url)
        self.assertEqual(er % (dest, dest, url, "bunzip2"), cmd)
        # test tar.bz2
        url = 'http://www.example.com/a.tar.bz2'
        cmd = sh._apply_source_cmd(dest, url)
        self.assertEqual(er % (dest, dest, url, "bunzip2"), cmd)
        # test zip
        er = "mkdir -p '%s'; cd '%s'; curl -s -o '%s' '%s' && unzip -o '%s'"
        url = 'http://www.example.com/a.zip'
        d = "/tmp/tmp2I0yNK"
        tmp = "%s/a.zip" % d
        with mock.patch.object(tempfile, 'mkdtemp') as mock_mkdtemp:
            mock_mkdtemp.return_value = d

            cmd = sh._apply_source_cmd(dest, url)
            self.assertEqual(er % (dest, dest, tmp, url, tmp), cmd)
            # test gz
            er = "mkdir -p '%s'; cd '%s'; curl -s '%s' | %s > '%s'"
            url = 'http://www.example.com/a.sh.gz'
            cmd = sh._apply_source_cmd(dest, url)
            self.assertEqual(er % (dest, dest, url, "gunzip", "a.sh"), cmd)
            # test bz2
            url = 'http://www.example.com/a.sh.bz2'
            cmd = sh._apply_source_cmd(dest, url)
            self.assertEqual(er % (dest, dest, url, "bunzip2", "a.sh"), cmd)
            # test other
            url = 'http://www.example.com/a.sh'
            cmd = sh._apply_source_cmd(dest, url)
            self.assertEqual("", cmd)
            mock_mkdtemp.assert_called_with()