# -*- coding: utf-8 -*- ''' :codeauthor: :email:`Jeff Schroeder ` ''' # Import Python libs from __future__ import absolute_import, unicode_literals, print_function import base64 from contextlib import contextmanager # Import Salt Testing Libs from tests.support.mixins import LoaderModuleMockMixin from tests.support.unit import skipIf, TestCase from tests.support.mock import ( MagicMock, patch) # Import Salt Libs import salt.utils.stringutils from salt.states import kubernetes from salt.ext import six @skipIf(kubernetes is False, "Probably Kubernetes client lib is not installed. \ Skipping test_kubernetes.py") class KubernetesTestCase(TestCase, LoaderModuleMockMixin): ''' Test cases for salt.states.kubernetes ''' def setup_loader_modules(self): return {kubernetes: {'__env__': 'base'}} @contextmanager def mock_func(self, func_name, return_value, test=False): ''' Mock any of the kubernetes state function return values and set the test options. ''' name = 'kubernetes.{0}'.format(func_name) mocked = {name: MagicMock(return_value=return_value)} with patch.dict(kubernetes.__salt__, mocked) as patched: with patch.dict(kubernetes.__opts__, {'test': test}): yield patched def make_configmap(self, name, namespace='default', data=None): return self.make_ret_dict( kind='ConfigMap', name=name, namespace=namespace, data=data, ) def make_secret(self, name, namespace='default', data=None): secret_data = self.make_ret_dict( kind='Secret', name=name, namespace=namespace, data=data, ) # Base64 all of the values just like kubectl does for key, value in six.iteritems(secret_data['data']): secret_data['data'][key] = base64.b64encode( salt.utils.stringutils.to_bytes(value) ) return secret_data def make_node_labels(self, name='minikube'): return { 'kubernetes.io/hostname': name, 'beta.kubernetes.io/os': 'linux', 'beta.kubernetes.io/arch': 'amd64', 'failure-domain.beta.kubernetes.io/region': 'us-west-1', } def make_node(self, name='minikube'): node_data = self.make_ret_dict(kind='Node', name='minikube') node_data.update({ 'api_version': 'v1', 'kind': 'Node', 'metadata': { 'annotations': { 'node.alpha.kubernetes.io/ttl': '0', }, 'labels': self.make_node_labels(name=name), 'name': name, 'namespace': None, 'self_link': '/api/v1/nodes/{name}'.format(name=name), 'uid': '7811b8ae-c1a1-11e7-a55a-0800279fb61e', }, 'spec': { 'external_id': name, }, 'status': {}, }) return node_data def make_namespace(self, name='default'): namespace_data = self.make_ret_dict(kind='Namespace', name=name) del namespace_data['data'] namespace_data.update({ 'status': { 'phase': 'Active', }, 'spec': { 'finalizers': ['kubernetes'], }, 'metadata': { 'name': name, 'namespace': None, 'labels': None, 'self_link': '/api/v1/namespaces/{namespace}'.format( namespace=name, ), 'annotations': None, 'uid': '752fceeb-c1a1-11e7-a55a-0800279fb61e', }, }) return namespace_data def make_ret_dict(self, kind, name, namespace=None, data=None): ''' Make a minimal example configmap or secret for using in mocks ''' assert kind in ('Secret', 'ConfigMap', 'Namespace', 'Node') if data is None: data = {} self_link = '/api/v1/namespaces/{namespace}/{kind}s/{name}'.format( namespace=namespace, kind=kind.lower(), name=name, ) return_data = { 'kind': kind, 'data': data, 'api_version': 'v1', 'metadata': { 'name': name, 'labels': None, 'namespace': namespace, 'self_link': self_link, 'annotations': { 'kubernetes.io/change-cause': 'salt-call state.apply', }, }, } return return_data def test_configmap_present__fail(self): error = kubernetes.configmap_present( name='testme', data={1: 1}, source='salt://beyond/oblivion.jinja', ) self.assertDictEqual( { 'changes': {}, 'result': False, 'name': 'testme', 'comment': "'source' cannot be used in combination with 'data'", }, error, ) def test_configmap_present__create_test_true(self): # Create a new configmap with test=True with self.mock_func('show_configmap', return_value=None, test=True): ret = kubernetes.configmap_present( name='example', data={'example.conf': '# empty config file'}, ) self.assertDictEqual( { 'comment': 'The configmap is going to be created', 'changes': {}, 'name': 'example', 'result': None, }, ret, ) def test_configmap_present__create(self): # Create a new configmap with self.mock_func('show_configmap', return_value=None): cm = self.make_configmap( name='test', namespace='default', data={'foo': 'bar'}, ) with self.mock_func('create_configmap', return_value=cm): actual = kubernetes.configmap_present( name='test', data={'foo': 'bar'}, ) self.assertDictEqual( { 'comment': '', 'changes': {'data': {'foo': 'bar'}}, 'name': 'test', 'result': True, }, actual, ) def test_configmap_present__create_no_data(self): # Create a new configmap with no 'data' attribute with self.mock_func('show_configmap', return_value=None): cm = self.make_configmap( name='test', namespace='default', ) with self.mock_func('create_configmap', return_value=cm): actual = kubernetes.configmap_present(name='test') self.assertDictEqual( { 'comment': '', 'changes': {'data': {}}, 'name': 'test', 'result': True, }, actual, ) def test_configmap_present__replace_test_true(self): cm = self.make_configmap( name='settings', namespace='saltstack', data={'foobar.conf': '# Example configuration'}, ) with self.mock_func('show_configmap', return_value=cm, test=True): ret = kubernetes.configmap_present( name='settings', namespace='saltstack', data={'foobar.conf': '# Example configuration'}, ) self.assertDictEqual( { 'comment': 'The configmap is going to be replaced', 'changes': {}, 'name': 'settings', 'result': None, }, ret, ) def test_configmap_present__replace(self): cm = self.make_configmap(name='settings', data={'action': 'make=war'}) # Replace an existing configmap with self.mock_func('show_configmap', return_value=cm): new_cm = cm.copy() new_cm.update({ 'data': {'action': 'make=peace'}, }) with self.mock_func('replace_configmap', return_value=new_cm): actual = kubernetes.configmap_present( name='settings', data={'action': 'make=peace'}, ) self.assertDictEqual( { 'comment': 'The configmap is already present. Forcing recreation', 'changes': { 'data': { 'action': 'make=peace', }, }, 'name': 'settings', 'result': True, }, actual, ) def test_configmap_absent__noop_test_true(self): # Nothing to delete with test=True with self.mock_func('show_configmap', return_value=None, test=True): actual = kubernetes.configmap_absent(name='NOT_FOUND') self.assertDictEqual( { 'comment': 'The configmap does not exist', 'changes': {}, 'name': 'NOT_FOUND', 'result': None, }, actual, ) def test_configmap_absent__test_true(self): # Configmap exists with test=True cm = self.make_configmap(name='deleteme', namespace='default') with self.mock_func('show_configmap', return_value=cm, test=True): actual = kubernetes.configmap_absent(name='deleteme') self.assertDictEqual( { 'comment': 'The configmap is going to be deleted', 'changes': {}, 'name': 'deleteme', 'result': None, }, actual, ) def test_configmap_absent__noop(self): # Nothing to delete with self.mock_func('show_configmap', return_value=None): actual = kubernetes.configmap_absent(name='NOT_FOUND') self.assertDictEqual( { 'comment': 'The configmap does not exist', 'changes': {}, 'name': 'NOT_FOUND', 'result': True, }, actual, ) def test_configmap_absent(self): # Configmap exists, delete it! cm = self.make_configmap(name='deleteme', namespace='default') with self.mock_func('show_configmap', return_value=cm): # The return from this module isn't used in the state with self.mock_func('delete_configmap', return_value={}): actual = kubernetes.configmap_absent(name='deleteme') self.assertDictEqual( { 'comment': 'ConfigMap deleted', 'changes': { 'kubernetes.configmap': { 'new': 'absent', 'old': 'present', }, }, 'name': 'deleteme', 'result': True, }, actual, ) def test_secret_present__fail(self): actual = kubernetes.secret_present( name='sekret', data={'password': 'monk3y'}, source='salt://nope.jinja', ) self.assertDictEqual( { 'changes': {}, 'result': False, 'name': 'sekret', 'comment': "'source' cannot be used in combination with 'data'", }, actual, ) def test_secret_present__exists_test_true(self): secret = self.make_secret(name='sekret') new_secret = secret.copy() new_secret.update({ 'data': {'password': 'uncle'}, }) # Secret exists already and needs replacing with test=True with self.mock_func('show_secret', return_value=secret): with self.mock_func('replace_secret', return_value=new_secret, test=True): actual = kubernetes.secret_present( name='sekret', data={'password': 'uncle'}, ) self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'sekret', 'comment': 'The secret is going to be replaced', }, actual, ) def test_secret_present__exists(self): # Secret exists and gets replaced secret = self.make_secret(name='sekret', data={'password': 'booyah'}) with self.mock_func('show_secret', return_value=secret): with self.mock_func('replace_secret', return_value=secret): actual = kubernetes.secret_present( name='sekret', data={'password': 'booyah'}, ) self.assertDictEqual( { 'changes': {'data': ['password']}, 'result': True, 'name': 'sekret', 'comment': "The secret is already present. Forcing recreation", }, actual, ) def test_secret_present__create(self): # Secret exists and gets replaced secret = self.make_secret(name='sekret', data={'password': 'booyah'}) with self.mock_func('show_secret', return_value=None): with self.mock_func('create_secret', return_value=secret): actual = kubernetes.secret_present( name='sekret', data={'password': 'booyah'}, ) self.assertDictEqual( { 'changes': {'data': ['password']}, 'result': True, 'name': 'sekret', 'comment': '', }, actual, ) def test_secret_present__create_no_data(self): # Secret exists and gets replaced secret = self.make_secret(name='sekret') with self.mock_func('show_secret', return_value=None): with self.mock_func('create_secret', return_value=secret): actual = kubernetes.secret_present(name='sekret') self.assertDictEqual( { 'changes': {'data': []}, 'result': True, 'name': 'sekret', 'comment': '', }, actual, ) def test_secret_present__create_test_true(self): # Secret exists and gets replaced with test=True secret = self.make_secret(name='sekret') with self.mock_func('show_secret', return_value=None): with self.mock_func('create_secret', return_value=secret, test=True): actual = kubernetes.secret_present(name='sekret') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'sekret', 'comment': 'The secret is going to be created', }, actual, ) def test_secret_absent__noop_test_true(self): with self.mock_func('show_secret', return_value=None, test=True): actual = kubernetes.secret_absent(name='sekret') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'sekret', 'comment': 'The secret does not exist', }, actual, ) def test_secret_absent__noop(self): with self.mock_func('show_secret', return_value=None): actual = kubernetes.secret_absent(name='passwords') self.assertDictEqual( { 'changes': {}, 'result': True, 'name': 'passwords', 'comment': 'The secret does not exist', }, actual, ) def test_secret_absent__delete_test_true(self): secret = self.make_secret(name='credentials', data={'redis': 'letmein'}) with self.mock_func('show_secret', return_value=secret): with self.mock_func('delete_secret', return_value=secret, test=True): actual = kubernetes.secret_absent(name='credentials') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'credentials', 'comment': 'The secret is going to be deleted', }, actual, ) def test_secret_absent__delete(self): secret = self.make_secret(name='foobar', data={'redis': 'letmein'}) deleted = { 'status': None, 'kind': 'Secret', 'code': None, 'reason': None, 'details': None, 'message': None, 'api_version': 'v1', 'metadata': { 'self_link': '/api/v1/namespaces/default/secrets/foobar', 'resource_version': '30292', }, } with self.mock_func('show_secret', return_value=secret): with self.mock_func('delete_secret', return_value=deleted): actual = kubernetes.secret_absent(name='foobar') self.assertDictEqual( { 'changes': { 'kubernetes.secret': { 'new': 'absent', 'old': 'present', }, }, 'result': True, 'name': 'foobar', 'comment': 'Secret deleted', }, actual, ) def test_node_label_present__add_test_true(self): labels = self.make_node_labels() with self.mock_func('node_labels', return_value=labels, test=True): actual = kubernetes.node_label_present( name='com.zoo-animal', node='minikube', value='monkey', ) self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'com.zoo-animal', 'comment': 'The label is going to be set', }, actual, ) def test_node_label_present__add(self): node_data = self.make_node() # Remove some of the defaults to make it simpler node_data['metadata']['labels'] = { 'beta.kubernetes.io/os': 'linux', } labels = node_data['metadata']['labels'] with self.mock_func('node_labels', return_value=labels): with self.mock_func('node_add_label', return_value=node_data): actual = kubernetes.node_label_present( name='failure-domain.beta.kubernetes.io/zone', node='minikube', value='us-central1-a', ) self.assertDictEqual( { 'comment': '', 'changes': { 'minikube.failure-domain.beta.kubernetes.io/zone': { 'new': { 'failure-domain.beta.kubernetes.io/zone': 'us-central1-a', 'beta.kubernetes.io/os': 'linux' }, 'old': { 'beta.kubernetes.io/os': 'linux', }, }, }, 'name': 'failure-domain.beta.kubernetes.io/zone', 'result': True, }, actual, ) def test_node_label_present__already_set(self): node_data = self.make_node() labels = node_data['metadata']['labels'] with self.mock_func('node_labels', return_value=labels): with self.mock_func('node_add_label', return_value=node_data): actual = kubernetes.node_label_present( name='failure-domain.beta.kubernetes.io/region', node='minikube', value='us-west-1', ) self.assertDictEqual( { 'changes': {}, 'result': True, 'name': 'failure-domain.beta.kubernetes.io/region', 'comment': 'The label is already set and has the specified value', }, actual, ) def test_node_label_present__update_test_true(self): node_data = self.make_node() labels = node_data['metadata']['labels'] with self.mock_func('node_labels', return_value=labels): with self.mock_func('node_add_label', return_value=node_data, test=True): actual = kubernetes.node_label_present( name='failure-domain.beta.kubernetes.io/region', node='minikube', value='us-east-1', ) self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'failure-domain.beta.kubernetes.io/region', 'comment': 'The label is going to be updated', }, actual, ) def test_node_label_present__update(self): node_data = self.make_node() # Remove some of the defaults to make it simpler node_data['metadata']['labels'] = { 'failure-domain.beta.kubernetes.io/region': 'us-west-1', } labels = node_data['metadata']['labels'] with self.mock_func('node_labels', return_value=labels): with self.mock_func('node_add_label', return_value=node_data): actual = kubernetes.node_label_present( name='failure-domain.beta.kubernetes.io/region', node='minikube', value='us-east-1', ) self.assertDictEqual( { 'changes': { 'minikube.failure-domain.beta.kubernetes.io/region': { 'new': {'failure-domain.beta.kubernetes.io/region': 'us-east-1'}, 'old': {'failure-domain.beta.kubernetes.io/region': 'us-west-1'}, } }, 'result': True, 'name': 'failure-domain.beta.kubernetes.io/region', 'comment': 'The label is already set, changing the value', }, actual, ) def test_node_label_absent__noop_test_true(self): labels = self.make_node_labels() with self.mock_func('node_labels', return_value=labels, test=True): actual = kubernetes.node_label_absent( name='non-existent-label', node='minikube', ) self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'non-existent-label', 'comment': 'The label does not exist', }, actual ) def test_node_label_absent__noop(self): labels = self.make_node_labels() with self.mock_func('node_labels', return_value=labels): actual = kubernetes.node_label_absent( name='non-existent-label', node='minikube', ) self.assertDictEqual( { 'changes': {}, 'result': True, 'name': 'non-existent-label', 'comment': 'The label does not exist', }, actual ) def test_node_label_absent__delete_test_true(self): labels = self.make_node_labels() with self.mock_func('node_labels', return_value=labels, test=True): actual = kubernetes.node_label_absent( name='failure-domain.beta.kubernetes.io/region', node='minikube', ) self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'failure-domain.beta.kubernetes.io/region', 'comment': 'The label is going to be deleted', }, actual ) def test_node_label_absent__delete(self): node_data = self.make_node() labels = node_data['metadata']['labels'].copy() node_data['metadata']['labels'].pop('failure-domain.beta.kubernetes.io/region') with self.mock_func('node_labels', return_value=labels): with self.mock_func('node_remove_label', return_value=node_data): actual = kubernetes.node_label_absent( name='failure-domain.beta.kubernetes.io/region', node='minikube', ) self.assertDictEqual( { 'result': True, 'changes': { 'kubernetes.node_label': { 'new': 'absent', 'old': 'present', } }, 'comment': 'Label removed from node', 'name': 'failure-domain.beta.kubernetes.io/region', }, actual ) def test_namespace_present__create_test_true(self): with self.mock_func('show_namespace', return_value=None, test=True): actual = kubernetes.namespace_present(name='saltstack') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'saltstack', 'comment': 'The namespace is going to be created', }, actual ) def test_namespace_present__create(self): namespace_data = self.make_namespace(name='saltstack') with self.mock_func('show_namespace', return_value=None): with self.mock_func('create_namespace', return_value=namespace_data): actual = kubernetes.namespace_present(name='saltstack') self.assertDictEqual( { 'changes': { 'namespace': { 'new': namespace_data, 'old': {}, }, }, 'result': True, 'name': 'saltstack', 'comment': '', }, actual ) def test_namespace_present__noop_test_true(self): namespace_data = self.make_namespace(name='saltstack') with self.mock_func('show_namespace', return_value=namespace_data, test=True): actual = kubernetes.namespace_present(name='saltstack') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'saltstack', 'comment': 'The namespace already exists', }, actual ) def test_namespace_present__noop(self): namespace_data = self.make_namespace(name='saltstack') with self.mock_func('show_namespace', return_value=namespace_data): actual = kubernetes.namespace_present(name='saltstack') self.assertDictEqual( { 'changes': {}, 'result': True, 'name': 'saltstack', 'comment': 'The namespace already exists', }, actual ) def test_namespace_absent__noop_test_true(self): with self.mock_func('show_namespace', return_value=None, test=True): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'salt', 'comment': 'The namespace does not exist', }, actual ) def test_namespace_absent__noop(self): with self.mock_func('show_namespace', return_value=None): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': {}, 'result': True, 'name': 'salt', 'comment': 'The namespace does not exist', }, actual ) def test_namespace_absent__delete_test_true(self): namespace_data = self.make_namespace(name='salt') with self.mock_func('show_namespace', return_value=namespace_data, test=True): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': {}, 'result': None, 'name': 'salt', 'comment': 'The namespace is going to be deleted', }, actual ) def test_namespace_absent__delete_code_200(self): namespace_data = self.make_namespace(name='salt') deleted = namespace_data.copy() deleted['code'] = 200 deleted.update({ 'code': 200, 'message': None, }) with self.mock_func('show_namespace', return_value=namespace_data): with self.mock_func('delete_namespace', return_value=deleted): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': { 'kubernetes.namespace': { 'new': 'absent', 'old': 'present', } }, 'result': True, 'name': 'salt', 'comment': 'Terminating', }, actual ) def test_namespace_absent__delete_status_terminating(self): namespace_data = self.make_namespace(name='salt') deleted = namespace_data.copy() deleted.update({ 'code': None, 'status': 'Terminating namespace', 'message': 'Terminating this shizzzle yo', }) with self.mock_func('show_namespace', return_value=namespace_data): with self.mock_func('delete_namespace', return_value=deleted): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': { 'kubernetes.namespace': { 'new': 'absent', 'old': 'present', } }, 'result': True, 'name': 'salt', 'comment': 'Terminating this shizzzle yo', }, actual ) def test_namespace_absent__delete_status_phase_terminating(self): # This is what kubernetes 1.8.0 looks like when deleting namespaces namespace_data = self.make_namespace(name='salt') deleted = namespace_data.copy() deleted.update({ 'code': None, 'message': None, 'status': {'phase': 'Terminating'}, }) with self.mock_func('show_namespace', return_value=namespace_data): with self.mock_func('delete_namespace', return_value=deleted): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': { 'kubernetes.namespace': { 'new': 'absent', 'old': 'present', } }, 'result': True, 'name': 'salt', 'comment': 'Terminating', }, actual ) def test_namespace_absent__delete_error(self): namespace_data = self.make_namespace(name='salt') deleted = namespace_data.copy() deleted.update({ 'code': 418, 'message': 'I\' a teapot!', 'status': None, }) with self.mock_func('show_namespace', return_value=namespace_data): with self.mock_func('delete_namespace', return_value=deleted): actual = kubernetes.namespace_absent(name='salt') self.assertDictEqual( { 'changes': {}, 'result': False, 'name': 'salt', 'comment': 'Something went wrong, response: {0}'.format( deleted, ), }, actual )