salt/tests/integration/states/test_pkgrepo.py

# -*- coding: utf-8 -*-
'''
tests for pkgrepo states
'''

# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals
import os

# Import Salt Testing libs
from tests.support.case import ModuleCase
from tests.support.mixins import SaltReturnAssertsMixin
from tests.support.unit import skipIf
from tests.support.helpers import (
    destructiveTest,
    requires_system_grains,
)

# Import Salt libs
import salt.utils.platform

# Import 3rd-party libs
from salt.ext import six


@destructiveTest
@skipIf(salt.utils.platform.is_windows(), 'minion is windows')
class PkgrepoTest(ModuleCase, SaltReturnAssertsMixin):
    '''
    pkgrepo state tests
    '''
    @requires_system_grains
    def test_pkgrepo_01_managed(self, grains):
        '''
        Test adding a repo
        '''
        os_grain = self.run_function('grains.item', ['os'])['os']
        os_release_info = tuple(self.run_function('grains.item', ['osrelease_info'])['osrelease_info'])
        if os_grain == 'Ubuntu' and os_release_info >= (15, 10):
            self.skipTest(
                'The PPA used for this test does not exist for Ubuntu Wily'
                ' (15.10) and later.'
            )

        if grains['os_family'] == 'Debian':
            try:
                from aptsources import sourceslist
            except ImportError:
                self.skipTest(
                    'aptsources.sourceslist python module not found'
                )
        ret = self.run_function('state.sls', mods='pkgrepo.managed', timeout=120)
        # If the below assert fails then no states were run, and the SLS in
        # tests/integration/files/file/base/pkgrepo/managed.sls needs to be
        # corrected.
        self.assertReturnNonEmptySaltType(ret)
        for state_id, state_result in six.iteritems(ret):
            self.assertSaltTrueReturn(dict([(state_id, state_result)]))

    def test_pkgrepo_02_absent(self):
        '''
        Test removing the repo from the above test
        '''
        os_grain = self.run_function('grains.item', ['os'])['os']
        os_release_info = tuple(self.run_function('grains.item', ['osrelease_info'])['osrelease_info'])
        if os_grain == 'Ubuntu' and os_release_info >= (15, 10):
            self.skipTest(
                'The PPA used for this test does not exist for Ubuntu Wily'
                ' (15.10) and later.'
            )

        ret = self.run_function('state.sls', mods='pkgrepo.absent', timeout=120)
        # If the below assert fails then no states were run, and the SLS in
        # tests/integration/files/file/base/pkgrepo/absent.sls needs to be
        # corrected.
        self.assertReturnNonEmptySaltType(ret)
        for state_id, state_result in six.iteritems(ret):
            self.assertSaltTrueReturn(dict([(state_id, state_result)]))

    @requires_system_grains
    def test_pkgrepo_03_with_comments(self, grains):
        '''
        Test adding a repo with comments
        '''
        os_family = grains['os_family'].lower()

        if os_family in ('redhat',):
            kwargs = {
                'name': 'examplerepo',
                'baseurl': 'http://example.com/repo',
                'enabled': False,
                'comments': ['This is a comment']
            }
        elif os_family in ('debian',):
            self.skipTest('Debian/Ubuntu test case needed')
        else:
            self.skipTest("No test case for os_family '{0}'".format(os_family))

        try:
            # Run the state to add the repo
            ret = self.run_state('pkgrepo.managed', **kwargs)
            self.assertSaltTrueReturn(ret)

            # Run again with modified comments
            kwargs['comments'].append('This is another comment')
            ret = self.run_state('pkgrepo.managed', **kwargs)
            self.assertSaltTrueReturn(ret)
            ret = ret[next(iter(ret))]
            self.assertEqual(
                ret['changes'],
                {
                    'comments': {
                        'old': ['This is a comment'],
                        'new': ['This is a comment',
                                'This is another comment']
                    }
                }
            )

            # Run a third time, no changes should be made
            ret = self.run_state('pkgrepo.managed', **kwargs)
            self.assertSaltTrueReturn(ret)
            ret = ret[next(iter(ret))]
            self.assertFalse(ret['changes'])
            self.assertEqual(
                ret['comment'],
                "Package repo '{0}' already configured".format(kwargs['name'])
            )
        finally:
            # Clean up
            self.run_state('pkgrepo.absent', name=kwargs['name'])

    @requires_system_grains
    def test_pkgrepo_04_apt_with_architectures(self, grains):
        '''
        Test managing a repo with architectures specified
        '''
        if grains['os_family'].lower() != 'debian':
            self.skipTest('APT-only test')

        name = 'deb {{arch}}http://foo.com/bar/latest {oscodename} main'.format(oscodename=grains['oscodename'])

        def _get_arch(arch):
            return '[arch={0}] '.format(arch) if arch else ''

        def _run(arch='', test=False):
            ret = self.run_state(
                'pkgrepo.managed',
                name=name.format(arch=_get_arch(arch)),
                file=fn_,
                refresh=False,
                test=test)
            return ret[next(iter(ret))]

        fn_ = salt.utils.files.mkstemp(dir='/etc/apt/sources.list.d', suffix='.list')

        try:
            # Run with test=True
            ret = _run(test=True)
            assert ret['changes'] == {'repo': name.format(arch='')}, ret['changes']
            assert 'would be' in ret['comment'], ret['comment']
            assert ret['result'] is None, ret['result']

            # Run for real
            ret = _run()
            assert ret['changes'] == {'repo': name.format(arch='')}, ret['changes']
            assert ret['comment'].startswith('Configured'), ret['comment']
            assert ret['result'] is True, ret['result']

            # Run again with test=True, should exit with no changes and a True
            # result.
            ret = _run(test=True)
            assert not ret['changes'], ret['changes']
            assert 'already' in ret['comment'], ret['comment']
            assert ret['result'] is True, ret['result']

            # Run for real again, results should be the same as above (i.e. we
            # should never get to the point where we exit with a None result).
            ret = _run()
            assert not ret['changes'], ret['changes']
            assert 'already' in ret['comment'], ret['comment']
            assert ret['result'] is True, ret['result']

            expected_changes = {
                'line': {
                    'new': name.format(arch=_get_arch('amd64')),
                    'old': name.format(arch=''),
                },
                'architectures': {
                    'new': ['amd64'],
                    'old': [],
                },
            }

            # Run with test=True and the architecture set. We should get a None
            # result with some expected changes.
            ret = _run(arch='amd64', test=True)
            assert ret['changes'] == expected_changes, ret['changes']
            assert 'would be' in ret['comment'], ret['comment']
            assert ret['result'] is None, ret['result']

            # Run for real, with the architecture set. We should get a True
            # result with the same changes.
            ret = _run(arch='amd64')
            assert ret['changes'] == expected_changes, ret['changes']
            assert ret['comment'].startswith('Configured'), ret['comment']
            assert ret['result'] is True, ret['result']

            # Run again with test=True, should exit with no changes and a True
            # result.
            ret = _run(arch='amd64', test=True)
            assert not ret['changes'], ret['changes']
            assert 'already' in ret['comment'], ret['comment']
            assert ret['result'] is True, ret['result']

            # Run for real again, results should be the same as above (i.e. we
            # should never get to the point where we exit with a None result).
            ret = _run(arch='amd64')
            assert not ret['changes'], ret['changes']
            assert 'already' in ret['comment'], ret['comment']
            assert ret['result'] is True, ret['result']

            expected_changes = {
                'line': {
                    'new': name.format(arch=''),
                    'old': name.format(arch=_get_arch('amd64')),
                },
                'architectures': {
                    'new': [],
                    'old': ['amd64'],
                },
            }

            # Run with test=True and the architecture set back to the original
            # value. We should get a None result with some expected changes.
            ret = _run(test=True)
            assert ret['changes'] == expected_changes, ret['changes']
            assert 'would be' in ret['comment'], ret['comment']
            assert ret['result'] is None, ret['result']

            # Run for real, with the architecture set. We should get a True
            # result with the same changes.
            ret = _run()
            assert ret['changes'] == expected_changes, ret['changes']
            assert ret['comment'].startswith('Configured'), ret['comment']
            assert ret['result'] is True, ret['result']

            # Run again with test=True, should exit with no changes and a True
            # result.
            ret = _run(test=True)
            assert not ret['changes'], ret['changes']
            assert 'already' in ret['comment'], ret['comment']
            assert ret['result'] is True, ret['result']

            # Run for real again, results should be the same as above (i.e. we
            # should never get to the point where we exit with a None result).
            ret = _run()
            assert not ret['changes'], ret['changes']
            assert 'already' in ret['comment'], ret['comment']
            assert ret['result'] is True, ret['result']
        finally:
            try:
                os.remove(fn_)
            except OSError:
                pass