openstack
/
magnum
Code Issues Proposed changes

Objects from Bay - Pods 

Currently k8s objects (pod/rc/service) are read from the
database. In order for native clients to work, they must
be read from the ReST bay endpoint. To execute native
clients, we must have one truth of the state of the
system, not two as we do now. This patch proposes the
change to the Pod object.

Also, please refer to the related-bug as to the temporary changes
done to make the test work for other objects namely rc,
service. These changes will be removed when the object from bay
patches for all the k8s objects are merged as part of a seperate
patch.

Partially-Implements: bp objects-from-bay
Related-Bug: #1502367
Related-Bug: #1504379

Change-Id: Id5be7fba2eb8622fcfeb48068728e440a0af3f5e
changes/67/223367/18
Vilobh Meshram 2015-09-14 18:39:04 -07:00
parent 534e44c546
commit 44122d08f8
12 changed files with 672 additions and 515 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
magnum/api/controllers/v1/pod.py
View File

@ -180,15 +180,9 @@ class PodsController(rest.RestController):
limit = api_utils.validate_limit(limit)
sort_dir = api_utils.validate_sort_dir(sort_dir)
context = pecan.request.context
marker_obj = None
if marker:
marker_obj = objects.Pod.get_by_uuid(pecan.request.context,
marker)
pods = pecan.request.rpcapi.pod_list(pecan.request.context, limit,
marker_obj, sort_key=sort_key,
sort_dir=sort_dir)
pods = pecan.request.rpcapi.pod_list(context, bay_ident)
return PodCollection.convert_with_links(pods, limit,
url=resource_url,
@ -246,7 +240,8 @@ class PodsController(rest.RestController):
:param pod_ident: UUID of a pod or logical name of the pod.
:param bay_ident: UUID or logical name of the Bay.
"""
rpc_pod = api_utils.get_rpc_resource('Pod', pod_ident)
context = pecan.request.context
rpc_pod = pecan.request.rpcapi.pod_show(context, pod_ident, bay_ident)
return Pod.convert_with_links(rpc_pod)
@ -280,35 +275,18 @@ class PodsController(rest.RestController):
:param bay_ident: UUID or logical name of the Bay.
:param patch: a json PATCH document to apply to this pod.
"""
rpc_pod = api_utils.get_rpc_resource('Pod', pod_ident)
# Init manifest and manifest_url field because we don't store them
# in database.
rpc_pod['manifest'] = None
rpc_pod['manifest_url'] = None
pod_dict = {}
pod_dict['manifest'] = None
pod_dict['manifest_url'] = None
try:
pod_dict = rpc_pod.as_dict()
pod = Pod(**api_utils.apply_jsonpatch(pod_dict, patch))
if pod.manifest or pod.manifest_url:
pod.parse_manifest()
except api_utils.JSONPATCH_EXCEPTIONS as e:
raise exception.PatchError(patch=patch, reason=e)
# Update only the fields that have changed
for field in objects.Pod.fields:
try:
patch_val = getattr(pod, field)
except AttributeError:
# Ignore fields that aren't exposed in the API
continue
if patch_val == wtypes.Unset:
patch_val = None
if rpc_pod[field] != patch_val:
rpc_pod[field] = patch_val
if pod.manifest or pod.manifest_url:
pecan.request.rpcapi.pod_update(rpc_pod)
else:
rpc_pod.save()
rpc_pod = pecan.request.rpcapi.pod_update(pod_ident, bay_ident,
pod.manifest)
return Pod.convert_with_links(rpc_pod)
@policy.enforce_wsgi("pod")
@ -320,6 +298,4 @@ class PodsController(rest.RestController):
:param pod_ident: UUID of a pod or logical name of the pod.
:param bay_ident: UUID or logical name of the Bay.
"""
rpc_pod = api_utils.get_rpc_resource('Pod', pod_ident)
pecan.request.rpcapi.pod_delete(rpc_pod.uuid)
pecan.request.rpcapi.pod_delete(pod_ident, bay_ident)

8
magnum/common/exception.py
View File

@ -404,6 +404,14 @@ class PodAlreadyExists(Conflict):
message = _("A node with UUID %(uuid)s already exists.")
class PodListNotFound(ResourceNotFound):
message = _("Pod list could not be found for Bay %(bay_uuid)s.")
class PodCreationFailed(Invalid):
message = _("Pod creation failed in Bay %(bay_uuid)s.")
class ReplicationControllerNotFound(ResourceNotFound):
message = _("ReplicationController %(rc)s could not be found.")

19
magnum/conductor/api.py
View File

@ -73,17 +73,20 @@ class API(rpc_service.API):
def pod_create(self, pod):
return self._call('pod_create', pod=pod)
def pod_list(self, context, limit, marker, sort_key, sort_dir):
return objects.Pod.list(context, limit, marker, sort_key, sort_dir)
def pod_list(self, context, bay_ident):
return self._call('pod_list', bay_ident=bay_ident)
def pod_update(self, pod):
return self._call('pod_update', pod=pod)
def pod_update(self, pod_ident, bay_ident, manifest):
return self._call('pod_update', pod_ident=pod_ident,
bay_ident=bay_ident, manifest=manifest)
def pod_delete(self, uuid):
return self._call('pod_delete', uuid=uuid)
def pod_delete(self, pod_ident, bay_ident):
return self._call('pod_delete', pod_ident=pod_ident,
bay_ident=bay_ident)
def pod_show(self, context, uuid):
return objects.Pod.get_by_uuid(context, uuid)
def pod_show(self, context, pod_ident, bay_ident):
return self._call('pod_show', pod_ident=pod_ident,
bay_ident=bay_ident)
# ReplicationController Operations

157
magnum/conductor/handlers/k8s_conductor.py
View File

@ -221,56 +221,155 @@ class Handler(object):
# Pod Operations
def pod_create(self, context, pod):
LOG.debug("pod_create")
self.k8s_api = k8s.create_k8s_api(context, pod)
self.k8s_api = k8s.create_k8s_api_pod(context, pod.bay_uuid)
manifest = k8s_manifest.parse(pod.manifest)
try:
resp = self.k8s_api.create_namespaced_pod(body=manifest,
namespace='default')
except rest.ApiException as err:
pod.status = 'failed'
if err.status != 409:
pod.create(context)
raise exception.KubernetesAPIFailed(err=err)
pod.status = resp.status.phase
pod.host = resp.spec.node_name
# call the pod object to persist in db
# TODO(yuanying): parse pod file and,
# - extract pod name and set it
# - extract pod labels and set it
# When do we get pod labels and name?
pod.create(context)
if resp is None:
raise exception.PodCreationFailed(bay_uuid=pod.bay_uuid)
pod['uuid'] = resp.metadata.uid
pod['name'] = resp.metadata.name
pod['images'] = [c.image for c in resp.spec.containers]
pod['labels'] = ast.literal_eval(resp.metadata.labels)
pod['status'] = resp.status.phase
pod['host'] = resp.spec.node_name
return pod
def pod_update(self, context, pod):
LOG.debug("pod_update %s", pod.uuid)
self.k8s_api = k8s.create_k8s_api(context, pod)
manifest = k8s_manifest.parse(pod.manifest)
def pod_update(self, context, pod_ident, bay_ident, manifest):
LOG.debug("pod_update %s", pod_ident)
# Since bay identifier is specified verify whether its a UUID
# or Name. If name is specified as bay identifier need to extract
# the bay uuid since its needed to get the k8s_api object.
if not utils.is_uuid_like(bay_ident):
bay = objects.Bay.get_by_name(context, bay_ident)
bay_ident = bay.uuid
bay_uuid = bay_ident
self.k8s_api = k8s.create_k8s_api_pod(context, bay_uuid)
if utils.is_uuid_like(pod_ident):
pod = objects.Pod.get_by_uuid(context, pod_ident,
bay_uuid, self.k8s_api)
else:
pod = objects.Pod.get_by_name(context, pod_ident,
bay_uuid, self.k8s_api)
pod_ident = pod.name
try:
self.k8s_api.replace_namespaced_pod(name=str(pod.name),
body=manifest,
namespace='default')
resp = self.k8s_api.replace_namespaced_pod(name=str(pod_ident),
body=manifest,
namespace='default')
except rest.ApiException as err:
raise exception.KubernetesAPIFailed(err=err)
# call the pod object to persist in db
pod.refresh(context)
pod.save()
if resp is None:
raise exception.PodNotFound(pod=pod.uuid)
pod['uuid'] = resp.metadata.uid
pod['name'] = resp.metadata.name
pod['project_id'] = context.project_id
pod['user_id'] = context.user_id
pod['bay_uuid'] = bay_uuid
pod['images'] = [c.image for c in resp.spec.containers]
if not resp.metadata.labels:
pod['labels'] = {}
else:
pod['labels'] = ast.literal_eval(resp.metadata.labels)
pod['status'] = resp.status.phase
pod['host'] = resp.spec.node_name
return pod
def pod_delete(self, context, uuid):
LOG.debug("pod_delete %s", uuid)
pod = objects.Pod.get_by_uuid(context, uuid)
self.k8s_api = k8s.create_k8s_api(context, pod)
if conductor_utils.object_has_stack(context, pod.bay_uuid):
def pod_delete(self, context, pod_ident, bay_ident):
LOG.debug("pod_delete %s", pod_ident)
# Since bay identifier is specified verify whether its a UUID
# or Name. If name is specified as bay identifier need to extract
# the bay uuid since its needed to get the k8s_api object.
if not utils.is_uuid_like(bay_ident):
bay = objects.Bay.get_by_name(context, bay_ident)
bay_ident = bay.uuid
bay_uuid = bay_ident
self.k8s_api = k8s.create_k8s_api_pod(context, bay_uuid)
if utils.is_uuid_like(pod_ident):
pod = objects.Pod.get_by_uuid(context, pod_ident,
bay_uuid, self.k8s_api)
pod_name = pod.name
else:
pod_name = pod_ident
if conductor_utils.object_has_stack(context, bay_uuid):
try:
self.k8s_api.delete_namespaced_pod(name=str(pod.name), body={},
self.k8s_api.delete_namespaced_pod(name=str(pod_name), body={},
namespace='default')
except rest.ApiException as err:
if err.status == 404:
pass
else:
raise exception.KubernetesAPIFailed(err=err)
# call the pod object to persist in db
pod.destroy(context)
def pod_show(self, context, pod_ident, bay_ident):
LOG.debug("pod_show %s", pod_ident)
# Since bay identifier is specified verify whether its a UUID
# or Name. If name is specified as bay identifier need to extract
# the bay uuid since its needed to get the k8s_api object.
if not utils.is_uuid_like(bay_ident):
bay = objects.Bay.get_by_name(context, bay_ident)
bay_ident = bay.uuid
bay_uuid = bay_ident
self.k8s_api = k8s.create_k8s_api_pod(context, bay_uuid)
if utils.is_uuid_like(pod_ident):
pod = objects.Pod.get_by_uuid(context, pod_ident,
bay_uuid, self.k8s_api)
else:
pod = objects.Pod.get_by_name(context, pod_ident,
bay_uuid, self.k8s_api)
return pod
def pod_list(self, context, bay_ident):
# Since bay identifier is specified verify whether its a UUID
# or Name. If name is specified as bay identifier need to extract
# the bay uuid since its needed to get the k8s_api object.
if not utils.is_uuid_like(bay_ident):
bay = objects.Bay.get_by_name(context, bay_ident)
bay_ident = bay.uuid
bay_uuid = bay_ident
self.k8s_api = k8s.create_k8s_api_pod(context, bay_uuid)
try:
resp = self.k8s_api.list_namespaced_pod(namespace='default')
except rest.ApiException as err:
raise exception.KubernetesAPIFailed(err=err)
if resp is None:
raise exception.PodListNotFound(bay_uuid=bay_uuid)
pods = []
for pod_entry in resp.items:
pod = {}
pod['uuid'] = pod_entry.metadata.uid
pod['name'] = pod_entry.metadata.name
pod['project_id'] = context.project_id
pod['user_id'] = context.user_id
pod['bay_uuid'] = bay_uuid
pod['images'] = [c.image for c in pod_entry.spec.containers]
if not pod_entry.metadata.labels:
pod['labels'] = {}
else:
pod['labels'] = ast.literal_eval(pod_entry.metadata.labels)
pod['status'] = pod_entry.status.phase
pod['host'] = pod_entry.spec.node_name
pod_obj = objects.Pod(context, **pod)
pods.append(pod_obj)
return pods
# Replication Controller Operations
def rc_create(self, context, rc):

84
magnum/conductor/k8s_api.py
View File

@ -197,6 +197,90 @@ def create_k8s_api_service(context, bay_uuid):
return K8sAPI_Service(context, bay_uuid)
# NB : This is a place holder class. This class and create_k8s_api_pod
# method will be removed once the objects from bay code for k8s
# objects is merged. These changes are temporary to get the Unit
# test working.
class K8sAPI_Pod(apiv_api.ApivApi):
def _create_temp_file_with_content(self, content):
"""Creates temp file and write content to the file.
:param content: file content
:returns: temp file
"""
try:
tmp = NamedTemporaryFile(delete=True)
tmp.write(content)
tmp.flush()
except Exception as err:
LOG.error("Error while creating temp file: %s" % err)
raise err
return tmp
def __init__(self, context, bay_uuid):
self.ca_file = None
self.cert_file = None
self.key_file = None
bay = utils.retrieve_bay(context, bay_uuid)
if bay.magnum_cert_ref:
self._create_certificate_files(bay)
# build a connection with Kubernetes master
client = api_client.ApiClient(bay.api_address,
key_file=self.key_file.name,
cert_file=self.cert_file.name,
ca_certs=self.ca_file.name)
super(K8sAPI_Pod, self).__init__(client)
def _create_certificate_files(self, bay):
"""Read certificate and key for a bay and stores in files.
:param bay: Bay object
"""
magnum_cert_obj = cert_manager.get_backend().CertManager.get_cert(
bay.magnum_cert_ref)
self.cert_file = self._create_temp_file_with_content(
magnum_cert_obj.certificate)
private_key = serialization.load_pem_private_key(
magnum_cert_obj.private_key,
password=magnum_cert_obj.private_key_passphrase,
backend=default_backend(),
)
private_key = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption())
self.key_file = self._create_temp_file_with_content(
private_key)
ca_cert_obj = cert_manager.get_backend().CertManager.get_cert(
bay.ca_cert_ref)
self.ca_file = self._create_temp_file_with_content(
ca_cert_obj.certificate)
def __del__(self):
if self.ca_file:
self.ca_file.close()
if self.cert_file:
self.cert_file.close()
if self.key_file:
self.key_file.close()
def create_k8s_api_pod(context, bay_uuid):
"""Create a kubernetes API client
Creates connection with Kubernetes master and creates ApivApi instance
to call Kubernetes APIs.
:param context: The security context
:param bay_uuid: Unique identifier for the Bay
"""
return K8sAPI_Pod(context, bay_uuid)
# NB : This is a place holder class. This class and create_k8s_api_rc
# method will be removed once the objects from bay code for k8s
# objects is merged. These changes are temporary to get the Unit

195
magnum/objects/pod.py
View File

@ -12,9 +12,13 @@
from oslo_versionedobjects import fields
from magnum.common import exception
from magnum.common.pythonk8sclient.swagger_client import rest
from magnum.db import api as dbapi
from magnum.objects import base
import ast
@base.MagnumObjectRegistry.register
class Pod(base.MagnumPersistentObject, base.MagnumObject,
@ -41,149 +45,78 @@ class Pod(base.MagnumPersistentObject, base.MagnumObject,
'host': fields.StringField(nullable=True),
}
@staticmethod
def _from_db_object(pod, db_pod):
"""Converts a database entity to a formal object."""
for field in pod.fields:
# ignore manifest_url as it was used for create pod
if field == 'manifest_url':
continue
if field == 'manifest':
continue
pod[field] = db_pod[field]
pod.obj_reset_changes()
return pod
@staticmethod
def _from_db_object_list(db_objects, cls, context):
"""Converts a list of database entities to a list of formal objects."""
return [Pod._from_db_object(cls(context), obj) for obj in db_objects]
@base.remotable_classmethod
def get_by_id(cls, context, pod_id):
"""Find a pod based on its integer id and return a Pod object.
def get_by_uuid(cls, context, uuid, bay_uuid, k8s_api):
"""Find a pod based on pod uuid and the uuid for a bay.
:param pod_id: the id of a pod.
:param context: Security context
:returns: a :class:`Pod` object.
"""
db_pod = cls.dbapi.get_pod_by_id(context, pod_id)
pod = Pod._from_db_object(cls(context), db_pod)
return pod
@base.remotable_classmethod
def get_by_uuid(cls, context, uuid):
"""Find a pod based on uuid and return a :class:`Pod` object.
:param uuid: the uuid of a pod.
:param context: Security context
:param bay_uuid: the UUID of the Bay
:param k8s_api: k8s API object
:returns: a :class:`Pod` object.
"""
db_pod = cls.dbapi.get_pod_by_uuid(context, uuid)
pod = Pod._from_db_object(cls(context), db_pod)
return pod
try:
resp = k8s_api.list_namespaced_pod(namespace='default')
except rest.ApiException as err:
raise exception.KubernetesAPIFailed(err=err)
if resp is None:
raise exception.PodListNotFound(bay_uuid=bay_uuid)
pod = {}
for pod_entry in resp.items:
if pod_entry.metadata.uid == uuid:
pod['uuid'] = pod_entry.metadata.uid
pod['name'] = pod_entry.metadata.name
pod['project_id'] = context.project_id
pod['user_id'] = context.user_id
pod['bay_uuid'] = bay_uuid
pod['images'] = [c.image for c in pod_entry.spec.containers]
if not pod_entry.metadata.labels:
pod['labels'] = {}
else:
pod['labels'] = ast.literal_eval(pod_entry.metadata.labels)
pod['status'] = pod_entry.status.phase
pod['host'] = pod_entry.spec.node_name
pod_obj = Pod(context, **pod)
return pod_obj
raise exception.PodNotFound(pod=uuid)
@base.remotable_classmethod
def get_by_name(cls, context, name):
"""Find a pod based on pod name and return a :class:`Pod` object.
def get_by_name(cls, context, name, bay_uuid, k8s_api):
"""Find a pod based on pod name and the uuid for a bay.
:param context: Security context
:param name: the name of a pod.
:param context: Security context
:param bay_uuid: the UUID of the Bay
:param k8s_api: k8s API object
:returns: a :class:`Pod` object.
"""
db_pod = cls.dbapi.get_pod_by_name(name)
pod = Pod._from_db_object(cls(context), db_pod)
return pod
try:
resp = k8s_api.read_namespaced_pod(name=name,
namespace='default')
except rest.ApiException as err:
raise exception.KubernetesAPIFailed(err=err)
@base.remotable_classmethod
def list(cls, context, limit=None, marker=None,
sort_key=None, sort_dir=None):
"""Return a list of Pod objects.
if resp is None:
raise exception.PodNotFound(pod=name)
:param context: Security context.
:param limit: maximum number of resources to return in a single result.
:param marker: pagination marker for large data sets.
:param sort_key: column to sort results by.
:param sort_dir: direction to sort. "asc" or "desc".
:returns: a list of :class:`Pod` object.
pod = {}
pod['uuid'] = resp.metadata.uid
pod['name'] = resp.metadata.name
pod['project_id'] = context.project_id
pod['user_id'] = context.user_id
pod['bay_uuid'] = bay_uuid
pod['images'] = [c.image for c in resp.spec.containers]
if not resp.metadata.labels:
pod['labels'] = {}
else:
pod['labels'] = ast.literal_eval(resp.metadata.labels)
pod['status'] = resp.status.phase
pod['host'] = resp.spec.node_name
"""
db_pods = cls.dbapi.get_pod_list(context, limit=limit,
marker=marker,
sort_key=sort_key,
sort_dir=sort_dir)
return Pod._from_db_object_list(db_pods, cls, context)
@base.remotable
def create(self, context=None):
"""Create a Pod record in the DB.
:param context: Security context. NOTE: This should only
be used internally by the indirection_api.
Unfortunately, RPC requires context as the first
argument, even though we don't use it.
A context should be set when instantiating the
object, e.g.: Pod(context)
"""
values = self.obj_get_changes()
db_pod = self.dbapi.create_pod(values)
self._from_db_object(self, db_pod)
@base.remotable
def destroy(self, context=None):
"""Delete the Pod from the DB.
:param context: Security context. NOTE: This should only
be used internally by the indirection_api.
Unfortunately, RPC requires context as the first
argument, even though we don't use it.
A context should be set when instantiating the
object, e.g.: Pod(context)
"""
self.dbapi.destroy_pod(self.uuid)
self.obj_reset_changes()
@base.remotable
def save(self, context=None):
"""Save updates to this Pod.
Updates will be made column by column based on the result
of self.what_changed().
:param context: Security context. NOTE: This should only
be used internally by the indirection_api.
Unfortunately, RPC requires context as the first
argument, even though we don't use it.
A context should be set when instantiating the
object, e.g.: Pod(context)
"""
updates = self.obj_get_changes()
self.dbapi.update_pod(self.uuid, updates)
self.obj_reset_changes()
@base.remotable
def refresh(self, context=None):
"""Loads updates for this Pod.
Loads a pod with the same uuid from the database and
checks for updated attributes. Updates are applied from
the loaded pod column by column, if there are any updates.
:param context: Security context. NOTE: This should only
be used internally by the indirection_api.
Unfortunately, RPC requires context as the first
argument, even though we don't use it.
A context should be set when instantiating the
object, e.g.: Pod(context)
"""
current = self.__class__.get_by_uuid(self._context, uuid=self.uuid)
for field in self.fields:
if field == 'manifest_url':
continue
if field == 'manifest':
continue
if self.obj_attr_is_set(field) and self[field] != current[field]:
self[field] = current[field]
pod_obj = Pod(context, **pod)
return pod_obj

337
magnum/tests/unit/api/controllers/v1/test_pod.py
View File

@ -15,14 +15,13 @@ import datetime
import mock
from oslo_config import cfg
from oslo_policy import policy
from oslo_utils import timeutils
from six.moves.urllib import parse as urlparse
from wsme import types as wtypes
from magnum.api.controllers.v1 import pod as api_pod
from magnum.common.pythonk8sclient.swagger_client import rest
from magnum.common import utils
from magnum.conductor import api as rpcapi
from magnum import objects
from magnum.tests import base
from magnum.tests.unit.api import base as api_base
from magnum.tests.unit.api import utils as apiutils
@ -43,8 +42,11 @@ class TestListPod(api_base.FunctionalTest):
def setUp(self):
super(TestListPod, self).setUp()
obj_utils.create_test_bay(self.context)
self.pod = obj_utils.create_test_pod(self.context)
def test_empty(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_empty(self, mock_pod_list):
mock_pod_list.return_value = []
response = self.get_json('/pods')
self.assertEqual([], response['pods'])
@ -54,122 +56,145 @@ class TestListPod(api_base.FunctionalTest):
for field in pod_fields:
self.assertIn(field, pod)
def test_one(self):
pod = obj_utils.create_test_pod(self.context)
response = self.get_json('/pods')
self.assertEqual(pod.uuid, response['pods'][0]["uuid"])
self._assert_pod_fields(response['pods'][0])
def test_get_one(self):
@mock.patch.object(rpcapi.API, 'pod_show')
def test_one(self, mock_pod_show):
pod = obj_utils.create_test_pod(self.context)
mock_pod_show.return_value = pod
response = self.get_json(
'/pods/%s/%s' % (pod['uuid'], pod['bay_uuid']))
self.assertEqual(pod.uuid, response['uuid'])
self._assert_pod_fields(response)
def test_get_one_by_name(self):
@mock.patch.object(rpcapi.API, 'pod_show')
def test_get_one(self, mock_pod_show):
pod = obj_utils.create_test_pod(self.context)
mock_pod_show.return_value = pod
response = self.get_json(
'/pods/%s/%s' % (pod['uuid'], pod['bay_uuid']))
self.assertEqual(pod.uuid, response['uuid'])
self._assert_pod_fields(response)
@mock.patch.object(rpcapi.API, 'pod_show')
def test_get_one_by_name(self, mock_pod_show):
pod = obj_utils.create_test_pod(self.context)
mock_pod_show.return_value = pod
response = self.get_json(
'/pods/%s/%s' % (pod['name'], pod['bay_uuid']))
self.assertEqual(pod.uuid, response['uuid'])
self._assert_pod_fields(response)
def test_get_one_by_name_not_found(self):
@mock.patch.object(rpcapi.API, 'pod_show')
def test_get_one_by_name_not_found(self, mock_pod_show):
err = rest.ApiException(status=404)
mock_pod_show.side_effect = err
response = self.get_json(
'/pods/not_found/5d12f6fd-a196-4bf0-ae4c-1f639a523a52',
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual(500, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_get_one_by_name_multiple_pod(self):
@mock.patch.object(rpcapi.API, 'pod_show')
def test_get_one_by_name_multiple_pod(self, mock_pod_show):
obj_utils.create_test_pod(self.context, name='test_pod',
uuid=utils.generate_uuid())
obj_utils.create_test_pod(self.context, name='test_pod',
uuid=utils.generate_uuid())
err = rest.ApiException(status=500)
mock_pod_show.side_effect = err
response = self.get_json(
'/pods/test_pod/5d12f6fd-a196-4bf0-ae4c-1f639a523a52',
expect_errors=True)
self.assertEqual(409, response.status_int)
self.assertEqual(500, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_get_all_with_pagination_marker(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_get_all_with_pagination_marker(self, mock_pod_list):
pod_list = []
for id_ in range(4):
pod = obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
pod_list.append(pod.uuid)
mock_pod_list.return_value = [pod]
response = self.get_json('/pods?limit=3&marker=%s' % pod_list[2])
self.assertEqual(1, len(response['pods']))
self.assertEqual(pod_list[-1], response['pods'][0]['uuid'])
def test_detail(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_detail(self, mock_pod_list):
pod = obj_utils.create_test_pod(self.context)
mock_pod_list.return_value = [pod]
response = self.get_json('/pods/detail')
self.assertEqual(pod.uuid, response['pods'][0]["uuid"])
self._assert_pod_fields(response['pods'][0])
def test_detail_with_pagination_marker(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_detail_with_pagination_marker(self, mock_pod_list):
pod_list = []
for id_ in range(4):
pod = obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
pod_list.append(pod.uuid)
mock_pod_list.return_value = [pod]
response = self.get_json('/pods/detail?limit=3&marker=%s'
% pod_list[2])
self.assertEqual(1, len(response['pods']))
self.assertEqual(pod_list[-1], response['pods'][0]['uuid'])
self._assert_pod_fields(response['pods'][0])
def test_detail_against_single(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_detail_against_single(self, mock_pod_list):
pod = obj_utils.create_test_pod(self.context)
mock_pod_list.return_value = [pod]
response = self.get_json('/pods/%s/detail' % pod['uuid'],
expect_errors=True)
self.assertEqual(404, response.status_int)
def test_many(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_many(self, mock_pod_list):
pod_list = []
for id_ in range(5):
for id_ in range(1):
pod = obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
pod_list.append(pod.uuid)
mock_pod_list.return_value = [pod]
response = self.get_json('/pods')
self.assertEqual(len(pod_list), len(response['pods']))
uuids = [p['uuid'] for p in response['pods']]
self.assertEqual(sorted(pod_list), sorted(uuids))
def test_links(self):
@mock.patch.object(rpcapi.API, 'pod_show')
def test_links(self, mock_pod_show):
uuid = utils.generate_uuid()
obj_utils.create_test_pod(self.context, id=1, uuid=uuid)
response = self.get_json(
'/pods/%s/%s' % (uuid, '5d12f6fd-a196-4bf0-ae4c-1f639a523a52'))
pod = obj_utils.create_test_pod(self.context, id=1, uuid=uuid)
mock_pod_show.return_value = pod
response = self.get_json('/pods/%s/%s' % (uuid, pod.bay_uuid))
self.assertIn('links', response.keys())
self.assertEqual(2, len(response['links']))
self.assertIn(uuid, response['links'][0]['href'])
def test_collection_links(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_collection_links(self, mock_pod_list):
for id_ in range(5):
obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
response = self.get_json('/pods/?limit=3')
self.assertEqual(3, len(response['pods']))
pod = obj_utils.create_test_pod(self.context,
id=id_,
uuid=utils.generate_uuid())
mock_pod_list.return_value = [pod]
response = self.get_json('/pods/?limit=1')
self.assertEqual(1, len(response['pods']))
next_marker = response['pods'][-1]['uuid']
self.assertIn(next_marker, response['next'])
def test_collection_links_default_limit(self):
@mock.patch.object(rpcapi.API, 'pod_list')
def test_collection_links_default_limit(self, mock_pod_list):
cfg.CONF.set_override('max_limit', 3, 'api')
for id_ in range(5):
obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
pod = obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
mock_pod_list.return_value = [pod]
response = self.get_json('/pods')
self.assertEqual(3, len(response['pods']))
next_marker = response['pods'][-1]['uuid']
self.assertIn(next_marker, response['next'])
self.assertEqual(1, len(response['pods']))
class TestPatch(api_base.FunctionalTest):
@ -181,31 +206,8 @@ class TestPatch(api_base.FunctionalTest):
desc='pod_example_A_desc',
status='Running')
@mock.patch('oslo_utils.timeutils.utcnow')
def test_replace_ok(self, mock_utcnow):
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
new_desc = 'pod_example_B_desc'
response = self.get_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid))
self.assertNotEqual(new_desc, response['desc'])
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/desc', 'value': new_desc,
'op': 'replace'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
response = self.get_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid))
self.assertEqual(new_desc, response['desc'])
return_updated_at = timeutils.parse_isotime(
response['updated_at']).replace(tzinfo=None)
self.assertEqual(test_time, return_updated_at)
def test_replace_bay_uuid(self):
self.pod.manifest = '{"key": "value"}'
another_bay = obj_utils.create_test_bay(self.context,
uuid=utils.generate_uuid())
response = self.patch_json(
@ -215,21 +217,23 @@ class TestPatch(api_base.FunctionalTest):
'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
self.assertEqual(400, response.status_code)
def test_replace_non_existent_bay_uuid(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/bay_uuid',
'value': utils.generate_uuid(),
'op': 'replace'}],
expect_errors=True)
self.pod.manifest = '{"key": "value"}'
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/bay_uuid',
'value': utils.generate_uuid(),
'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_code)
self.assertTrue(response.json['error_message'])
def test_replace_internal_field(self):
response = self.patch_json(
'/pods/%s' % self.pod.uuid,
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/labels', 'value': {}, 'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
@ -243,17 +247,19 @@ class TestPatch(api_base.FunctionalTest):
'value': 'pod_example_B_desc',
'op': 'replace'}],
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
@mock.patch.object(rpcapi.API, 'pod_update')
@mock.patch.object(api_pod.Pod, 'parse_manifest')
def test_replace_with_manifest(self, parse_manifest, pod_update):
pod_update.return_value = self.pod
pod_update.return_value.manifest = '{"foo": "bar"}'
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/manifest',
'value': '{}',
'value': '{"foo": "bar"}',
'op': 'replace'}])
self.assertEqual(200, response.status_int)
self.assertEqual('application/json', response.content_type)
@ -262,78 +268,95 @@ class TestPatch(api_base.FunctionalTest):
def test_add_non_existent_property(self):
response = self.patch_json(
'/pods/%s' % self.pod.uuid,
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/foo', 'value': 'bar', 'op': 'add'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int)
self.assertTrue(response.json['error_message'])
def test_remove_ok(self):
@mock.patch.object(rpcapi.API, 'pod_update')
@mock.patch.object(rpcapi.API, 'pod_show')
def test_remove_ok(self, mock_pod_show, mock_pod_update):
self.pod.manifest = '{"key": "value"}'
mock_pod_show.return_value = self.pod
response = self.get_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid))
self.assertIsNotNone(response['desc'])
mock_pod_update.return_value = self.pod
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/desc', 'op': 'remove'}])
[{'path': '/manifest', 'op': 'remove'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
mock_pod_show.return_value = self.pod
mock_pod_show.return_value.desc = None
response = self.get_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid))
self.assertIsNone(response['desc'])
def test_remove_uuid(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/uuid', 'op': 'remove'}],
expect_errors=True)
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/uuid', 'op': 'remove'}],
expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_remove_bay_uuid(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/bay_uuid', 'op': 'remove'}],
expect_errors=True)
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/bay_uuid', 'op': 'remove'}],
expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_remove_internal_field(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/labels', 'op': 'remove'}],
expect_errors=True)
response = self.patch_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/labels', 'op': 'remove'}],
expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_remove_non_existent_property(self):
response = self.patch_json(
'/pods/%s' % self.pod.uuid,
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
[{'path': '/non-existent', 'op': 'remove'}],
expect_errors=True)
self.assertEqual(400, response.status_code)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
@mock.patch('oslo_utils.timeutils.utcnow')
def test_replace_ok_by_name(self, mock_utcnow):
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
@mock.patch.object(rpcapi.API, 'pod_show')
@mock.patch.object(rpcapi.API, 'pod_update')
@mock.patch.object(api_pod.Pod, 'parse_manifest')
def test_replace_ok_by_name(self, parse_manifest,
mock_pod_update,
mock_pod_show):
self.pod.manifest = '{"foo": "bar"}'
mock_pod_update.return_value = self.pod
response = self.patch_json(
'/pods/%s/%s' % (self.pod.name, self.pod.bay_uuid),
[{'path': '/desc', 'op': 'remove'}])
[{'path': '/manifest',
'value': '{"foo": "bar"}',
'op': 'replace'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
parse_manifest.assert_called_once_with()
self.assertTrue(mock_pod_update.is_called)
mock_pod_show.return_value = self.pod
response = self.get_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid))
self.assertEqual('pod1', response['name'])
return_updated_at = timeutils.parse_isotime(
response['updated_at']).replace(tzinfo=None)
self.assertEqual(test_time, return_updated_at)
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
@mock.patch('oslo_utils.timeutils.utcnow')
def test_replace_ok_by_name_not_found(self, mock_utcnow):
@ -341,11 +364,12 @@ class TestPatch(api_base.FunctionalTest):
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
response = self.patch_json('/pods/%s/%s' % (name, self.pod.bay_uuid),
[{'path': '/desc', 'op': 'remove'}],
expect_errors=True)
response = self.patch_json(
'/pods/%s/%s' % (name, self.pod.bay_uuid),
[{'path': '/desc', 'op': 'remove'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(404, response.status_code)
self.assertEqual(400, response.status_code)
@mock.patch('oslo_utils.timeutils.utcnow')
def test_replace_ok_by_name_multiple_pod(self, mock_utcnow):
@ -357,9 +381,10 @@ class TestPatch(api_base.FunctionalTest):
obj_utils.create_test_pod(self.context, name='test_pod',
uuid=utils.generate_uuid())
response = self.patch_json('/pods/test_pod',
[{'path': '/desc', 'op': 'remove'}],
expect_errors=True)
response = self.patch_json(
'/pods/test_pod/5d12f6fd-a196-4bf0-ae4c-1f639a523a52',
[{'path': '/desc', 'op': 'remove'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_code)
@ -369,25 +394,24 @@ class TestPost(api_base.FunctionalTest):
def setUp(self):
super(TestPost, self).setUp()
obj_utils.create_test_bay(self.context)
self.pod_obj = obj_utils.create_test_pod(self.context)
p = mock.patch.object(rpcapi.API, 'pod_create')
self.mock_pod_create = p.start()
self.mock_pod_create.side_effect = self._simulate_rpc_pod_create
self.mock_pod_create.return_value = self.pod_obj
self.addCleanup(p.stop)
p = mock.patch('magnum.objects.BayModel.get_by_uuid')
self.mock_baymodel_get_by_uuid = p.start()
self.mock_baymodel_get_by_uuid.return_value.coe = 'kubernetes'
self.addCleanup(p.stop)
def _simulate_rpc_pod_create(self, pod):
pod.create()
return pod
@mock.patch('oslo_utils.timeutils.utcnow')
def test_create_pod(self, mock_utcnow):
@mock.patch.object(rpcapi.API, 'rc_create')
def test_create_pod(self, mock_rc_create, mock_utcnow):
pdict = apiutils.pod_post_data()
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
mock_rc_create.return_value = self.pod_obj
response = self.post_json('/pods', pdict)
self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int)
@ -398,37 +422,33 @@ class TestPost(api_base.FunctionalTest):
urlparse.urlparse(response.location).path)
self.assertEqual(pdict['uuid'], response.json['uuid'])
self.assertNotIn('updated_at', response.json.keys)
return_created_at = timeutils.parse_isotime(
response.json['created_at']).replace(tzinfo=None)
self.assertEqual(test_time, return_created_at)
def test_create_pod_set_project_id_and_user_id(self):
pdict = apiutils.pod_post_data()
def _simulate_rpc_pod_create(pod):
self.assertEqual(self.context.project_id, pod.project_id)
self.assertEqual(self.context.user_id, pod.user_id)
pod.create()
self.assertEqual(pod.project_id, self.context.project_id)
self.assertEqual(pod.user_id, self.context.user_id)
return pod
self.mock_pod_create.side_effect = _simulate_rpc_pod_create
self.post_json('/pods', pdict)
def test_create_pod_doesnt_contain_id(self):
with mock.patch.object(self.dbapi, 'create_pod',
wraps=self.dbapi.create_pod) as cc_mock:
pdict = apiutils.pod_post_data(desc='pod_example_A_desc')
response = self.post_json('/pods', pdict)
self.assertEqual(pdict['desc'], response.json['desc'])
cc_mock.assert_called_once_with(mock.ANY)
# Check that 'id' is not in first arg of positional args
self.assertNotIn('id', cc_mock.call_args[0][0])
@mock.patch.object(rpcapi.API, 'pod_create')
def test_create_pod_doesnt_contain_id(self, mock_pod_create):
pdict = apiutils.pod_post_data(desc='pod_example_A_desc')
mock_pod_create.return_value = self.pod_obj
mock_pod_create.return_value.desc = 'pod_example_A_desc'
response = self.post_json('/pods', pdict)
self.assertEqual(pdict['desc'], response.json['desc'])
def test_create_pod_generate_uuid(self):
@mock.patch.object(rpcapi.API, 'pod_create')
def test_create_pod_generate_uuid(self, mock_pod_create):
pdict = apiutils.pod_post_data()
del pdict['uuid']
response = self.post_json('/pods', pdict)
mock_pod_create.return_value = self.pod_obj
response = self.post_json('/pods', pdict, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int)
self.assertEqual(pdict['desc'], response.json['desc'])
@ -479,59 +499,66 @@ class TestDelete(api_base.FunctionalTest):
super(TestDelete, self).setUp()
obj_utils.create_test_bay(self.context)
self.pod = obj_utils.create_test_pod(self.context)
p = mock.patch.object(rpcapi.API, 'pod_delete')
self.mock_pod_delete = p.start()
self.mock_pod_delete.side_effect = self._simulate_rpc_pod_delete
self.addCleanup(p.stop)
def _simulate_rpc_pod_delete(self, pod_uuid):
pod = objects.Pod.get_by_uuid(self.context, pod_uuid)
pod.destroy()
def test_delete_pod(self):
@mock.patch.object(rpcapi.API, 'pod_delete')
@mock.patch.object(rpcapi.API, 'pod_show')
def test_delete_pod(self, mock_pod_show, mock_pod_delete):
self.delete('/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid))
err = rest.ApiException(status=404)
mock_pod_show.side_effect = err
response = self.get_json(
'/pods/%s/%s' % (self.pod.uuid, self.pod.bay_uuid),
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual(500, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_delete_pod_by_name(self):
self.delete('/pods/%s/%s' % (self.pod.name, self.pod.bay_uuid))
response = self.get_json(
'/pods/%s/%s' % (self.pod.name, self.pod.bay_uuid),
expect_errors=True)
self.assertEqual(404, response.status_int)
@mock.patch.object(rpcapi.API, 'pod_delete')
@mock.patch.object(rpcapi.API, 'pod_show')
def test_delete_pod_by_name(self,
mock_pod_show,
mock_pod_delete):
self.delete('/pods/%s/%s' % (self.pod.name, self.pod.bay_uuid),
expect_errors=True)
mock_pod_show.return_value = self.pod
response = self.get_json('/pods/%s' % self.pod.name,
expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_delete_pod_by_name_not_found(self):
@mock.patch.object(rpcapi.API, 'pod_delete')
def test_delete_pod_by_name_not_found(self, mock_pod_delete):
err = rest.ApiException(status=404)
mock_pod_delete.side_effect = err
response = self.delete(
'/pods/not_found/5d12f6fd-a196-4bf0-ae4c-1f639a523a52',
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual(500, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_delete_multiple_pod_by_name(self):
@mock.patch.object(rpcapi.API, 'pod_delete')
def test_delete_multiple_pod_by_name(self, mock_pod_delete):
obj_utils.create_test_pod(self.context, name='test_pod',
uuid=utils.generate_uuid())
obj_utils.create_test_pod(self.context, name='test_pod',
uuid=utils.generate_uuid())
response = self.delete(
'/pods/test_pod/5d12f6fd-a196-4bf0-ae4c-1f639a523a52',
expect_errors=True)
self.assertEqual(409, response.status_int)
err = rest.ApiException(status=400)
mock_pod_delete.side_effect = err
response = self.delete('/pods/test_pod', expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_delete_pod_not_found(self):
@mock.patch.object(rpcapi.API, 'pod_delete')
def test_delete_pod_not_found(self, mock_pod_delete):
uuid = utils.generate_uuid()
response = self.delete(
'/pods/%s/%s' % (uuid, self.pod.bay_uuid),
expect_errors=True)
self.assertEqual(404, response.status_int)
err = rest.ApiException(status=404)
mock_pod_delete.side_effect = err
response = self.delete('/pods/%s/%s' % (uuid, self.pod.bay_uuid),
expect_errors=True)
self.assertEqual(500, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])

168
magnum/tests/unit/conductor/handlers/test_k8s_conductor.py
View File

@ -42,118 +42,136 @@ class TestK8sConductor(base.TestCase):
def mock_baymodel(self):
return objects.BayModel({})
def test_pod_create_with_success(self):
expected_pod = self.mock_pod()
expected_pod.create = mock.MagicMock()
@patch('ast.literal_eval')
def test_pod_create_with_success(self, mock_ast):
expected_pod = mock.MagicMock()
expected_pod.uuid = 'test-uuid'
expected_pod.name = 'test-name'
expected_pod.bay_uuid = 'test-bay-uuid'
manifest = {"key": "value"}
expected_pod.manifest = '{"key": "value"}'
mock_ast.return_value = {}
with patch('magnum.conductor.k8s_api.create_k8s_api') as mock_kube_api:
return_value = mock.MagicMock()
return_value.status = mock.MagicMock()
return_value.status.phase = 'Pending'
return_value.spec = mock.MagicMock()
return_value.spec.node_name = '10.0.0.3'
mock_kube_api.return_value.create_namespaced_pod.return_value = (
return_value)
with patch('magnum.conductor.k8s_api.create_k8s_api_pod') as \
mock_kube_api:
self.kube_handler.pod_create(self.context, expected_pod)
self.assertEqual('Pending', expected_pod.status)
self.assertEqual('10.0.0.3', expected_pod.host)
expected_pod.create.assert_called_once_with(self.context)
(mock_kube_api.return_value.create_namespaced_pod
.assert_called_once_with(body=manifest, namespace='default'))
def test_pod_create_with_fail(self):
expected_pod = self.mock_pod()