ocloud_global_id = 4e24b97c-8c49-4c4f-b53e-3de5235a4e37
smo_register_url = http://127.0.0.1:8090/register
smo_token_data = smo_token_payload
+
+
[API]
test = "hello"
from o2common.config import config
from o2common.service.unit_of_work import AbstractUnitOfWork
-from o2ims.adapter import ocloud_repository, alarm_repository, alarm_loader
+from o2ims.adapter import ocloud_repository, alarm_repository
from o2dms.adapter import dms_repository
from o2common.helper import o2logging
self.alarm_probable_causes = alarm_repository\
.AlarmProbableCauseSqlAlchemyRepository(self.session)
- # config file
- self.alarm_dictionaries = alarm_loader\
- .AlarmDictionaryConfigFileRepository()
return super().__enter__()
def __exit__(self, *args):
def get_o2ims_api_base():
- return get_root_api_base() + 'o2ims-infrastructureInventory/v1'
+ return get_root_api_base() + 'o2ims-infrastructureInventory'
def get_o2ims_monitoring_api_base():
- return get_root_api_base() + 'o2ims-infrastructureMonitoring/v1'
+ return get_root_api_base() + 'o2ims-infrastructureMonitoring'
def get_o2dms_api_base():
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+
+from o2common.config import config, conf
+from o2ims.adapter import alarm_loader
+from o2ims.adapter.clients.alarm_dict_client import \
+ load_alarm_dictionary_from_conf_file
+
+# config file
+conf.alarm_dictionaries = alarm_loader\
+ .AlarmDictionaryConfigFileRepository()
+load_alarm_dictionary_from_conf_file(config.get_alarm_yaml_filename())
import uuid as uuid_gen
from o2common.service import unit_of_work
-from o2common.config import config
+from o2common.config import config, conf
from o2ims.domain import alarm_obj as alarm
from o2common.helper import o2logging
logger = o2logging.get_logger(__name__)
-def load_alarm_dictionary_from_conf_file(conf_path: str,
- uow: unit_of_work.AbstractUnitOfWork):
+def load_alarm_dictionary_from_conf_file(conf_path: str):
logger.info("Converting alarm.yaml to dict: ")
raise RuntimeError(exp)
for dictionary in list(dictionaries.keys()):
- with uow:
- # res_type = uow.resource_types.get_by_name(dictionary)
- # logger.info('res_type: ' + res_type.resourceTypeName)
- alarm_dict = alarm.AlarmDictionary(dictionary)
- alarm_dict.entityType = dictionary
- alarm_dict.alarmDictionaryVersion = \
- dictionaries[dictionary]['version']
- alarm_dict.alarmDefinition = \
- dictionaries[dictionary]['alarmDefinition']
- uow.alarm_dictionaries.add(alarm_dict)
+ # res_type = uow.resource_types.get_by_name(dictionary)
+ # logger.info('res_type: ' + res_type.resourceTypeName)
+ alarm_dict = alarm.AlarmDictionary(dictionary)
+ alarm_dict.entityType = dictionary
+ alarm_dict.alarmDictionaryVersion = \
+ dictionaries[dictionary]['version']
+ alarm_dict.alarmDefinition = \
+ dictionaries[dictionary]['alarmDefinition']
+ conf.alarm_dictionaries.add(alarm_dict)
def prettyDict(dict):
from o2ims.domain import ocloud as ocloudModel
from o2ims.domain import subscription_obj as subModel
from o2ims.domain import alarm_obj as alarmModel
-from o2ims.domain.resource_type import ResourceTypeEnum
+from o2ims.domain.resource_type import ResourceTypeEnum, ResourceKindEnum
# from o2ims.domain.alarm_obj import AlarmLastChangeEnum, PerceivedSeverityEnum
from o2common.helper import o2logging
Column("version_number", Integer),
Column("oCloudId", String(255), primary_key=True),
- Column("globalcloudId", String(255)),
+ Column("globalCloudId", String(255)),
Column("name", String(255)),
Column("description", String(255)),
- Column("serviceUri", String(255))
+ Column("serviceUri", String(255)),
+ Column("smoRegistrationService", String(255))
# Column("extensions", String(1024))
)
resourcetype = Table(
- "resourcetype",
+ "resourceType",
metadata,
Column("updatetime", DateTime),
Column("createtime", DateTime),
Column("hash", String(255)),
+ Column("version_number", Integer),
Column("resourceTypeId", String(255), primary_key=True),
Column("resourceTypeEnum", Enum(
ResourceTypeEnum, native_enum=False), nullable=False),
- Column("oCloudId", ForeignKey("ocloud.oCloudId")),
Column("name", String(255)),
+ Column("description", String(255)),
Column("vendor", String(255)),
Column("model", String(255)),
Column("version", String(255)),
- Column("description", String(255)),
+ Column("resourceKind", Enum(ResourceKindEnum)),
+ Column("resourceClass", Enum(ResourceTypeEnum)),
# Column("extensions", String(1024))
+
+ Column("oCloudId", ForeignKey("ocloud.oCloudId")),
)
resourcepool = Table(
- "resourcepool",
+ "resourcePool",
metadata,
Column("updatetime", DateTime),
Column("createtime", DateTime),
Column("version_number", Integer),
Column("resourceId", String(255), primary_key=True),
- Column("resourceTypeId", ForeignKey("resourcetype.resourceTypeId")),
- Column("resourcePoolId", ForeignKey("resourcepool.resourcePoolId")),
+ Column("resourceTypeId", ForeignKey("resourceType.resourceTypeId")),
+ Column("resourcePoolId", ForeignKey("resourcePool.resourcePoolId")),
Column("name", String(255)),
- # Column("globalAssetId", String(255)),
+ Column("globalAssetId", String(255)),
Column("parentId", String(255)),
Column("description", String(255)),
Column("elements", Text())
)
deploymentmanager = Table(
- "deploymentmanager",
+ "deploymentManager",
metadata,
Column("updatetime", DateTime),
Column("createtime", DateTime),
Column("oCloudId", ForeignKey("ocloud.oCloudId")),
Column("name", String(255)),
Column("description", String(255)),
- Column("deploymentManagementServiceEndpoint", String(255)),
+ Column("serviceUri", String(255)),
Column("supportedLocations", String(255)),
Column("capabilities", String(255)),
Column("capacity", String(255)),
Column("hash", String(255)),
Column("alarmEventRecordId", String(255), primary_key=True),
- Column("resourceTypeId", ForeignKey("resourcetype.resourceTypeId")),
+ Column("resourceTypeId", ForeignKey("resourceType.resourceTypeId")),
Column("resourceId", ForeignKey("resource.resourceId")),
Column("alarmDefinitionId", ForeignKey(
"alarmDefinition.alarmDefinitionId")),
import json
from o2common.domain.base import AgRoot, Serializer
+from o2common.config import conf as CONF
# from dataclasses import dataclass
# from datetime import date
# from typing import Optional, List, Set
-from .resource_type import ResourceTypeEnum
+from .resource_type import ResourceKindEnum, ResourceTypeEnum
# from uuid import UUID
capacity: str = '', profile: str = '') -> None:
super().__init__()
self.deploymentManagerId = id
- self.version_number = 0
- self.oCloudId = ocloudid
self.name = name
self.description = description
- self.deploymentManagementServiceEndpoint = dmsendpoint
+ self.oCloudId = ocloudid
+ self.serviceUri = dmsendpoint
self.supportedLocations = supportedLocations
self.capabilities = capabilities
self.capacity = capacity
self.profile = profile
self.extensions = []
+ self.version_number = 0
+
def serialize(self):
d = Serializer.serialize(self)
description: str = '') -> None:
super().__init__()
self.resourcePoolId = id
- self.version_number = 0
- self.oCloudId = ocloudid
self.globalLocationId = gLocationId
self.name = name
- self.location = location
self.description = description
+ self.oCloudId = ocloudid
+ self.location = location
+ self.resources = ''
self.extensions = []
+ self.version_number = 0
+
class ResourceType(AgRoot, Serializer):
def __init__(self, typeid: str, name: str, typeEnum: ResourceTypeEnum,
description: str = '') -> None:
super().__init__()
self.resourceTypeId = typeid
- self.version_number = 0
- self.oCloudId = ocloudid
self.resourceTypeEnum = typeEnum
self.name = name
+ self.description = description
self.vender = vender
self.model = model
self.version = version
- self.description = description
+ self.alarmDictionary = {}
+ self.resourceKind = ResourceKindEnum.UNDEFINED
+ self.resourceClass = ResourceTypeEnum.UNDEFINED
self.extensions = []
+ self.version_number = 0
+
+ def serialize(self):
+ d = Serializer.serialize(self)
+
+ d["alarmDictionary"] = CONF.alarm_dictionaries.get(
+ d['name']).serialize()
+ return d
+
class Resource(AgRoot, Serializer):
def __init__(self, resourceId: str, resourceTypeId: str,
description: str = '') -> None:
super().__init__()
self.resourceId = resourceId
- self.version_number = 0
+ self.description = description
self.resourceTypeId = resourceTypeId
+ self.globalAssetId = gAssetId
self.resourcePoolId = resourcePoolId
+ self.elements = elements
+ self.extensions = []
+
self.name = name
- self.globalAssetId = gAssetId
self.parentId = parentId
- self.elements = elements
- self.description = description
self.children = []
- self.extensions = []
+
+ self.version_number = 0
def set_children(self, children: list):
self.children = children
description: str = '', version_number: int = 0) -> None:
super().__init__()
self.oCloudId = ocloudid
- self.globalcloudId = globalcloudId
- self.version_number = version_number
+ self.globalCloudId = globalcloudId
self.name = name
self.description = description
self.serviceUri = imsendpoint
+ self.resourceTypes = []
self.resourcePools = []
self.deploymentManagers = []
- self.resourceTypes = []
+ self.smoRegistrationService = ''
self.extensions = []
+ self.version_number = version_number
+
# def addDeploymentManager(self,
# deploymentManager: DeploymentManager):
class ResourceTypeEnum(Enum):
+ UNDEFINED = 0
OCLOUD = 1
RESOURCE_POOL = 2
DMS = 3
PSERVER_ETH = 16
+class ResourceKindEnum(Enum):
+ UNDEFINED = 0
+ PHYSICAL = 1
+ LOGICAL = 2
+
+
class InvalidOcloudState(Exception):
pass
filter: str = '') -> None:
super().__init__()
self.subscriptionId = id
- self.version_number = 0
self.callback = callback
self.consumerSubscriptionId = consumersubid
self.filter = filter
+ self.version_number = 0
+
class NotificationEventEnum(str, Enum):
CREATE = 'CREATE'
res = uow.session.execute(
'''
SELECT "resourceTypeId", "oCloudId", "name"
- FROM resourcetype
+ FROM "resourceType"
WHERE "resourceTypeEnum" = :resource_type_enum
''',
dict(resource_type_enum=stxobj.type.name)
res = uow.session.execute(
'''
SELECT "resourceTypeId", "oCloudId", "name"
- FROM resourcetype
+ FROM "resourceType"
WHERE "resourceTypeEnum" = :resource_type_enum
''',
dict(resource_type_enum=stxobj.type.name)
res = uow.session.execute(
'''
SELECT "resourceTypeId", "oCloudId", "name"
- FROM resourcetype
+ FROM "resourceType"
WHERE "resourceTypeEnum" = :resource_type_enum
''',
dict(resource_type_enum=stxobj.type.name)
res = uow.session.execute(
'''
SELECT "resourceTypeId", "oCloudId", "name"
- FROM resourcetype
+ FROM "resourceType"
WHERE "resourceTypeEnum" = :resource_type_enum
''',
dict(resource_type_enum=stxobj.type.name)
res = uow.session.execute(
'''
SELECT "resourceTypeId", "oCloudId", "name"
- FROM resourcetype
+ FROM "resourceType"
WHERE "resourceTypeEnum" = :resource_type_enum
''',
dict(resource_type_enum=stxobj.type.name)
res = uow.session.execute(
'''
SELECT "resourceTypeId", "oCloudId", "name"
- FROM resourcetype
+ FROM "resourceType"
WHERE "resourceTypeEnum" = :resource_type_enum
''',
dict(resource_type_enum=stxobj.type.name)
ocloud_route.configure_api_route()
alarm_route.configure_api_route()
- app.add_namespace(api_ns.api_ims_inventory_v1, path=apiims)
- app.add_namespace(api_ns.api_monitoring_v1, path=apimonitoring)
+ app.add_namespace(api_ns.api_ims_inventory, path=apiims)
+ app.add_namespace(api_ns.api_ims_monitoring, path=apimonitoring)
from flask_restx import fields
-from o2ims.views.api_ns import api_monitoring_v1
+from o2ims.views.api_ns import api_ims_monitoring as api_monitoring_v1
class AlarmDTO:
from o2common.service.messagebus import MessageBus
from o2common.views.pagination_route import link_header, PAGE_PARAM
from o2ims.views import alarm_view
-from o2ims.views.api_ns import api_monitoring_v1
+from o2ims.views.api_ns import api_ims_monitoring as api_monitoring_v1
from o2ims.views.alarm_dto import AlarmDTO, SubscriptionDTO
from o2common.helper import o2logging
bus = MessageBus.get_instance()
+# ---------- API versions ---------- #
+@api_monitoring_v1.route("/v1/api_versions")
+class VersionRouter(Resource):
+ def get(self):
+ return {
+ 'uriPrefix': request.base_url.rsplit('/', 1)[0],
+ 'apiVersions': [{
+ 'version': '1',
+ # 'isDeprecated': 'False',
+ # 'retirementDate': ''
+ }]
+ }
+
+
# ---------- Alarm Event Record ---------- #
-@api_monitoring_v1.route("/alarms")
+@api_monitoring_v1.route("/v1/alarms")
@api_monitoring_v1.param(PAGE_PARAM,
'Page number of the results to fetch.' +
' Default: 1',
return link_header(request.full_path, ret)
-@api_monitoring_v1.route("/alarms/<alarmEventRecordId>")
+@api_monitoring_v1.route("/
v1/alarms/<alarmEventRecordId>")
@api_monitoring_v1.param('alarmEventRecordId', 'ID of the alarm event record')
@api_monitoring_v1.response(404, 'Alarm Event Record not found')
@api_monitoring_v1.param(
# ---------- Alarm Subscriptions ---------- #
-@api_monitoring_v1.route("/alarmSubscriptions")
+@api_monitoring_v1.route("/v1/alarmSubscriptions")
class SubscriptionsListRouter(Resource):
model = SubscriptionDTO.subscription_get
return result, 201
-@api_monitoring_v1.route("/alarmSubscriptions/<alarmSubscriptionID>")
+@api_monitoring_v1.route("/
v1/alarmSubscriptions/<alarmSubscriptionID>")
@api_monitoring_v1.param('alarmSubscriptionID', 'ID of the Alarm Subscription')
@api_monitoring_v1.response(404, 'Alarm Subscription not found')
class SubscriptionGetDelRouter(Resource):
# See the License for the specific language governing permissions and
# limitations under the License.
+from flask import request
+from flask_restx import Resource
from o2common.views.route import O2Namespace
-api_ims_inventory_v1 = O2Namespace(
+api_ims_inventory = O2Namespace(
"O2IMS_Inventory",
description='IMS Inventory related operations.')
"PROVISION",
description='Provision related operations.')
-api_monitoring_v1 = O2Namespace(
+api_ims_monitoring = O2Namespace(
"O2IMS_InfrastructureMonitoring",
description='O2 IMS Monitoring related operations.')
+
+
+@api_ims_inventory.route('/api_versions')
+class InventoryVersion(Resource):
+ def get():
+ return {
+ 'uriPrefix': request.base_url.rsplit('/', 1)[0],
+ 'apiVersions': [{
+ 'version': '1',
+ # 'isDeprecated': 'False',
+ # 'retirementDate': ''
+ }]
+ }
+
+
+@api_ims_monitoring.route('/api_versions')
+class MonitoringVersion(Resource):
+ def get():
+ return {
+ 'uriPrefix': request.base_url.rsplit('/', 1)[0],
+ 'apiVersions': [{
+ 'version': '1',
+ # 'isDeprecated': 'False',
+ # 'retirementDate': ''
+ }]
+ }
from flask_restx import fields
-from o2ims.views.api_ns import api_ims_inventory_v1
+from o2ims.views.api_ns import api_ims_inventory as api_ims_inventory_v1
class OcloudDTO:
"OcloudDto",
{
'oCloudId': fields.String(required=True),
- 'globalCloudId': fields.String,
+ 'globalcloudId': fields.String(attribute='globalCloudId'),
'name': fields.String,
'description': fields.String,
- 'infrastructureManagementServiceEndpoint': fields.String,
+ 'serviceUri': fields.String(attribute='serviceUri'),
+ # 'infrastructureManagementServiceEndpoint': fields.String(
+ # attribute='serviceUri'),
+ # 'infrastructureMangementServiceEndPoint': fields.String(
+ # attribute='serviceUri'),
+ # 'resourceTypes': fields.String,
+ # 'resourcePools': fields.String,
+ # 'deploymentManagers': fields.String,
+ # 'smoRegistrationService': fields.String
+ 'extensions': fields.String
}
)
class ResourceTypeDTO:
+ alarm_dictionary = api_ims_inventory_v1.model(
+ "AlarmDictionaryDto",
+ {
+ 'id': fields.String,
+ 'alarmDictionaryVersion': fields.String,
+ 'alarmDictionarySchemVersion': fields.String,
+ 'entityType': fields.String,
+ 'vendor': fields.String,
+ 'managementInterfaceId': fields.String,
+ 'pkNotificationField': fields.String,
+ 'alarmDefinition': fields.String,
+ }
+ )
resource_type_get = api_ims_inventory_v1.model(
"ResourceTypeGetDto",
'resourceTypeId': fields.String(required=True,
description='Resource type ID'),
'name': fields.String,
+ 'description': fields.String,
'vendor': fields.String,
+ 'model': fields.String,
'version': fields.String,
- 'description': fields.String,
+ 'alarmDictionary': fields.Nested(alarm_dictionary, False, True),
+ # 'resourceKind': fields.String,
+ # 'resourceClass': fields.String,
+ 'extensions': fields.String
}
)
{
'resourcePoolId': fields.String(required=True,
description='Resource pool ID'),
- 'name': fields.String,
'globalLocationId': fields.String,
- 'location': fields.String,
+ 'name': fields.String,
'description': fields.String,
+ 'oCloudId': fields.String,
+ 'location': fields.String,
+ # 'resources': fields.String,
+ 'extensions': fields.String
}
)
description='Resource ID'),
'resourceTypeId': fields.String,
'resourcePoolId': fields.String,
- 'name': fields.String,
+ 'globalAssetId': fields.String,
+ # 'name': fields.String,
'parentId': fields.String,
'description': fields.String,
+ # 'elements': fields.String,
+ 'extensions': fields.String
}
)
description='Resource ID'),
'resourceTypeId': fields.String,
'resourcePoolId': fields.String,
- 'name': fields.String,
+ 'globalAssetId': fields.String,
+ # 'name': fields.String,
'parentId': fields.String,
'description': fields.String,
- 'elements': fields.String,
+ # 'elements': fields.String,
+ 'extensions': fields.String
}
if iteration_number:
- resource_json_mapping['children'] = fields.List(
+ resource_json_mapping['elements'] = fields.List(
fields.Nested(ResourceDTO.recursive_resource_mapping(
- iteration_number-1)))
+ iteration_number-1)), attribute='children')
return api_ims_inventory_v1.model(
'ResourceGetDto' + str(iteration_number), resource_json_mapping)
- # def _recursive_resource_mapping(self, iteration_number=2):
- # resource_json_mapping = {
- # 'resourceId': fields.String(required=True,
- # description='Resource ID'),
- # 'resourceTypeId': fields.String,
- # 'resourcePoolId': fields.String,
- # 'name': fields.String,
- # 'parentId': fields.String,
- # 'description': fields.String,
- # }
- # if iteration_number:
- # resource_json_mapping['children'] = fields.List(
- # fields.Nested(self._recursive_resource_mapping(
- # iteration_number-1)))
- # # print(type(resource_json_mapping['children']))
- # if resource_json_mapping['children'] is None:
- # del resource_json_mapping['children']
- # return resource_json_mapping
-
- # def get_resource_get(self):
- # return api_ims_inventory_v1.model(
- # 'ResourceGetDto',
- # {
- # 'resourceId': fields.String(required=True,
- # description='Resource ID'),
- # 'resourceTypeId': fields.String,
- # 'resourcePoolId': fields.String,
- # 'name': fields.String,
- # 'parentId': fields.String,
- # 'description': fields.String,
- # 'children': fields.List(fields.Nested(
- # self._recursive_resource_mapping()))
- # }
- # )
-
class DeploymentManagerDTO:
description='Deployment manager ID'),
'name': fields.String,
'description': fields.String,
- 'deploymentManagementServiceEndpoint': fields.String,
- 'supportedLocations': fields.String,
- 'capabilities': fields.String,
- 'capacity': fields.String,
+ 'serviceUri': fields.String(attribute='serviceUri'),
+ # 'deploymentManagementServiceEndpoint': fields.String(
+ # attribute='serviceUri'),
+ # 'supportedLocations': fields.String,
+ # 'capabilities': fields.String,
+ # 'capacity': fields.String,
'profileSupportList': fields.List(
fields.String,
description='Profile support list, use default for the return \
endpoint'),
+ 'extensions': fields.String
}
)
profile = api_ims_inventory_v1.model("DeploymentManagerGetDtoProfile", {
'cluster_api_endpoint': fields.String(
- attributes='cluster_api_endpoint'),
- 'cluster_ca_cert': fields.String(attributes='cluster_ca_cert'),
- 'admin_user': fields.String(attributes='admin_user'),
- 'admin_client_cert': fields.String(attributes='admin_client_cert'),
- 'admin_client_key': fields.String(attributes='admin_client_key'),
- # 'kube_config_file': fields.String(attributes='kube_config_file')
+ attribute='cluster_api_endpoint'),
+ 'cluster_ca_cert': fields.String(attribute='cluster_ca_cert'),
+ 'admin_user': fields.String(attribute='admin_user'),
+ 'admin_client_cert': fields.String(attribute='admin_client_cert'),
+ 'admin_client_key': fields.String(attribute='admin_client_key'),
+ # 'kube_config_file': fields.String(attribute='kube_config_file')
'helmcli_host_with_port': fields.String(
- attributes='helmcli_host_with_port'),
- 'helmcli_username': fields.String(attributes='helmcli_username'),
- 'helmcli_password': fields.String(attributes='helmcli_password'),
- 'helmcli_kubeconfig': fields.String(attributes='helmcli_kubeconfig'),
+ attribute='helmcli_host_with_port'),
+ 'helmcli_username': fields.String(attribute='helmcli_username'),
+ 'helmcli_password': fields.String(attribute='helmcli_password'),
+ 'helmcli_kubeconfig': fields.String(attribute='helmcli_kubeconfig'),
})
deployment_manager_get = api_ims_inventory_v1.model(
description='Deployment manager ID'),
'name': fields.String,
'description': fields.String,
- 'deploymentManagementServiceEndpoint': fields.String,
- 'supportedLocations': fields.String,
- 'capabilities': fields.String,
- 'capacity': fields.String,
+ 'serviceUri': fields.String(attribute='serviceUri'),
+ # 'deploymentManagementServiceEndpoint': fields.String(
+ # attribute='serviceUri'),
+ # 'supportedLocations': fields.String,
+ # 'capabilities': fields.String,
+ # 'capacity': fields.String,
'profileName': fields.String,
'profileData': fields.Nested(profile, False, True),
+ 'extensions': fields.String
}
)
from o2common.service.messagebus import MessageBus
from o2common.views.pagination_route import link_header, PAGE_PARAM
from o2ims.views import ocloud_view
-from o2ims.views.api_ns import api_ims_inventory_v1
+from o2ims.views.api_ns import api_ims_inventory as api_ims_inventory_v1
from o2ims.views.ocloud_dto import OcloudDTO, ResourceTypeDTO,\
ResourcePoolDTO, ResourceDTO, DeploymentManagerDTO, SubscriptionDTO
bus = MessageBus.get_instance()
+# ---------- API versions ---------- #
+@api_ims_inventory_v1.route("/v1/api_versions")
+class VersionRouter(Resource):
+ def get(self):
+ return {
+ 'uriPrefix': request.base_url.rsplit('/', 1)[0],
+ 'apiVersions': [{
+ 'version': '1',
+ # 'isDeprecated': 'False',
+ # 'retirementDate': ''
+ }]
+ }
+
+
# ---------- OClouds ---------- #
-@api_ims_inventory_v1.route("/")
+@api_ims_inventory_v1.route("/v1/")
@api_ims_inventory_v1.response(404, 'oCloud not found')
@api_ims_inventory_v1.param(
'all_fields',
# ---------- ResourceTypes ---------- #
-@api_ims_inventory_v1.route("/resourceTypes")
+@api_ims_inventory_v1.route("/v1/resourceTypes")
@api_ims_inventory_v1.param(PAGE_PARAM,
'Page number of the results to fetch.' +
' Default: 1',
return link_header(request.full_path, ret)
-@api_ims_inventory_v1.route("/resourceTypes/<resourceTypeID>")
+@api_ims_inventory_v1.route("/v1/resourceTypes/<resourceTypeID>")
@api_ims_inventory_v1.param('resourceTypeID', 'ID of the resource type')
@api_ims_inventory_v1.response(404, 'Resource type not found')
@api_ims_inventory_v1.param(
# ---------- ResourcePools ---------- #
-@api_ims_inventory_v1.route("/resourcePools")
+@api_ims_inventory_v1.route("/v1/resourcePools")
@api_ims_inventory_v1.param(PAGE_PARAM,
'Page number of the results to fetch.' +
' Default: 1',
return link_header(request.full_path, ret)
-@api_ims_inventory_v1.route("/resourcePools/<resourcePoolID>")
+@api_ims_inventory_v1.route("/v1/resourcePools/<resourcePoolID>")
@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool')
@api_ims_inventory_v1.response(404, 'Resource pool not found')
@api_ims_inventory_v1.param(
# ---------- Resources ---------- #
-@api_ims_inventory_v1.route("/resourcePools/<resourcePoolID>/resources")
+@api_ims_inventory_v1.route("/v1/resourcePools/<resourcePoolID>/resources")
@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool')
# @api_ims_inventory_v1.param('sort', 'sort by column name',
# _in='query')
@api_ims_inventory_v1.route(
- "/resourcePools/<resourcePoolID>/resources/<resourceID>")
+ "/v1/resourcePools/<resourcePoolID>/resources/<resourceID>")
@api_ims_inventory_v1.param('resourcePoolID', 'ID of the resource pool')
@api_ims_inventory_v1.param('resourceID', 'ID of the resource')
@api_ims_inventory_v1.response(404, 'Resource not found')
# ---------- DeploymentManagers ---------- #
-@api_ims_inventory_v1.route("/deploymentManagers")
+@api_ims_inventory_v1.route("/v1/deploymentManagers")
@api_ims_inventory_v1.param(PAGE_PARAM,
'Page number of the results to fetch.' +
' Default: 1',
return link_header(request.full_path, ret)
-@api_ims_inventory_v1.route("/deploymentManagers/<deploymentManagerID>")
+@api_ims_inventory_v1.route("/v1/deploymentManagers/<deploymentManagerID>")
@api_ims_inventory_v1.param('deploymentManagerID',
'ID of the deployment manager')
@api_ims_inventory_v1.param('profile', 'DMS profile: value supports "sol018"',
# ---------- Subscriptions ---------- #
-@api_ims_inventory_v1.route("/subscriptions")
+@api_ims_inventory_v1.route("/v1/subscriptions")
class SubscriptionsListRouter(Resource):
model = SubscriptionDTO.subscription_get
return result, 201
-@api_ims_inventory_v1.route("/subscriptions/<subscriptionID>")
+@api_ims_inventory_v1.route("/v1/subscriptions/<subscriptionID>")
@api_ims_inventory_v1.param('subscriptionID', 'ID of the subscription')
@api_ims_inventory_v1.response(404, 'Subscription not found')
class SubscriptionGetDelRouter(Resource):
if ocloud.DeploymentManagerProfileDefault == profile:
pass
elif ocloud.DeploymentManagerProfileSOL018 == profile:
- result['deploymentManagementServiceEndpoint'] = \
+ result['serviceUri'] = \
profile_data['cluster_api_endpoint']
result['profileData'] = profile_data
elif ocloud.DeploymentManagerProfileSOL018HelmCLI == profile:
- result['deploymentManagementServiceEndpoint'] = \
+ result['serviceUri'] = \
profile_data['cluster_api_endpoint']
helmcli_profile = dict()
"alarmSubscriptionId")) == subscription_id1
-def test_alarm_dictionary(mock_uow):
- session, uow = mock_uow
- alarm_dict1 = alarm_obj.AlarmDictionary('test1')
- alarm_dict1.entityType = 'test1'
- with uow:
- uow.alarm_dictionaries.add(alarm_dict1)
-
- alarm_dict2 = uow.alarm_dictionaries.get('test1')
- assert alarm_dict1 == alarm_dict2
-
- dict_list = uow.alarm_dictionaries.list()
- assert len(dict_list) > 0
-
-
def test_flask_get_list(mock_flask_uow):
session, app = mock_flask_uow
order_by = MagicMock()
order_by.limit.return_value.offset.return_value = []
session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
- apibase = config.get_o2ims_monitoring_api_base()
+ apibase = config.get_o2ims_monitoring_api_base() + '/v1'
with app.test_client() as client:
# Get list and return empty list
session.return_value.query.return_value.filter_by.return_value.\
first.return_value = None
- apibase = config.get_o2ims_monitoring_api_base()
+ apibase = config.get_o2ims_monitoring_api_base() + '/v1'
with app.test_client() as client:
# Get one and return 404
def test_flask_post(mock_flask_uow):
session, app = mock_flask_uow
- apibase = config.get_o2ims_monitoring_api_base()
+ apibase = config.get_o2ims_monitoring_api_base() + '/v1'
with app.test_client() as client:
session.return_value.execute.return_value = []
def test_flask_delete(mock_flask_uow):
session, app = mock_flask_uow
- apibase = config.get_o2ims_monitoring_api_base()
+ apibase = config.get_o2ims_monitoring_api_base() + '/v1'
with app.test_client() as client:
session.return_value.execute.return_value.first.return_value = {}
def test_flask_not_allowed(mock_flask_uow):
_, app = mock_flask_uow
- apibase = config.get_o2ims_monitoring_api_base()
+ apibase = config.get_o2ims_monitoring_api_base() + '/v1'
with app.test_client() as client:
# Testing resource type not support method
session.return_value.query.return_value.filter_by.return_value.first.\
return_value.serialize.return_value = {
"deploymentManagerId": deployment_manager_id1,
- "deploymentManagementServiceEndpoint": dms_endpoint,
+ "serviceUri": dms_endpoint,
"profile": {}
}
assert str(deployment_manager_res.get(
"deploymentManagerId")) == deployment_manager_id1
assert str(deployment_manager_res.get(
- 'deploymentManagementServiceEndpoint')) == dms_endpoint
+ 'serviceUri')) == dms_endpoint
assert deployment_manager_res.get('profile') is None
# profile sol018
deployment_manager_res = ocloud_view.deployment_manager_one(
deployment_manager_id1, uow, profile=profileName)
assert str(deployment_manager_res.get(
- 'deploymentManagementServiceEndpoint')) == cluster_endpoint
+ 'serviceUri')) == cluster_endpoint
assert str(deployment_manager_res.get(
"profileName")) == profileName
order_by.limit.return_value.offset.return_value = []
session.return_value.query.return_value.filter.return_value.\
order_by.return_value = order_by
- apibase = config.get_o2ims_api_base()
+ apibase = config.get_o2ims_api_base() + '/v1'
# TODO: workaround for sqlalchemy not mapping with resource object
setattr(ocloud.Resource, 'resourcePoolId', '')
session.return_value.query.return_value.filter_by.return_value.\
first.return_value = None
- apibase = config.get_o2ims_api_base()
+ apibase = config.get_o2ims_api_base() + '/v1'
with app.test_client() as client:
# Get one and return 404
def test_flask_post(mock_flask_uow):
session, app = mock_flask_uow
- apibase = config.get_o2ims_api_base()
+ apibase = config.get_o2ims_api_base() + '/v1'
with app.test_client() as client:
session.return_value.execute.return_value = []
def test_flask_delete(mock_flask_uow):
session, app = mock_flask_uow
- apibase = config.get_o2ims_api_base()
+ apibase = config.get_o2ims_api_base() + '/v1'
with app.test_client() as client:
session.return_value.execute.return_value.first.return_value = {}
def test_flask_not_allowed(mock_flask_uow):
_, app = mock_flask_uow
- apibase = config.get_o2ims_api_base()
+ apibase = config.get_o2ims_api_base() + '/v1'
with app.test_client() as client:
# Testing resource type not support method
[testenv:code]
setenv =
O2APP_CONFIG=configs/o2app.conf
+ ALARM_YAML=configs/alarm.yaml
commands =
pytest tests/unit
Code Review / pti / o2.git / commitdiff