Change-Id: I5c0133ea163b98c74fb668f2b77aaa95ec1cc3ec
Issue-ID: MODELING-315
Signed-off-by: hongyuzhao <zhao.hongyu@zte.com.cn>
params = auth_info.get("paramsBasic", {})
username = params.get("userName")
password = params.get("password")
params = auth_info.get("paramsBasic", {})
username = params.get("userName")
password = params.get("password")
- resp = requests.post(callbackuri, data=notification, headers={'Connection': 'close'},
- auth=HTTPBasicAuth(username, password))
+ resp = requests.post(callbackuri,
+ data=notification,
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ auth=HTTPBasicAuth(username, password),
+ verify=False)
elif const.OAUTH2_CLIENT_CREDENTIALS in auth_info.get("authType", ''):
# todo
pass
elif const.OAUTH2_CLIENT_CREDENTIALS in auth_info.get("authType", ''):
# todo
pass
- resp = requests.post(callbackuri, data=notification, headers={'Connection': 'close'})
- if resp.status_code != status.HTTP_204_NO_CONTENT:
- logger.error("Sending notification to %s failed: %s" % (callbackuri, resp.text))
+ resp = requests.post(callbackuri,
+ data=notification,
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ verify=False)
+
+ if resp.status_code == status.HTTP_204_NO_CONTENT:
+ logger.error("Sending notification to %s failed: %s" % (callbackuri, resp))
else:
logger.info("Sending notification to %s successfully.", callbackuri)
except:
else:
logger.info("Sending notification to %s successfully.", callbackuri)
except:
logger.debug("Create Subscription --> Test Callback URI --"
"Sending GET request to %s" % self.callback_uri)
try:
logger.debug("Create Subscription --> Test Callback URI --"
"Sending GET request to %s" % self.callback_uri)
try:
- response = requests.get(self.callback_uri, timeout=2)
+ response = requests.get(self.callback_uri, timeout=2, verify=False)
if response.status_code != status.HTTP_204_NO_CONTENT:
raise CatalogException("callbackUri %s returns %s status "
"code." % (self.callback_uri,
if response.status_code != status.HTTP_204_NO_CONTENT:
raise CatalogException("callbackUri %s returns %s status "
"code." % (self.callback_uri,
params = self.authentication.get("paramsBasic", {})
username = params.get("userName")
password = params.get("password")
params = self.authentication.get("paramsBasic", {})
username = params.get("userName")
password = params.get("password")
- response = requests.get(self.callback_uri, auth=HTTPBasicAuth(username, password), timeout=2)
+ response = requests.get(self.callback_uri, auth=HTTPBasicAuth(username, password), timeout=2,
+ verify=False)
elif const.OAUTH2_CLIENT_CREDENTIALS in self.authentication.get("authType", ''):
# todo
pass
elif const.OAUTH2_CLIENT_CREDENTIALS in self.authentication.get("authType", ''):
# todo
pass
- response = requests.get(self.callback_uri, timeout=2)
+ response = requests.get(self.callback_uri, timeout=2, verify=False)
if response.status_code != status.HTTP_204_NO_CONTENT:
raise VnfPkgSubscriptionException(
"callbackUri %s returns %s status code." % (
if response.status_code != status.HTTP_204_NO_CONTENT:
raise VnfPkgSubscriptionException(
"callbackUri %s returns %s status code." % (
class TestNsdmSubscription(TestCase):
class TestNsdmSubscription(TestCase):
def setUp(self):
self.client = APIClient()
NsdmSubscriptionModel.objects.all().delete()
def setUp(self):
self.client = APIClient()
NsdmSubscriptionModel.objects.all().delete()
expected_data = {
"status": 404,
"detail": "Subscription(" + self.subscription_id + ") "
expected_data = {
"status": 404,
"detail": "Subscription(" + self.subscription_id + ") "
}
response = self.client.get('/api/nsd/v1/'
'subscriptions/' + self.subscription_id,
}
response = self.client.get('/api/nsd/v1/'
'subscriptions/' + self.subscription_id,
}
mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
auth=HTTPBasicAuth("username", "password"),
}
mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
auth=HTTPBasicAuth("username", "password"),
- headers={'Connection': 'close'})
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ verify=False)
class NotificationTest(TestCase):
class NotificationTest(TestCase):
- mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})
+ mock_requests_post.assert_called_with(expect_callbackuri,
+ data=expect_notification,
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ verify=False
+ )
@mock.patch("requests.post")
@mock.patch("uuid.uuid4")
@mock.patch("requests.post")
@mock.patch("uuid.uuid4")
}
}
mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
}
}
mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
- headers={'Connection': 'close'})
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ verify=False)
}
}
mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"], data=expect_notification,
}
}
mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"], data=expect_notification,
- headers={'Connection': 'close'}, auth=HTTPBasicAuth("admin", "pwd1234"))
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ auth=HTTPBasicAuth("admin", "pwd1234"),
+ verify=False)
def test_service_query_single_subscription_not_found(self):
try:
def test_service_query_single_subscription_not_found(self):
try:
}
}
mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
}
}
mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
- headers={'Connection': 'close'})
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ verify=False)