from copy import deepcopy
from http import HTTPStatus
import logging
-from os import mkdir
+from os import makedirs
+from os import path
import queue
-from shutil import rmtree
import threading
import time
import traceback
from importlib_metadata import entry_points
from osm_common.dbbase import DbException
from osm_ng_ro.vim_admin import LockRenew
-from osm_ro_plugin import sdnconn, vimconn
+from osm_ro_plugin import sdnconn
+from osm_ro_plugin import vimconn
from osm_ro_plugin.sdn_dummy import SdnDummyConnector
from osm_ro_plugin.vim_dummy import VimDummyConnector
import yaml
try:
flavor_data = task["find_params"]["flavor_data"]
vim_flavor_id = target_vim.get_flavor_id_from_data(flavor_data)
- except vimconn.VimConnNotFoundException:
- self.logger.warning("VimConnNotFoundException occured.")
+ except vimconn.VimConnNotFoundException as flavor_not_found_msg:
+ self.logger.warning(
+ f"VimConnNotFoundException occured: {flavor_not_found_msg}"
+ )
if not vim_flavor_id and task.get("params"):
# CREATE
self.task_lock.release()
return False
- def _process_vim_config(self, target_id, db_vim):
+ def _process_vim_config(self, target_id: str, db_vim: dict) -> None:
"""
Process vim config, creating vim configuration files as ca_cert
:param target_id: vim/sdn/wim + id
return
file_name = ""
+ work_dir = "/app/osm_ro/certs"
try:
if db_vim["config"].get("ca_cert_content"):
- file_name = "{}:{}".format(target_id, self.worker_index)
+ file_name = f"{work_dir}/{target_id}:{self.worker_index}"
- try:
- mkdir(file_name)
- except FileExistsError:
- self.logger.exception(
- "FileExistsError occured while processing vim_config."
- )
+ if not path.isdir(file_name):
+ makedirs(file_name)
file_name = file_name + "/ca_cert"
self.vim_targets.remove(target_id)
self.logger.info("Unloaded {}".format(target_id))
- rmtree("{}:{}".format(target_id, self.worker_index))
- except FileNotFoundError:
- # This is raised by rmtree if folder does not exist.
- self.logger.exception("FileNotFoundError occured while unloading VIM.")
except Exception as e:
self.logger.error("Cannot unload {}: {}".format(target_id, e))