from catalog.packages.biz.nsdm_subscription import NsdmSubscription
from catalog.pub.database.models import NsdmSubscriptionModel
-from catalog.packages.biz.notificationsutil import NotificationsUtil, prepare_nsd_notification, prepare_pnfd_notification
+from catalog.packages.biz.notificationsutil import NsdNotifications, PnfNotifications
from catalog.packages import const
from catalog.pub.config import config as pub_config
import catalog.pub.utils.timeutil
def test_nsdm_get_subscription_failure_bad_request(self):
response = self.client.get("/api/nsd/v1/subscriptions/123",
format='json')
- self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
@mock.patch.object(NsdmSubscription, 'query_single_subscription')
def test_nsdmsubscription_getsingle_when_catch_exception(
def test_nsdm_delete_subscription_failure_bad_request(self):
response = self.client.delete("/api/nsd/v1/subscriptions/123",
format='json')
- self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+ self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
@mock.patch.object(NsdmSubscription, 'delete_single_subscription')
def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
'timeStamp': "nowtime()",
'nsdInfoId': "d0ea5ec3-0b98-438a-9bea-488230cff174",
'nsdId': "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
- 'onboardingFailureDetails': None,
- 'nsdOperationalState': None,
"subscriptionId": "1111",
'_links': {
'subscription': {
def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
mock_nowtime.return_value = "nowtime()"
mock_uuid.return_value = "1111"
- notification_content = prepare_nsd_notification("nsdinfoid1", "nsdid1",
- const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
- "NSD(nsdid1) already exists.", operational_state=None)
- filters = {
- 'nsdInfoId': 'nsdInfoId',
- 'nsdId': 'nsdId',
- }
- NotificationsUtil().send_notification(notification_content, filters, False)
+ notify = NsdNotifications(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
+ nsd_info_id="nsdinfoid1",
+ nsd_id="nsdid1",
+ failure_details="NSD(nsdid1) already exists.", operational_state=None)
+ notify.send_notification()
expect_callbackuri = "http://127.0.0.1/self"
expect_notification = {
'id': "1111",
'nsdInfoId': "nsdinfoid1",
'nsdId': "nsdid1",
'onboardingFailureDetails': "NSD(nsdid1) already exists.",
- 'nsdOperationalState': None,
"subscriptionId": "1",
'_links': {
'subscription': {
def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
mock_nowtime.return_value = "nowtime()"
mock_uuid.return_value = "1111"
- notification_content = prepare_pnfd_notification("pnfdInfoIds1", 'pnfdId1',
- const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING)
- filters = {
- 'pnfdId': 'pnfdId',
- 'pnfdInfoIds': 'pnfdInfoIds',
- }
- NotificationsUtil().send_notification(notification_content, filters, False)
+ notify = PnfNotifications(const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
+ pnfd_info_id="pnfdInfoIds1",
+ pnfd_id='pnfdId1',
+ failure_details=None)
+ notify.send_notification()
expect_callbackuri = "http://127.0.0.1/self"
expect_notification = {
'id': "1111",
'timeStamp': "nowtime()",
'pnfdInfoIds': "pnfdInfoIds1",
'pnfdId': "pnfdId1",
- 'onboardingFailureDetails': None,
"subscriptionId": "1",
'_links': {
'subscription': {