openstack/zun
Code Issues Proposed changes

Rename kw to kwargs

Browse Source 
Change-Id: I0530dd63f5960c9df1e791a419c5c43f19b689ec
Partial-Bug: #1702587
This commit is contained in:
Kien Nguyen 2017-07-07 11:51:57 +07:00
parent f869fd1115
commit c2ea4329fa
7 changed files with 208 additions and 205 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
zun/api/controllers/v1/containers.py
View File

@ -355,7 +355,7 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
def start(self, container_id, **kw):
def start(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:start")
utils.validate_container_state(container, 'start')
@ -369,7 +369,7 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@validation.validate_query_param(pecan.request, schema.query_param_stop)
def stop(self, container_id, timeout=None, **kw):
def stop(self, container_id, timeout=None, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:stop")
utils.validate_container_state(container, 'stop')
@ -383,7 +383,7 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@validation.validate_query_param(pecan.request, schema.query_param_reboot)
def reboot(self, container_id, timeout=None, **kw):
def reboot(self, container_id, timeout=None, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:reboot")
utils.validate_container_state(container, 'reboot')
@ -396,7 +396,7 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
def pause(self, container_id, **kw):
def pause(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:pause")
utils.validate_container_state(container, 'pause')
@ -409,7 +409,7 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
def unpause(self, container_id, **kw):
def unpause(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:unpause")
utils.validate_container_state(container, 'unpause')
@ -447,7 +447,7 @@ class ContainersController(base.Controller):
@exception.wrap_pecan_controller_exception
@validation.validate_query_param(pecan.request,
schema.query_param_execute_command)
def execute(self, container_id, run=True, interactive=False, **kw):
def execute(self, container_id, run=True, interactive=False, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:execute")
utils.validate_container_state(container, 'execute')
@ -458,17 +458,18 @@ class ContainersController(base.Controller):
msg = _('Valid run values are true, false, 0, 1, yes and no')
raise exception.InvalidValue(msg)
LOG.debug('Calling compute.container_exec with %s command %s'
% (container.uuid, kw['command']))
% (container.uuid, kwargs['command']))
context = pecan.request.context
compute_api = pecan.request.compute_api
return compute_api.container_exec(context, container, kw['command'],
return compute_api.container_exec(context, container,
kwargs['command'],
run, interactive)
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@validation.validate_query_param(pecan.request,
schema.query_param_execute_resize)
def execute_resize(self, container_id, exec_id, **kw):
def execute_resize(self, container_id, exec_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(),
"container:execute_resize")
@ -477,20 +478,22 @@ class ContainersController(base.Controller):
context = pecan.request.context
compute_api = pecan.request.compute_api
return compute_api.container_exec_resize(
context, container, exec_id, kw.get('h', None), kw.get('w', None))
context, container, exec_id, kwargs.get('h', None),
kwargs.get('w', None))
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@validation.validated(schema.query_param_signal)
def kill(self, container_id, **kw):
def kill(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:kill")
utils.validate_container_state(container, 'kill')
LOG.debug('Calling compute.container_kill with %s signal %s'
% (container.uuid, kw.get('signal', kw.get('signal'))))
% (container.uuid,
kwargs.get('signal', kwargs.get('signal'))))
context = pecan.request.context
compute_api = pecan.request.compute_api
compute_api.container_kill(context, container, kw.get('signal'))
compute_api.container_kill(context, container, kwargs.get('signal'))
pecan.response.status = 202
@pecan.expose('json')
@ -513,15 +516,15 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@validation.validate_query_param(pecan.request, schema.query_param_resize)
def resize(self, container_id, **kw):
def resize(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:resize")
utils.validate_container_state(container, 'resize')
LOG.debug('Calling tty resize with %s ' % (container.uuid))
context = pecan.request.context
compute_api = pecan.request.compute_api
compute_api.container_resize(context, container, kw.get('h', None),
kw.get('w', None))
compute_api.container_resize(context, container, kwargs.get('h', None),
kwargs.get('w', None))
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@ -538,30 +541,30 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
def get_archive(self, container_id, **kw):
def get_archive(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:get_archive")
utils.validate_container_state(container, 'get_archive')
LOG.debug('Calling compute.container_get_archive with %s path %s'
% (container.uuid, kw['path']))
% (container.uuid, kwargs['path']))
context = pecan.request.context
compute_api = pecan.request.compute_api
data, stat = compute_api.container_get_archive(
context, container, kw['path'])
context, container, kwargs['path'])
return {"data": data, "stat": stat}
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
def put_archive(self, container_id, **kw):
def put_archive(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:put_archive")
utils.validate_container_state(container, 'put_archive')
LOG.debug('Calling compute.container_put_archive with %s path %s'
% (container.uuid, kw['path']))
% (container.uuid, kwargs['path']))
context = pecan.request.context
compute_api = pecan.request.compute_api
compute_api.container_put_archive(context, container,
kw['path'], kw['data'])
kwargs['path'], kwargs['data'])
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@ -578,7 +581,7 @@ class ContainersController(base.Controller):
@pecan.expose('json')
@exception.wrap_pecan_controller_exception
@validation.validate_query_param(pecan.request, schema.query_param_commit)
def commit(self, container_id, **kw):
def commit(self, container_id, **kwargs):
container = _get_container(container_id)
check_policy_on_container(container.as_dict(), "container:commit")
utils.validate_container_state(container, 'commit')
@ -587,5 +590,5 @@ class ContainersController(base.Controller):
compute_api = pecan.request.compute_api
pecan.response.status = 202
return compute_api.container_commit(context, container,
kw.get('repository', None),
kw.get('tag', None))
kwargs.get('repository', None),
kwargs.get('tag', None))

4
zun/api/http_error.py
View File

@ -38,11 +38,11 @@ class HTTPNotAcceptableAPIVersion(exc.HTTPNotAcceptable):
min_version = ''
def __init__(self, detail=None, headers=None, comment=None,
body_template=None, max_version='', min_version='', **kw):
body_template=None, max_version='', min_version='', **kwargs):
super(HTTPNotAcceptableAPIVersion, self).__init__(
detail=detail, headers=headers, comment=comment,
body_template=body_template, **kw)
body_template=body_template, **kwargs)
self.max_version = max_version
self.min_version = min_version

14
zun/common/exception.py
View File

@ -60,16 +60,16 @@ def wrap_exception(notifier=None, event_type=None):
it to the notification system.
"""
def inner(f):
def wrapped(self, context, *args, **kw):
def wrapped(self, context, *args, **kwargs):
# Don't store self or context in the payload, it now seems to
# contain confidential information.
try:
return f(self, context, *args, **kw)
return f(self, context, *args, **kwargs)
except Exception as e:
with excutils.save_and_reraise_exception():
if notifier:
call_dict = inspect.getcallargs(f, self, context,
*args, **kw)
*args, **kwargs)
payload = dict(exception=e,
private=dict(args=call_dict)
)
@ -104,9 +104,9 @@ def wrap_controller_exception(func, func_server_error, func_client_error):
"""
@functools.wraps(func)
def wrapped(*args, **kw):
def wrapped(*args, **kwargs):
try:
return func(*args, **kw)
return func(*args, **kwargs)
except Exception as excp:
if isinstance(excp, ZunException):
http_error_code = excp.code
@ -155,9 +155,9 @@ def wrap_pecan_controller_exception(func):
def wrap_keystone_exception(func):
"""Wrap keystone exceptions and throw Zun specific exceptions."""
@functools.wraps(func)
def wrapped(*args, **kw):
def wrapped(*args, **kwargs):
try:
return func(*args, **kw)
return func(*args, **kwargs)
except keystone_exceptions.AuthorizationFailure:
raise AuthorizationFailure(
client=func.__name__, message="reason: %s" % sys.exc_info()[1])

6
zun/tests/base.py
View File

@ -103,10 +103,10 @@ class TestCase(base.BaseTestCase):
objects_base.ZunObjectRegistry._registry._obj_classes \
= self._base_test_obj_backup
def config(self, **kw):
def config(self, **kwargs):
"""Override config options for a test."""
group = kw.pop('group', None)
for k, v in kw.items():
group = kwargs.pop('group', None)
for k, v in kwargs.items():
CONF.set_override(k, v, group)
def get_path(self, project_file=None):

22
zun/tests/unit/api/utils.py
View File

@ -17,18 +17,18 @@ import datetime
import pytz
def zservice_get_data(**kw):
def zservice_get_data(**kwargs):
"""Simulate what the RPC layer will get from DB """
faketime = datetime.datetime(2001, 1, 1, tzinfo=pytz.UTC)
return {
'binary': kw.get('binary', 'fake-binary'),
'host': kw.get('host', 'fake-host'),
'id': kw.get('id', 13),
'report_count': kw.get('report_count', 13),
'disabled': kw.get('disabled', False),
'disabled_reason': kw.get('disabled_reason'),
'forced_down': kw.get('forced_down', False),
'last_seen_up': kw.get('last_seen_up', faketime),
'created_at': kw.get('created_at', faketime),
'updated_at': kw.get('updated_at', faketime),
'binary': kwargs.get('binary', 'fake-binary'),
'host': kwargs.get('host', 'fake-host'),
'id': kwargs.get('id', 13),
'report_count': kwargs.get('report_count', 13),
'disabled': kwargs.get('disabled', False),
'disabled_reason': kwargs.get('disabled_reason'),
'forced_down': kwargs.get('forced_down', False),
'last_seen_up': kwargs.get('last_seen_up', faketime),
'created_at': kwargs.get('created_at', faketime),
'updated_at': kwargs.get('updated_at', faketime),
}

306
zun/tests/unit/db/utils.py
View File

@ -21,30 +21,31 @@ from zun.db import api as db_api
CONF = cfg.CONF
def get_test_container(**kw):
def get_test_container(**kwargs):
return {
'id': kw.get('id', 42),
'uuid': kw.get('uuid', 'ea8e2a25-2901-438d-8157-de7ffd68d051'),
'container_id': kw.get('container_id', 'ddcb39a3fcec'),
'name': kw.get('name', 'container1'),
'project_id': kw.get('project_id', 'fake_project'),
'user_id': kw.get('user_id', 'fake_user'),
'image': kw.get('image', 'ubuntu'),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'command': kw.get('command', 'fake_command'),
'status': kw.get('status', 'Running'),
'status_reason': kw.get('status_reason', 'Created Successfully'),
'task_state': kw.get('task_state', 'container_creating'),
'environment': kw.get('environment', {'key1': 'val1', 'key2': 'val2'}),
'cpu': kw.get('cpu', 1.0),
'memory': kw.get('memory', '512m'),
'workdir': kw.get('workdir', '/home/ubuntu'),
'ports': kw.get('ports', [80, 443]),
'hostname': kw.get('hostname', 'testhost'),
'labels': kw.get('labels', {'key1': 'val1', 'key2': 'val2'}),
'meta': kw.get('meta', {'key1': 'val1', 'key2': 'val2'}),
'addresses': kw.get('addresses', {
'id': kwargs.get('id', 42),
'uuid': kwargs.get('uuid', 'ea8e2a25-2901-438d-8157-de7ffd68d051'),
'container_id': kwargs.get('container_id', 'ddcb39a3fcec'),
'name': kwargs.get('name', 'container1'),
'project_id': kwargs.get('project_id', 'fake_project'),
'user_id': kwargs.get('user_id', 'fake_user'),
'image': kwargs.get('image', 'ubuntu'),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
'command': kwargs.get('command', 'fake_command'),
'status': kwargs.get('status', 'Running'),
'status_reason': kwargs.get('status_reason', 'Created Successfully'),
'task_state': kwargs.get('task_state', 'container_creating'),
'environment': kwargs.get('environment', {'key1': 'val1',
'key2': 'val2'}),
'cpu': kwargs.get('cpu', 1.0),
'memory': kwargs.get('memory', '512m'),
'workdir': kwargs.get('workdir', '/home/ubuntu'),
'ports': kwargs.get('ports', [80, 443]),
'hostname': kwargs.get('hostname', 'testhost'),
'labels': kwargs.get('labels', {'key1': 'val1', 'key2': 'val2'}),
'meta': kwargs.get('meta', {'key1': 'val1', 'key2': 'val2'}),
'addresses': kwargs.get('addresses', {
'private': [
{
'OS-EXT-IPS-MAC:mac_addr': 'fa:16:3e:04:da:76',
@ -54,65 +55,65 @@ def get_test_container(**kw):
},
],
}),
'image_pull_policy': kw.get('image_pull_policy', 'always'),
'host': kw.get('host', 'localhost'),
'restart_policy': kw.get('restart_policy',
{'Name': 'no', 'MaximumRetryCount': '0'}),
'status_detail': kw.get('status_detail', 'up from 5 hours'),
'interactive': kw.get('interactive', True),
'image_pull_policy': kwargs.get('image_pull_policy', 'always'),
'host': kwargs.get('host', 'localhost'),
'restart_policy': kwargs.get('restart_policy',
{'Name': 'no', 'MaximumRetryCount': '0'}),
'status_detail': kwargs.get('status_detail', 'up from 5 hours'),
'interactive': kwargs.get('interactive', True),
'image_driver': 'glance',
'websocket_url': 'ws://127.0.0.1:6784/4c03164962fa/attach/'
'ws?logs=0&stream=1&stdin=1&stdout=1&stderr=1',
'ws?logs=0&stream=1&stdin=1&stdout=1&stderr=1',
'websocket_token': '7878038e-957c-4d52-ae19-1e9561784e7b',
'security_groups': kw.get('security_groups', ['default'])
'security_groups': kwargs.get('security_groups', ['default'])
}
def create_test_container(**kw):
def create_test_container(**kwargs):
"""Create test container entry in DB and return Container DB object.
Function to be used to create test Container objects in the database.
:param kw: kwargs with overriding values for container's attributes.
:param kwargs: kwargs with overriding values for container's attributes.
:returns: Test Container DB object.
"""
container = get_test_container(**kw)
container = get_test_container(**kwargs)
# Let DB generate ID if it isn't specified explicitly
if CONF.db_type == 'sql' and 'id' not in kw:
if CONF.db_type == 'sql' and 'id' not in kwargs:
del container['id']
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_container(kw['context'], container)
return dbapi.create_container(kwargs['context'], container)
def get_test_image(**kw):
def get_test_image(**kwargs):
return {
'id': kw.get('id', 42),
'uuid': kw.get('uuid', 'ea8e2a25-2901-438d-8157-de7ffd68d051'),
'repo': kw.get('repo', 'image1'),
'tag': kw.get('tag', 'latest'),
'image_id': kw.get('image_id', 'sha256:c54a2cc56cbb2f0400'),
'size': kw.get('size', '1848'),
'project_id': kw.get('project_id', 'fake_project'),
'user_id': kw.get('user_id', 'fake_user'),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'id': kwargs.get('id', 42),
'uuid': kwargs.get('uuid', 'ea8e2a25-2901-438d-8157-de7ffd68d051'),
'repo': kwargs.get('repo', 'image1'),
'tag': kwargs.get('tag', 'latest'),
'image_id': kwargs.get('image_id', 'sha256:c54a2cc56cbb2f0400'),
'size': kwargs.get('size', '1848'),
'project_id': kwargs.get('project_id', 'fake_project'),
'user_id': kwargs.get('user_id', 'fake_user'),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_image(**kw):
def create_test_image(**kwargs):
"""Create test image entry in DB and return Image DB object.
Function to be used to create test Image objects in the database.
:param kw: kwargs with overriding values for image's attributes.
:param kwargs: kwargs with overriding values for image's attributes.
:returns: Test Image DB object.
"""
image = get_test_image(**kw)
image = get_test_image(**kwargs)
# Let DB generate ID if it isn't specified explicitly
if 'id' not in kw:
if 'id' not in kwargs:
del image['id']
if 'repo' not in kw:
if 'repo' not in kwargs:
image['repo'] = _generate_repo_for_image()
dbapi = db_api._get_dbdriver_instance()
return dbapi.pull_image(kw['context'], image)
return dbapi.pull_image(kwargs['context'], image)
def _generate_repo_for_image():
@ -122,127 +123,127 @@ def _generate_repo_for_image():
return name + '-image'
def get_test_zun_service(**kw):
def get_test_zun_service(**kwargs):
return {
'id': kw.get('id', 23),
'uuid': kw.get('uuid', '2e8e2a25-2901-438d-8157-de7ffd68d066'),
'host': kw.get('host', 'fakehost'),
'binary': kw.get('binary', 'fake-bin'),
'disabled': kw.get('disabled', False),
'disabled_reason': kw.get('disabled_reason', 'fake-reason'),
'last_seen_up': kw.get('last_seen_up'),
'forced_down': kw.get('forced_down', False),
'report_count': kw.get('report_count', 13),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'id': kwargs.get('id', 23),
'uuid': kwargs.get('uuid', '2e8e2a25-2901-438d-8157-de7ffd68d066'),
'host': kwargs.get('host', 'fakehost'),
'binary': kwargs.get('binary', 'fake-bin'),
'disabled': kwargs.get('disabled', False),
'disabled_reason': kwargs.get('disabled_reason', 'fake-reason'),
'last_seen_up': kwargs.get('last_seen_up'),
'forced_down': kwargs.get('forced_down', False),
'report_count': kwargs.get('report_count', 13),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_zun_service(**kw):
zun_service = get_test_zun_service(**kw)
def create_test_zun_service(**kwargs):
zun_service = get_test_zun_service(**kwargs)
# Let DB generate ID if it isn't specifiled explicitly
if CONF.db_type == 'sql' and 'id' not in kw:
if CONF.db_type == 'sql' and 'id' not in kwargs:
del zun_service['id']
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_zun_service(zun_service)
def get_test_resource_provider(**kw):
def get_test_resource_provider(**kwargs):
return {
'id': kw.get('id', 42),
'uuid': kw.get('uuid', 'e166bf0e-66db-409d-aa4d-3af94efd8bcf'),
'name': kw.get('name', 'provider1'),
'root_provider': kw.get('root_provider',
'd3d4c98a-8c95-4d3c-8605-ea38ea036556'),
'parent_provider': kw.get('parent_provider',
'2c4de408-6c4f-4257-8274-f2d2192fe871'),
'can_host': kw.get('can_host', 0),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'id': kwargs.get('id', 42),
'uuid': kwargs.get('uuid', 'e166bf0e-66db-409d-aa4d-3af94efd8bcf'),
'name': kwargs.get('name', 'provider1'),
'root_provider': kwargs.get('root_provider',
'd3d4c98a-8c95-4d3c-8605-ea38ea036556'),
'parent_provider': kwargs.get('parent_provider',
'2c4de408-6c4f-4257-8274-f2d2192fe871'),
'can_host': kwargs.get('can_host', 0),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_resource_provider(**kw):
provider = get_test_resource_provider(**kw)
def create_test_resource_provider(**kwargs):
provider = get_test_resource_provider(**kwargs)
# Let DB generate ID if it isn't specified explicitly
if CONF.db_type == 'sql' and 'id' not in kw:
if CONF.db_type == 'sql' and 'id' not in kwargs:
del provider['id']
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_resource_provider(kw['context'], provider)
return dbapi.create_resource_provider(kwargs['context'], provider)
def get_test_resource_class(**kw):
def get_test_resource_class(**kwargs):
return {
'id': kw.get('id', 42),
'uuid': kw.get('uuid', '1136bf0e-66db-409d-aa4d-3af94eed8bcc'),
'name': kw.get('name', 'VCPU'),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'id': kwargs.get('id', 42),
'uuid': kwargs.get('uuid', '1136bf0e-66db-409d-aa4d-3af94eed8bcc'),
'name': kwargs.get('name', 'VCPU'),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_resource_class(**kw):
resource = get_test_resource_class(**kw)
def create_test_resource_class(**kwargs):
resource = get_test_resource_class(**kwargs)
# Let DB generate ID if it isn't specified explicitly
if CONF.db_type == 'sql' and 'id' not in kw:
if CONF.db_type == 'sql' and 'id' not in kwargs:
del resource['id']
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_resource_class(kw['context'], resource)
return dbapi.create_resource_class(kwargs['context'], resource)
def get_test_inventory(**kw):
def get_test_inventory(**kwargs):
return {
'id': kw.get('id', 42),
'resource_provider_id': kw.get('resource_provider_id', 1),
'resource_class_id': kw.get('resource_class_id', 2),
'total': kw.get('total', 4),
'reserved': kw.get('reserved', 1),
'min_unit': kw.get('min_unit', 0),
'max_unit': kw.get('max_unit', 4),
'step_size': kw.get('step_size', 1),
'allocation_ratio': kw.get('allocation_ratio', 1.0),
'is_nested': kw.get('is_nested', True),
'blob': kw.get('blob', [1, 2, 5]),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'id': kwargs.get('id', 42),
'resource_provider_id': kwargs.get('resource_provider_id', 1),
'resource_class_id': kwargs.get('resource_class_id', 2),
'total': kwargs.get('total', 4),
'reserved': kwargs.get('reserved', 1),
'min_unit': kwargs.get('min_unit', 0),
'max_unit': kwargs.get('max_unit', 4),
'step_size': kwargs.get('step_size', 1),
'allocation_ratio': kwargs.get('allocation_ratio', 1.0),
'is_nested': kwargs.get('is_nested', True),
'blob': kwargs.get('blob', [1, 2, 5]),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_inventory(**kw):
inventory = get_test_inventory(**kw)
def create_test_inventory(**kwargs):
inventory = get_test_inventory(**kwargs)
# Let DB generate ID if it isn't specified explicitly
if CONF.db_type == 'sql' and 'id' not in kw:
if CONF.db_type == 'sql' and 'id' not in kwargs:
del inventory['id']
provider_id = inventory.pop('resource_provider_id')
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_inventory(kw['context'], provider_id, inventory)
return dbapi.create_inventory(kwargs['context'], provider_id, inventory)
def get_test_allocation(**kw):
def get_test_allocation(**kwargs):
return {
'id': kw.get('id', 42),
'resource_provider_id': kw.get('resource_provider_id', 1),
'resource_class_id': kw.get('resource_class_id', 2),
'consumer_id': kw.get('consumer_id',
'cb775e13-72e2-4bfa-9646-020725e1325e'),
'used': kw.get('used', 1),
'is_nested': kw.get('is_nested', 0),
'blob': kw.get('blob', [1, 2, 5]),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'id': kwargs.get('id', 42),
'resource_provider_id': kwargs.get('resource_provider_id', 1),
'resource_class_id': kwargs.get('resource_class_id', 2),
'consumer_id': kwargs.get('consumer_id',
'cb775e13-72e2-4bfa-9646-020725e1325e'),
'used': kwargs.get('used', 1),
'is_nested': kwargs.get('is_nested', 0),
'blob': kwargs.get('blob', [1, 2, 5]),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_allocation(**kw):
allocation = get_test_allocation(**kw)
def create_test_allocation(**kwargs):
allocation = get_test_allocation(**kwargs)
# Let DB generate ID if it isn't specified explicitly
if CONF.db_type == 'sql' and 'id' not in kw:
if CONF.db_type == 'sql' and 'id' not in kwargs:
del allocation['id']
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_allocation(kw['context'], allocation)
return dbapi.create_allocation(kwargs['context'], allocation)
def get_test_numa_topology(**kw):
def get_test_numa_topology(**kwargs):
return {
"nodes": [
{
@ -259,39 +260,39 @@ def get_test_numa_topology(**kw):
}
def get_test_compute_node(**kw):
def get_test_compute_node(**kwargs):
return {
'uuid': kw.get('uuid', '24a5b17a-f2eb-4556-89db-5f4169d13982'),
'hostname': kw.get('hostname', 'localhost'),
'numa_topology': kw.get('numa_topology', get_test_numa_topology()),
'mem_total': kw.get('mem_total', 1024),
'mem_free': kw.get('mem_free', 512),
'mem_available': kw.get('mem_available', 512),
'mem_used': kw.get('mem_used', 512),
'total_containers': kw.get('total_containers', 10),
'running_containers': kw.get('running_containers', 8),
'paused_containers': kw.get('paused_containers', 0),
'stopped_containers': kw.get('stopped_containers', 2),
'cpus': kw.get('cpus', 48),
'cpu_used': kw.get('cpu_used', 6.5),
'architecture': kw.get('architecture', 'x86_64'),
'os_type': kw.get('os_type', 'linux'),
'os': kw.get('os', 'Centos'),
'kernel_version': kw.get('kernel_version', '3.10.0-123.el7.x86_64'),
'labels': kw.get('labels', {"dev.type": "product"}),
'created_at': kw.get('created_at'),
'updated_at': kw.get('updated_at'),
'uuid': kwargs.get('uuid', '24a5b17a-f2eb-4556-89db-5f4169d13982'),
'hostname': kwargs.get('hostname', 'localhost'),
'numa_topology': kwargs.get('numa_topology', get_test_numa_topology()),
'mem_total': kwargs.get('mem_total', 1024),
'mem_free': kwargs.get('mem_free', 512),
'mem_available': kwargs.get('mem_available', 512),
'mem_used': kwargs.get('mem_used', 512),
'total_containers': kwargs.get('total_containers', 10),
'running_containers': kwargs.get('running_containers', 8),
'paused_containers': kwargs.get('paused_containers', 0),
'stopped_containers': kwargs.get('stopped_containers', 2),
'cpus': kwargs.get('cpus', 48),
'cpu_used': kwargs.get('cpu_used', 6.5),
'architecture': kwargs.get('architecture', 'x86_64'),
'os_type': kwargs.get('os_type', 'linux'),
'os': kwargs.get('os', 'Centos'),
'kernel_version': kwargs.get('kernel_version',
'3.10.0-123.el7.x86_64'),
'labels': kwargs.get('labels', {"dev.type": "product"}),
'created_at': kwargs.get('created_at'),
'updated_at': kwargs.get('updated_at'),
}
def create_test_compute_node(**kw):
compute_host = get_test_compute_node(**kw)
def create_test_compute_node(**kwargs):
compute_host = get_test_compute_node(**kwargs)
dbapi = db_api._get_dbdriver_instance()
return dbapi.create_compute_node(kw['context'], compute_host)
return dbapi.create_compute_node(kwargs['context'], compute_host)
class FakeEtcdMultipleResult(object):
def __init__(self, value):
self.children = []
for v in value:
@ -301,6 +302,5 @@ class FakeEtcdMultipleResult(object):
class FakeEtcdResult(object):
def __init__(self, value):
self.value = json.dump_as_bytes(value)

8
zun/tests/unit/objects/utils.py
View File

@ -16,24 +16,24 @@ from zun import objects
from zun.tests.unit.db import utils as db_utils
def create_test_container(context, **kw):
def create_test_container(context, **kwargs):
"""Create and return a test container object.
Create a container in the DB and return a container object with
appropriate attributes.
"""
container = get_test_container(context, **kw)
container = get_test_container(context, **kwargs)
container.create(context)
return container
def get_test_container(context, **kw):
def get_test_container(context, **kwargs):
"""Return a test container object with appropriate attributes.
NOTE: The object leaves the attributes marked as changed, such
that a create() could be used to commit it to the DB.
"""
db_container = db_utils.get_test_container(**kw)
db_container = db_utils.get_test_container(**kwargs)
container = objects.Container(context)
for key in db_container:
setattr(container, key, db_container[key])