Remove unused files that used kubectl
Now the kubectl is not used in magnum, we can remove the unused file. Change-Id: Ia65554004a307f0941dc3398106b39267efb6402 Partially-Implements: blueprint python-k8sclient
This commit is contained in:
parent
5638ee05cf
commit
32bf5a52e6
|
@ -1,207 +0,0 @@
|
|||
# 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.
|
||||
|
||||
"""Magnum Kubernetes RPC handler."""
|
||||
|
||||
import tempfile
|
||||
|
||||
from oslo_log import log as logging
|
||||
|
||||
from magnum.common import exception
|
||||
from magnum.i18n import _LE
|
||||
from magnum.openstack.common import utils
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _k8s_create(api_address, resource):
|
||||
data = resource.manifest
|
||||
definition_url = resource.manifest_url
|
||||
if data is not None:
|
||||
return _k8s_create_with_data(api_address, data)
|
||||
else:
|
||||
return _k8s_create_with_path(api_address, definition_url)
|
||||
|
||||
|
||||
def _k8s_create_with_path(api_address, resource_file):
|
||||
return utils.trycmd('kubectl', 'create',
|
||||
'-s', api_address,
|
||||
'-f', resource_file)
|
||||
|
||||
|
||||
def _k8s_create_with_data(api_address, resource_data):
|
||||
with tempfile.NamedTemporaryFile() as f:
|
||||
f.write(resource_data)
|
||||
f.flush()
|
||||
return _k8s_create_with_path(api_address, f.name)
|
||||
|
||||
|
||||
def _k8s_update(api_address, resource):
|
||||
data = resource.manifest
|
||||
definition_url = resource.manifest_url
|
||||
if data is not None:
|
||||
return _k8s_update_with_data(api_address, data)
|
||||
else:
|
||||
return _k8s_update_with_path(api_address, definition_url)
|
||||
|
||||
|
||||
def _k8s_update_with_path(api_address, resource_file):
|
||||
return utils.trycmd('kubectl', 'update',
|
||||
'-s', api_address,
|
||||
'-f', resource_file)
|
||||
|
||||
|
||||
def _k8s_update_with_data(api_address, resource_data):
|
||||
with tempfile.NamedTemporaryFile() as f:
|
||||
f.write(resource_data)
|
||||
f.flush()
|
||||
return _k8s_update_with_path(api_address, f.name)
|
||||
|
||||
|
||||
class KubeClient(object):
|
||||
"""These are the backend operations. They are executed by the backend
|
||||
service. API calls via AMQP (within the ReST API) trigger the
|
||||
handlers to be called.
|
||||
|
||||
This handler acts as an interface to executes kubectl command line
|
||||
services.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super(KubeClient, self).__init__()
|
||||
|
||||
def service_create(self, api_address, service):
|
||||
LOG.debug("service_create with contents %s" % service.as_dict())
|
||||
try:
|
||||
out, err = _k8s_create(api_address, service)
|
||||
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't create service with contents %(content)s "
|
||||
"due to error %(error)s") %
|
||||
{'content': service, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
def service_update(self, api_address, service):
|
||||
LOG.debug("service_update with contents %s" % service.as_dict())
|
||||
try:
|
||||
out, err = _k8s_update(api_address, service)
|
||||
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't update service with contents %(content)s "
|
||||
"due to error %(error)s") %
|
||||
{'content': service, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
def service_delete(self, api_address, name):
|
||||
LOG.debug("service_delete %s" % name)
|
||||
try:
|
||||
out, err = utils.trycmd('kubectl', 'delete', 'service', name,
|
||||
'-s', api_address)
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't delete service %(service)s due to error "
|
||||
"%(error)s") % {'service': name, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
# Pod Operations
|
||||
def pod_create(self, api_address, pod):
|
||||
LOG.debug("pod_create contents %s" % pod.as_dict())
|
||||
try:
|
||||
out, err = _k8s_create(api_address, pod)
|
||||
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't create pod with contents %(content)s "
|
||||
"due to error %(error)s") %
|
||||
{'content': pod, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
def pod_update(self, api_address, pod):
|
||||
LOG.debug("pod_update contents %s" % pod.as_dict())
|
||||
try:
|
||||
out, err = _k8s_update(api_address, pod)
|
||||
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't update pod with %(content)s due to error "
|
||||
"%(error)s") % {'content': pod, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
def pod_delete(self, api_address, name):
|
||||
LOG.debug("pod_delete %s" % name)
|
||||
try:
|
||||
out, err = utils.trycmd('kubectl', 'delete', 'pod', name,
|
||||
'-s', api_address,)
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't delete pod %(pod)s due to error "
|
||||
"%(error)s") % {'pod': name, 'error': e})
|
||||
return False
|
||||
|
||||
if err:
|
||||
if ('"%s" not found' % name) in err:
|
||||
raise exception.PodNotFound(pod=name)
|
||||
else:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
# Replication Controller Operations
|
||||
def rc_create(self, api_address, rc):
|
||||
LOG.debug("rc_create contents %s" % rc.as_dict())
|
||||
try:
|
||||
out, err = _k8s_create(api_address, rc)
|
||||
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't create rc with contents %(content)s due "
|
||||
"error %(error)s") % {'content': rc, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
def rc_update(self, api_address, rc):
|
||||
LOG.debug("rc_update contents %s" % rc.as_dict())
|
||||
try:
|
||||
out, err = _k8s_update(api_address, rc)
|
||||
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't update rc with contents %(content)s due "
|
||||
"to error %(error)s") % {'content': rc, 'error': e})
|
||||
return False
|
||||
return True
|
||||
|
||||
def rc_delete(self, api_address, name):
|
||||
LOG.debug("rc_delete %s" % name)
|
||||
try:
|
||||
out, err = utils.trycmd('kubectl', 'delete', 'rc', name,
|
||||
'-s', api_address)
|
||||
if err:
|
||||
return False
|
||||
except Exception as e:
|
||||
LOG.error(_LE("Couldn't delete rc %(rc)s due to error %(error)s")
|
||||
% {'rc': name, 'error': e})
|
||||
return False
|
||||
return True
|
|
@ -181,7 +181,6 @@ class Handler(object):
|
|||
# TODO(yuanying): parse pod file and,
|
||||
# - extract pod name and set it
|
||||
# - extract pod labels and set it
|
||||
# TODO(yuanying): Should kube_utils support definition_url?
|
||||
# When do we get pod labels and name?
|
||||
pod.create(context)
|
||||
return pod
|
||||
|
|
|
@ -1,552 +0,0 @@
|
|||
# Copyright 2014 NEC Corporation. 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.
|
||||
|
||||
from magnum.common import exception
|
||||
from magnum.conductor.handlers.common import kube_utils
|
||||
from magnum.tests import base
|
||||
|
||||
import mock
|
||||
from mock import patch
|
||||
|
||||
|
||||
class TestKubeUtils(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(TestKubeUtils, self).setUp()
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create_with_data')
|
||||
def test_k8s_create_data(self,
|
||||
mock_create_with_data):
|
||||
expected_data = 'data'
|
||||
api_address = 'api_address'
|
||||
mock_resource = mock.MagicMock()
|
||||
mock_resource.manifest = expected_data
|
||||
mock_resource.manifest_url = None
|
||||
|
||||
kube_utils._k8s_create(api_address, mock_resource)
|
||||
mock_create_with_data.assert_called_once_with(api_address,
|
||||
expected_data)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create_with_path')
|
||||
def test_k8s_create_url(self,
|
||||
mock_create_with_path):
|
||||
expected_url = 'url'
|
||||
api_address = 'api_address'
|
||||
mock_resource = mock.MagicMock()
|
||||
mock_resource.manifest = None
|
||||
mock_resource.manifest_url = expected_url
|
||||
|
||||
kube_utils._k8s_create(api_address, mock_resource)
|
||||
mock_create_with_path.assert_called_once_with(api_address,
|
||||
expected_url)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_k8s_create_with_path(self, mock_trycmd):
|
||||
expected_api_address = 'api_address'
|
||||
expected_pod_file = 'pod_file'
|
||||
expected_command = [
|
||||
'kubectl', 'create',
|
||||
'-s', expected_api_address,
|
||||
'-f', expected_pod_file
|
||||
]
|
||||
|
||||
kube_utils._k8s_create_with_path(expected_api_address,
|
||||
expected_pod_file)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create_with_path')
|
||||
@patch('tempfile.NamedTemporaryFile')
|
||||
def test_k8s_create_with_data(self,
|
||||
mock_named_tempfile,
|
||||
mock_k8s_create):
|
||||
expected_api_address = 'api_address'
|
||||
expected_data = 'resource_data'
|
||||
expected_filename = 'resource_file'
|
||||
|
||||
mock_file = mock.MagicMock()
|
||||
mock_file.name = expected_filename
|
||||
mock_named_tempfile.return_value.__enter__.return_value = mock_file
|
||||
|
||||
kube_utils._k8s_create_with_data(
|
||||
expected_api_address,
|
||||
expected_data)
|
||||
|
||||
mock_file.write.assert_called_once_with(expected_data)
|
||||
mock_k8s_create.assert_called_once_with(expected_api_address,
|
||||
expected_filename)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update_with_data')
|
||||
def test_k8s_update_data(self,
|
||||
mock_update_with_data):
|
||||
expected_data = 'data'
|
||||
api_address = 'api_address'
|
||||
mock_resource = mock.MagicMock()
|
||||
mock_resource.manifest = expected_data
|
||||
mock_resource.manifest_url = None
|
||||
|
||||
kube_utils._k8s_update(api_address, mock_resource)
|
||||
mock_update_with_data.assert_called_once_with(api_address,
|
||||
expected_data)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update_with_path')
|
||||
def test_k8s_update_url(self,
|
||||
mock_update_with_path):
|
||||
expected_url = 'url'
|
||||
api_address = 'api_address'
|
||||
mock_resource = mock.MagicMock()
|
||||
mock_resource.manifest = None
|
||||
mock_resource.manifest_url = expected_url
|
||||
|
||||
kube_utils._k8s_update(api_address, mock_resource)
|
||||
mock_update_with_path.assert_called_once_with(api_address,
|
||||
expected_url)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_k8s_update_with_path(self, mock_trycmd):
|
||||
expected_api_address = 'api_address'
|
||||
expected_pod_file = 'pod_file'
|
||||
expected_command = [
|
||||
'kubectl', 'update',
|
||||
'-s', expected_api_address,
|
||||
'-f', expected_pod_file
|
||||
]
|
||||
|
||||
kube_utils._k8s_update_with_path(expected_api_address,
|
||||
expected_pod_file)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update_with_path')
|
||||
@patch('tempfile.NamedTemporaryFile')
|
||||
def test_k8s_update_with_data(self,
|
||||
mock_named_tempfile,
|
||||
mock_k8s_update):
|
||||
expected_api_address = 'api_address'
|
||||
expected_data = 'resource_data'
|
||||
expected_filename = 'resource_file'
|
||||
|
||||
mock_file = mock.MagicMock()
|
||||
mock_file.name = expected_filename
|
||||
mock_named_tempfile.return_value.__enter__.return_value = mock_file
|
||||
|
||||
kube_utils._k8s_update_with_data(
|
||||
expected_api_address,
|
||||
expected_data)
|
||||
|
||||
mock_file.write.assert_called_once_with(expected_data)
|
||||
mock_k8s_update.assert_called_once_with(expected_api_address,
|
||||
expected_filename)
|
||||
|
||||
|
||||
class KubeClientTestCase(base.TestCase):
|
||||
def setUp(self):
|
||||
super(KubeClientTestCase, self).setUp()
|
||||
self.kube_client = kube_utils.KubeClient()
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_pod_create(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_content = mock.MagicMock(manifest='pod_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_pod_content,
|
||||
]
|
||||
mock_k8s_create.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.pod_create(expected_api_address,
|
||||
expected_pod_content)
|
||||
self.assertTrue(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_pod_create_with_err(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_content = mock.MagicMock(manifest='pod_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_pod_content,
|
||||
]
|
||||
mock_k8s_create.return_value = ("", "create failed")
|
||||
|
||||
result = self.kube_client.pod_create(expected_api_address,
|
||||
expected_pod_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_pod_create_failure_exception(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_content = mock.MagicMock(manifest='pod_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_pod_content,
|
||||
]
|
||||
mock_k8s_create.side_effect = Exception()
|
||||
result = self.kube_client.pod_create(expected_api_address,
|
||||
expected_pod_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_pod_update(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_content = mock.MagicMock(manifest='pod_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_pod_content,
|
||||
]
|
||||
mock_k8s_update.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.pod_update(expected_api_address,
|
||||
expected_pod_content)
|
||||
self.assertTrue(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_pod_update_with_err(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_content = mock.MagicMock(manifest='pod_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_pod_content,
|
||||
]
|
||||
mock_k8s_update.return_value = ("", "create failed")
|
||||
|
||||
result = self.kube_client.pod_update(expected_api_address,
|
||||
expected_pod_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_pod_update_failure_exception(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_content = mock.MagicMock(manifest='pod_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_pod_content,
|
||||
]
|
||||
mock_k8s_update.side_effect = Exception()
|
||||
result = self.kube_client.pod_update(expected_api_address,
|
||||
expected_pod_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_pod_delete(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_name = 'test-pod'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'pod', expected_pod_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.pod_delete(expected_api_address,
|
||||
expected_pod_name)
|
||||
self.assertTrue(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_pod_delete_failure_err_not_empty(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_name = 'test-pod'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'pod', expected_pod_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.return_value = ("", "error")
|
||||
|
||||
result = self.kube_client.pod_delete(expected_api_address,
|
||||
expected_pod_name)
|
||||
self.assertFalse(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_pod_delete_failure_exception(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_name = 'test-pod'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'pod', expected_pod_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.side_effect = Exception()
|
||||
|
||||
result = self.kube_client.pod_delete(expected_api_address,
|
||||
expected_pod_name)
|
||||
self.assertFalse(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_pod_delete_not_found_old(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_name = 'test-pod'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'pod', expected_pod_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.return_value = ("", 'pod "test-pod" not found')
|
||||
|
||||
self.assertRaises(exception.PodNotFound, self.kube_client.pod_delete,
|
||||
expected_api_address, expected_pod_name)
|
||||
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_pod_delete_not_found_new(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_pod_name = 'test-pod'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'pod', expected_pod_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.return_value = ("", 'pods "test-pod" not found')
|
||||
|
||||
self.assertRaises(exception.PodNotFound, self.kube_client.pod_delete,
|
||||
expected_api_address, expected_pod_name)
|
||||
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_service_delete(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_name = 'test-service'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'service', expected_service_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.service_delete(expected_api_address,
|
||||
expected_service_name)
|
||||
self.assertTrue(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_service_delete_failure_exception(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_name = 'test-service'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'service', expected_service_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.side_effect = Exception()
|
||||
|
||||
result = self.kube_client.service_delete(expected_api_address,
|
||||
expected_service_name)
|
||||
self.assertFalse(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_service_update(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_content = mock.MagicMock(manifest='service_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_service_content,
|
||||
]
|
||||
mock_k8s_update.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.service_update(expected_api_address,
|
||||
expected_service_content)
|
||||
self.assertTrue(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_service_update_with_err(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_content = mock.MagicMock(manifest='service_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_service_content,
|
||||
]
|
||||
mock_k8s_update.return_value = ("", "create failed")
|
||||
|
||||
result = self.kube_client.service_update(expected_api_address,
|
||||
expected_service_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_service_update_failure_exception(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_content = mock.MagicMock(manifest='service_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_service_content,
|
||||
]
|
||||
mock_k8s_update.side_effect = Exception()
|
||||
result = self.kube_client.service_update(expected_api_address,
|
||||
expected_service_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_service_create(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_content = mock.MagicMock(manifest='service_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_service_content,
|
||||
]
|
||||
mock_k8s_create.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.service_create(expected_api_address,
|
||||
expected_service_content)
|
||||
self.assertTrue(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_service_create_with_err(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_content = mock.MagicMock(manifest='service_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_service_content,
|
||||
]
|
||||
mock_k8s_create.return_value = ("", "create failed")
|
||||
|
||||
result = self.kube_client.service_create(expected_api_address,
|
||||
expected_service_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_service_create_failure_exception(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_service_content = mock.MagicMock(manifest='service_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_service_content,
|
||||
]
|
||||
mock_k8s_create.side_effect = Exception()
|
||||
result = self.kube_client.service_create(expected_api_address,
|
||||
expected_service_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_rc_update(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_content = mock.MagicMock(manifest='rc_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_rc_content,
|
||||
]
|
||||
mock_k8s_update.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.rc_update(expected_api_address,
|
||||
expected_rc_content)
|
||||
self.assertTrue(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_rc_update_with_err(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_content = mock.MagicMock(manifest='rc_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_rc_content,
|
||||
]
|
||||
mock_k8s_update.return_value = ("", "update failed")
|
||||
|
||||
result = self.kube_client.rc_update(expected_api_address,
|
||||
expected_rc_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_update')
|
||||
def test_rc_update_failure_exception(self, mock_k8s_update):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_content = mock.MagicMock(manifest='rc_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_rc_content,
|
||||
]
|
||||
mock_k8s_update.side_effect = Exception()
|
||||
result = self.kube_client.rc_update(expected_api_address,
|
||||
expected_rc_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_update.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_rc_create(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_content = mock.MagicMock(manifest='rc_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_rc_content,
|
||||
]
|
||||
mock_k8s_create.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.rc_create(expected_api_address,
|
||||
expected_rc_content)
|
||||
self.assertTrue(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_rc_create_with_err(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_content = mock.MagicMock(manifest='rc_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_rc_content,
|
||||
]
|
||||
mock_k8s_create.return_value = ("", "create failed")
|
||||
|
||||
result = self.kube_client.rc_create(expected_api_address,
|
||||
expected_rc_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.conductor.handlers.common.kube_utils._k8s_create')
|
||||
def test_rc_create_failure_exception(self, mock_k8s_create):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_content = mock.MagicMock(manifest='rc_content')
|
||||
expected_command = [
|
||||
expected_api_address,
|
||||
expected_rc_content,
|
||||
]
|
||||
mock_k8s_create.side_effect = Exception()
|
||||
result = self.kube_client.rc_create(expected_api_address,
|
||||
expected_rc_content)
|
||||
self.assertFalse(result)
|
||||
mock_k8s_create.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_rc_delete(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_name = 'test-rc'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'rc', expected_rc_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.return_value = ("", "")
|
||||
|
||||
result = self.kube_client.rc_delete(expected_api_address,
|
||||
expected_rc_name)
|
||||
self.assertTrue(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
||||
|
||||
@patch('magnum.openstack.common.utils.trycmd')
|
||||
def test_rc_delete_failure_exception(self, mock_trycmd):
|
||||
expected_api_address = 'master-address'
|
||||
expected_rc_name = 'test-rc'
|
||||
expected_command = [
|
||||
'kubectl', 'delete', 'rc', expected_rc_name,
|
||||
'-s', expected_api_address
|
||||
]
|
||||
mock_trycmd.side_effect = Exception()
|
||||
|
||||
result = self.kube_client.rc_delete(expected_api_address,
|
||||
expected_rc_name)
|
||||
self.assertFalse(result)
|
||||
mock_trycmd.assert_called_once_with(*expected_command)
|
Loading…
Reference in New Issue