Merge "PTP interfaces: CLI, REST API"

This commit is contained in:
Zuul 2021-11-04 17:24:54 +00:00 committed by Gerrit Code Review
commit c0003fd203
13 changed files with 786 additions and 13 deletions

View File

@ -70,6 +70,7 @@ from cgtsclient.v1 import pci_device
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 registry_image
from cgtsclient.v1 import remotelogging
from cgtsclient.v1 import restore
@ -120,6 +121,7 @@ class Client(http.HTTPClient):
self.intp = intp.intpManager(self)
self.ptp = ptp.ptpManager(self)
self.ptp_instance = ptp_instance.PtpInstanceManager(self)
self.ptp_interface = ptp_interface.PtpInterfaceManager(self)
self.iextoam = iextoam.iextoamManager(self)
self.controller_fs = controller_fs.ControllerFsManager(self)
self.storage_backend = storage_backend.StorageBackendManager(self)

View File

@ -29,8 +29,8 @@ def do_ptp_instance_list(cc, args):
ihost = ihost_utils._find_ihost(cc, instance.host_uuid)
setattr(instance, 'hostname', ihost.hostname)
field_labels = ['name', 'service', 'hostname']
fields = ['name', 'service', 'hostname']
field_labels = ['uuid', 'name', 'service', 'hostname']
fields = ['uuid', 'name', 'service', 'hostname']
utils.print_list(ptp_instances, fields, field_labels)

View File

@ -0,0 +1,56 @@
########################################################################
#
# 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 = ['interface_uuid', 'ptp_instance_uuid']
class PtpInterface(base.Resource):
def __repr__(self):
return "<PtpInterface %s>" % self._info
class PtpInterfaceManager(base.Manager):
resource_class = PtpInterface
@staticmethod
def _path(ptp_interface_id=None):
return 'v1/ptp_interfaces/%s' % ptp_interface_id if ptp_interface_id \
else 'v1/ptp_interfaces'
def list(self, q=None):
return self._list(options.build_url(self._path(), q), "ptp_interfaces")
def list_by_host(self, host_id):
path = 'v1/ihosts/%s/ptp_interfaces' % host_id
return self._list(path, "ptp_interfaces")
def list_by_interface(self, host_id, interface_id):
path = 'v1/ihosts/%s/ptp_interfaces?interface_uuid=%s' % (host_id, interface_id)
return self._list(path, "ptp_interfaces")
def get(self, ptp_interface_id):
try:
return self._list(self._path(ptp_interface_id))[0]
except IndexError:
return None
def create(self, **kwargs):
body = {}
for (key, value) in kwargs.items():
if key in CREATION_ATTRIBUTES:
body[key] = value
else:
raise exc.InvalidAttribute('Invalid attribute: %s' % key)
return self._create(self._path(), body)
def delete(self, ptp_interface_id):
return self._delete(self._path(ptp_interface_id))

View File

@ -0,0 +1,110 @@
########################################################################
#
# 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 ihost as ihost_utils
from cgtsclient.v1 import iinterface as iinterface_utils
from cgtsclient.v1 import ptp_instance as ptp_instance_utils
def _print_ptp_interface_show(ptp_interface_obj):
fields = ['uuid', 'ifname', 'ptp_instance_name',
'hostname', 'created_at']
data = [(f, getattr(ptp_interface_obj, f, '')) for f in fields]
utils.print_tuple_list(data)
@utils.arg('ptp_interface_uuid',
metavar='<ptp_interface_uuid>',
help="UUID of a PTP interface")
def do_ptp_interface_show(cc, args):
"""Show PTP interface attributes."""
ptp_interface = cc.ptp_interface.get(args.ptp_interface_uuid)
host_id = str(getattr(ptp_interface, 'forihostid', ''))
ihost = ihost_utils._find_ihost(cc, host_id)
setattr(ptp_interface, 'hostname', ihost.hostname)
_print_ptp_interface_show(ptp_interface)
@utils.arg('hostnameorid',
metavar='<hostname or id>',
help="Hostname or ID of a host")
@utils.arg('ifnameorid',
metavar='<interface name or uuid>',
nargs='?',
help="Interface name [OPTIONAL]")
def do_ptp_interface_list(cc, args):
"""List PTP interfaces on the specified host,
or a subset of PTP interfaces associated
with a given underlying interface.
"""
ihost = ihost_utils._find_ihost(cc, args.hostnameorid)
if args.ifnameorid:
validate_interface = iinterface_utils._find_interface(cc, ihost, args.ifnameorid)
ptp_interfaces = cc.ptp_interface.list_by_interface(ihost.uuid, validate_interface.uuid)
else:
ptp_interfaces = cc.ptp_interface.list_by_host(ihost.uuid)
# Add a hostname column using the forihostid field
for i in ptp_interfaces[:]:
host_id = str(getattr(i, 'forihostid', ''))
ihost = ihost_utils._find_ihost(cc, host_id)
setattr(i, 'hostname', ihost.hostname)
field_labels = ['uuid', 'hostname', 'ifname', 'ptp_instance_name']
fields = ['uuid', 'hostname', 'ifname', 'ptp_instance_name']
utils.print_list(ptp_interfaces, fields, field_labels)
@utils.arg('ptp_interface_uuid',
metavar='<ptp_interface_uuid>',
help="UUID of a PTP instance")
def do_ptp_interface_delete(cc, args):
"""Delete a PTP interface"""
cc.ptp_interface.delete(args.ptp_interface_uuid)
print('Deleted PTP interface: %s' % (args.ptp_interface_uuid))
@utils.arg('hostnameorid',
metavar='<hostname or id>',
help="The hostname or id associated with the interface and ptp instance [REQUIRED]")
@utils.arg('ifnameorid',
metavar='<interface name or uuid>',
help="Name or UUID of an interface [REQUIRED]")
@utils.arg('ptpinstancenameorid',
metavar='<ptp instance name or uuid>',
help="Name or UUID of a PTP instance [REQUIRED]")
def do_ptp_interface_add(cc, args):
"""Add a PTP interface."""
field_list = ['interface_uuid', 'ptp_instance_uuid']
validate_ihost = ihost_utils._find_ihost(cc, args.hostnameorid)
validate_ptp_instance = ptp_instance_utils._find_ptp_instance(cc, args.ptpinstancenameorid)
validate_interface = iinterface_utils._find_interface(cc, validate_ihost, args.ifnameorid)
if validate_ihost.uuid != validate_ptp_instance.host_uuid:
raise exc.CommandError('PTP instance %s is not on host %s.'
% (validate_ptp_instance.uuid, validate_ihost.hostname))
# 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))
data["interface_uuid"] = validate_interface.uuid
data["ptp_instance_uuid"] = validate_ptp_instance.uuid
ptp_interface = cc.ptp_interface.create(**data)
uuid = getattr(ptp_interface, 'uuid', '')
try:
ptp_interface = cc.ptp_interface.get(uuid)
except exc.HTTPNotFound:
raise exc.CommandError('Created PTP interface UUID not found: %s'
% uuid)
setattr(ptp_interface, 'hostname', validate_ihost.hostname)
_print_ptp_interface_show(ptp_interface)

View File

@ -55,6 +55,7 @@ from cgtsclient.v1 import partition_shell
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_shell
from cgtsclient.v1 import registry_image_shell
from cgtsclient.v1 import remotelogging_shell
@ -77,6 +78,7 @@ COMMAND_MODULES = [
intp_shell,
ptp_shell,
ptp_instance_shell,
ptp_interface_shell,
iextoam_shell,
controller_fs_shell,
storage_backend_shell,

View File

@ -66,6 +66,7 @@ from sysinv.api.controllers.v1 import pci_device
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 pv
from sysinv.api.controllers.v1 import registry_image
from sysinv.api.controllers.v1 import remotelogging
@ -150,6 +151,9 @@ class V1(base.APIBase):
ptp_instances = [link.Link]
"Links to the ptp_instances resource"
ptp_interfaces = [link.Link]
"Links to the ptp_interfaces resource"
iextoam = [link.Link]
"Links to the iextoam resource"
@ -457,6 +461,14 @@ class V1(base.APIBase):
'ptp_instances', '',
bookmark=True)]
v1.ptp_interfaces = [link.Link.make_link('self', pecan.request.host_url,
'ptp_interfaces', ''),
link.Link.make_link('bookmark',
pecan.request.host_url,
'ptp_interfaces', '',
bookmark=True)
]
v1.iextoam = [link.Link.make_link('self', pecan.request.host_url,
'iextoam', ''),
link.Link.make_link('bookmark',
@ -900,6 +912,7 @@ class Controller(rest.RestController):
intp = ntp.NTPController()
ptp = ptp.PTPController()
ptp_instances = ptp_instance.PtpInstanceController()
ptp_interfaces = ptp_interface.PtpInterfaceController()
iextoam = network_oam.OAMNetworkController()
controller_fs = controller_fs.ControllerFsController()
storage_backend = storage_backend.StorageBackendController()

View File

@ -87,7 +87,7 @@ from sysinv.api.controllers.v1 import interface_datanetwork
from sysinv.api.controllers.v1 import vim_api
from sysinv.api.controllers.v1 import patch_api
from sysinv.api.controllers.v1 import ptp_instance
from sysinv.api.controllers.v1 import ptp_interface
from sysinv.common import ceph
from sysinv.common import constants
from sysinv.common import device
@ -1131,6 +1131,9 @@ class HostController(rest.RestController):
ptp_instances = ptp_instance.PtpInstanceController(from_ihosts=True)
"Expose PTP instance as a sub-element of ihosts"
ptp_interfaces = ptp_interface.PtpInterfaceController(from_ihosts=True)
"Expose PTP interface as a sub-element of ihosts"
_custom_actions = {
'detail': ['GET'],
'bulk_add': ['POST'],
@ -6470,12 +6473,12 @@ class HostController(rest.RestController):
if ihost['clock_synchronization'] == constants.PTP:
# Ensure we have at least one PTP interface
host_interfaces = pecan.request.dbapi.iinterface_get_by_ihost(host_uuid)
ptp_interfaces = []
ptp_ifaces = []
for interface in host_interfaces:
if interface.ptp_role != constants.INTERFACE_PTP_ROLE_NONE:
ptp_interfaces.append(interface)
ptp_ifaces.append(interface)
if not ptp_interfaces:
if not ptp_ifaces:
raise wsme.exc.ClientSideError(
_("Hosts with PTP clock synchronization must have at least one PTP interface configured"))
@ -6486,8 +6489,8 @@ class HostController(rest.RestController):
address_interfaces = set()
for address in addresses:
address_interfaces.add(address.ifname)
for ptp_interface in ptp_interfaces:
if ptp_interface.ifname not in address_interfaces:
for ptp_if in ptp_ifaces:
if ptp_if.ifname not in address_interfaces:
raise wsme.exc.ClientSideError(
_("All PTP interfaces must have an associated address when PTP transport is UDP"))

View File

@ -0,0 +1,207 @@
########################################################################
#
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
########################################################################
import pecan
from pecan import rest
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
from oslo_log import log
from sysinv.api.controllers.v1 import base
from sysinv.api.controllers.v1 import collection
from sysinv.api.controllers.v1 import link
from sysinv.api.controllers.v1 import types
from sysinv.api.controllers.v1 import utils
from sysinv.common import exception
from sysinv import objects
LOG = log.getLogger(__name__)
class PtpInterfacePatchType(types.JsonPatchType):
@staticmethod
def mandatory_attrs():
return []
class PtpInterface(base.APIBase):
"""API representation of a PTP interface.
This class enforces type checking and value constraints, and converts
between the interna object model and the API representation of a PTP
interface.
"""
uuid = types.uuid
"Unique UUID for this PTP interface"
interface_uuid = types.uuid
"ID for the interface associated with the PTP interface"
ptp_instance_id = int
"ID for the PTP instance this interface is associated with"
links = [link.Link]
"A list containing a self link and associated ptp interface links"
ptp_instance_uuid = types.uuid
"The UUID of the host this PTP interface belongs to"
ifname = wtypes.text
"The name of the underlying interface"
forihostid = int
"The foreign key host id"
ptp_instance_name = wtypes.text
"The name of the associated PTP instance"
created_at = wtypes.datetime.datetime
def __init__(self, **kwargs):
self.fields = list(objects.ptp_interface.fields.keys())
for k in self.fields:
if not hasattr(self, k):
continue
setattr(self, k, kwargs.get(k, wtypes.Unset))
@classmethod
def convert_with_links(cls, rpc_ptp_interface, expand=True):
ptp_interface = PtpInterface(**rpc_ptp_interface.as_dict())
if not expand:
ptp_interface.unset_fields_except(['uuid',
'ptp_instance_id',
'forihostid',
'ptp_instance_name',
'ifname',
'interface_uuid',
'created_at'])
return ptp_interface
class PtpInterfaceCollection(collection.Collection):
"""API representation of a collection of PTP interfaces."""
ptp_interfaces = [PtpInterface]
"A list containing PtpInterface objects"
def __init__(self, **kwargs):
self._type = 'ptp_interfaces'
@classmethod
def convert_with_links(cls, rpc_ptp_interfaces, limit, url=None,
expand=False, **kwargs):
collection = PtpInterfaceCollection()
collection.ptp_interfaces = [PtpInterface.convert_with_links(p, expand)
for p in rpc_ptp_interfaces]
collection.next = collection.get_next(limit, url=url, **kwargs)
return collection
LOCK_NAME = 'PtpInterfaceController'
class PtpInterfaceController(rest.RestController):
"""REST controller for ptp interfaces."""
def __init__(self, from_ihosts=False):
self._from_ihosts = from_ihosts
def _get_ptp_interfaces_collection(self, host_uuid=None, marker=None,
limit=None, sort_key=None,
sort_dir=None, expand=False,
resource_url=None, interface_uuid=None):
limit = utils.validate_limit(limit)
sort_dir = utils.validate_sort_dir(sort_dir)
marker_obj = None
if marker:
marker_obj = objects.ptp_interface.get_by_uuid(pecan.request.context,
marker)
if self._from_ihosts or host_uuid is not None:
if interface_uuid is not None:
ptp_interfaces = pecan.request.dbapi.ptp_interfaces_get_by_interface(
interface_uuid, limit,
marker_obj,
sort_key,
sort_dir)
else:
ptp_interfaces = pecan.request.dbapi.ptp_interfaces_get_by_host(
host_uuid, limit,
marker_obj,
sort_key,
sort_dir)
else:
ptp_interfaces = pecan.request.dbapi.ptp_interfaces_get_list()
return PtpInterfaceCollection.convert_with_links(ptp_interfaces,
limit,
url=resource_url,
expand=expand,
sort_key=sort_key,
sort_dir=sort_dir)
@wsme_pecan.wsexpose(PtpInterfaceCollection, types.uuid, types.uuid, int,
wtypes.text, wtypes.text, types.uuid)
def get_all(self, host_uuid, marker=None, limit=None,
sort_key='id', sort_dir='asc', interface_uuid=None):
"""Retrieve a list of PTP interfaces."""
return self._get_ptp_interfaces_collection(host_uuid, marker, limit,
sort_key=sort_key,
sort_dir=sort_dir,
expand=False,
interface_uuid=interface_uuid)
@wsme_pecan.wsexpose(PtpInterface, types.uuid)
def get_one(self, ptp_interface_uuid):
"""Retrieve information about the given PTP interface"""
rpc_ptp_interface = objects.ptp_interface.get_by_uuid(pecan.request.context,
ptp_interface_uuid)
return PtpInterface.convert_with_links(rpc_ptp_interface)
@wsme_pecan.wsexpose(PtpInterface, body=PtpInterface)
def post(self, ptp_interface):
"""Create a new PTP interface"""
return self._create_ptp_interface(ptp_interface)
def _create_ptp_interface(self, ptp_interface):
# Create a new PTP interface
ptp_interface_dict = ptp_interface.as_dict()
instance_uuid = ptp_interface_dict.pop('ptp_instance_uuid', None)
instance = objects.ptp_instance.get_by_uuid(pecan.request.context,
instance_uuid)
interface_uuid = ptp_interface_dict.pop('interface_uuid', None)
interface = pecan.request.dbapi.iinterface_get(interface_uuid)
ptp_interface_dict['interface_id'] = interface['id']
ptp_interface_dict['ptp_instance_id'] = instance['id']
check = pecan.request.dbapi.ptp_interfaces_get_by_instance_and_interface(
ptp_interface_dict["ptp_instance_id"],
ptp_interface_dict["interface_id"])
if len(check) != 0:
raise exception.PtpInterfaceAlreadyExists()
result = pecan.request.dbapi.ptp_interface_create(ptp_interface_dict)
return PtpInterface.convert_with_links(result)
@wsme_pecan.wsexpose(None, types.uuid, status_code=204)
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)
except exception.PtpInterfaceNotFound:
raise
pecan.request.dbapi.ptp_interface_destroy(ptp_interface.uuid)

View File

@ -2034,6 +2034,23 @@ class Connection(object):
:returns: A list of PTP interface associations.
"""
@abc.abstractmethod
def ptp_interfaces_get_by_host(self, host_uuid, limit=None,
marker=None, sort_key=None,
sort_dir=None):
"""Returns a list of the PTP associations for a given host.
:param host_uuid: The id or uuid of a host.
:param limit: Maximum number of PTP associations 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 associations (instances) for the host
"""
@abc.abstractmethod
def ptp_interfaces_get_by_interface(self, interface_id, limit=None,
marker=None, sort_key=None,
@ -2066,6 +2083,25 @@ class Connection(object):
:returns: A list of PTP associations (interfaces) for the PTP instance.
"""
@abc.abstractmethod
def ptp_interfaces_get_by_instance_and_interface(self, ptp_instance_id,
interface_id,
limit=None,
marker=None,
sort_key=None,
sort_dir=None):
"""Returns a list of one PTP interface for a given instance and interface.
:param ptp_instance_id: The id of a PTP instance.
:param interface_id: The UUID of the underlying interface.
: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 interfaces with the given instance and interface.
"""
@abc.abstractmethod
def ptp_interface_destroy(self, ptp_interface_id):
"""Destroys a PTP interface association.

View File

@ -3815,7 +3815,6 @@ class Connection(api.Connection):
def _ptp_interface_get(self, ptp_interface_id):
query = model_query(models.PtpInterfaces)
query = add_identity_filter(query, ptp_interface_id)
try:
return query.one()
except NoResultFound:
@ -3825,7 +3824,8 @@ class Connection(api.Connection):
def ptp_interface_create(self, values):
if not values.get('uuid'):
values['uuid'] = uuidutils.generate_uuid()
ptp_interface = models.PtpInterfaces(**values)
ptp_interface = models.PtpInterfaces()
ptp_interface.update(values)
with _session_for_write() as session:
try:
session.add(ptp_interface)
@ -3853,6 +3853,18 @@ class Connection(api.Connection):
return _paginate_query(models.PtpInterfaces, limit, marker,
sort_key, sort_dir, query)
@objects.objectify(objects.ptp_interface)
def ptp_interfaces_get_by_host(self, host_uuid, limit=None,
marker=None, sort_key=None,
sort_dir=None):
ihost_obj = self.ihost_get(host_uuid)
query = model_query(models.PtpInterfaces)
query = (query.join(models.Interfaces).
join(models.ihost,
models.ihost.id == models.Interfaces.forihostid))
query, field = add_filter_by_many_identities(query, models.ihost, [ihost_obj.uuid])
return _paginate_query(models.PtpInterfaces, limit, marker, sort_key, sort_dir, query)
@objects.objectify(objects.ptp_interface)
def ptp_interfaces_get_by_interface(self, interface_id, limit=None,
marker=None, sort_key=None,
@ -3865,6 +3877,21 @@ class Connection(api.Connection):
return _paginate_query(models.PtpInterfaces, limit, marker,
sort_key, sort_dir, query)
@objects.objectify(objects.ptp_interface)
def ptp_interfaces_get_by_instance_and_interface(self, ptp_instance_id,
interface_id,
limit=None,
marker=None,
sort_key=None,
sort_dir=None):
ptp_instance_obj = self.ptp_instance_get(ptp_instance_id)
ptp_interface_obj = self.iinterface_get(interface_id)
query = model_query(models.PtpInterfaces)
query = query.filter_by(interface_id=ptp_interface_obj.id)
query = query.filter_by(ptp_instance_id=ptp_instance_obj.id)
return _paginate_query(models.PtpInterfaces, limit, marker,
sort_key, sort_dir, query)
@objects.objectify(objects.ptp_interface)
def ptp_interfaces_get_by_instance(self, ptp_instance_id, limit=None,
marker=None, sort_key=None,

View File

@ -23,14 +23,25 @@ class PtpInterface(base.SysinvObject):
'interface_id': utils.int_or_none,
'ptp_instance_uuid': utils.str_or_none,
'ptp_instance_id': utils.int_or_none
'ptp_instance_id': utils.int_or_none,
'ptp_instance_name': utils.str_or_none,
'ifname': utils.str_or_none,
'forihostid': utils.int_or_none,
}
_foreign_fields = {
'interface_uuid': 'interface:uuid',
'ptp_instance_uuid': 'ptp_instance:uuid'
'interface_id': 'interface:id',
'ptp_instance_uuid': 'ptp_instance:uuid',
'ptp_instance_name': 'ptp_instance:name',
'ifname': 'interface:ifname',
'forihostid': 'interface:forihostid',
'ptp_instance_host': 'ptp_instance:host_id'
}
@base.remotable_classmethod
def get_by_uuid(cls, context, uuid):
return cls.dbapi.ptp_get_interface(uuid)
return cls.dbapi.ptp_interface_get(uuid)

View File

@ -0,0 +1,267 @@
########################################################################
#
# Copyright (c) 2021 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
########################################################################
from oslo_utils import uuidutils
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 BasePtpInterfaceTestCase(base.FunctionalTest, dbbase.BaseHostTestCase):
# Generic header passed in most API calls
API_HEADERS = {'User-Agent': 'sysinv-test'}
# Prefix for the URL
API_PREFIX = '/ptp_interfaces'
# Python table key for the list of results
RESULT_KEY = 'ptp_interfaces'
# Field that is known to exist for inputs and outputs
COMMON_FIELD = 'interface_uuid'
# Can perform API operations on thie object at a sublevel of host
HOST_PREFIX = '/ihosts'
# Attributes that should be populated by an API query
expected_api_fields = ['uuid', 'interface_id', 'ptp_instance_id']
# Attributes that should NOT be populated by an API query
hidden_api_fields = ['host_id']
def setUp(self):
super(BasePtpInterfaceTestCase, self).setUp()
self.controller = self._create_test_host(constants.CONTROLLER)
self.worker = self._create_test_host(constants.WORKER)
def get_single_url(self, ptp_interface_uuid):
return '%s/%s' % (self.API_PREFIX, ptp_interface_uuid)
def get_host_scoped_url(self, host_uuid):
return '%s/%s%s' % (self.HOST_PREFIX, host_uuid, self.API_PREFIX)
def get_host_scoped_url_interface(self, host_uuid, interface_uuid):
return '%s/%s%s?interface_uuid=%s' % (self.HOST_PREFIX,
host_uuid,
self.API_PREFIX,
interface_uuid)
def get_post_object(self, interface_uuid=None, ptp_instance_uuid=None):
ptp_interface_db = {
'interface_uuid': interface_uuid,
'ptp_instance_uuid': ptp_instance_uuid
}
return ptp_interface_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 TestCreatePtpInterface(BasePtpInterfaceTestCase):
def setUp(self):
super(TestCreatePtpInterface, self).setUp()
self.test_interface = dbutils.create_test_interface(
ifname='ptp0',
ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.controller.id,
ihost_uuid=self.controller.uuid)
self.test_instance = dbutils.create_test_ptp_instance(
name='testInstance',
service='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)
response = self.post_json(self.API_PREFIX, ptp_interface_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_interface_db[self.COMMON_FIELD])
def _create_ptp_interface_failed(self, interface_uuid, ptp_instance_uuid,
status_code, error_message):
ptp_interface_db = self.get_post_object(interface_uuid,
ptp_instance_uuid)
response = self.post_json(self.API_PREFIX, ptp_interface_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_interface_ok(self):
self._create_ptp_interface_success(self.test_interface.uuid,
self.test_instance.uuid)
def test_create_ptp_interface_invalid_interface(self):
self._create_ptp_interface_failed(
'32dbb999-6c10-448d-aeca-964c50af6384',
self.test_instance.uuid,
status_code=http_client.BAD_REQUEST,
error_message='No entry found for interface 32dbb999-6c10-448d-aeca-964c50af6384')
def test_create_ptp_interface_invalid_instance(self):
self._create_ptp_interface_failed(
self.test_interface.uuid,
'32dbb999-6c10-448d-aeca-964c50af6384',
status_code=http_client.NOT_FOUND,
error_message='No PTP instance with id 32dbb999-6c10-448d-aeca-964c50af6384 found.')
def test_create_ptp_interface_duplicate(self):
self._create_ptp_interface_success(self.test_interface.uuid,
self.test_instance.uuid)
self._create_ptp_interface_failed(
interface_uuid=self.test_interface.uuid,
ptp_instance_uuid=self.test_instance.uuid,
status_code=http_client.INTERNAL_SERVER_ERROR,
error_message='')
class TestGetPtpInterface(BasePtpInterfaceTestCase):
def setUp(self):
super(TestGetPtpInterface, self).setUp()
self.test_interface = dbutils.create_test_interface(
ifname='ptp0',
ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.controller.id,
ihost_uuid=self.controller.uuid)
self.test_instance = dbutils.create_test_ptp_instance(
name='testInstance',
service='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)
def test_get_ptp_interface_found(self):
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):
fake_uuid = 'f4c56ddf-aef3-46ed-b9aa-126a1faafd40'
error_message = 'No PTP interface 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 TestListPtpInterface(BasePtpInterfaceTestCase):
def setUp(self):
super(TestListPtpInterface, self).setUp()
self.test_interface = dbutils.create_test_interface(
ifname='ptp0',
ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.worker.id,
ihost_uuid=self.worker.uuid)
self.dummy_interface = dbutils.create_test_interface(
ifname='ptp1',
ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.worker.id,
ihost_uuid=self.worker.uuid)
self.test_instance_ptp4l = dbutils.create_test_ptp_instance(
name='ptp4lInstance',
service='ptp4l',
host_id=self.worker.id)
self.test_instance_phc2sys = dbutils.create_test_ptp_instance(
name='phc2sysInstance',
service='phc2sys',
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)
self.phc2sys_ptp_interface = dbutils.create_test_ptp_interface(
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)
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'])
elif result['uuid'] == self.phc2sys_ptp_interface.uuid:
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))
for result in response[self.RESULT_KEY]:
self.assertIn(self.COMMON_FIELD, result)
self.assertNotIn(self.dummy_interface.uuid, result)
def test_list_ptp_interface_empty(self):
response = self.get_json(self.get_host_scoped_url(self.controller.uuid))
self.assertEqual([], response[self.RESULT_KEY])
class TestDeletePtpInterface(BasePtpInterfaceTestCase):
""" 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(TestDeletePtpInterface, self).setUp()
self.test_interface = dbutils.create_test_interface(
ifname='ptp0',
ifclass=constants.INTERFACE_CLASS_PLATFORM,
forihostid=self.worker.id,
ihost_uuid=self.worker.uuid)
self.test_instance_ptp4l = dbutils.create_test_ptp_instance(
name='ptp4lInstance',
service='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)
def test_delete_ptp_interface(self):
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)
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

View File

@ -536,6 +536,45 @@ def create_test_ptp(**kw):
return dbapi.ptp_create(ptp)
# Create test ptp_instance object
def get_test_ptp_instance(**kw):
ptp_instance = {
'id': kw.get('id'),
'uuid': kw.get('uuid'),
'name': kw.get('name'),
'service': kw.get('service'),
'host_id': kw.get('host_id'),
}
return ptp_instance
def create_test_ptp_instance(**kw):
ptp_instance = get_test_ptp_instance(**kw)
# Let DB generate ID if it isn't specified explicitly
if 'id' not in kw:
del ptp_instance['id']
dbapi = db_api.get_instance()
return dbapi.ptp_instance_create(ptp_instance)
# Create test ptp_interface object
def get_test_ptp_interface(**kw):
ptp_interface = {
'uuid': kw.get('uuid'),
'interface_id': kw.get('interface_id'),
'ptp_instance_id': kw.get('ptp_instance_id')
}
return ptp_interface
def create_test_ptp_interface(**kw):
ptp_interface = get_test_ptp_interface(**kw)
if 'uuid' in kw:
del ptp_interface['uuid']
dbapi = db_api.get_instance()
return dbapi.ptp_interface_create(ptp_interface)
# Create test dns object
def get_test_dns(**kw):
dns = {