salt/tests/integration/minion/blackout.py

# -*- coding: utf-8 -*-
'''
Tests for minion blackout
'''

# Import Python libs
from __future__ import absolute_import
import os
from time import sleep
import textwrap

# Import Salt Testing libs
from salttesting.helpers import destructiveTest, ensure_in_syspath

ensure_in_syspath('../')

# Import Salt libs
import integration
import salt.utils


BLACKOUT_PILLAR = os.path.join(integration.PILLAR_DIR, 'base', 'blackout.sls')


@destructiveTest
class MinionBlackoutTestCase(integration.ModuleCase):
    '''
    Test minion blackout functionality
    '''
    def begin_blackout(self, blackout_data='minion_blackout: True'):
        '''
        setup minion blackout mode
        '''
        salt.utils.fopen(BLACKOUT_PILLAR, 'w').write(blackout_data)
        self.run_function('saltutil.refresh_pillar')
        sleep(5)  # wait for minion to enter blackout mode

    def end_blackout(self):
        '''
        takedown minion blackout mode
        '''
        with salt.utils.fopen(BLACKOUT_PILLAR, 'w') as blackout_pillar:
            blackout_pillar.write(textwrap.dedent('''\
                minion_blackout: False
                '''))
        self.run_function('saltutil.refresh_pillar')
        sleep(5)  # wait for minion to exit blackout mode

    def test_blackout(self):
        '''
        Test that basic minion blackout functionality works
        '''
        try:
            self.begin_blackout()
            blackout_ret = self.run_function('test.ping')
            self.assertIn('Minion in blackout mode.', blackout_ret)
        finally:
            self.end_blackout()

        ret = self.run_function('test.ping')
        self.assertEqual(ret, True)

    def test_blackout_whitelist(self):
        '''
        Test that minion blackout whitelist works
        '''
        try:
            self.begin_blackout(textwrap.dedent('''\
                minion_blackout: True
                minion_blackout_whitelist:
                  - test.ping
                  - test.fib
                '''))

            ping_ret = self.run_function('test.ping')
            self.assertEqual(ping_ret, True)

            fib_ret = self.run_function('test.fib', [7])
            self.assertTrue(isinstance(fib_ret, list))
            self.assertEqual(fib_ret[0], 13)
        finally:
            self.end_blackout()

    def test_blackout_nonwhitelist(self):
        '''
        Test that minion refuses to run non-whitelisted functions during
        blackout whitelist
        '''
        try:
            self.begin_blackout(textwrap.dedent('''\
                minion_blackout: True
                minion_blackout_whitelist:
                  - test.ping
                  - test.fib
                '''))

            state_ret = self.run_function('state.apply')
            self.assertIn('Minion in blackout mode.', state_ret)

            cloud_ret = self.run_function('cloud.query', ['list_nodes_full'])
            self.assertIn('Minion in blackout mode.', cloud_ret)
        finally:
            self.end_blackout()


if __name__ == '__main__':
    from integration import run_tests
    run_tests(MinionBlackoutTestCase, needs_daemon=True)