callback = data.get('callback', None)
if not callback:
raise BadRequestException('The callback parameter is required')
+
result = alarm_view.subscription_create(data, bus.uow)
return result, 201
import uuid as uuid
from o2common.service import unit_of_work
-from o2common.views.pagination_view import Pagination
from o2common.views.view import gen_filter, check_filter
+from o2common.views.pagination_view import Pagination
+from o2common.views.route_exception import BadRequestException
+
from o2ims.views.alarm_dto import SubscriptionDTO
from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
sub_uuid, subscriptionDto['callback'],
consumer_subs_id, filter)
with uow:
+ args = list()
+ args.append(getattr(AlarmSubscription, 'callback')
+ == subscriptionDto['callback'])
+ args.append(getattr(AlarmSubscription, 'filter') == filter)
+ args.append(getattr(AlarmSubscription,
+ 'consumerSubscriptionId') == consumer_subs_id)
+ count, _ = uow.alarm_subscriptions.list_with_count(*args)
+ if count > 0:
+ raise BadRequestException("The value of parameters is duplicated")
uow.alarm_subscriptions.add(subscription)
uow.commit()
first = uow.alarm_subscriptions.get(sub_uuid)
'helmcli_kubeconfig': fields.String(attribute='helmcli_kubeconfig'),
})
+ extensions = api_ims_inventory_v1.model("DeploymentManagerExtensions", {
+ 'profileName': fields.String,
+ 'profileData': fields.Nested(profile, False, True),
+ })
+
deployment_manager_get = api_ims_inventory_v1.model(
"DeploymentManagerGetDto",
{
# 'supportedLocations': fields.String,
# 'capabilities': fields.String,
# 'capacity': fields.String,
- 'profileName': fields.String,
- 'profileData': fields.Nested(profile, False, True),
- 'extensions': fields.String
+ 'extensions': fields.Nested(extensions, True, True)
},
mask='{deploymentManagerId,name,description,oCloudId,serviceUri,' +\
- 'profileName,profileData}'
+ 'extensions/profileName,extensions/profileData}'
)
callback = data.get('callback', None)
if not callback:
raise BadRequestException('The callback parameter is required')
+
result = ocloud_view.subscription_create(data, bus.uow)
return result, 201
from o2common.service import unit_of_work
from o2common.config import config
-from o2common.views.pagination_view import Pagination
from o2common.views.view import gen_filter, check_filter
+from o2common.views.pagination_view import Pagination
+from o2common.views.route_exception import BadRequestException
+
from o2ims.domain import ocloud
from o2ims.views.ocloud_dto import SubscriptionDTO
from o2ims.domain.subscription_obj import Subscription
return None
profile_data = result.pop("profile", None)
- result['profileName'] = profile
profiles = config.get_dms_support_profiles()
if profile not in profiles:
return ""
+ extensions = {
+ 'profileName': profile
+ }
if ocloud.DeploymentManagerProfileDefault == profile \
or ocloud.DeploymentManagerProfileSOL018 == profile:
result['serviceUri'] = \
profile_data['cluster_api_endpoint']
- result['profileData'] = profile_data
+ extensions['profileData'] = profile_data
elif ocloud.DeploymentManagerProfileSOL018HelmCLI == profile:
result['serviceUri'] = \
profile_data['cluster_api_endpoint']
config.get_helmcli_access()
helmcli_profile["helmcli_kubeconfig"] = _gen_kube_config(
deploymentManagerId, profile_data)
- result['profileData'] = helmcli_profile
+ extensions['profileData'] = helmcli_profile
else:
return ""
+ result['extensions'] = extensions
return result
sub_uuid, subscriptionDto['callback'],
consumer_subs_id, filter)
with uow:
+ args = list()
+ args.append(getattr(Subscription, 'callback')
+ == subscriptionDto['callback'])
+ args.append(getattr(Subscription, 'filter') == filter)
+ args.append(getattr(Subscription,
+ 'consumerSubscriptionId') == consumer_subs_id)
+ count, _ = uow.alarm_subscriptions.list_with_count(*args)
+ if count > 0:
+ raise BadRequestException("The value of parameters is duplicated")
uow.subscriptions.add(subscription)
uow.commit()
first = uow.subscriptions.get(sub_uuid)
assert resp.status_code == 404
-def test_flask_post(mock_flask_uow):
+def test_flask_post(mock_flask_uow, mappers):
session, app = mock_flask_uow
apibase = config.get_o2ims_monitoring_api_base() + '/v1'
+ order_by = MagicMock()
+ order_by.count.return_value = 0
+ order_by.limit.return_value.offset.return_value = []
+ session.return_value.query.return_value.filter.return_value.\
+ order_by.return_value = order_by
+
with app.test_client() as client:
session.return_value.execute.return_value = []
resp = client.post(apibase+'/alarmSubscriptions', json={
'callback': sub_callback,
'consumerSubscriptionId': 'consumerSubId1',
- 'filter': ''
+ 'filter': '(eq,resourceTypeId,xxx)'
})
assert resp.status_code == 201
assert 'alarmSubscriptionId' in resp.get_json()
deployment_manager_id1, uow, profile=profileName)
assert str(deployment_manager_res.get(
'serviceUri')) == cluster_endpoint
- assert str(deployment_manager_res.get(
+ assert str(deployment_manager_res.get("extensions").get(
"profileName")) == profileName
# profile wrong name
assert resp.status_code == 404
-def test_flask_post(mock_flask_uow):
+def test_flask_post(mock_flask_uow, mappers):
session, app = mock_flask_uow
apibase = config.get_o2ims_api_base() + '/v1'
+ order_by = MagicMock()
+ order_by.count.return_value = 0
+ order_by.limit.return_value.offset.return_value = []
+ session.return_value.query.return_value.filter.return_value.\
+ order_by.return_value = order_by
+
with app.test_client() as client:
session.return_value.execute.return_value = []
resp = client.post(apibase+'/subscriptions', json={
'callback': sub_callback,
'consumerSubscriptionId': 'consumerSubId1',
- 'filter': ''
+ 'filter': '(eq,resourceTypeId,xxx)'
})
assert resp.status_code == 201
assert 'subscriptionId' in resp.get_json()