starlingx
/
config
Code Issues Proposed changes

[PTP dual NIC config] PTP parameter: CLI, REST API 

New shell commands for management of PTP parameters that will replace
those currently stored using service parameters:
- "system ptp-parameter-add <name> <value> <owner-type> <owner-uuid>"
- "system ptp-parameter-show <uuid>"
- "system ptp-parameter-list [--type <owner-type>
                              --foreign_uuid <owner-uuid>]"
- "system ptp-parameter-modify <uuid> <value>"
- "system ptp-parameter-delete <uuid>"
Also added:
- REST API to manage the PTP parameter table contents;
- Unit tests for PTP parameter operations (CRUD).

Test Plan:

PASS: New unit tests for PTP parameter API.
PASS: Integration test with latest changes in PTP interfaces

Regression:

PASS: Existing unit tests still running with former PTP implementation.

Story: 2009248
Task: 43623
Signed-off-by: Douglas Henrique Koerich <douglashenrique.koerich@windriver.com>
Change-Id: I04d1dbfd859838a350fe2f4f38ae6d596e1b4142
This commit is contained in:
Douglas Henrique Koerich 2021-11-04 18:23:03 -03:00
parent 73be929787
commit d532a61fb6
23 changed files with 1212 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
sysinv/cgts-client/cgts-client/cgtsclient/v1/client.py
View File

@ -71,6 +71,7 @@ from cgtsclient.v1 import port
from cgtsclient.v1 import ptp
from cgtsclient.v1 import ptp_instance
from cgtsclient.v1 import ptp_interface
from cgtsclient.v1 import ptp_parameter
from cgtsclient.v1 import registry_image
from cgtsclient.v1 import remotelogging
from cgtsclient.v1 import restore
@ -122,6 +123,7 @@ class Client(http.HTTPClient):
self.ptp = ptp.ptpManager(self)
self.ptp_instance = ptp_instance.PtpInstanceManager(self)
self.ptp_interface = ptp_interface.PtpInterfaceManager(self)
self.ptp_parameter = ptp_parameter.PtpParameterManager(self)
self.iextoam = iextoam.iextoamManager(self)
self.controller_fs = controller_fs.ControllerFsManager(self)
self.storage_backend = storage_backend.StorageBackendManager(self)

8
sysinv/cgts-client/cgts-client/cgtsclient/v1/ptp_instance_shell.py
View File

@ -25,10 +25,6 @@ def _print_ptp_instance_show(ptp_instance_obj):
def do_ptp_instance_list(cc, args):
"""List all PTP instances, in any host."""
ptp_instances = cc.ptp_instance.list()
for instance in ptp_instances[:]:
ihost = ihost_utils._find_ihost(cc, instance.host_uuid)
setattr(instance, 'hostname', ihost.hostname)
field_labels = ['uuid', 'name', 'service', 'hostname']
fields = ['uuid', 'name', 'service', 'hostname']
utils.print_list(ptp_instances, fields, field_labels)
@ -53,8 +49,6 @@ def do_host_ptp_instance_list(cc, args):
def do_ptp_instance_show(cc, args):
"""Show PTP instance attributes."""
ptp_instance = ptp_instance_utils._find_ptp_instance(cc, args.nameoruuid)
ihost = ihost_utils._find_ihost(cc, ptp_instance.host_uuid)
setattr(ptp_instance, 'hostname', ihost.hostname)
_print_ptp_instance_show(ptp_instance)
@ -87,8 +81,6 @@ def do_ptp_instance_add(cc, args):
except exc.HTTPNotFound:
raise exc.CommandError('PTP instance just created not found: %s' %
uuid)
if ptp_instance:
setattr(ptp_instance, 'hostname', ihost.hostname)
_print_ptp_instance_show(ptp_instance)

69
sysinv/cgts-client/cgts-client/cgtsclient/v1/ptp_parameter.py Normal file
View File

@ -0,0 +1,69 @@
########################################################################
#
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
########################################################################
from cgtsclient.common import base
from cgtsclient import exc
from cgtsclient.v1 import options
CREATION_ATTRIBUTES = ['name', 'value', 'type', 'foreign_uuid']
class PtpParameter(base.Resource):
def __repr__(self):
return "<PtpParameter %s>" % self._info
class PtpParameterManager(base.Manager):
resource_class = PtpParameter
@staticmethod
def _path(ptp_parameter_id=None):
return '/v1/ptp_parameters/%s' % ptp_parameter_id if ptp_parameter_id \
else '/v1/ptp_parameters'
def list(self, q=None):
return self._list(options.build_url(self._path(), q), "ptp_parameters")
def list_by_ptp_instance(self, ptp_instance_uuid):
path = '/v1/ptp_instances/%s/ptp_parameters' % ptp_instance_uuid
return self._list(path, "ptp_parameters")
def list_by_interface(self, interface_uuid):
path = '/v1/iinterfaces/%s/ptp_parameters' % interface_uuid
return self._list(path, "ptp_parameters")
def get(self, ptp_parameter_id):
try:
return self._list(self._path(ptp_parameter_id))[0]
except IndexError:
return None
def create(self, **kwargs):
data = {}
for (key, value) in kwargs.items():
if key in CREATION_ATTRIBUTES:
data[key] = value
else:
raise exc.InvalidAttribute('%s' % key)
return self._create(self._path(), data)
def update(self, ptp_parameter_id, patch):
return self._update(self._path(ptp_parameter_id), patch)
def delete(self, ptp_parameter_id):
return self._delete(self._path(ptp_parameter_id))
def _find_ptp_parameter(cc, id):
try:
parameter = cc.ptp_parameter.get(id)
except exc.HTTPNotFound:
raise exc.CommandError('PTP parameter not found: %s' % id)
else:
return parameter

151
sysinv/cgts-client/cgts-client/cgtsclient/v1/ptp_parameter_shell.py Normal file
View File

@ -0,0 +1,151 @@
########################################################################
#
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
########################################################################
from cgtsclient.common import utils
from cgtsclient import exc
from cgtsclient.v1 import ptp_parameter as ptp_parameter_utils
def _owner_formatter(values):
result = []
result.append(str(values['name'] +
" of type " + values['type'] +
" at " + values['hostname']))
return result
def _print_ptp_parameter_show(ptp_parameter_obj):
fields = ['uuid', 'name', 'value', 'type',
'owner', 'foreign_uuid', 'created_at', 'updated_at']
labels = ['uuid', 'name', 'value', 'type',
'owned_by', 'owner_id', 'created_at', 'updated_at']
data = [(f, getattr(ptp_parameter_obj, f, '')) for f in fields]
utils.print_tuple_list(data, labels,
formatters={'owner': _owner_formatter})
@utils.arg('-t', '--type',
metavar='<owner type>',
choices=['ptp-instance', 'ptp-interface'],
help='List PTP parameters for a specific owner type')
@utils.arg('-u', '--foreign_uuid',
metavar='<owner uuid>',
help='List PTP parameters associated to specified owner')
def do_ptp_parameter_list(cc, args):
"""List all PTP parameters, in any host."""
missing = ((args.type is None) and (args.foreign_uuid is not None)) or \
((args.type is not None) and (args.foreign_uuid is None))
if missing:
raise exc.CommandError("Both 'type' and 'foreign_uuid' "
"must be provided")
ptp_parameters = None
if args.type == 'ptp-instance':
ptp_parameters = cc.ptp_parameter.list_by_ptp_instance(
args.foreign_uuid)
elif args.type == 'ptp-interface':
ptp_parameters = cc.ptp_parameter.list_by_interface(
args.foreign_uuid)
if ptp_parameters:
fields = ['uuid', 'name', 'value']
labels = ['uuid', 'name', 'value']
else:
ptp_parameters = cc.ptp_parameter.list()
for ptp_parameter in ptp_parameters:
owner_dict = getattr(ptp_parameter, 'owner', '')
setattr(ptp_parameter, 'owner_name', owner_dict['name'])
setattr(ptp_parameter, 'owner_host', owner_dict['hostname'])
fields = ['uuid',
'name',
'value',
'type',
'owner_name',
'owner_host',
'foreign_uuid']
labels = ['uuid',
'name',
'value',
'owner_type',
'owner_name',
'owner_host',
'owner_uuid']
utils.print_list(ptp_parameters, fields, labels)
@utils.arg('uuid',
metavar='<uuid>',
help="UUID of PTP parameter")
def do_ptp_parameter_show(cc, args):
"""Show PTP parameter attributes."""
ptp_parameter = ptp_parameter_utils._find_ptp_parameter(cc, args.uuid)
_print_ptp_parameter_show(ptp_parameter)
@utils.arg('name',
metavar='<name>',
help="Name of PTP parameter [REQUIRED]")
@utils.arg('value',
metavar='<value>',
help="Value of PTP parameter [REQUIRED]")
@utils.arg('type',
metavar='<owner type>',
choices=['ptp-instance', 'ptp-interface'],
help="Type of parameter owner ('ptp-instance' or 'ptp-interface') "
"[REQUIRED]")
@utils.arg('foreign_uuid',
metavar='<owner uuid>',
help="UUID of parameter owner [REQUIRED]")
def do_ptp_parameter_add(cc, args):
"""Add a PTP parameter."""
field_list = ['name', 'value', 'type', 'foreign_uuid']
# Prune input fields down to required/expected values
data = dict((k, v) for (k, v) in vars(args).items()
if k in field_list and not (v is None))
ptp_parameter = cc.ptp_parameter.create(**data)
uuid = getattr(ptp_parameter, 'uuid', '')
try:
ptp_parameter = cc.ptp_parameter.get(uuid)
except exc.HTTPNotFound:
raise exc.CommandError('PTP parameter just created not found: %s' %
uuid)
_print_ptp_parameter_show(ptp_parameter)
@utils.arg('uuid',
metavar='<uuid>',
help="UUID of PTP parameter")
@utils.arg('value',
metavar='<new value>',
help="New value of parameter")
def do_ptp_parameter_modify(cc, args):
"""Change PTP parameter value."""
field_list = ['value']
data = dict((k, v) for (k, v) in vars(args).items()
if k in field_list and not (v is None))
patch = []
for (k, v) in data.items():
patch.append({'op': 'replace', 'path': '/' + k, 'value': v})
ptp_parameter = cc.ptp_parameter.update(args.uuid, patch)
_print_ptp_parameter_show(ptp_parameter)
@utils.arg('uuid',
metavar='<uuid>',
help="UUID of PTP parameter")
def do_ptp_parameter_delete(cc, args):
"""Delete a PTP parameter."""
cc.ptp_parameter.delete(args.uuid)
print('Deleted PTP parameter: %s' % args.uuid)

2
sysinv/cgts-client/cgts-client/cgtsclient/v1/shell.py
View File

@ -56,6 +56,7 @@ from cgtsclient.v1 import pci_device_shell
from cgtsclient.v1 import port_shell
from cgtsclient.v1 import ptp_instance_shell
from cgtsclient.v1 import ptp_interface_shell
from cgtsclient.v1 import ptp_parameter_shell
from cgtsclient.v1 import ptp_shell
from cgtsclient.v1 import registry_image_shell
from cgtsclient.v1 import remotelogging_shell
@ -79,6 +80,7 @@ COMMAND_MODULES = [
ptp_shell,
ptp_instance_shell,
ptp_interface_shell,
ptp_parameter_shell,
iextoam_shell,
controller_fs_shell,
storage_backend_shell,

12
sysinv/sysinv/sysinv/sysinv/api/controllers/v1/__init__.py
View File

@ -67,6 +67,7 @@ from sysinv.api.controllers.v1 import port
from sysinv.api.controllers.v1 import ptp
from sysinv.api.controllers.v1 import ptp_instance
from sysinv.api.controllers.v1 import ptp_interface
from sysinv.api.controllers.v1 import ptp_parameter
from sysinv.api.controllers.v1 import pv
from sysinv.api.controllers.v1 import registry_image
from sysinv.api.controllers.v1 import remotelogging
@ -154,6 +155,9 @@ class V1(base.APIBase):
ptp_interfaces = [link.Link]
"Links to the ptp_interfaces resource"
ptp_parameters = [link.Link]
"Links to the ptp_parameters resource"
iextoam = [link.Link]
"Links to the iextoam resource"
@ -469,6 +473,13 @@ class V1(base.APIBase):
bookmark=True)
]
v1.ptp_parameters = [link.Link.make_link('self', pecan.request.host_url,
'ptp_parameters', ''),
link.Link.make_link('bookmark',
pecan.request.host_url,
'ptp_parameters', '',
bookmark=True)]
v1.iextoam = [link.Link.make_link('self', pecan.request.host_url,
'iextoam', ''),
link.Link.make_link('bookmark',
@ -913,6 +924,7 @@ class Controller(rest.RestController):
ptp = ptp.PTPController()
ptp_instances = ptp_instance.PtpInstanceController()
ptp_interfaces = ptp_interface.PtpInterfaceController()
ptp_parameters = ptp_parameter.PtpParameterController()
iextoam = network_oam.OAMNetworkController()
controller_fs = controller_fs.ControllerFsController()
storage_backend = storage_backend.StorageBackendController()

4
sysinv/sysinv/sysinv/sysinv/api/controllers/v1/interface.py
View File

@ -46,6 +46,7 @@ from sysinv.api.controllers.v1 import types
from sysinv.api.controllers.v1 import utils
from sysinv.api.controllers.v1 import interface_network
from sysinv.api.controllers.v1 import interface_datanetwork
from sysinv.api.controllers.v1 import ptp_parameter
from sysinv.common import constants
from sysinv.common import exception
from sysinv.common import utils as cutils
@ -302,6 +303,9 @@ class InterfaceController(rest.RestController):
routes = route.RouteController(parent="iinterfaces")
"Expose routes as a sub-element of interface"
ptp_parameters = ptp_parameter.PtpParameterController(parent="iinterface")
"Expose PTP parameters as a sub-element of interface"
interface_networks = interface_network.InterfaceNetworkController(
parent="iinterfaces")
"Expose interface_networks as a sub-element of interface"

52
sysinv/sysinv/sysinv/sysinv/api/controllers/v1/ptp_instance.py
View File

@ -15,8 +15,10 @@ from oslo_log import log
from sysinv._i18n import _
from sysinv.api.controllers.v1 import base
from sysinv.api.controllers.v1 import collection
from sysinv.api.controllers.v1 import ptp_parameter
from sysinv.api.controllers.v1 import types
from sysinv.api.controllers.v1 import utils
from sysinv.common import constants
from sysinv.common import exception
from sysinv.common import utils as cutils
from sysinv import objects
@ -53,10 +55,16 @@ class PtpInstance(base.APIBase):
host_uuid = types.uuid
"UUID of the host the PTP instance is associated to"
hostname = wtypes.text
"Name of the host the PTP instance is associated to"
name = wtypes.text
"Name given to the PTP instance"
service = wtypes.Enum(str, 'ptp4l', 'phc2sys', 'ts2phc')
service = wtypes.Enum(str,
constants.PTP_INSTANCE_TYPE_PTP4L,
constants.PTP_INSTANCE_TYPE_PHC2SYS,
constants.PTP_INSTANCE_TYPE_TS2PHC)
"Type of service of the PTP instance"
capabilities = {wtypes.text: utils.ValidTypes(wtypes.text,
@ -76,6 +84,7 @@ class PtpInstance(base.APIBase):
if not expand:
ptp_instance.unset_fields_except(['uuid',
'host_uuid',
'hostname',
'name',
'service',
'capabilities',
@ -114,11 +123,15 @@ LOCK_NAME = 'PtpInstanceController'
class PtpInstanceController(rest.RestController):
"""REST controller for PTP instance."""
ptp_parameters = ptp_parameter.PtpParameterController(
parent="ptp_instance")
"Expose PTP parameters as a sub-element of PTP instances"
def __init__(self, from_ihosts=False):
self._from_ihosts = from_ihosts
def _get_ptp_instance_collection(
self, host_uuid, marker=None, limit=None, sort_key=None,
self, host_uuid=None, marker=None, limit=None, sort_key=None,
sort_dir=None, expand=False, resource_url=None):
LOG.debug("PtpInstanceController._get_ptp_instance_collection: "
"from_ihosts %s host_uuid %s" % (self._from_ihosts,
@ -186,6 +199,13 @@ class PtpInstanceController(rest.RestController):
ptp_instance_dict = ptp_instance.as_dict()
LOG.debug("PtpInstanceController.post: %s" % ptp_instance_dict)
# Get rid of hostname (if any) to create the PTP instance
try:
ptp_instance_dict.pop('hostname')
except KeyError:
LOG.debug("PtpInstanceController.post: no hostname in %s" %
ptp_instance_dict)
# Replace host UUID by host ID
host_uuid = ptp_instance_dict.pop('host_uuid')
try:
@ -195,8 +215,8 @@ class PtpInstanceController(rest.RestController):
raise wsme.exc.ClientSideError(msg)
ptp_instance_dict['host_id'] = ihost_obj['id']
result = pecan.request.dbapi.ptp_instance_create(ptp_instance_dict)
return PtpInstance.convert_with_links(result)
return PtpInstance.convert_with_links(
pecan.request.dbapi.ptp_instance_create(ptp_instance_dict))
@cutils.synchronized(LOCK_NAME)
@wsme_pecan.wsexpose(None, types.uuid, status_code=204)
@ -208,14 +228,24 @@ class PtpInstanceController(rest.RestController):
# Only allow delete if there are no associated interfaces and
# parameters
parameters = pecan.request.dbapi.ptp_parameters_get_by_foreign_uuid(
parameters = pecan.request.dbapi.ptp_parameters_get_by_owner(
ptp_instance_uuid)
interfaces = pecan.request.dbapi.ptp_interfaces_get_by_instance(
ptp_instance_uuid)
if parameters or interfaces:
if parameters:
names = [str(p['name']) for p in parameters]
raise wsme.exc.ClientSideError(
_("PTP instance %s has still parameters or associated "
"interfaces. Check both ptp-interfaces and ptp-parameters.")
% ptp_instance_uuid)
"PTP instance %s is still associated with PTP parameter(s): %s"
% (ptp_instance_uuid, names))
ptp_instance_obj = objects.ptp_instance.get_by_uuid(
pecan.request.context, ptp_instance_uuid)
interfaces = pecan.request.dbapi.ptp_interfaces_get_by_instance(
ptp_instance_obj.id)
if interfaces:
names = [str(i['ifname']) for i in interfaces]
raise wsme.exc.ClientSideError(
"PTP instance %s is still associated with PTP interface(s): %s"
% (ptp_instance_uuid, names))
LOG.debug("PtpInstanceController.delete: all clear for %s" %
ptp_instance_uuid)
pecan.request.dbapi.ptp_instance_destroy(ptp_instance_uuid)

17
sysinv/sysinv/sysinv/sysinv/api/controllers/v1/ptp_interface.py
View File

@ -9,6 +9,7 @@
import pecan
from pecan import rest
import six
import wsme
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
@ -206,8 +207,20 @@ class PtpInterfaceController(rest.RestController):
def delete(self, ptp_interface_uuid):
"""Delete a PTP interface."""
try:
ptp_interface = objects.ptp_interface.get_by_uuid(pecan.request.context,
ptp_interface_uuid)
ptp_interface = objects.ptp_interface.get_by_uuid(
pecan.request.context, ptp_interface_uuid)
except exception.PtpInterfaceNotFound:
raise
# Only allow delete if there are no associated parameters
parameters = pecan.request.dbapi.ptp_parameters_get_by_owner(
ptp_interface_uuid)
if parameters:
names = [str(p['name']) for p in parameters]
raise wsme.exc.ClientSideError(
"PTP interface %s has PTP parameter(s): %s"
% (ptp_interface_uuid, names))
LOG.debug("PtpInterfaceController.delete: all clear for %s" %
ptp_interface_uuid)
pecan.request.dbapi.ptp_interface_destroy(ptp_interface.uuid)

262
sysinv/sysinv/sysinv/sysinv/api/controllers/v1/ptp_parameter.py Normal file
View File

@ -0,0 +1,262 @@
#
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
import jsonpatch
import pecan
from pecan import rest
import wsme
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
from oslo_log import log
from sysinv._i18n import _
from sysinv.api.controllers.v1 import base
from sysinv.api.controllers.v1 import collection
from sysinv.api.controllers.v1 import types
from sysinv.api.controllers.v1 import utils
from sysinv.common import constants
from sysinv.common import exception
from sysinv.common import utils as cutils
from sysinv import objects
LOG = log.getLogger(__name__)
class PtpParameterPatchType(types.JsonPatchType):
@staticmethod
def mandatory_attrs():
return []
class PtpParameter(base.APIBase):
"""API representation of a PTP parameter.
This class enforces type checking and value constraints, and converts
between the internal object model and the API representation of
a PTP parameter.
"""
created_at = wtypes.datetime.datetime
"Timestamp of creation of this PTP parameter"
updated_at = wtypes.datetime.datetime
"Timestamp of update of this PTP parameter"
id = int
"Unique ID for this PTP parameter"
uuid = types.uuid
"Unique UUID for this PTP parameter"
name = wtypes.text
"Name of PTP parameter"
value = wtypes.text
"Value of PTP parameter"
type = wtypes.Enum(str,
constants.PTP_PARAMETER_OWNER_INSTANCE,
constants.PTP_PARAMETER_OWNER_INTERFACE)
"Type of owner of this PTP parameter"
foreign_uuid = types.uuid
"UUID of the owner of this PTP parameter"
owner = types.MultiType([dict])
"Owner information: name, type, hostname"
def __init__(self, **kwargs):
self.fields = list(objects.ptp_parameter.fields.keys())
for k in self.fields:
if not hasattr(self, k):
continue
setattr(self, k, kwargs.get(k))
@classmethod
def convert_with_links(cls, rpc_ptp_parameter, expand=True):
ptp_parameter = PtpParameter(**rpc_ptp_parameter.as_dict())
if not expand:
ptp_parameter.unset_fields_except(['uuid',
'name',
'value',
'type',
'foreign_uuid',
'owner',
'created_at',
'updated_at'])
LOG.debug("PtpParameter.convert_with_links: converted %s" %
ptp_parameter.as_dict())
return ptp_parameter
class PtpParameterCollection(collection.Collection):
"""API representation of a collection of PTP parameters."""
ptp_parameters = [PtpParameter]
"A list containing PTP parameter objects"
def __init__(self, **kwargs):
self._type = 'ptp_parameters'
@classmethod
def convert_with_links(cls, rpc_ptp_parameters, limit, url=None,
expand=False, **kwargs):
collection = PtpParameterCollection()
collection.ptp_parameters = [PtpParameter.convert_with_links(p, expand)
for p in rpc_ptp_parameters]
collection.next = collection.get_next(limit, url=url, **kwargs)
return collection
LOCK_NAME = 'PtpParameterController'
class PtpParameterController(rest.RestController):
"""REST controller for PTP parameter."""
def __init__(self, parent=None):
self._parent = parent
def _get_ptp_parameter_collection(
self, parent_uuid=None, type=None, marker=None, limit=None,
sort_key=None, sort_dir=None, expand=False, resource_url=None):
LOG.debug("PtpParameterController._get_ptp_parameter_collection: "
"parent %s uuid %s type %s" %
(self._parent, parent_uuid, type))
if self._parent and not parent_uuid:
raise exception.InvalidParameterValue(_(
"Parent id not specified."))
limit = utils.validate_limit(limit)
sort_dir = utils.validate_sort_dir(sort_dir)
LOG.debug("PtpParameterController._get_ptp_parameter_collection: "
"marker %s, limit %s, sort_dir %s" % (marker, limit,
sort_dir))
marker_obj = None
if marker:
marker_obj = objects.ptp_parameter.get_by_uuid(
pecan.request.context, marker)
if parent_uuid:
ptp_parameters = pecan.request.dbapi.ptp_parameters_get_by_owner(
parent_uuid, limit, marker_obj, sort_key=sort_key,
sort_dir=sort_dir)
elif type is not None:
ptp_parameters = pecan.request.dbapi.ptp_parameters_get_by_type(
type, limit, marker_obj, sort_key=sort_key, sort_dir=sort_dir)
else:
ptp_parameters = pecan.request.dbapi.ptp_parameters_get_list(
limit, marker_obj, sort_key=sort_key, sort_dir=sort_dir)
return PtpParameterCollection.convert_with_links(
ptp_parameters, limit, url=resource_url, expand=expand,
sort_key=sort_key, sort_dir=sort_dir)
@wsme_pecan.wsexpose(PtpParameterCollection, types.uuid, types.uuid,
int, wtypes.text, wtypes.text)
def get_all(self, uuid=None, marker=None, limit=None,
sort_key='id', sort_dir='asc'):
"""Retrieve a list of PTP parameters."""
type = None
LOG.debug("PtpParameterController.get_all: uuid=%s, type=%s" %
(uuid, type))
return self._get_ptp_parameter_collection(uuid, type,
marker, limit,
sort_key=sort_key,
sort_dir=sort_dir)
@wsme_pecan.wsexpose(PtpParameter, types.uuid)
def get_one(self, ptp_parameter_uuid):
"""Retrieve a single PTP parameter."""
LOG.debug("PtpParameterController.get_one: uuid=%s" %
ptp_parameter_uuid)
try:
ptp_parameter = objects.ptp_parameter.get_by_uuid(
pecan.request.context,
ptp_parameter_uuid)
except exception.InvalidParameterValue:
raise wsme.exc.ClientSideError(
_("No PTP parameter found for %s" % ptp_parameter_uuid))
return PtpParameter.convert_with_links(ptp_parameter)
def _check_foreign_exists(self, type, uuid):
LOG.debug("PtpParameterController._check_foreign_exists: "
"type %s uuid %s" % (type, uuid))
try:
if type == constants.PTP_PARAMETER_OWNER_INSTANCE:
try:
pecan.request.dbapi.ptp_instance_get(uuid)
except exception.PtpInstanceNotFound:
raise exception.NotFound
elif type == constants.PTP_PARAMETER_OWNER_INTERFACE:
try:
pecan.request.dbapi.ptp_interface_get(uuid)
except exception.PtpInterfaceNotFound:
raise exception.NotFound
except exception.NotFound:
raise wsme.exc.ClientSideError(
_("No foreign object found with id %s" % uuid))
@cutils.synchronized(LOCK_NAME)
@wsme_pecan.wsexpose(PtpParameter, body=PtpParameter)
def post(self, ptp_parameter):
"""Create a new PTP parameter."""
ptp_parameter_dict = ptp_parameter.as_dict()
LOG.debug("PtpParameterController.post: %s" % ptp_parameter_dict)
self._check_foreign_exists(ptp_parameter_dict['type'],
ptp_parameter_dict['foreign_uuid'])
# Get rid of owner details to create the PTP parameter
try:
ptp_parameter_dict.pop('owner')
except KeyError:
LOG.debug("PtpParameterController.post: no owner data in %s" %
ptp_parameter_dict)
result = pecan.request.dbapi.ptp_parameter_create(ptp_parameter_dict)
return PtpParameter.convert_with_links(result)
@cutils.synchronized(LOCK_NAME)
@wsme.validate(types.uuid, [PtpParameterPatchType])
@wsme_pecan.wsexpose(PtpParameter, types.uuid,
body=[PtpParameterPatchType])
def patch(self, uuid, patch):
"""Update the value of an existing PTP parameter."""
if self._parent:
raise exception.OperationNotPermitted
ptp_parameter = objects.ptp_parameter.get_by_uuid(
pecan.request.context, uuid)
patch_obj = jsonpatch.JsonPatch(patch)
try:
patched_parameter = PtpParameter(
**jsonpatch.apply_patch(ptp_parameter.as_dict(), patch_obj))
except utils.JSONPATCH_EXCEPTIONS as e:
raise exception.PatchError(patch=patch, reason=e)
# Update only the fields that have changed
for field in objects.ptp_parameter.fields:
if ptp_parameter[field] != getattr(patched_parameter, field):
ptp_parameter[field] = getattr(patched_parameter, field)
ptp_parameter.save()
return PtpParameter.convert_with_links(ptp_parameter)
@cutils.synchronized(LOCK_NAME)
@wsme_pecan.wsexpose(None, types.uuid, status_code=204)
def delete(self, ptp_parameter_uuid):
"""Delete a PTP parameter."""
LOG.debug("PtpParameterController.delete: %s" % ptp_parameter_uuid)
if self._parent:
raise exception.OperationNotPermitted
pecan.request.dbapi.ptp_parameter_destroy(ptp_parameter_uuid)

13
sysinv/sysinv/sysinv/sysinv/common/constants.py
View File

@ -1850,6 +1850,19 @@ PTP_TRANSPORT_UDP = 'udp'
PTP_TRANSPORT_L2 = 'l2'
PTP_NETWORK_TRANSPORT_IEEE_802_3 = 'L2'
# PTP instance types
PTP_INSTANCE_TYPE_PTP4L = 'ptp4l'
PTP_INSTANCE_TYPE_PHC2SYS = 'phc2sys'
PTP_INSTANCE_TYPE_TS2PHC = 'ts2phc'
# PTP instances created during migration from service parameters
PTP_INSTANCE_DEFAULT_PTP4L = 'default-ptp4l'
PTP_INSTANCE_DEFAULT_PHC2SYS = 'default-phc2sys'
# PTP parameter: owner types
PTP_PARAMETER_OWNER_INSTANCE = 'ptp-instance'
PTP_PARAMETER_OWNER_INTERFACE = 'ptp-interface'
# Backup & Restore
FIX_INSTALL_UUID_INTERVAL_SECS = 30

28
sysinv/sysinv/sysinv/sysinv/db/api.py
View File

@ -2141,6 +2141,7 @@ class Connection(object):
{
'name': 'domain',
'value': '24',
'type': 'ptp-instance',
'foreign_uuid': 'c2abca03-2f33-413e-b60d-85133a4a37b6'
}
:returns: A PTP parameter.
@ -2177,18 +2178,35 @@ class Connection(object):
"""
@abc.abstractmethod
def ptp_parameters_get_by_foreign_uuid(self, uuid, limit=None, marker=None,
sort_key=None, sort_dir=None):
"""Returns a list of PTP parameters for a given foreign UUID.
def ptp_parameters_get_by_type(self, type, limit=None, marker=None,
sort_key=None, sort_dir=None):
"""Returns a list of all PTP parameters for a given owner type.
:param uuid: The uuid of a PTP instance or PTP interface association.
:param type: Type of the parameter owner (either 'ptp-instance' or
'ptp-interface')
:param limit: Maximum number of PTP parameters to return.
:param marker: The last item of the previous page; we return the next
result set.
:param sort_key: Attribute by which results should be sorted
:param sort_dir: direction in which results should be sorted
(asc, desc)
:returns: A list of PTP parameters.
:returns: A list of PTP parameters for a specific owner type.
"""
@abc.abstractmethod
def ptp_parameters_get_by_owner(self, uuid, limit=None, marker=None,
sort_key=None, sort_dir=None):
"""Returns a list of the PTP parameters for a given owner identified by
its UUID.
:param uuid: UUID of the parameter owner.
:param limit: Maximum number of PTP parameters to return.
:param marker: The last item of the previous page; we return the next
result set.
:param sort_key: Attribute by which results should be sorted
:param sort_dir: direction in which results should be sorted
(asc, desc)
:returns: A list of PTP parameters of the specified owner.
"""
@abc.abstractmethod

81
sysinv/sysinv/sysinv/sysinv/db/sqlalchemy/api.py
View File

@ -1178,6 +1178,43 @@ def add_deviceimage_filter(query, value):
return add_identity_filter(query, value, use_name=True)
def add_ptp_instance_filter_by_host(query, hostid):
"""Adds a ptp-instance-specific ihost filter to a query.
Filters results by host id if supplied value is an integer,
otherwise attempts to filter results by host uuid.
:param query: Initial query to add filter to.
:param hostid: host id or uuid to filter results by.
:return: Modified query.
"""
if utils.is_int_like(hostid):
return query.filter_by(host_id=hostid)
elif utils.is_uuid_like(hostid):
query = query.join(models.ihost)
return query.filter(models.ihost.uuid == hostid)
LOG.debug("ptp_instance_filter_by_host: "
"No match for supplied filter id (%s)" % str(hostid))
def add_ptp_parameter_filter_by_owner(query, type, uuid):
if type == constants.PTP_PARAMETER_OWNER_INSTANCE:
query = query.join(models.PtpInstances,
models.PtpInstances.uuid == uuid)
elif type == constants.PTP_PARAMETER_OWNER_INTERFACE:
query = (query.join(models.PtpInterfaces,
models.PtpInterfaces.uuid == uuid)
.join(models.Interfaces, models.Interfaces.id ==
models.PtpInterfaces.interface_id))
else:
LOG.error("ptp_parameter_filter_by_owner: "
"Invalid owner type (%s)" % type)
raise exception.Invalid()
return query
class Connection(api.Connection):
"""SqlAlchemy connection."""
@ -3758,7 +3795,7 @@ class Connection(api.Connection):
query = model_query(models.PtpInstances)
if name is not None:
query = query.filter_by(name=name)
if service is not None:
elif service is not None:
query = query.filter_by(service=service)
try:
return query.one()
@ -3775,10 +3812,8 @@ class Connection(api.Connection):
@objects.objectify(objects.ptp_instance)
def ptp_instances_get_by_ihost(self, ihost_id, limit=None, marker=None,
sort_key=None, sort_dir=None):
# ihost_get() to raise an exception if the ihost is not found
ihost_obj = self.ihost_get(ihost_id)
query = model_query(models.PtpInstances)
query = query.filter_by(host_id=ihost_obj.id)
query = add_ptp_instance_filter_by_host(query, ihost_id)
return _paginate_query(models.PtpInstances, limit, marker,
sort_key, sort_dir, query)
@ -3968,20 +4003,32 @@ class Connection(api.Connection):
sort_key, sort_dir, query)
@objects.objectify(objects.ptp_parameter)
def ptp_parameters_get_by_foreign_uuid(self, uuid, limit=None, marker=None,
sort_key=None, sort_dir=None):
# Look for foreign UUID in PTP instances first, then in PTP interfaces
# if not found
try:
foreign_obj = self.ptp_instance_get(uuid)
except exception.PtpInstanceNotFound:
try:
foreign_obj = self.ptp_interface_get(uuid)
except exception.PtpInterfaceNotFound:
raise exception.NotFound()
def ptp_parameters_get_by_type(self, type, limit=None, marker=None,
sort_key=None, sort_dir=None):
query = model_query(models.PtpParameters)
query = query.filter_by(foreign_uuid=foreign_obj.uuid)
query = query.filter_by(type=type)
return _paginate_query(models.PtpParameters, limit, marker,
sort_key, sort_dir, query)
def _ptp_parameter_get_type(self, uuid):
type = None
query = model_query(models.PtpParameters)
query = query.filter_by(foreign_uuid=uuid)
ptp_parameter_object = query.first()
if ptp_parameter_object:
type = ptp_parameter_object.type
return type
@objects.objectify(objects.ptp_parameter)
def ptp_parameters_get_by_owner(self, uuid, limit=None, marker=None,
sort_key=None, sort_dir=None):
type = self._ptp_parameter_get_type(uuid)
if not type:
return []
query = model_query(models.PtpParameters)
query = query.filter_by(foreign_uuid=uuid)
query = add_ptp_parameter_filter_by_owner(query, type, uuid)
return _paginate_query(models.PtpParameters, limit, marker,
sort_key, sort_dir, query)

9
sysinv/sysinv/sysinv/sysinv/db/sqlalchemy/migrate_repo/versions/120_ptp_instances.py
View File

@ -13,7 +13,7 @@
# from datetime import datetime
from sqlalchemy import Integer, String, DateTime, Text
from sqlalchemy import Column, MetaData, Table, ForeignKey
from sqlalchemy import Column, MetaData, Table, ForeignKey, UniqueConstraint
from sysinv.db.sqlalchemy.models import UUID_LENGTH
@ -101,9 +101,12 @@ def upgrade(migrate_engine):
Column('id', Integer, primary_key=True, nullable=False),
Column('uuid', String(UUID_LENGTH), unique=True),
Column('name', String(255)),
Column('name', String(255), nullable=False),
Column('value', String(255)),
Column('foreign_uuid', String(UUID_LENGTH)),
Column('type', String(255)),
Column('foreign_uuid', String(UUID_LENGTH), nullable=False),
UniqueConstraint('name', 'foreign_uuid', name='u_paramnameforeign'),
mysql_engine=ENGINE,
mysql_charset=CHARSET,

46
sysinv/sysinv/sysinv/sysinv/db/sqlalchemy/models.py
View File

@ -262,7 +262,7 @@ class inode(Base):
forihostid = Column(Integer, ForeignKey('i_host.id', ondelete='CASCADE'))
host = relationship("ihost", backref="nodes", lazy="joined", join_depth=1)
host = relationship("ihost", backref="nodes", lazy="joined", cascade="all")
UniqueConstraint('numa_node', 'forihostid', name='u_hostnuma')
@ -369,7 +369,7 @@ class Interfaces(Base):
join_depth=1)
host = relationship("ihost", backref="interfaces",
lazy="joined", join_depth=1)
lazy="joined", cascade="all")
addresses = relationship("Addresses",
backref=backref("interface", lazy="joined"),
@ -481,10 +481,10 @@ class Ports(Base):
capabilities = Column(JSONEncodedDict)
# JSON{'speed':1000,'MTU':9600, 'duplex':'', 'autonegotiation':'false'}
node = relationship("inode", backref="ports", lazy="joined", join_depth=1)
host = relationship("ihost", backref="ports", lazy="joined", join_depth=1)
node = relationship("inode", backref="ports", lazy="joined", cascade="all")
host = relationship("ihost", backref="ports", lazy="joined", cascade="all")
interface = relationship("Interfaces", backref="port",
lazy="joined", join_depth=1)
lazy="joined", cascade="all")
UniqueConstraint('pciaddr', 'dev_id', 'host_id', name='u_pciaddrdevihost')
@ -798,8 +798,8 @@ class PtpInstances(Base):
# capabilities not used yet: JSON{'':"", '':''}
capabilities = Column(JSONEncodedDict)
host = relationship("ihost", backref="ptp_instances", lazy="joined",
join_depth=1)
host = relationship("ihost", backref="ptp_instance", lazy="joined",
cascade="all")
class PtpInterfaces(Base):
@ -811,15 +811,16 @@ class PtpInterfaces(Base):
interface_id = Column(Integer,
ForeignKey('interfaces.id', ondelete='CASCADE'))
ptp_instance_id = Column(Integer,
ForeignKey('ptp_instances.id', ondelete='CASCADE'))
ForeignKey('ptp_instances.id',
ondelete='CASCADE'))
# capabilities not used yet: JSON{'':"", '':''}
capabilities = Column(JSONEncodedDict)
interface = relationship("Interfaces", backref="ptp_interfaces",
lazy="joined", join_depth=1)
ptp_instance = relationship("PtpInstances", backref="ptp_interfaces",
lazy="joined", join_depth=1)
interface = relationship("Interfaces", backref="ptp_interface",
lazy="joined", cascade="all")
ptp_instance = relationship("PtpInstances", backref="ptp_interface",
lazy="joined", cascade="all")
class PtpParameters(Base):
@ -829,10 +830,25 @@ class PtpParameters(Base):
uuid = Column(String(UUID_LENGTH), unique=True)
name = Column(String(255), nullable=False)
value = Column(String(255), nullable=False)
value = Column(String(255))
# Either a "PtpInstance" or "PtpInterface" uuid:
foreign_uuid = Column(String(UUID_LENGTH))
type = Column(String(255))
foreign_uuid = Column(String(UUID_LENGTH), nullable=False)
ptp_instance = relationship(
"PtpInstances",
primaryjoin="PtpParameters.foreign_uuid == foreign(PtpInstances.uuid)",
lazy="subquery",
cascade="all")
ptp_interface = relationship(
"PtpInterfaces",
primaryjoin="PtpParameters.foreign_uuid == "
"foreign(PtpInterfaces.uuid)",
lazy="subquery",
cascade="all")
UniqueConstraint('name', 'foreign_uuid', name='u_paramnameforeign')
class StorageTier(Base):

8
sysinv/sysinv/sysinv/sysinv/objects/ptp_instance.py
View File

@ -22,13 +22,17 @@ class PtpInstance(base.SysinvObject):
'name': utils.str_or_none,
'service': utils.str_or_none,
'host_uuid': utils.str_or_none,
'host_id': utils.int_or_none,
'host_uuid': utils.str_or_none,
'hostname': utils.str_or_none,
'capabilities': utils.dict_or_none
}
_foreign_fields = {
'host_uuid': 'host:uuid'
'host_uuid': 'host:uuid',
'hostname': 'host:hostname'
}
@base.remotable_classmethod

33
sysinv/sysinv/sysinv/sysinv/objects/ptp_parameter.py
View File

@ -6,11 +6,38 @@
#
########################################################################
from sysinv.common import constants
from sysinv.db import api as db_api
from sysinv.objects import base
from sysinv.objects import utils
def get_owner(field, db_object):
owner = {}
"""Retrieves the owner details based on type and uuid."""
if db_object['type'] == constants.PTP_PARAMETER_OWNER_INSTANCE:
ptp_instances = getattr(db_object, 'ptp_instance')
if ptp_instances:
owner['name'] = ptp_instances[0].name
owner['type'] = ptp_instances[0].service
host = getattr(ptp_instances[0], 'host')
if host:
owner['hostname'] = host.hostname
elif db_object['type'] == constants.PTP_PARAMETER_OWNER_INTERFACE:
ptp_interfaces = getattr(db_object, 'ptp_interface')
if ptp_interfaces:
interface = getattr(ptp_interfaces[0], 'interface')
if interface:
owner['name'] = interface.ifname
owner['type'] = interface.iftype
host = getattr(interface, 'host')
if host:
owner['hostname'] = host.hostname
return owner
class PtpParameter(base.SysinvObject):
dbapi = db_api.get_instance()
@ -22,10 +49,14 @@ class PtpParameter(base.SysinvObject):
'name': utils.str_or_none,
'value': utils.str_or_none,
'foreign_uuid': utils.str_or_none
'type': utils.str_or_none,
'foreign_uuid': utils.str_or_none,
'owner': dict
}
_foreign_fields = {
'owner': get_owner
}
@base.remotable_classmethod

143
sysinv/sysinv/sysinv/sysinv/tests/api/test_ptp_instance.py
View File

@ -3,12 +3,11 @@
# SPDX-License-Identifier: Apache-2.0
#
from oslo_utils import uuidutils
from six.moves import http_client
from sysinv.common import constants
from sysinv.db import api as dbapi
from sysinv.tests.api import base
from sysinv.tests.db import base as dbbase
from sysinv.tests.db import utils as dbutils
class BasePtpInstanceTestCase(base.FunctionalTest, dbbase.BaseHostTestCase):
@ -27,32 +26,6 @@ class BasePtpInstanceTestCase(base.FunctionalTest, dbbase.BaseHostTestCase):
# Can perform API operations on this object at a sublevel of host
HOST_PREFIX = '/ihosts'
# Attributes that should be populated by an API query
expected_api_fields = ['uuid', 'name', 'service']
# Attributes that should NOT be populated by an API query
hidden_api_fields = ['host_id']
def _get_ptp_instance(self, **kw):
instance = {
'id': kw.get('id'),
'uuid': kw.get('uuid'),
'name': kw.get('name', None),
'service': kw.get('service', 'ptp4l'),
'host_id': kw.get('host_id', None)
}
return instance
def _create_ptp_instance(self, **kw):
instance = self._get_ptp_instance(**kw)
# Let DB generate ID if isn't specified
if 'id' not in kw:
del instance['id']
if 'uuid' in kw:
del instance['uuid']
db_api = dbapi.get_instance()
return db_api.ptp_instance_create(instance)
def setUp(self):
super(BasePtpInstanceTestCase, self).setUp()
self.controller = self._create_test_host(constants.CONTROLLER)
@ -64,39 +37,38 @@ class BasePtpInstanceTestCase(base.FunctionalTest, dbbase.BaseHostTestCase):
def get_host_scoped_url(self, host_uuid):
return '%s/%s%s' % (self.HOST_PREFIX, host_uuid, self.API_PREFIX)
def get_post_object(self, name='test_instance', service='ptp4l',
host_id=None, host_uuid=None):
ptp_instance_db = self._get_ptp_instance(name=name,
service=service,
host_id=host_id)
def get_post_object(self, name='test_instance',
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=None, host_uuid=None, hostname=None):
ptp_instance_db = dbutils.get_test_ptp_instance(name=name,
service=service,
host_id=host_id)
ptp_instance_db['host_uuid'] = host_uuid
ptp_instance_db['hostname'] = hostname
return ptp_instance_db
def assert_fields(self, api_object):
assert(uuidutils.is_uuid_like(api_object['uuid']))
for field in self.expected_api_fields:
self.assertIn(field, api_object)
for field in self.hidden_api_fields:
self.assertNotIn(field, api_object)
class TestCreatePtpInstance(BasePtpInstanceTestCase):
name = 'ptp-name'
service = 'ptp4l'
name = constants.PTP_INSTANCE_DEFAULT_PTP4L
service = constants.PTP_INSTANCE_TYPE_PTP4L
host_id = None
host_uuid = None
hostname = None
def setUp(self):
super(TestCreatePtpInstance, self).setUp()
self.host_id = self.controller.id
self.host_uuid = self.controller.uuid
self._create_ptp_instance(name=self.name, service=self.service,
host_id=self.host_id)
self.hostname = self.controller.hostname
dbutils.create_test_ptp_instance(name=self.name, service=self.service,
host_id=self.host_id)
def _create_ptp_instance_success(self, name, service, host_id, host_uuid):
def _create_ptp_instance_success(self, name, service, host_id, host_uuid,
hostname):
ptp_instance_db = self.get_post_object(name=name, service=service,
host_id=host_id,
host_uuid=host_uuid)
host_uuid=host_uuid,
hostname=hostname)
response = self.post_json(self.API_PREFIX, ptp_instance_db,
headers=self.API_HEADERS)
self.assertEqual('application/json', response.content_type)
@ -105,10 +77,11 @@ class TestCreatePtpInstance(BasePtpInstanceTestCase):
ptp_instance_db[self.COMMON_FIELD])
def _create_ptp_instance_failed(self, name, service, host_id, host_uuid,
status_code, error_message):
hostname, status_code, error_message):
ptp_instance_db = self.get_post_object(name=name, service=service,
host_id=host_id,
host_uuid=host_uuid)
host_uuid=host_uuid,
hostname=hostname)
response = self.post_json(self.API_PREFIX, ptp_instance_db,
headers=self.API_HEADERS, expect_errors=True)
self.assertEqual('application/json', response.content_type)
@ -116,9 +89,11 @@ class TestCreatePtpInstance(BasePtpInstanceTestCase):
self.assertIn(error_message, response.json['error_message'])
def test_create_ptp_instance_ok(self):
self._create_ptp_instance_success('test-instance', 'ptp4l',
self._create_ptp_instance_success('test-instance',
constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.controller.id,
host_uuid=self.controller.uuid)
host_uuid=self.controller.uuid,
hostname=self.controller.hostname)
def test_create_ptp_instance_invalid_service(self):
self._create_ptp_instance_failed(
@ -126,6 +101,7 @@ class TestCreatePtpInstance(BasePtpInstanceTestCase):
'invalid',
host_id=self.controller.id,
host_uuid=self.controller.uuid,
hostname=self.controller.hostname,
status_code=http_client.BAD_REQUEST,
error_message='Invalid input for field/attribute service')
@ -137,6 +113,7 @@ class TestCreatePtpInstance(BasePtpInstanceTestCase):
service=self.service,
host_id=self.host_id,
host_uuid=self.host_uuid,
hostname=self.controller.hostname,
status_code=http_client.CONFLICT,
error_message=error_message)
@ -145,9 +122,10 @@ class TestCreatePtpInstance(BasePtpInstanceTestCase):
error_message = '%s could not be found' % bad_uuid
self._create_ptp_instance_failed(
'test-invalid',
'phc2sys',
constants.PTP_INSTANCE_TYPE_PHC2SYS,
host_id=99,
host_uuid='f4c56ddf-aef3-46ed-b9aa-126a1faafd40',
host_uuid=bad_uuid,
hostname='badhost',
status_code=http_client.NOT_FOUND,
error_message=error_message)
@ -157,8 +135,10 @@ class TestGetPtpInstance(BasePtpInstanceTestCase):
super(TestGetPtpInstance, self).setUp()
def test_get_ptp_instance_found(self):
ptp_instance = self._create_ptp_instance(
name='fake-ptp4l', service='ptp4l', host_id=self.controller.id)
ptp_instance = dbutils.create_test_ptp_instance(
name=constants.PTP_INSTANCE_DEFAULT_PTP4L,
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.controller.id)
uuid = ptp_instance['uuid']
response = self.get_json(self.get_single_url(uuid))
self.assertIn(self.COMMON_FIELD, response)
@ -180,13 +160,15 @@ class TestListPtpInstance(BasePtpInstanceTestCase):
self._create_test_ptp_instances()
def _create_test_ptp_instances(self, name_prefix='test', host_id=None):
services = ['ptp4l', 'phc2sys', 'ts2phc']
services = [constants.PTP_INSTANCE_TYPE_PTP4L,
constants.PTP_INSTANCE_TYPE_PHC2SYS,
constants.PTP_INSTANCE_TYPE_TS2PHC]
instances = []
if not host_id:
host_id = self.controller.id
for service in services:
name = '%s-%s' % (name_prefix, service)
instance = self._create_ptp_instance(
instance = dbutils.create_test_ptp_instance(
name=name, service=service, host_id=host_id)
instances.append(instance)
return instances
@ -214,30 +196,57 @@ class TestDeletePtpInstance(BasePtpInstanceTestCase):
python2 and python3 libraries may return different
content_type (None, or empty json) when NO_CONTENT returned.
"""
ptp_instance = None
uuid = None
def setUp(self):
super(TestDeletePtpInstance, self).setUp()
self.ptp_instance = dbutils.create_test_ptp_instance(
name=constants.PTP_INSTANCE_DEFAULT_PTP4L,
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.controller.id)
self.uuid = self.ptp_instance['uuid']
def test_delete_ptp_instance(self):
ptp_instance = self._create_ptp_instance(
name='fake-phc2sys', service='phc2sys', host_id=self.controller.id)
uuid = ptp_instance['uuid']
response = self.delete(self.get_single_url(uuid),
def test_delete_ptp_instance_ok(self):
response = self.delete(self.get_single_url(self.uuid),
headers=self.API_HEADERS)
self.assertEqual(response.status_code, http_client.NO_CONTENT)
# Check the instance was indeed removed
error_message = 'No PTP instance with id %s found' % uuid
response = self.get_json(self.get_single_url(uuid),
error_message = 'No PTP instance with id %s found' % self.uuid
response = self.get_json(self.get_single_url(self.uuid),
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.NOT_FOUND)
self.assertIn(error_message, response.json['error_message'])
def test_delete_ptp_instance_with_parameters_failed(self):
# TODO: implement when PTP parameters API is available
pass
ptp_parameter = dbutils.create_test_ptp_parameter(
name='fake-param', value='fake-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.uuid)
self.assertEqual(self.uuid, ptp_parameter['foreign_uuid'])
response = self.delete(self.get_single_url(self.uuid),
headers=self.API_HEADERS, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.BAD_REQUEST)
self.assertIn('still associated with PTP parameter',
response.json['error_message'])
def test_delete_ptp_instance_with_interfaces_failed(self):
# TODO: implement when PTP interfaces API is available
pass
interface = dbutils.create_test_interface(
ifname='fake0', ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.controller.id, ihost_uuid=self.controller.uuid)
ptp_interface = dbutils.create_test_ptp_interface(
interface_id=interface['id'],
ptp_instance_id=self.ptp_instance['id'])
self.assertEqual(self.ptp_instance['id'],
ptp_interface['ptp_instance_id'])
response = self.delete(self.get_single_url(self.uuid),
headers=self.API_HEADERS, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.BAD_REQUEST)
self.assertIn('still associated with PTP interface',
response.json['error_message'])

75
sysinv/sysinv/sysinv/sysinv/tests/api/test_ptp_interface.py
View File

@ -81,12 +81,12 @@ class TestCreatePtpInterface(BasePtpInterfaceTestCase):
self.test_instance = dbutils.create_test_ptp_instance(
name='testInstance',
service='ptp4l',
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.controller.id)
def _create_ptp_interface_success(self, interface_uuid, ptp_instance_uuid):
ptp_interface_db = self.get_post_object(interface_uuid,
ptp_instance_uuid)
ptp_instance_uuid)
response = self.post_json(self.API_PREFIX, ptp_interface_db,
headers=self.API_HEADERS)
self.assertEqual('application/json', response.content_type)
@ -125,7 +125,7 @@ class TestCreatePtpInterface(BasePtpInterfaceTestCase):
def test_create_ptp_interface_duplicate(self):
self._create_ptp_interface_success(self.test_interface.uuid,
self.test_instance.uuid)
self.test_instance.uuid)
self._create_ptp_interface_failed(
interface_uuid=self.test_interface.uuid,
@ -145,16 +145,17 @@ class TestGetPtpInterface(BasePtpInterfaceTestCase):
self.test_instance = dbutils.create_test_ptp_instance(
name='testInstance',
service='ptp4l',
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.controller.id)
self.test_ptp_interface = dbutils.create_test_ptp_interface(
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance.id)
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance.id)
def test_get_ptp_interface_found(self):
response = self.get_json(self.get_single_url(self.test_ptp_interface.uuid))
response = self.get_json(
self.get_single_url(self.test_ptp_interface.uuid))
self.assertIn(self.COMMON_FIELD, response)
def test_get_ptp_interface_not_found(self):
@ -185,7 +186,7 @@ class TestListPtpInterface(BasePtpInterfaceTestCase):
self.test_instance_ptp4l = dbutils.create_test_ptp_instance(
name='ptp4lInstance',
service='ptp4l',
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.worker.id)
self.test_instance_phc2sys = dbutils.create_test_ptp_instance(
@ -194,28 +195,30 @@ class TestListPtpInterface(BasePtpInterfaceTestCase):
host_id=self.worker.id)
self.ptp4l_ptp_interface = dbutils.create_test_ptp_interface(
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance_ptp4l.id)
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance_ptp4l.id)
self.phc2sys_ptp_interface = dbutils.create_test_ptp_interface(
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance_phc2sys.id)
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance_phc2sys.id)
self.dummy_ptp_interface = dbutils.create_test_ptp_interface(
interface_id=self.dummy_interface.id,
ptp_instance_id=self.test_instance_ptp4l.id)
interface_id=self.dummy_interface.id,
ptp_instance_id=self.test_instance_ptp4l.id)
def test_list_ptp_interface_host(self):
response = self.get_json(self.get_host_scoped_url(self.worker.uuid))
for result in response[self.RESULT_KEY]:
self.assertEqual(self.worker.id, result['forihostid'])
if result['uuid'] == self.ptp4l_ptp_interface.uuid \
or result['uuid'] == self.dummy_interface.uuid:
self.assertEqual(self.test_instance_ptp4l.id, result['ptp_instance_id'])
or result['uuid'] == self.dummy_interface.uuid:
self.assertEqual(self.test_instance_ptp4l.id,
result['ptp_instance_id'])
elif result['uuid'] == self.phc2sys_ptp_interface.uuid:
self.assertEqual(self.test_instance_phc2sys.id, result['ptp_instance_id'])
self.assertEqual(self.test_instance_phc2sys.id,
result['ptp_instance_id'])
def test_list_ptp_interface_interface(self):
response = self.get_json(self.get_host_scoped_url_interface(self.worker.uuid,
self.test_interface.uuid))
response = self.get_json(self.get_host_scoped_url_interface(
self.worker.uuid, self.test_interface.uuid))
for result in response[self.RESULT_KEY]:
self.assertIn(self.COMMON_FIELD, result)
self.assertNotIn(self.dummy_interface.uuid, result)
@ -243,25 +246,39 @@ class TestDeletePtpInterface(BasePtpInterfaceTestCase):
self.test_instance_ptp4l = dbutils.create_test_ptp_instance(
name='ptp4lInstance',
service='ptp4l',
service=constants.PTP_INSTANCE_TYPE_PTP4L,
host_id=self.worker.id)
self.test_ptp_interface = dbutils.create_test_ptp_interface(
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance_ptp4l.id)
interface_id=self.test_interface.id,
ptp_instance_id=self.test_instance_ptp4l.id)
def test_delete_ptp_interface(self):
response = self.delete(self.get_single_url(self.test_ptp_interface.uuid),
headers=self.API_HEADERS)
response = self.delete(
self.get_single_url(self.test_ptp_interface.uuid),
headers=self.API_HEADERS)
self.assertEqual(response.status_code, http_client.NO_CONTENT)
error_message = 'No PTP interface with id %s found' % self.test_ptp_interface.uuid
response = self.get_json(self.get_single_url(self.test_ptp_interface.uuid),
expect_errors=True)
error_message = \
'No PTP interface with id %s found' % self.test_ptp_interface.uuid
response = self.get_json(
self.get_single_url(self.test_ptp_interface.uuid),
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.NOT_FOUND)
self.assertIn(error_message, response.json['error_message'])
def test_delete_ptp_interface_with_parameters_failed(self):
# TODO: implement when PTP parameters API is available
pass
ptp_parameter = dbutils.create_test_ptp_parameter(
name='fake-param', value='fake-value',
type=constants.PTP_PARAMETER_OWNER_INTERFACE,
foreign_uuid=self.test_ptp_interface.uuid)
self.assertEqual(self.test_ptp_interface.uuid,
ptp_parameter['foreign_uuid'])
response = self.delete(
self.get_single_url(self.test_ptp_interface.uuid),
headers=self.API_HEADERS, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.BAD_REQUEST)
self.assertIn('has PTP parameter', response.json['error_message'])

297
sysinv/sysinv/sysinv/sysinv/tests/api/test_ptp_parameter.py Normal file
View File

@ -0,0 +1,297 @@
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
from six.moves import http_client
from sysinv.common import constants
from sysinv.tests.api import base
from sysinv.tests.db import base as dbbase
from sysinv.tests.db import utils as dbutils
class BasePtpParameterTestCase(base.FunctionalTest, dbbase.BaseHostTestCase):
# Generic header passed to most API calls
API_HEADERS = {'User-Agent': 'sysinv-test'}
# Prefix for the URL
API_PREFIX = '/ptp_parameters'
# Python table key for the list of results
RESULT_KEY = 'ptp_parameters'
# Field that is known to exist for inputs and outputs
COMMON_FIELD = 'name'
# Can perform API operations on this object at a sublevel of PTP instances
PTP_INSTANCE_PREFIX = '/ptp_instances'
# Can perform API operations on this object at a sublevel of interfaces
INTERFACE_PREFIX = '/iinterfaces'
def setUp(self):
super(BasePtpParameterTestCase, self).setUp()
self.controller = self._create_test_host(constants.CONTROLLER)
self.ptp_instances = self._create_test_ptp_instance(self.controller)
self.platform_interfaces = \
self._create_test_host_platform_interface(self.controller)
self.ptp_interfaces = self._create_test_ptp_interface(
self.ptp_instances, self.platform_interfaces)
def get_single_url(self, ptp_parameter_uuid):
return '%s/%s' % (self.API_PREFIX, ptp_parameter_uuid)
def get_instance_scoped_url(self, ptp_instance_uuid):
return '%s/%s%s' % (self.PTP_INSTANCE_PREFIX, ptp_instance_uuid,
self.API_PREFIX)
def get_interface_scoped_url(self, interface_uuid):
return '%s/%s%s' % (self.INTERFACE_PREFIX, interface_uuid,
self.API_PREFIX)
def get_post_object(self, name='test_parameter', value='test_value',
type=None, foreign_uuid=None):
return dbutils.get_test_ptp_parameter(name=name,
value=value,
type=type,
foreign_uuid=foreign_uuid)
class TestCreatePtpParameter(BasePtpParameterTestCase):
name = 'test-param'
value = 'test-value'
type = constants.PTP_PARAMETER_OWNER_INSTANCE
foreign_uuid = None
def setUp(self):
super(TestCreatePtpParameter, self).setUp()
self.foreign_uuid = self.ptp_instances[0].uuid
dbutils.create_test_ptp_parameter(name=self.name,
value=self.value,
type=self.type,
foreign_uuid=self.foreign_uuid)
def _create_ptp_parameter_success(self, name, value, type, foreign_uuid):
ptp_parameter_db = self.get_post_object(name=name,
value=value,
type=type,
foreign_uuid=foreign_uuid)
response = self.post_json(self.API_PREFIX, ptp_parameter_db,
headers=self.API_HEADERS)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.OK)
self.assertEqual(response.json[self.COMMON_FIELD],
ptp_parameter_db[self.COMMON_FIELD])
def _create_ptp_parameter_failed(self, name, value, type, foreign_uuid,
status_code, error_message):
ptp_parameter_db = self.get_post_object(name=name,
value=value,
type=type,
foreign_uuid=foreign_uuid)
response = self.post_json(self.API_PREFIX, ptp_parameter_db,
headers=self.API_HEADERS, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, status_code)
self.assertIn(error_message, response.json['error_message'])
def test_create_ptp_parameter_instance_ok(self):
self._create_ptp_parameter_success(
name='instance-param', value='instance-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.ptp_instances[0].uuid)
def test_create_ptp_parameter_interface_ok(self):
self._create_ptp_parameter_success(
name='interface-param', value='interface-value',
type=constants.PTP_PARAMETER_OWNER_INTERFACE,
foreign_uuid=self.ptp_interfaces[0].uuid)
def test_create_ptp_parameter_invalid_type(self):
self._create_ptp_parameter_failed(
name='fake-param', value='fake-value',
type='invalid',
foreign_uuid=self.ptp_instances[0].uuid,
status_code=http_client.BAD_REQUEST,
error_message='Invalid input for field/attribute type')
def test_create_ptp_parameter_invalid_uuid(self):
bad_uuid = 'f4c56ddf-aef3-46ed-b9aa-126a1faafd40'
error_message = 'No foreign object found with id %s' % bad_uuid
self._create_ptp_parameter_failed(
name='fake-param', value='fake-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=bad_uuid,
status_code=http_client.BAD_REQUEST,
error_message=error_message)
def test_create_ptp_parameter_duplicate(self):
self._create_ptp_parameter_failed(
name=self.name,
value='another-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.foreign_uuid,
status_code=http_client.CONFLICT,
error_message='already exists')
class TestGetPtpParameter(BasePtpParameterTestCase):
def setUp(self):
super(TestGetPtpParameter, self).setUp()
def test_get_ptp_parameter_found(self):
ptp_parameter = dbutils.create_test_ptp_parameter(
name='fake-param', value='fake-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.ptp_instances[0].uuid)
uuid = ptp_parameter['uuid']
response = self.get_json(self.get_single_url(uuid))
self.assertIn(self.COMMON_FIELD, response)
def test_get_ptp_parameter_not_found(self):
fake_uuid = 'f4c56ddf-aef3-46ed-b9aa-126a1faafd40'
error_message = 'No PTP parameter with id %s found' % fake_uuid
response = self.get_json(self.get_single_url(fake_uuid),
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.NOT_FOUND)
self.assertIn(error_message, response.json['error_message'])
class TestListPtpParameter(BasePtpParameterTestCase):
def setUp(self):
super(TestListPtpParameter, self).setUp()
self._create_test_ptp_parameters(
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
prefix='ptp')
self._create_test_ptp_parameters(
type=constants.PTP_PARAMETER_OWNER_INTERFACE,
prefix='iface')
def _create_test_ptp_parameters(self, type, prefix='test',
foreign_uuid=None):
parameters = []
if not foreign_uuid:
if type == constants.PTP_PARAMETER_OWNER_INSTANCE:
foreign_uuid = self.ptp_instances[0].uuid
elif type == constants.PTP_PARAMETER_OWNER_INTERFACE:
foreign_uuid = self.ptp_interfaces[0].uuid
else:
return parameters
for i in range(2):
name = '%s-name%s' % (prefix, i)
value = '%s-value%s' % (prefix, i)
parameter = dbutils.create_test_ptp_parameter(
name=name, value=value, type=type, foreign_uuid=foreign_uuid)
parameters.append(parameter)
return parameters
def test_list_ptp_parameter_all(self):
response = self.get_json(self.API_PREFIX)
for result in response[self.RESULT_KEY]:
self.assertIn(self.COMMON_FIELD, result)
def test_list_ptp_parameter_empty(self):
fake_uuid = 'f4c56ddf-aef3-46ed-b9aa-126a1faafd40'
response = self.get_json(self.get_instance_scoped_url(fake_uuid))
self.assertEqual([], response[self.RESULT_KEY])
def test_list_ptp_parameter_by_type(self):
self._create_test_ptp_parameters(
constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.ptp_instances[1].uuid)
"""
TODO: needs investigation of the reason to get this:
webtest.app.AppError: Bad response: 400 Bad Request (not 200 OK or 3xx
redirect for http://
localhost/v1/ptp_parameters?q.field=type&q.value=ptp-instance&q.op=eq)
'{"error_message": "{\\"debuginfo\\": null, \\"faultcode\\":
\\"Client\\", \\"faultstring\\": \\"Unknown argument: \\\\\\"q.field,
q.value, q.op\\\\\\"\\"}"}'
query = [{
'field': 'type',
'value': constants.PTP_PARAMETER_OWNER_INSTANCE,
'op': 'eq'
}]
response = self.get_json(self.API_PREFIX, q=query)
for result in response[self.RESULT_KEY]:
self.assertEqual(constants.PTP_PARAMETER_OWNER_INSTANCE,
result['type'])
"""
def test_list_ptp_parameter_by_instance(self):
self._create_test_ptp_parameters(
constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.ptp_instances[1].uuid)
response = self.get_json(self.get_instance_scoped_url(
self.ptp_instances[1].uuid))
for result in response[self.RESULT_KEY]:
self.assertEqual(self.ptp_instances[1].uuid,
result['foreign_uuid'])
def test_list_ptp_parameter_by_interface(self):
self._create_test_ptp_parameters(
constants.PTP_PARAMETER_OWNER_INTERFACE,
foreign_uuid=self.ptp_interfaces[1].uuid)
response = self.get_json(self.get_interface_scoped_url(
self.ptp_interfaces[1].uuid))
for result in response[self.RESULT_KEY]:
self.assertEqual(self.ptp_interfaces[1].uuid,
result['foreign_uuid'])
class TestUpdatePtpParameter(BasePtpParameterTestCase):
def setUp(self):
super(TestUpdatePtpParameter, self).setUp()
def test_update_ptp_parameter(self):
ptp_parameter = dbutils.create_test_ptp_parameter(
name='fake-param', value='fake-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.ptp_instances[0].uuid)
uuid = ptp_parameter['uuid']
response = self.patch_json(self.get_single_url(uuid),
[{'path': '/value',
'value': 'changed-value',
'op': 'replace'}],
headers=self.API_HEADERS)
self.assertEqual(response.content_type, 'application/json')
self.assertEqual(response.status_code, http_client.OK)
# Check the parameter was indeed updated
response = self.get_json(self.get_single_url(uuid))
self.assertEqual(response['value'], 'changed-value')
class TestDeletePtpParameter(BasePtpParameterTestCase):
""" Tests deletion.
Typically delete APIs return NO CONTENT.
python2 and python3 libraries may return different
content_type (None, or empty json) when NO_CONTENT returned.
"""
def setUp(self):
super(TestDeletePtpParameter, self).setUp()
def test_delete_ptp_parameter(self):
ptp_parameter = dbutils.create_test_ptp_parameter(
name='fake-param', value='fake-value',
type=constants.PTP_PARAMETER_OWNER_INSTANCE,
foreign_uuid=self.ptp_instances[0].uuid)
uuid = ptp_parameter['uuid']
response = self.delete(self.get_single_url(uuid),
headers=self.API_HEADERS)
self.assertEqual(response.status_code, http_client.NO_CONTENT)
# Check the instance was indeed removed
error_message = 'No PTP parameter with id %s found' % uuid
response = self.get_json(self.get_single_url(uuid),
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(response.status_code, http_client.NOT_FOUND)
self.assertIn(error_message, response.json['error_message'])

23
sysinv/sysinv/sysinv/sysinv/tests/db/base.py
View File

@ -447,6 +447,29 @@ class BaseHostTestCase(BaseSystemTestCase):
index = index + 1
return ifaces
def _create_test_ptp_instance(self, host):
services = [constants.PTP_INSTANCE_TYPE_PTP4L,
constants.PTP_INSTANCE_TYPE_PHC2SYS]
names = [constants.PTP_INSTANCE_DEFAULT_PTP4L,
constants.PTP_INSTANCE_DEFAULT_PHC2SYS]
ptp_instances = []
for svc, nm in zip(services, names):
instance = dbutils.create_test_ptp_instance(
name=nm, service=svc, host_id=host['id'])
ptp_instances.append(instance)
return ptp_instances
def _create_test_ptp_interface(self,
ptp_instances,
platform_interfaces):
ptp_interfaces = []
for ptp_instance in ptp_instances:
ptp_interface = dbutils.create_test_ptp_interface(
ptp_instance_id=ptp_instance['id'],
interface_id=platform_interfaces[0]['id'])
ptp_interfaces.append(ptp_interface)
return ptp_interfaces
class ControllerHostTestCase(BaseHostTestCase):

1
sysinv/sysinv/sysinv/sysinv/tests/db/sqlalchemy/test_migrations.py
View File

@ -2030,6 +2030,7 @@ class TestMigrations(BaseMigrationTestCase, WalkVersionsMixin):
'uuid': 'String',
'name': 'String',
'value': 'String',
'type': 'String',
'foreign_uuid': 'String',
}
for column, column_type in ptp_parameters_columns.items():

46
sysinv/sysinv/sysinv/sysinv/tests/db/utils.py
View File

@ -536,25 +536,27 @@ def create_test_ptp(**kw):
return dbapi.ptp_create(ptp)
# Create test ptp_instance object
# Utility functions to create a PTP instance for testing
def get_test_ptp_instance(**kw):
ptp_instance = {
instance = {
'id': kw.get('id'),
'uuid': kw.get('uuid'),
'name': kw.get('name'),
'service': kw.get('service'),
'host_id': kw.get('host_id'),
'name': kw.get('name', None),
'service': kw.get('service', constants.PTP_INSTANCE_TYPE_PTP4L),
'host_id': kw.get('host_id', None)
}
return ptp_instance
return instance
def create_test_ptp_instance(**kw):
ptp_instance = get_test_ptp_instance(**kw)
# Let DB generate ID if it isn't specified explicitly
instance = get_test_ptp_instance(**kw)
# Let DB generate ID if isn't specified
if 'id' not in kw:
del ptp_instance['id']
del instance['id']
if 'uuid' in kw:
del instance['uuid']
dbapi = db_api.get_instance()
return dbapi.ptp_instance_create(ptp_instance)
return dbapi.ptp_instance_create(instance)
# Create test ptp_interface object
@ -575,6 +577,30 @@ def create_test_ptp_interface(**kw):
return dbapi.ptp_interface_create(ptp_interface)
# Utility functions to create a PTP parameter for testing
def get_test_ptp_parameter(**kw):
parameter = {
'id': kw.get('id'),
'uuid': kw.get('uuid'),
'name': kw.get('name', None),
'value': kw.get('value', None),
'type': kw.get('type', None),
'foreign_uuid': kw.get('foreign_uuid', None)
}
return parameter
def create_test_ptp_parameter(**kw):
parameter = get_test_ptp_parameter(**kw)
# Let DB generate ID if isn't specified
if 'id' not in kw:
del parameter['id']
if 'uuid' in kw:
del parameter['uuid']
dbapi = db_api.get_instance()
return dbapi.ptp_parameter_create(parameter)
# Create test dns object
def get_test_dns(**kw):
dns = {