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)
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
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()
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()