# See the License for the specific language governing permissions and
# limitations under the License.
-import ssl
import json
-from retry import retry
-from urllib.parse import urlparse
from o2common.config import config, conf
from o2common.domain.filter import gen_orm_filter
from o2common.service.unit_of_work import AbstractUnitOfWork
-from o2common.service.command.handler import get_https_conn_default, \
- get_http_conn, get_https_conn_selfsigned, post_data
+from o2common.adapter.notifications import AbstractNotifications
from o2ims.domain import commands, ocloud as cloud
from o2ims.domain.subscription_obj import Message2SMO, NotificationEventEnum
def registry_to_smo(
cmd: commands.Register2SMO,
uow: AbstractUnitOfWork,
+ notifications: AbstractNotifications,
):
logger.debug('In registry_to_smo')
data = cmd.data
logger.warning('Ocloud {} does not exists.'.format(data.id))
return
logger.debug('O-Cloud Global UUID: {}'.format(ocloud.globalCloudId))
- # ocloud_dict = ocloud.serialize()
- ocloud_dict = {
- 'oCloudId': ocloud.oCloudId,
- 'globalcloudId': ocloud.globalCloudId,
- 'globalCloudId': ocloud.globalCloudId,
- 'name': ocloud.name,
- 'description': ocloud.description,
- 'serviceUri': ocloud.serviceUri
- }
+ ocloud_dict = ocloud.get_notification_dict()
if data.notificationEventType == NotificationEventEnum.CREATE:
- register_smo(uow, ocloud_dict)
+ register_smo(notifications, ocloud_dict)
elif data.notificationEventType in [NotificationEventEnum.MODIFY,
NotificationEventEnum.DELETE]:
_notify_ocloud(uow, data, ocloud_dict)
self.value = value
-def register_smo(uow, ocloud_data):
- call_res, status = call_smo(ocloud_data)
+def register_smo(notifications, ocloud_data):
+ call_res = call_smo(notifications, ocloud_data)
logger.debug('Call SMO response is {}'.format(call_res))
if call_res is True:
- logger.info('Register to smo success response is {}'.format(status))
+ logger.info('Register to smo success response')
else:
raise RegIMSToSMOExp('Register o2ims to SMO failed')
# TODO: record the result for the smo register
callback_smo(sub, msg, ocloud_dict)
-@retry((ConnectionRefusedError), tries=2, delay=2)
-def call_smo(reg_data: dict):
+def call_smo(notifications: AbstractNotifications, reg_data: dict):
smo_token = conf.DEFAULT.smo_token_data
smo_token_info = {
'iss': 'o2ims',
'smo_token_algo': 'RS256'
}
- callback_data = json.dumps({
+ callback = {
'globalCloudId': reg_data['globalCloudId'],
'oCloudId': reg_data['oCloudId'],
'IMS_EP': config.get_api_url(),
'smo_token_data': smo_token_info
- })
+ }
logger.info('callback URL: {}'.format(conf.DEFAULT.smo_register_url))
- logger.debug('callback data: {}'.format(callback_data))
- o = urlparse(conf.DEFAULT.smo_register_url)
- if o.scheme == 'https':
- conn = get_https_conn_default(o.netloc)
- else:
- conn = get_http_conn(o.netloc)
-
- try:
- return post_data(conn, o.path, callback_data)
- except ssl.SSLCertVerificationError as e:
- logger.debug('Try to register to smo with \
- trusted ca failed: {}'.format(e))
- if 'self signed' in str(e):
- conn = get_https_conn_selfsigned(o.netloc)
- try:
- return post_data(conn, o.path, callback_data)
- except Exception as e:
- logger.info(
- 'Register to smo with self-signed ca failed: {}'.format(e))
- # TODO: write the status to extension db table.
- return False, None
- return False, None
- except Exception as e:
- logger.critical('Register to smo except: {}'.format(e))
- return False, None
+ logger.debug('callback data: {}'.format(json.dumps(callback)))
+ return notifications.send(conf.DEFAULT.smo_register_url, callback)