1 # Copyright (C) 2021-2022 Wind River Systems, Inc.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
20 from urllib.parse import urlparse
21 from retry import retry
23 from o2common.service.unit_of_work import AbstractUnitOfWork
24 from o2common.config import config, conf
25 from o2common.service.command.handler import get_https_conn_default
26 from o2common.service.command.handler import get_http_conn
27 from o2common.service.command.handler import get_https_conn_selfsigned
28 from o2common.service.command.handler import post_data
30 from o2ims.domain import commands
31 from o2ims.domain.subscription_obj import NotificationEventEnum
33 from o2common.helper import o2logging
34 logger = o2logging.get_logger(__name__)
38 cmd: commands.Register2SMO,
39 uow: AbstractUnitOfWork,
41 logger.info('In registry_to_smo')
43 logger.info('The Register2SMO notificationEventType is {}'.format(
44 data.notificationEventType))
46 ocloud = uow.oclouds.get(data.id)
49 logger.debug('O-Cloud Global UUID: {}'.format(ocloud.globalCloudId))
50 ocloud_dict = ocloud.serialize()
51 if data.notificationEventType == NotificationEventEnum.CREATE:
52 register_smo(uow, ocloud_dict)
55 class RegIMSToSMOExp(Exception):
56 def __init__(self, value):
60 def register_smo(uow, ocloud_data):
61 call_res, status = call_smo(ocloud_data)
62 logger.debug('Call SMO response is {}'.format(call_res))
64 logger.info('Register to smo success response is {}'.format(status))
66 raise RegIMSToSMOExp('Register o2ims to SMO failed')
67 # TODO: record the result for the smo register
70 # def retry(fun, max_tries=2):
71 # for i in range(max_tries):
74 # # await asyncio.sleep(5*i)
76 # logger.debug('retry function result: {}'.format(res))
82 @retry((ConnectionRefusedError), tries=2, delay=2)
83 def call_smo(reg_data: dict):
84 smo_token = conf.DEFAULT.smo_token_data
88 'smo_token_payload': smo_token,
89 'smo_token_type': 'jwt',
90 'smo_token_expiration': '',
91 'smo_token_algo': 'RS256'
94 callback_data = json.dumps({
95 'globalCloudId': reg_data['globalCloudId'],
96 'oCloudId': reg_data['oCloudId'],
97 'IMS_EP': config.get_api_url(),
98 'smo_token_data': smo_token_info
100 logger.info('URL: {}, data: {}'.format(
101 conf.DEFAULT.smo_register_url, callback_data))
102 o = urlparse(conf.DEFAULT.smo_register_url)
103 if o.scheme == 'https':
104 conn = get_https_conn_default(o.netloc)
106 conn = get_http_conn(o.netloc)
109 return post_data(conn, o.path, callback_data)
110 except ssl.SSLCertVerificationError as e:
111 logger.debug('Try to register to smo with \
112 trusted ca failed: {}'.format(e))
113 if 'self signed' in str(e):
114 conn = get_https_conn_selfsigned(o.netloc)
116 return post_data(conn, o.path, callback_data)
117 except Exception as e:
119 'Register to smo with self-signed ca failed: {}'.format(e))
120 # TODO: write the status to extension db table.
123 except Exception as e:
124 logger.critical('Register to smo except: {}'.format(e))