EMS simulator extension for PNF SW Upgrade
Change-Id: I3a8c706373f4004850c2403f4aee0d1f28aad464
Issue-ID: INT-1208
Signed-off-by: Enbo Wang <wangenbo@huawei.com>
diff --git a/test/mocks/emssimulator/install.sh b/test/mocks/emssimulator/install.sh
index e3069aa..64982d9 100755
--- a/test/mocks/emssimulator/install.sh
+++ b/test/mocks/emssimulator/install.sh
@@ -4,13 +4,13 @@
cd swm/sw_server_simulator
for i in `ls -1`; do
- if [ -d $i ]; then
- cd $i
- echo $i
- zip -r ${i}.zip *
- mv ${i}.zip ..
- cd $OLDPWD
- fi
+ if [ -d $i ]; then
+ cd $i
+ echo $i
+ zip -r ${i}.zip *
+ mv ${i}.zip ..
+ cd $OLDPWD
+ fi
done
cd $old_pwd
diff --git a/test/mocks/emssimulator/swm/activateNESw b/test/mocks/emssimulator/swm/activateNESw
index 67d233e..3531f16 100755
--- a/test/mocks/emssimulator/swm/activateNESw
+++ b/test/mocks/emssimulator/swm/activateNESw
@@ -1,117 +1,28 @@
-#!/usr/bin/python
+#!/usr/bin/python3
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
import sys
import argparse
import json
-import os
-import shutil
-import random
-import time
import conf
-import ems_util
-
-
-def do_activate_sw(sw_version_to_be_activated, ne_info):
- """
- return err, reason
- """
-
- installed_sw = ne_info.get("installedSw", {})
- if sw_version_to_be_activated in installed_sw:
- target_sw_version = installed_sw[sw_version_to_be_activated]["version"]
- else:
- target_sw_version = sw_version_to_be_activated
-
- sw_install_dir_in_ne = conf.PNF_SIMULATORS_DIR + '/' + ne_info['omIP'] + conf.PNF_SW_INSTALL_DIR
- target_sw_dir = sw_install_dir_in_ne + '/' + target_sw_version
- if not os.path.isdir(target_sw_dir):
- return True, "SW to be activated does not install"
-
- if "targetSwVersion" in ne_info:
- if ne_info["targetSwVersion"] != target_sw_version:
- return True, "Conflicted targetVersion with to be activated %s" % target_sw_version
- del ne_info["targetSwVersion"]
-
- old_sw_version = ne_info.get("oldSwVersion", "")
-
- if target_sw_version != ne_info["currentSwVersion"]:
- ne_info["oldSwVersion"] = ne_info["currentSwVersion"]
- ne_info["currentSwVersion"] = target_sw_version
- ne_info["status"] = conf.STATUS_ACTIVATING
- ems_util.update_ne_info(ne_info)
-
- if target_sw_version != old_sw_version:
- old_sw_dir = sw_install_dir_in_ne + '/' + old_sw_version
- if old_sw_version and os.path.isdir(old_sw_dir):
- shutil.rmtree(old_sw_dir, ignore_errors=True)
-
- old_cwd = os.getcwd()
- os.chdir(sw_install_dir_in_ne)
- if os.path.islink(conf.CURRENT_VERSION_DIR):
- os.remove(conf.CURRENT_VERSION_DIR)
- os.symlink(target_sw_version, conf.CURRENT_VERSION_DIR)
- os.chdir(old_cwd)
-
- if "downloadedSwLocation" in ne_info:
- if os.path.isdir(ne_info["downloadedSwLocation"]):
- shutil.rmtree(ne_info["downloadedSwLocation"], ignore_errors=True)
- del ne_info["downloadedSwLocation"]
-
- return False, None
-
-
-def generate_notification(activate_process_id, activate_status, sw_version, failure_reason):
- notification = {
- "objectClass": "EMSClass",
- "objectInstance": "EMSInstance",
- "notificationId": random.randint(1, conf.MAX_INT),
- "eventTime": time.asctime(),
- "systemDN": "emssimulator",
- "notificationType": "notifyActivateNESwStatusChanged",
- "activateProcessId": activate_process_id,
- "activateOperationStatus": activate_status,
- "swVersion": sw_version
- }
-
- if failure_reason:
- notification["failureReason"] = failure_reason
-
- return notification
-
-
-def activate_ne_sw(sw_version_to_be_activated, ne_id):
- ne_info = ems_util.get_ne_info_from_db_by_id(ne_id)
-
- activate_process_id = random.randint(1, conf.MAX_INT)
- result = conf.REQ_SUCCESS
- ret_value = {
- "activateProcessId": activate_process_id,
- "result": result
- }
-
- if not ne_info:
- ret_value["result"] = conf.REQ_FAILURE
- ret_value["reason"] = "Can not find NE %s" % ne_id
- return ret_value
-
- err, reason = do_activate_sw(sw_version_to_be_activated, ne_info)
-
- if not err:
- ne_info["status"] = conf.STATUS_ACTIVATED
- ems_util.update_ne_info(ne_info)
- activate_status = "NE_SWACTIVATION_SUCCESSFUL"
- else:
- ret_value["result"] = conf.REQ_FAILURE
- ret_value["reason"] = reason
-
- activate_status = "NE_SWACTIVATION_FAILED"
-
- notification = generate_notification(activate_process_id, activate_status, sw_version_to_be_activated, reason)
- ems_util.send_notification(notification, activate_process_id)
-
- # for automated software management, there is no listOfStepNumbersAndDurations
- return ret_value
+import activate_n_e_sw
def main():
@@ -122,8 +33,8 @@
args = parser.parse_args()
- ret_value = activate_ne_sw(args.swVersionToBeActivated, args.neIdentifier)
- print json.dumps(ret_value)
+ _, ret_value = activate_n_e_sw.activate(args.swVersionToBeActivated, args.neIdentifier)
+ print(json.dumps(ret_value))
if ret_value["result"] == conf.REQ_SUCCESS:
sys.exit(conf.RET_CODE_SUCCESS)
diff --git a/test/mocks/emssimulator/swm/activate_n_e_sw.py b/test/mocks/emssimulator/swm/activate_n_e_sw.py
new file mode 100644
index 0000000..26214fa
--- /dev/null
+++ b/test/mocks/emssimulator/swm/activate_n_e_sw.py
@@ -0,0 +1,141 @@
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+import os
+import logging
+import shutil
+import random
+import time
+
+import conf
+import ems_util
+
+
+OPERATION_NAME = "activateNESw"
+logging.basicConfig(level=logging.INFO, format=conf.LOGGER_FORMAT, filename=ems_util.get_log_file(OPERATION_NAME))
+logger = logging.getLogger(OPERATION_NAME)
+
+
+def do_activate(sw_version_to_be_activated, ne_info):
+ """
+ return err, reason
+ """
+
+ logger.info("swVersionToBeActivated: %s" % sw_version_to_be_activated)
+
+ installed_sw = ne_info.get("installedSw", {})
+ if sw_version_to_be_activated in installed_sw:
+ target_sw_version = installed_sw[sw_version_to_be_activated]["version"]
+ else:
+ target_sw_version = sw_version_to_be_activated
+
+ sw_install_dir_in_ne = ems_util.get_install_dir(ne_info['omIP'])
+ logger.info("SW has been installed at %s" % sw_install_dir_in_ne)
+
+ if "targetSwVersion" in ne_info:
+ if ne_info["targetSwVersion"] != target_sw_version:
+ msg = "Conflicted targetVersion with to be activated %s" % target_sw_version
+ logger.error(msg)
+ return True, msg
+ del ne_info["targetSwVersion"]
+
+ old_sw_version = ne_info.get("oldSwVersion", "")
+
+ if target_sw_version != ne_info["currentSwVersion"]:
+ ne_info["oldSwVersion"] = ne_info["currentSwVersion"]
+ ne_info["currentSwVersion"] = target_sw_version
+ ne_info["status"] = conf.STATUS_ACTIVATING
+ ems_util.update_ne_info(ne_info)
+
+ if target_sw_version != old_sw_version:
+ old_sw_dir = os.path.join(sw_install_dir_in_ne, old_sw_version)
+ if old_sw_version and os.path.isdir(old_sw_dir):
+ shutil.rmtree(old_sw_dir, ignore_errors=True)
+
+ old_cwd = os.getcwd()
+ os.chdir(sw_install_dir_in_ne)
+ if os.path.islink(conf.CURRENT_VERSION_DIR):
+ os.remove(conf.CURRENT_VERSION_DIR)
+ os.symlink(target_sw_version, conf.CURRENT_VERSION_DIR)
+ os.chdir(old_cwd)
+
+ if "downloadedSwLocation" in ne_info:
+ if os.path.isdir(ne_info["downloadedSwLocation"]):
+ shutil.rmtree(ne_info["downloadedSwLocation"], ignore_errors=True)
+ del ne_info["downloadedSwLocation"]
+
+ return False, None
+
+
+def generate_notification(activate_process_id, activate_status, sw_version, failure_reason):
+ notification = {
+ "objectClass": conf.OBJECT_CLASS,
+ "objectInstance": conf.OBJECT_INSTANCE,
+ "notificationId": random.randint(1, conf.MAX_INT),
+ "eventTime": time.asctime(),
+ "systemDN": conf.SYSTEM_DN,
+ "notificationType": "notifyActivateNESwStatusChanged",
+ "activateProcessId": activate_process_id,
+ "activateOperationStatus": activate_status,
+ "swVersion": sw_version
+ }
+
+ if failure_reason:
+ logger.error(failure_reason)
+ notification["failureReason"] = failure_reason
+
+ return notification
+
+
+def activate(sw_version_to_be_activated, ne_id):
+ ne_info = ems_util.get_ne_info_from_db_by_id(ne_id)
+
+ activate_process_id = random.randint(1, conf.MAX_INT)
+ result = conf.REQ_SUCCESS
+ ret_value = {
+ "activateProcessId": activate_process_id,
+ "result": result
+ }
+
+ if not ne_info:
+ ret_value["result"] = conf.REQ_FAILURE
+ ret_value["reason"] = "Can not find NE %s" % ne_id
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ err, reason = do_activate(sw_version_to_be_activated, ne_info)
+
+ if not err:
+ ne_info["status"] = conf.STATUS_ACTIVATED
+ ems_util.update_ne_info(ne_info)
+
+ logger.info("Activate SW success")
+ activate_status = "NE_SWACTIVATION_SUCCESSFUL"
+ else:
+ ret_value["result"] = conf.REQ_FAILURE
+ ret_value["reason"] = reason
+
+ logger.error("Activate SW failure, reason: %s" % ret_value["reason"])
+ activate_status = "NE_SWACTIVATION_FAILED"
+
+ notification = generate_notification(activate_process_id, activate_status, sw_version_to_be_activated, reason)
+ ems_util.send_notification(notification, activate_process_id)
+
+ # for automated software management, there is no listOfStepNumbersAndDurations
+ return notification, ret_value
diff --git a/test/mocks/emssimulator/swm/conf.py b/test/mocks/emssimulator/swm/conf.py
index d7ba5b4..5aac383 100644
--- a/test/mocks/emssimulator/swm/conf.py
+++ b/test/mocks/emssimulator/swm/conf.py
@@ -1,22 +1,46 @@
-#!/usr/bin/python
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
import sys
+import os
SWM_DIR = sys.path[0]
-NE_INFO_TABLE = SWM_DIR + "/ems_db/ne_info_table.json"
-SW_SERVER_SIMULATOR = SWM_DIR + "/sw_server_simulator"
-PNF_SIMULATORS_DIR = SWM_DIR + "/pnf_simulators"
-PNF_SW_DOWNLOAD_DIR = "/opt/download"
-PNF_SW_INSTALL_DIR = "/opt/install"
-PNF_SW_FALLBACK_DIR = "/opt/fallback"
+LOGGER_FORMAT = "[%(asctime)-15s] %(levelname)s [%(name)s]: %(message)s"
+LOGGER_FILE_DIR = os.path.join(SWM_DIR, "log")
+
+NE_INFO_TABLE = os.path.join(SWM_DIR, "ems_db", "ne_info_table.json")
+SW_SERVER_SIMULATOR = os.path.join(SWM_DIR, "sw_server_simulator")
+PNF_SIMULATORS_DIR = os.path.join(SWM_DIR, "pnf_simulators")
+COMMON_PATH = "opt"
+PNF_SW_DOWNLOAD_DIR = "download"
+PNF_SW_INSTALL_DIR = "install"
MANIFEST_FILE = "manifest.json"
-INSTALLED_SW = "installed_sw.json"
+INSTALLED_SW_FILE = "installed_sw.json"
CURRENT_VERSION_DIR = "current"
NOTIFICATION_DIR = "/tmp"
MAX_INT = (2**32) - 1
+OBJECT_CLASS = "NRCellDU"
+OBJECT_INSTANCE = "DC=com, SubNetwork=1, ManagedElement=123, GNBDUFunction=1, NRCellDU=1"
+SYSTEM_DN = "DC=com, SubNetwork=1, ManagedElement=123"
+
STATUS_DOWNLOADING = "Downloading"
STATUS_INSTALLING = "Installing"
STATUS_ACTIVATING = "Activating"
@@ -34,4 +58,3 @@
RESULT_SUCCESS = "Success"
RESULT_FAILURE = "Failure"
RESULT_PARTLY = "Partly successful"
-
diff --git a/test/mocks/emssimulator/swm/downloadNESw b/test/mocks/emssimulator/swm/downloadNESw
index 06a8d6b..9079448 100755
--- a/test/mocks/emssimulator/swm/downloadNESw
+++ b/test/mocks/emssimulator/swm/downloadNESw
@@ -1,134 +1,29 @@
-#!/usr/bin/python
+#!/usr/bin/python3
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
import sys
import argparse
import json
-import os
-import shutil
-import random
-import time
-import tempfile
import conf
-import ems_util
-
-
-def do_download_sw(sw_info, sw_download_dir):
- """
- return err, reason, file_location_in_nes
- """
-
- sw_location = sw_info['swLocation']
-
- # Use copy file from SW_SERVER_SIMULATOR to simulate download file
- sw_file_name = sw_location.split('/')[-1]
-
- file_location_in_server = conf.SW_SERVER_SIMULATOR + '/' + sw_file_name
- file_location_in_ne = sw_download_dir + '/' + sw_file_name
-
- try:
- shutil.copy(file_location_in_server, sw_download_dir)
- except IOError as e:
- return True, "Download %s to %s error: %s" % (sw_file_name, sw_download_dir, str(e)), file_location_in_ne
-
- return False, None, file_location_in_ne
-
-
-def generate_notification(download_process_id, download_status, downloaded_ne_sw_info, failed_sw_info):
- notification = {
- "objectClass": "EMSClass",
- "objectInstance": "EMSInstance",
- "notificationId": random.randint(1, conf.MAX_INT),
- "eventTime": time.asctime(),
- "systemDN": "emssimulator",
- "notificationType": "notifyDownloadNESwStatusChanged",
- "downloadProcessId": download_process_id,
- "downloadOperationStatus": download_status
- }
-
- if downloaded_ne_sw_info:
- notification["downloadedNESwInfo"] = downloaded_ne_sw_info
-
- if failed_sw_info:
- notification["failedSwInfo"] = failed_sw_info
-
- return notification
-
-
-def download_ne_sw(sw_to_be_downloaded, ne_id):
- ne_info = ems_util.get_ne_info_from_db_by_id(ne_id)
-
- download_process_id = random.randint(1, conf.MAX_INT)
- result = conf.REQ_SUCCESS
- ret_value = {
- "downloadProcessId": download_process_id,
- "result": result
- }
-
- if not ne_info:
- ret_value["result"] = conf.REQ_FAILURE
- ret_value["reason"] = "Can not find NE %s" % ne_id
- return ret_value
-
- ne_info["status"] = conf.STATUS_DOWNLOADING
- ems_util.update_ne_info(ne_info)
-
- num_sw_to_be_downloaded = len(sw_to_be_downloaded)
-
- downloaded_ne_sw_info = []
- failed_sw_info = []
-
- sw_download_parent_dir = conf.PNF_SIMULATORS_DIR + '/' + ne_info['omIP'] + conf.PNF_SW_DOWNLOAD_DIR
- sw_download_dir = ne_info.get("downloadedSwLocation", "")
- try:
- if not os.path.isdir(sw_download_parent_dir):
- os.makedirs(sw_download_parent_dir)
-
- if sw_download_dir and not os.path.isdir(sw_download_dir):
- os.makedirs(sw_download_dir)
- except OSError as e:
- ret_value["result"] = conf.REQ_FAILURE
- ret_value["reason"] = str(e)
- return ret_value
-
- if not sw_download_dir:
- sw_download_dir = tempfile.mkdtemp(dir=sw_download_parent_dir)
-
- for sw_info in sw_to_be_downloaded:
- err, reason, file_location = do_download_sw(sw_info, sw_download_dir)
- if not err:
- downloaded_ne_sw_info.append(file_location)
- else:
- result = conf.REQ_FAILURE
- failed_sw_entry = {
- "failedSw": file_location,
- "failureReason": reason
- }
- failed_sw_info.append(failed_sw_entry)
-
- num_downloaded_ne_sw = len(downloaded_ne_sw_info)
-
- if num_downloaded_ne_sw == num_sw_to_be_downloaded:
- download_status = "NE_SWDOWNLOAD_SUCCESSFUL"
- elif num_downloaded_ne_sw == 0:
- download_status = "NE_SWDOWNLOAD_FAILED"
- else:
- download_status = "NE_SWDOWNLOAD_PARTIALLY_SUCCESSFUL"
-
- notification = generate_notification(download_process_id, download_status, downloaded_ne_sw_info, failed_sw_info)
- ems_util.send_notification(notification, download_process_id)
-
- if result == conf.REQ_SUCCESS:
- ne_info["downloadedSwLocation"] = sw_download_dir
- ems_util.update_ne_info(ne_info)
- else:
- shutil.rmtree(sw_download_dir, ignore_errors=True)
-
- ret_value["result"] = result
- ret_value["reason"] = json.dumps(failed_sw_info)
-
- # for automated software management, there is no listOfStepNumbersAndDurations
- return ret_value
+import download_n_e_sw
+import install_n_e_sw
def main():
@@ -141,10 +36,35 @@
sw_to_be_downloaded = json.loads(args.swToBeDownloaded)
- ret_value = download_ne_sw(sw_to_be_downloaded, args.neIdentifier)
- print json.dumps(ret_value)
+ all_installed_ne_sw_list = []
+ all_failed_sw_info = []
- if ret_value["result"] == conf.REQ_SUCCESS:
+ download_notification, download_ret_value = download_n_e_sw.download(sw_to_be_downloaded, args.neIdentifier)
+
+ downloaded_ne_sw_list = download_notification.get("downloadedNESwInfo", [])
+ failed_downloaded_sw_list = download_notification.get("failedSwInfo", [])
+ all_failed_sw_info.extend(failed_downloaded_sw_list)
+
+ for downloaded_ne_sw in downloaded_ne_sw_list:
+ install_notification, _ = install_n_e_sw.install(downloaded_ne_sw, args.neIdentifier)
+ installed_ne_sw_list = install_notification.get("installedNESwInfo", [])
+ failed_installed_sw_list = install_notification.get("failedSwInfo", [])
+
+ all_installed_ne_sw_list.extend(installed_ne_sw_list)
+ all_failed_sw_info.extend(failed_installed_sw_list)
+
+ num_all_installed_ne_sw_list = len(all_installed_ne_sw_list)
+ num_sw_to_be_downloaded = len(sw_to_be_downloaded)
+
+ if num_all_installed_ne_sw_list == num_sw_to_be_downloaded:
+ download_ret_value["result"] = conf.REQ_SUCCESS
+ else:
+ download_ret_value["result"] = conf.REQ_FAILURE
+ download_ret_value["reason"] = "Failed downloaded or installed SW: %s" % json.dumps(all_failed_sw_info)
+
+ print(json.dumps(download_ret_value))
+
+ if download_ret_value["result"] == conf.REQ_SUCCESS:
sys.exit(conf.RET_CODE_SUCCESS)
else:
sys.exit(conf.RET_CODE_FAILURE)
diff --git a/test/mocks/emssimulator/swm/download_n_e_sw.py b/test/mocks/emssimulator/swm/download_n_e_sw.py
new file mode 100644
index 0000000..05ae4bb
--- /dev/null
+++ b/test/mocks/emssimulator/swm/download_n_e_sw.py
@@ -0,0 +1,169 @@
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+import os
+import logging
+import json
+import shutil
+import random
+import time
+import tempfile
+
+import conf
+import ems_util
+
+
+OPERATION_NAME = "downloadNESw"
+logging.basicConfig(level=logging.INFO, format=conf.LOGGER_FORMAT, filename=ems_util.get_log_file(OPERATION_NAME))
+logger = logging.getLogger(OPERATION_NAME)
+
+
+def do_download(sw_info, sw_download_dir):
+ """
+ return err, reason, file_location_in_nes
+ """
+
+ sw_location = sw_info['swLocation']
+
+ logger.info("swLocationToBeDownloaded: %s" % sw_location)
+
+ # Use copy file from SW_SERVER_SIMULATOR to simulate download file
+ sw_file_name = sw_location.split('/')[-1]
+
+ file_location_in_server = os.path.join(conf.SW_SERVER_SIMULATOR, sw_file_name)
+ file_location_in_ne = os.path.join(sw_download_dir, sw_file_name)
+
+ try:
+ shutil.copy(file_location_in_server, sw_download_dir)
+ except IOError as e:
+ msg = "Download %s to %s error: %s" % (sw_file_name, sw_download_dir, str(e))
+ logger.error(msg)
+ return True, msg, file_location_in_ne
+
+ return False, None, file_location_in_ne
+
+
+def generate_notification(download_process_id, download_status, downloaded_ne_sw_info, failed_sw_info):
+ notification = {
+ "objectClass": conf.OBJECT_CLASS,
+ "objectInstance": conf.OBJECT_INSTANCE,
+ "notificationId": random.randint(1, conf.MAX_INT),
+ "eventTime": time.asctime(),
+ "systemDN": conf.SYSTEM_DN,
+ "notificationType": "notifyDownloadNESwStatusChanged",
+ "downloadProcessId": download_process_id,
+ "downloadOperationStatus": download_status
+ }
+
+ if downloaded_ne_sw_info:
+ notification["downloadedNESwInfo"] = downloaded_ne_sw_info
+
+ if failed_sw_info:
+ notification["failedSwInfo"] = failed_sw_info
+
+ return notification
+
+
+def download(sw_to_be_downloaded, ne_id):
+ ne_info = ems_util.get_ne_info_from_db_by_id(ne_id)
+
+ download_process_id = random.randint(1, conf.MAX_INT)
+ result = conf.REQ_SUCCESS
+ ret_value = {
+ "downloadProcessId": download_process_id,
+ "result": result
+ }
+
+ if not ne_info:
+ ret_value["result"] = conf.REQ_FAILURE
+ ret_value["reason"] = "Can not find NE %s" % ne_id
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ ne_info["status"] = conf.STATUS_DOWNLOADING
+ ems_util.update_ne_info(ne_info)
+
+ num_sw_to_be_downloaded = len(sw_to_be_downloaded)
+
+ downloaded_ne_sw_info = []
+ failed_sw_info = []
+
+ sw_download_parent_dir = ems_util.get_download_dir(ne_info['omIP'])
+ logger.info("SW will be downloaded to %s" % sw_download_parent_dir)
+
+ sw_download_dir = ne_info.get("downloadedSwLocation", "")
+ try:
+ if not os.path.isdir(sw_download_parent_dir):
+ os.makedirs(sw_download_parent_dir)
+
+ if sw_download_dir and not os.path.isdir(sw_download_dir):
+ os.makedirs(sw_download_dir)
+ except OSError as e:
+ ret_value["result"] = conf.REQ_FAILURE
+ ret_value["reason"] = str(e)
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ if not sw_download_dir:
+ sw_download_dir = tempfile.mkdtemp(dir=sw_download_parent_dir)
+
+ for sw_info in sw_to_be_downloaded:
+ err, reason, file_location = do_download(sw_info, sw_download_dir)
+ if not err:
+ logger.info("Downloaded SW file location: %s" % file_location)
+ downloaded_ne_sw_info.append(file_location)
+ else:
+ result = conf.REQ_FAILURE
+ failed_sw_entry = {
+ "failedSw": file_location,
+ "failureReason": reason
+ }
+
+ logger.error("Failed downloaded SW: %s" % str(failed_sw_entry))
+ failed_sw_info.append(failed_sw_entry)
+
+ num_downloaded_ne_sw = len(downloaded_ne_sw_info)
+
+ if num_downloaded_ne_sw == num_sw_to_be_downloaded:
+ download_status = "NE_SWDOWNLOAD_SUCCESSFUL"
+ elif num_downloaded_ne_sw == 0:
+ download_status = "NE_SWDOWNLOAD_FAILED"
+ else:
+ download_status = "NE_SWDOWNLOAD_PARTIALLY_SUCCESSFUL"
+ logger.info("Download SW status: %s" % download_status)
+
+ notification = generate_notification(download_process_id, download_status, downloaded_ne_sw_info, failed_sw_info)
+ ems_util.send_notification(notification, download_process_id)
+
+ if result == conf.REQ_SUCCESS:
+ ne_info["downloadedSwLocation"] = sw_download_dir
+ ems_util.update_ne_info(ne_info)
+
+ logger.info("Download SW success")
+ else:
+ shutil.rmtree(sw_download_dir, ignore_errors=True)
+
+ ret_value["result"] = result
+ ret_value["reason"] = json.dumps(failed_sw_info)
+
+ logger.info("Download SW failure, reason: %s" % ret_value["reason"])
+
+ # for automated software management, there is no listOfStepNumbersAndDurations
+ return notification, ret_value
diff --git a/test/mocks/emssimulator/swm/ems_db/ne_info_table.json b/test/mocks/emssimulator/swm/ems_db/ne_info_table.json
index eebabf9..5bf014f 100644
--- a/test/mocks/emssimulator/swm/ems_db/ne_info_table.json
+++ b/test/mocks/emssimulator/swm/ems_db/ne_info_table.json
@@ -7,12 +7,12 @@
"currentSwVersion": "v1",
"installedSw": {
"ran_du_pkg2-v1": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg2",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg2",
"version": "v1",
"name": "ran_du_pkg2"
},
"ran_du_pkg1-v1": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg1",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg1",
"version": "v1",
"name": "ran_du_pkg1"
}
@@ -27,12 +27,12 @@
"currentSwVersion": "v2",
"installedSw": {
"ran_du_pkg1-v2": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg1",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg1",
"version": "v2",
"name": "ran_du_pkg1"
},
"ran_du_pkg2-v2": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg2",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg2",
"version": "v2",
"name": "ran_du_pkg2"
}
diff --git a/test/mocks/emssimulator/swm/ems_util.py b/test/mocks/emssimulator/swm/ems_util.py
index 6d0d310..9bfb2cd 100644
--- a/test/mocks/emssimulator/swm/ems_util.py
+++ b/test/mocks/emssimulator/swm/ems_util.py
@@ -1,5 +1,22 @@
-#!/usr/bin/python
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+import os
import time
import json
import jsonpath
@@ -7,6 +24,18 @@
import conf
+def get_log_file(operation):
+ return os.path.join(conf.LOGGER_FILE_DIR, "%s.txt" % operation)
+
+
+def get_download_dir(om_ip):
+ return os.path.join(conf.PNF_SIMULATORS_DIR, om_ip, conf.COMMON_PATH, conf.PNF_SW_DOWNLOAD_DIR)
+
+
+def get_install_dir(om_ip):
+ return os.path.join(conf.PNF_SIMULATORS_DIR, om_ip, conf.COMMON_PATH, conf.PNF_SW_INSTALL_DIR)
+
+
def get_ne_info_list_from_db(ne_filter):
with open(conf.NE_INFO_TABLE) as f_ne_info:
ne_info_table = json.load(f_ne_info)
@@ -47,7 +76,7 @@
def send_notification(notification, process_id):
- notification_file = conf.NOTIFICATION_DIR + '/%s-%d' % (notification['notificationType'], process_id)
+ notification_file = os.path.join(conf.NOTIFICATION_DIR, '%s-%d' % (notification['notificationType'], process_id))
with open(notification_file, 'w') as f_notification:
f_notification.write(json.dumps(notification))
diff --git a/test/mocks/emssimulator/swm/installNESw b/test/mocks/emssimulator/swm/installNESw
index 84e2fb9..e56f799 100755
--- a/test/mocks/emssimulator/swm/installNESw
+++ b/test/mocks/emssimulator/swm/installNESw
@@ -1,188 +1,28 @@
-#!/usr/bin/python
+#!/usr/bin/python3
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
import sys
import argparse
import json
-import os
-import shutil
-import random
-import time
-import tempfile
-import zipfile
import conf
-import ems_util
-
-
-def do_install_sw(sw_to_be_installed, ne_info):
- """
- return err, reason, installed_ne_sw
- """
-
- sw_install_dir_in_ne = conf.PNF_SIMULATORS_DIR + '/' + ne_info['omIP'] + conf.PNF_SW_INSTALL_DIR
-
- if sw_to_be_installed.startswith('/'):
- file_location = sw_to_be_installed
- else:
- sw_download_dir_in_ne = ne_info.get("downloadedSwLocation", "")
- file_location = sw_download_dir_in_ne + '/' + sw_to_be_installed
-
- if not os.access(file_location, os.R_OK):
- return True, "Missing to be installed SW file %s" % file_location, None
-
- try:
- if not os.path.isdir(sw_install_dir_in_ne):
- os.makedirs(sw_install_dir_in_ne)
- except OSError as e:
- return True, str(e), None
-
- temp_dir = tempfile.mkdtemp(dir=sw_install_dir_in_ne)
- if file_location.endswith(".zip"):
- with zipfile.ZipFile(file_location) as sw_zip:
- sw_zip.extractall(temp_dir)
- else:
- return True, "Only support zip file", None
-
- manifest_location = temp_dir + '/' + conf.MANIFEST_FILE
- if os.access(manifest_location, os.R_OK):
- with open(manifest_location) as f_manifest:
- manifest = json.load(f_manifest)
- else:
- shutil.rmtree(temp_dir, ignore_errors=True)
- return True, "Missing manifest file in %s" % file_location, None
-
- try:
- target_sw_name = manifest["name"]
- target_sw_version = manifest["version"]
- except KeyError as e:
- shutil.rmtree(temp_dir, ignore_errors=True)
- return True, "Missing key %s in %s of %s" % (str(e), conf.MANIFEST_FILE, file_location), None
-
- if "targetSwVersion" in ne_info and ne_info["targetSwVersion"] != target_sw_version:
- shutil.rmtree(temp_dir, ignore_errors=True)
- return True, "Conflicted targetVersion for %s" % file_location, None
-
- ne_info["targetSwVersion"] = target_sw_version
- ems_util.update_ne_info(ne_info)
-
- target_sw_parent_dir = sw_install_dir_in_ne + '/' + target_sw_version
- try:
- if not os.path.isdir(target_sw_parent_dir):
- os.makedirs(target_sw_parent_dir)
- except OSError as e:
- shutil.rmtree(temp_dir, ignore_errors=True)
- return True, str(e), None
-
- target_sw_dir = target_sw_parent_dir + '/' + target_sw_name
- if os.path.isdir(target_sw_dir):
- shutil.rmtree(target_sw_dir, ignore_errors=True)
-
- try:
- shutil.move(temp_dir, target_sw_dir)
- except shutil.Error as e:
- shutil.rmtree(temp_dir, ignore_errors=True)
- return True, str(e), None
-
- installed_ne_sw = target_sw_name + '-' + target_sw_version
-
- installed_sw_db = target_sw_parent_dir + '/' + conf.INSTALLED_SW
- if os.path.isfile(installed_sw_db):
- with open(installed_sw_db) as f_installed_sw:
- installed_sw_table = json.load(f_installed_sw)
- if not installed_sw_table:
- installed_sw_table = {}
- else:
- installed_sw_table = {}
-
- target_sw_info = {
- "name": target_sw_name,
- "version": target_sw_version,
- "installedLocation": target_sw_dir
- }
- installed_sw_table[installed_ne_sw] = target_sw_info
-
- with open(installed_sw_db, 'w') as f_installed_sw:
- json.dump(installed_sw_table, f_installed_sw, indent=2)
-
- ne_info["installedSw"] = installed_sw_table
-
- return False, None, installed_ne_sw
-
-
-def generate_notification(install_process_id, install_status, installed_ne_sw_info, failed_sw_info):
- notification = {
- "objectClass": "EMSClass",
- "objectInstance": "EMSInstance",
- "notificationId": random.randint(1, conf.MAX_INT),
- "eventTime": time.asctime(),
- "systemDN": "emssimulator",
- "notificationType": "notifyInstallNESwStatusChanged",
- "installProcessId": install_process_id,
- "installOperationStatus": install_status
- }
-
- if installed_ne_sw_info:
- notification["installedNESwInfo"] = installed_ne_sw_info
-
- if failed_sw_info:
- notification["failedSwInfo"] = failed_sw_info
-
- return notification
-
-
-def install_ne_sw(sw_to_be_installed, ne_id):
- ne_info = ems_util.get_ne_info_from_db_by_id(ne_id)
-
- install_process_id = random.randint(1, conf.MAX_INT)
- result = conf.REQ_SUCCESS
- ret_value = {
- "installProcessId": install_process_id,
- "result": result
- }
-
- if not ne_info:
- ret_value["result"] = conf.REQ_FAILURE
- ret_value["reason"] = "Can not find NE %s" % ne_id
- return ret_value
-
- ne_info["status"] = conf.STATUS_INSTALLING
- ems_util.update_ne_info(ne_info)
-
- installed_ne_sw_info = []
- failed_sw_info = []
-
- err, reason, installed_ne_sw = do_install_sw(sw_to_be_installed, ne_info)
-
- if not err:
- installed_ne_sw_info.append(installed_ne_sw)
- else:
- result = conf.REQ_FAILURE
- failed_sw_entry = {
- "failedSw": installed_ne_sw,
- "failureReason": reason
- }
- failed_sw_info.append(failed_sw_entry)
-
- num_installed_ne_sw = len(installed_ne_sw_info)
-
- if num_installed_ne_sw == 1:
- install_status = "NE_SWINSTALLATION_SUCCESSFUL"
- elif num_installed_ne_sw == 0:
- install_status = "NE_SWINSTALLATION_FAILED"
- else:
- install_status = "NE_SWINSTALLATION_PARTIALLY_SUCCESSFUL"
-
- notification = generate_notification(install_process_id, install_status, installed_ne_sw_info, failed_sw_info)
- ems_util.send_notification(notification, install_process_id)
-
- if result == conf.REQ_SUCCESS:
- ems_util.update_ne_info(ne_info)
- else:
- ret_value["result"] = result
- ret_value["reason"] = json.dumps(failed_sw_info)
-
- # for automated software management, there is no listOfStepNumbersAndDurations
- return ret_value
+import install_n_e_sw
def main():
@@ -193,8 +33,8 @@
args = parser.parse_args()
- ret_value = install_ne_sw(args.swToBeInstalled, args.neIdentifier)
- print json.dumps(ret_value)
+ _, ret_value = install_n_e_sw.install(args.swToBeInstalled, args.neIdentifier)
+ print(json.dumps(ret_value))
if ret_value["result"] == conf.REQ_SUCCESS:
sys.exit(conf.RET_CODE_SUCCESS)
diff --git a/test/mocks/emssimulator/swm/install_n_e_sw.py b/test/mocks/emssimulator/swm/install_n_e_sw.py
new file mode 100644
index 0000000..57c59a3
--- /dev/null
+++ b/test/mocks/emssimulator/swm/install_n_e_sw.py
@@ -0,0 +1,234 @@
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+import os
+import logging
+import json
+import shutil
+import random
+import time
+import tempfile
+import zipfile
+
+import conf
+import ems_util
+
+
+OPERATION_NAME = "installNESw"
+logging.basicConfig(level=logging.INFO, format=conf.LOGGER_FORMAT, filename=ems_util.get_log_file(OPERATION_NAME))
+logger = logging.getLogger(OPERATION_NAME)
+
+
+def do_install(sw_to_be_installed, ne_info):
+ """
+ return err, reason, installed_ne_sw
+ """
+
+ logger.info("swToBeInstalled: %s" % sw_to_be_installed)
+
+ sw_install_dir_in_ne = ems_util.get_install_dir(ne_info['omIP'])
+
+ if sw_to_be_installed.startswith('/'):
+ file_location = sw_to_be_installed
+ else:
+ sw_download_dir_in_ne = ne_info.get("downloadedSwLocation", "")
+ file_location = os.path.join(sw_download_dir_in_ne, sw_to_be_installed)
+
+ if not os.access(file_location, os.R_OK):
+ msg = "Missing to be installed SW file %s" % file_location
+ logger.error(msg)
+ return True, msg, None
+
+ try:
+ if not os.path.isdir(sw_install_dir_in_ne):
+ os.makedirs(sw_install_dir_in_ne)
+ except OSError as e:
+ msg = str(e)
+ logger.error(msg)
+ return True, msg, None
+
+ temp_dir = tempfile.mkdtemp(dir=sw_install_dir_in_ne)
+ if file_location.endswith(".zip"):
+ with zipfile.ZipFile(file_location) as sw_zip:
+ sw_zip.extractall(temp_dir)
+ else:
+ msg = "Only support zip file"
+ logger.error(msg)
+ return True, msg, None
+
+ manifest_location = os.path.join(temp_dir, conf.MANIFEST_FILE)
+ if os.access(manifest_location, os.R_OK):
+ with open(manifest_location) as f_manifest:
+ manifest = json.load(f_manifest)
+ else:
+ shutil.rmtree(temp_dir, ignore_errors=True)
+ msg = "Missing manifest file in %s" % file_location
+ logger.error(msg)
+ return True, msg, None
+
+ try:
+ target_sw_name = manifest["name"]
+ target_sw_version = manifest["version"]
+ except KeyError as e:
+ shutil.rmtree(temp_dir, ignore_errors=True)
+ msg = "Missing key %s in %s of %s" % (str(e), conf.MANIFEST_FILE, file_location)
+ logger.error(msg)
+ return True, msg, None
+
+ if "targetSwVersion" in ne_info and ne_info["targetSwVersion"] != target_sw_version:
+ shutil.rmtree(temp_dir, ignore_errors=True)
+ msg = "Conflicted targetVersion for %s" % file_location
+ logger.error(msg)
+ return True, msg, None
+
+ ne_info["targetSwVersion"] = target_sw_version
+ ems_util.update_ne_info(ne_info)
+
+ target_sw_parent_dir = os.path.join(sw_install_dir_in_ne, target_sw_version)
+ try:
+ if not os.path.isdir(target_sw_parent_dir):
+ os.makedirs(target_sw_parent_dir)
+ except OSError as e:
+ shutil.rmtree(temp_dir, ignore_errors=True)
+ msg = str(e)
+ logger.error(msg)
+ return True, msg, None
+
+ target_sw_dir = os.path.join(target_sw_parent_dir, target_sw_name)
+ if os.path.isdir(target_sw_dir):
+ shutil.rmtree(target_sw_dir, ignore_errors=True)
+
+ try:
+ shutil.move(temp_dir, target_sw_dir)
+ except shutil.Error as e:
+ shutil.rmtree(temp_dir, ignore_errors=True)
+ msg = str(e)
+ logger.error(msg)
+ return True, msg, None
+ logger.info("Install SW to %s" % target_sw_dir)
+
+ installed_ne_sw = target_sw_name + '-' + target_sw_version
+ logger.info("Installed SW: %s" % installed_ne_sw)
+
+ installed_sw_db = os.path.join(target_sw_parent_dir, conf.INSTALLED_SW_FILE)
+ if os.path.isfile(installed_sw_db):
+ with open(installed_sw_db) as f_installed_sw:
+ installed_sw_table = json.load(f_installed_sw)
+ if not installed_sw_table:
+ installed_sw_table = {}
+ else:
+ installed_sw_table = {}
+
+ target_sw_info = {
+ "name": target_sw_name,
+ "version": target_sw_version,
+ "installedLocation": target_sw_dir
+ }
+ installed_sw_table[installed_ne_sw] = target_sw_info
+
+ with open(installed_sw_db, 'w') as f_installed_sw:
+ json.dump(installed_sw_table, f_installed_sw, indent=2)
+
+ ne_info["installedSw"] = installed_sw_table
+
+ return False, None, installed_ne_sw
+
+
+def generate_notification(install_process_id, install_status, installed_ne_sw_info, failed_sw_info):
+ notification = {
+ "objectClass": conf.OBJECT_CLASS,
+ "objectInstance": conf.OBJECT_INSTANCE,
+ "notificationId": random.randint(1, conf.MAX_INT),
+ "eventTime": time.asctime(),
+ "systemDN": conf.SYSTEM_DN,
+ "notificationType": "notifyInstallNESwStatusChanged",
+ "installProcessId": install_process_id,
+ "installOperationStatus": install_status
+ }
+
+ if installed_ne_sw_info:
+ notification["installedNESwInfo"] = installed_ne_sw_info
+
+ if failed_sw_info:
+ notification["failedSwInfo"] = failed_sw_info
+
+ return notification
+
+
+def install(sw_to_be_installed, ne_id):
+ ne_info = ems_util.get_ne_info_from_db_by_id(ne_id)
+
+ install_process_id = random.randint(1, conf.MAX_INT)
+ result = conf.REQ_SUCCESS
+ ret_value = {
+ "installProcessId": install_process_id,
+ "result": result
+ }
+
+ if not ne_info:
+ ret_value["result"] = conf.REQ_FAILURE
+ ret_value["reason"] = "Can not find NE %s" % ne_id
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ ne_info["status"] = conf.STATUS_INSTALLING
+ ems_util.update_ne_info(ne_info)
+
+ installed_ne_sw_info = []
+ failed_sw_info = []
+
+ err, reason, installed_ne_sw = do_install(sw_to_be_installed, ne_info)
+
+ if not err:
+ installed_ne_sw_info.append(installed_ne_sw)
+ else:
+ result = conf.REQ_FAILURE
+ failed_sw_entry = {
+ "failedSw": installed_ne_sw,
+ "failureReason": reason
+ }
+
+ logger.error("Failed installed SW: %s" % str(failed_sw_entry))
+ failed_sw_info.append(failed_sw_entry)
+
+ num_installed_ne_sw = len(installed_ne_sw_info)
+
+ if num_installed_ne_sw == 1:
+ install_status = "NE_SWINSTALLATION_SUCCESSFUL"
+ elif num_installed_ne_sw == 0:
+ install_status = "NE_SWINSTALLATION_FAILED"
+ else:
+ install_status = "NE_SWINSTALLATION_PARTIALLY_SUCCESSFUL"
+ logger.info("Install SW status: %s" % install_status)
+
+ notification = generate_notification(install_process_id, install_status, installed_ne_sw_info, failed_sw_info)
+ ems_util.send_notification(notification, install_process_id)
+
+ if result == conf.REQ_SUCCESS:
+ ems_util.update_ne_info(ne_info)
+
+ logger.info("Install SW success")
+ else:
+ ret_value["result"] = result
+ ret_value["reason"] = json.dumps(failed_sw_info)
+
+ logger.info("Install SW failure, reason: %s" % ret_value["reason"])
+
+ # for automated software management, there is no listOfStepNumbersAndDurations
+ return notification, ret_value
diff --git a/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.1/opt/install/v1/installed_sw.json b/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.1/opt/install/v1/installed_sw.json
index 830c0e5..e7f13a5 100644
--- a/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.1/opt/install/v1/installed_sw.json
+++ b/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.1/opt/install/v1/installed_sw.json
@@ -1,11 +1,11 @@
{
"ran_du_pkg2-v1": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg2",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg2",
"version": "v1",
"name": "ran_du_pkg2"
},
"ran_du_pkg1-v1": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg1",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.1/opt/install/v1/ran_du_pkg1",
"version": "v1",
"name": "ran_du_pkg1"
}
diff --git a/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v1/installed_sw.json b/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v1/installed_sw.json
index ae18314..bd63618 100644
--- a/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v1/installed_sw.json
+++ b/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v1/installed_sw.json
@@ -1,11 +1,11 @@
{
"ran_du_pkg2-v1": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.2/opt/install/v1/ran_du_pkg2",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.2/opt/install/v1/ran_du_pkg2",
"version": "v1",
"name": "ran_du_pkg2"
},
"ran_du_pkg1-v1": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.2/opt/install/v1/ran_du_pkg1",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.2/opt/install/v1/ran_du_pkg1",
"version": "v1",
"name": "ran_du_pkg1"
}
diff --git a/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v2/installed_sw.json b/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v2/installed_sw.json
index 12c76a8..3fed183 100644
--- a/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v2/installed_sw.json
+++ b/test/mocks/emssimulator/swm/pnf_simulators/192.168.1.2/opt/install/v2/installed_sw.json
@@ -1,11 +1,11 @@
{
"ran_du_pkg1-v2": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg1",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg1",
"version": "v2",
"name": "ran_du_pkg1"
},
"ran_du_pkg2-v2": {
- "installedLocation": "/home/ubuntu/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg2",
+ "installedLocation": "/home/admin/swm/pnf_simulators/192.168.1.2/opt/install/v2/ran_du_pkg2",
"version": "v2",
"name": "ran_du_pkg2"
}
diff --git a/test/mocks/emssimulator/swm/swFallback b/test/mocks/emssimulator/swm/swFallback
index 9d6608c..506eee0 100755
--- a/test/mocks/emssimulator/swm/swFallback
+++ b/test/mocks/emssimulator/swm/swFallback
@@ -1,108 +1,29 @@
-#!/usr/bin/python
+#!/usr/bin/python3
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+import sys
import argparse
import json
-import sys
-import os
-import shutil
import conf
import ems_util
-
-
-def sw_fallback(ne_info_list):
- ne_list = []
- num_failure = 0
-
- for ne_info in ne_info_list:
- if ne_info.get("status") == conf.STATUS_DOWNLOADING:
- ne_info["status"] = conf.STATUS_ACTIVATED
- ems_util.update_ne_info(ne_info)
-
- ne_entry = {
- "nEIdentification": ne_info["nEIdentification"],
- "swFallbackStatus": "fallbackSuccessful"
- }
- ne_list.append(ne_entry)
- continue
-
- sw_install_dir_in_ne = conf.PNF_SIMULATORS_DIR + '/' + ne_info['omIP'] + conf.PNF_SW_INSTALL_DIR
-
- if ne_info.get("status") == conf.STATUS_INSTALLING:
- old_sw_version = ne_info.get("currentSwVersion", "")
- current_sw_version = ne_info.get("targetSwVersion", "")
- else:
- old_sw_version = ne_info.get("oldSwVersion", "")
- current_sw_version = ne_info.get("currentSwVersion", "")
-
- old_sw_dir = sw_install_dir_in_ne + '/' + old_sw_version
-
- if not old_sw_version or not os.path.isdir(old_sw_dir):
- ne_entry = {
- "nEIdentification": ne_info["nEIdentification"],
- "swFallbackStatus": "fallbackUnsuccessful"
- }
- ne_list.append(ne_entry)
-
- num_failure += 1
- continue
-
- current_sw_dir = sw_install_dir_in_ne + '/' + current_sw_version
-
- if current_sw_version and os.path.isdir(current_sw_dir) and current_sw_dir != old_sw_dir:
- shutil.rmtree(current_sw_dir, ignore_errors=True)
-
- old_cwd = os.getcwd()
- os.chdir(sw_install_dir_in_ne)
- if os.path.islink(conf.CURRENT_VERSION_DIR):
- os.remove(conf.CURRENT_VERSION_DIR)
- os.symlink(old_sw_version, conf.CURRENT_VERSION_DIR)
- os.chdir(old_cwd)
-
- installed_sw_db = old_sw_dir + '/' + conf.INSTALLED_SW
- if os.path.isfile(installed_sw_db):
- with open(installed_sw_db) as f_installed_sw:
- installed_sw_table = json.load(f_installed_sw)
- if not installed_sw_table:
- installed_sw_table = {}
- else:
- installed_sw_table = {}
-
- ne_info["installedSw"] = installed_sw_table
- if "oldSwVersion" in ne_info:
- ne_info["currentSwVersion"] = ne_info["oldSwVersion"]
- del ne_info["oldSwVersion"]
-
- if "targetSwVersion" in ne_info:
- del ne_info["targetSwVersion"]
-
- if "downloadedSwLocation" in ne_info:
- if os.path.isdir(ne_info["downloadedSwLocation"]):
- shutil.rmtree(ne_info["downloadedSwLocation"], ignore_errors=True)
- del ne_info["downloadedSwLocation"]
-
- ne_info["status"] = conf.STATUS_ACTIVATED
- ems_util.update_ne_info(ne_info)
-
- ne_entry = {
- "nEIdentification": ne_info["nEIdentification"],
- "swFallbackStatus": "fallbackSuccessful"
- }
- ne_list.append(ne_entry)
-
- if num_failure == 0:
- result = conf.RESULT_SUCCESS
- elif num_failure == len(ne_info_list):
- result = conf.RESULT_FAILURE
- else:
- result = conf.RESULT_PARTLY
-
- ret_value = {
- "nEList": ne_list,
- "result": result
- }
-
- return ret_value
+import sw_fallback
def main():
@@ -114,8 +35,8 @@
ne_info_list = ems_util.get_ne_info_list_from_db(args.filter)
- ret_value = sw_fallback(ne_info_list)
- print json.dumps(ret_value)
+ ret_value = sw_fallback.fallback(ne_info_list)
+ print(json.dumps(ret_value))
if ret_value["result"] == conf.RESULT_SUCCESS:
sys.exit(conf.RET_CODE_SUCCESS)
diff --git a/test/mocks/emssimulator/swm/sw_fallback.py b/test/mocks/emssimulator/swm/sw_fallback.py
new file mode 100644
index 0000000..3037d45
--- /dev/null
+++ b/test/mocks/emssimulator/swm/sw_fallback.py
@@ -0,0 +1,129 @@
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+import os
+import logging
+import json
+import shutil
+
+import conf
+import ems_util
+
+
+OPERATION_NAME = "swFallback"
+logging.basicConfig(level=logging.INFO, format=conf.LOGGER_FORMAT, filename=ems_util.get_log_file(OPERATION_NAME))
+logger = logging.getLogger(OPERATION_NAME)
+
+
+def fallback(ne_info_list):
+ logger.info("NE info list: %s" % ne_info_list)
+
+ ne_list = []
+ num_failure = 0
+
+ for ne_info in ne_info_list:
+ if ne_info.get("status") == conf.STATUS_DOWNLOADING:
+ ne_info["status"] = conf.STATUS_ACTIVATED
+ ems_util.update_ne_info(ne_info)
+
+ ne_entry = {
+ "nEIdentification": ne_info["nEIdentification"],
+ "swFallbackStatus": "fallbackSuccessful"
+ }
+ ne_list.append(ne_entry)
+ continue
+
+ sw_install_dir_in_ne = ems_util.get_install_dir(ne_info['omIP'])
+
+ if ne_info.get("status") == conf.STATUS_INSTALLING:
+ old_sw_version = ne_info.get("currentSwVersion", "")
+ current_sw_version = ne_info.get("targetSwVersion", "")
+ else:
+ old_sw_version = ne_info.get("oldSwVersion", "")
+ current_sw_version = ne_info.get("currentSwVersion", "")
+
+ old_sw_dir = os.path.join(sw_install_dir_in_ne, old_sw_version)
+
+ if not old_sw_version or not os.path.isdir(old_sw_dir):
+ ne_entry = {
+ "nEIdentification": ne_info["nEIdentification"],
+ "swFallbackStatus": "fallbackUnsuccessful"
+ }
+ logger.error("oldSwVersion (%s) or oldSwDirectory (%s) is none" % (old_sw_version, old_sw_dir))
+ ne_list.append(ne_entry)
+
+ num_failure += 1
+ continue
+
+ current_sw_dir = os.path.join(sw_install_dir_in_ne, current_sw_version)
+
+ if current_sw_version and os.path.isdir(current_sw_dir) and current_sw_dir != old_sw_dir:
+ shutil.rmtree(current_sw_dir, ignore_errors=True)
+
+ old_cwd = os.getcwd()
+ os.chdir(sw_install_dir_in_ne)
+ if os.path.islink(conf.CURRENT_VERSION_DIR):
+ os.remove(conf.CURRENT_VERSION_DIR)
+ os.symlink(old_sw_version, conf.CURRENT_VERSION_DIR)
+ os.chdir(old_cwd)
+
+ installed_sw_db = os.path.join(old_sw_dir, conf.INSTALLED_SW_FILE)
+ if os.path.isfile(installed_sw_db):
+ with open(installed_sw_db) as f_installed_sw:
+ installed_sw_table = json.load(f_installed_sw)
+ if not installed_sw_table:
+ installed_sw_table = {}
+ else:
+ installed_sw_table = {}
+
+ ne_info["installedSw"] = installed_sw_table
+ if "oldSwVersion" in ne_info:
+ ne_info["currentSwVersion"] = ne_info["oldSwVersion"]
+ del ne_info["oldSwVersion"]
+
+ if "targetSwVersion" in ne_info:
+ del ne_info["targetSwVersion"]
+
+ if "downloadedSwLocation" in ne_info:
+ if os.path.isdir(ne_info["downloadedSwLocation"]):
+ shutil.rmtree(ne_info["downloadedSwLocation"], ignore_errors=True)
+ del ne_info["downloadedSwLocation"]
+
+ ne_info["status"] = conf.STATUS_ACTIVATED
+ ems_util.update_ne_info(ne_info)
+
+ ne_entry = {
+ "nEIdentification": ne_info["nEIdentification"],
+ "swFallbackStatus": "fallbackSuccessful"
+ }
+ ne_list.append(ne_entry)
+
+ if num_failure == 0:
+ result = conf.RESULT_SUCCESS
+ elif num_failure == len(ne_info_list):
+ result = conf.RESULT_FAILURE
+ else:
+ result = conf.RESULT_PARTLY
+ logger.info("Fallback SW result: %s" % result)
+
+ ret_value = {
+ "nEList": ne_list,
+ "result": result
+ }
+
+ return ret_value
diff --git a/test/mocks/emssimulator/swm/upgrade-post-check b/test/mocks/emssimulator/swm/upgrade-post-check
index 799afbc..e5dfae5 100755
--- a/test/mocks/emssimulator/swm/upgrade-post-check
+++ b/test/mocks/emssimulator/swm/upgrade-post-check
@@ -1,74 +1,52 @@
-#!/usr/bin/python
+#!/usr/bin/python3
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
import sys
import json
import conf
-import ems_util
-
-
-def upgrade_postcheck(pnf_id, old_sw_version, target_sw_version, rule_name, tmp_file=None):
- ne_info = ems_util.get_ne_info_from_db_by_id(pnf_id)
-
- if not ne_info:
- ret_value = {
- "result": conf.RESULT_FAILURE,
- "reason": "Can not find NE %s" % pnf_id
- }
- return ret_value
-
- old_sw_version_in_db = ne_info.get("oldSwVersion", "")
- current_sw_version_in_db = ne_info.get("currentSwVersion", "")
-
- if old_sw_version != old_sw_version_in_db:
- ret_value = {
- "result": conf.RESULT_FAILURE,
- "reason": "Old SW version %s in PNF is not matched with oldSwVersion %s" %
- (old_sw_version_in_db, old_sw_version)
- }
- return ret_value
-
- if target_sw_version != current_sw_version_in_db:
- ret_value = {
- "result": conf.RESULT_FAILURE,
- "reason": "Current SW version %s in PNF is not matched with targetSwVersion %s" %
- (current_sw_version_in_db, target_sw_version)
- }
- return ret_value
-
- ne_info["checkStatus"] = conf.STATUS_POSTCHECKED
- ems_util.update_ne_info(ne_info)
-
- ret_value = {
- "result": conf.RESULT_SUCCESS
- }
-
- return ret_value
+import upgrade_post_check
def main():
- # {{pnfId}} {{oldSwVersion}} {{targetSwVersion}} {{ruleName}} /tmp/tmp-{{Id}}
+ # {{pnfName}} {{oldSwVersion}} {{targetSwVersion}} {{ruleName}} {{additionalDataFile}}
if len(sys.argv) < 5:
ret_value = {
"result": conf.RESULT_FAILURE,
"reason": "Missing parameters"
}
- print json.dumps(ret_value)
+ print(json.dumps(ret_value))
sys.exit(conf.RET_CODE_FAILURE)
-
- if len(sys.argv) >= 5:
- pnf_id = sys.argv[1]
+ else:
+ pnf_name = sys.argv[1]
old_sw_version = sys.argv[2]
target_sw_version = sys.argv[3]
rule_name = sys.argv[4]
- tmp_file = None
+ additional_data_file = None
if len(sys.argv) >= 6:
- tmp_file = sys.argv[5]
+ additional_data_file = sys.argv[5]
- ret_value = upgrade_postcheck(pnf_id, old_sw_version, target_sw_version, rule_name, tmp_file)
- print json.dumps(ret_value)
+ ret_value = upgrade_post_check.post_check(
+ pnf_name, old_sw_version, target_sw_version, rule_name, additional_data_file)
+ print(json.dumps(ret_value))
if ret_value["result"] == conf.RESULT_SUCCESS:
sys.exit(conf.RET_CODE_SUCCESS)
diff --git a/test/mocks/emssimulator/swm/upgrade-pre-check b/test/mocks/emssimulator/swm/upgrade-pre-check
index 6bf867f..5b0726b 100755
--- a/test/mocks/emssimulator/swm/upgrade-pre-check
+++ b/test/mocks/emssimulator/swm/upgrade-pre-check
@@ -1,65 +1,52 @@
-#!/usr/bin/python
+#!/usr/bin/python3
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
import sys
import json
import conf
-import ems_util
-
-
-def upgrade_precheck(pnf_id, old_sw_version, target_sw_version, rule_name, tmp_file=None):
- ne_info = ems_util.get_ne_info_from_db_by_id(pnf_id)
-
- if not ne_info:
- ret_value = {
- "result": conf.RESULT_FAILURE,
- "reason": "Can not find NE %s" % pnf_id
- }
- return ret_value
-
- current_sw_version_in_db = ne_info.get("currentSwVersion", "")
-
- if old_sw_version != current_sw_version_in_db:
- ret_value = {
- "result": conf.RESULT_FAILURE,
- "reason": "Current SW version %s in PNF is not matched with oldSwVersion %s" %
- (current_sw_version_in_db, old_sw_version)
- }
- return ret_value
-
- ne_info["checkStatus"] = conf.STATUS_PRECHECKED
- ems_util.update_ne_info(ne_info)
-
- ret_value = {
- "result": conf.RESULT_SUCCESS
- }
-
- return ret_value
+import upgrade_pre_check
def main():
- # {{pnfId}} {{oldSwVersion}} {{targetSwVersion}} {{ruleName}} /tmp/tmp-{{Id}}
+ # {{pnfName}} {{oldSwVersion}} {{targetSwVersion}} {{ruleName}} {{additionalDataFile}}
if len(sys.argv) < 5:
ret_value = {
"result": conf.RESULT_FAILURE,
"reason": "Missing parameters"
}
- print json.dumps(ret_value)
+ print(json.dumps(ret_value))
sys.exit(conf.RET_CODE_FAILURE)
-
- if len(sys.argv) >= 5:
- pnf_id = sys.argv[1]
+ else:
+ pnf_name = sys.argv[1]
old_sw_version = sys.argv[2]
target_sw_version = sys.argv[3]
rule_name = sys.argv[4]
- tmp_file = None
+ additional_data_file = None
if len(sys.argv) >= 6:
- tmp_file = sys.argv[5]
+ additional_data_file = sys.argv[5]
- ret_value = upgrade_precheck(pnf_id, old_sw_version, target_sw_version, rule_name, tmp_file)
- print json.dumps(ret_value)
+ ret_value = upgrade_pre_check.pre_check(
+ pnf_name, old_sw_version, target_sw_version, rule_name, additional_data_file)
+ print(json.dumps(ret_value))
if ret_value["result"] == conf.RESULT_SUCCESS:
sys.exit(conf.RET_CODE_SUCCESS)
diff --git a/test/mocks/emssimulator/swm/upgrade_post_check.py b/test/mocks/emssimulator/swm/upgrade_post_check.py
new file mode 100644
index 0000000..5140e06
--- /dev/null
+++ b/test/mocks/emssimulator/swm/upgrade_post_check.py
@@ -0,0 +1,76 @@
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+import logging
+
+import conf
+import ems_util
+
+
+OPERATION_NAME = "upgrade-post-check"
+logging.basicConfig(level=logging.INFO, format=conf.LOGGER_FORMAT, filename=ems_util.get_log_file(OPERATION_NAME))
+logger = logging.getLogger(OPERATION_NAME)
+
+
+def post_check(pnf_name, old_sw_version, target_sw_version, rule_name, additional_data_file=None):
+ logger.info("PostCheck for oldSwVersion: %s, targetSwVersion: %s, ruleName: %s, additionalDataFile: %s" %
+ (old_sw_version, target_sw_version, rule_name, additional_data_file))
+
+ ne_info = ems_util.get_ne_info_from_db_by_id(pnf_name)
+
+ if not ne_info:
+ ret_value = {
+ "result": conf.RESULT_FAILURE,
+ "reason": "Can not find NE %s" % pnf_name
+ }
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ old_sw_version_in_db = ne_info.get("oldSwVersion", "")
+ current_sw_version_in_db = ne_info.get("currentSwVersion", "")
+
+ if old_sw_version != old_sw_version_in_db:
+ ret_value = {
+ "result": conf.RESULT_FAILURE,
+ "reason": "Old SW version %s in PNF is not matched with oldSwVersion %s" %
+ (old_sw_version_in_db, old_sw_version)
+ }
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ if target_sw_version != current_sw_version_in_db:
+ ret_value = {
+ "result": conf.RESULT_FAILURE,
+ "reason": "Current SW version %s in PNF is not matched with targetSwVersion %s" %
+ (current_sw_version_in_db, target_sw_version)
+ }
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ ne_info["checkStatus"] = conf.STATUS_POSTCHECKED
+ ems_util.update_ne_info(ne_info)
+ logger.info("PostCheck SW success, check status: %s" % ne_info["checkStatus"])
+
+ ret_value = {
+ "result": conf.RESULT_SUCCESS
+ }
+
+ return ret_value
diff --git a/test/mocks/emssimulator/swm/upgrade_pre_check.py b/test/mocks/emssimulator/swm/upgrade_pre_check.py
new file mode 100644
index 0000000..ee0423a
--- /dev/null
+++ b/test/mocks/emssimulator/swm/upgrade_pre_check.py
@@ -0,0 +1,65 @@
+# ============LICENSE_START=======================================================
+# ONAP - SO
+# ================================================================================
+# Copyright (C) 2020 Huawei Technologies Co., Ltd. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+
+import logging
+
+import conf
+import ems_util
+
+
+OPERATION_NAME = "upgrade-pre-check"
+logging.basicConfig(level=logging.INFO, format=conf.LOGGER_FORMAT, filename=ems_util.get_log_file(OPERATION_NAME))
+logger = logging.getLogger(OPERATION_NAME)
+
+
+def pre_check(pnf_name, old_sw_version, target_sw_version, rule_name, additional_data_file=None):
+ logger.info("PreCheck for oldSwVersion: %s, targetSwVersion: %s, ruleName: %s, additionalDataFile: %s" %
+ (old_sw_version, target_sw_version, rule_name, additional_data_file))
+
+ ne_info = ems_util.get_ne_info_from_db_by_id(pnf_name)
+
+ if not ne_info:
+ ret_value = {
+ "result": conf.RESULT_FAILURE,
+ "reason": "Can not find NE %s" % pnf_name
+ }
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ current_sw_version_in_db = ne_info.get("currentSwVersion", "")
+
+ if old_sw_version != current_sw_version_in_db:
+ ret_value = {
+ "result": conf.RESULT_FAILURE,
+ "reason": "Current SW version %s in PNF is not matched with oldSwVersion %s" %
+ (current_sw_version_in_db, old_sw_version)
+ }
+
+ logger.error(ret_value["reason"])
+ return ret_value
+
+ ne_info["checkStatus"] = conf.STATUS_PRECHECKED
+ ems_util.update_ne_info(ne_info)
+ logger.info("PreCheck SW success, check status: %s" % ne_info["checkStatus"])
+
+ ret_value = {
+ "result": conf.RESULT_SUCCESS
+ }
+
+ return ret_value