Fixing Bug 2103 and Bug 2134 in NBI

[osm/NBI.git] / osm_nbi / base_topic.py

diff --git a/osm_nbi/base_topic.py b/osm_nbi/base_topic.py
index 8c67c2d..591d71a 100644 (file)
--- a/osm_nbi/base_topic.py
+++ b/osm_nbi/base_topic.py
@@ -17,7 +17,7 @@ import logging
 from uuid import uuid4
 from http import HTTPStatus
 from time import time
-from osm_common.dbbase import deep_update_rfc7396
+from osm_common.dbbase import deep_update_rfc7396, DbException
 from osm_nbi.validation import validate_input, ValidationError, is_valid_uuid
 from yaml import safe_load, YAMLError
 
@@ -29,6 +29,20 @@ class EngineException(Exception):
         self.http_code = http_code
         super(Exception, self).__init__(message)
 
+class NBIBadArgumentsException(Exception):
+    """
+    Bad argument values exception
+    """
+
+    def __init__(self, message: str = "", bad_args: list = None):
+        Exception.__init__(self, message)
+        self.message = message
+        self.bad_args = bad_args
+
+    def __str__(self):
+        return "{}, Bad arguments: {}".format(
+            self.message, self.bad_args
+        )
 
 def deep_get(target_dict, key_list):
     """
@@ -45,6 +59,79 @@ def deep_get(target_dict, key_list):
     return target_dict
 
 
+def detect_descriptor_usage(
+    descriptor: dict, db_collection: str, db: object
+) -> bool:
+    """Detect the descriptor usage state.
+
+    Args:
+        descriptor (dict):   VNF or NS Descriptor as dictionary
+        db_collection (str):   collection name which is looked for in DB
+        db (object):   name of db object
+
+    Returns:
+        True if descriptor is in use else None
+
+    """
+    try:
+        if not descriptor:
+            raise NBIBadArgumentsException(
+                "Argument is mandatory and can not be empty", "descriptor"
+            )
+
+        if not db:
+            raise NBIBadArgumentsException("A valid DB object should be provided", "db")
+
+        search_dict = {
+            "vnfds": ("vnfrs", "vnfd-id"),
+            "nsds": ("nsrs", "nsd-id"),
+        }
+
+        if db_collection not in search_dict:
+            raise NBIBadArgumentsException("db_collection should be equal to vnfds or nsds", "db_collection")
+
+        record_list = db.get_list(
+            search_dict[db_collection][0],
+            {search_dict[db_collection][1]: descriptor["_id"]},
+        )
+
+        if record_list:
+            return True
+
+    except (DbException, KeyError, NBIBadArgumentsException) as error:
+        raise EngineException(f"Error occured while detecting the descriptor usage: {error}")
+
+
+def update_descriptor_usage_state(
+    descriptor: dict, db_collection: str, db: object
+) -> None:
+    """Updates the descriptor usage state.
+
+    Args:
+        descriptor (dict):   VNF or NS Descriptor as dictionary
+        db_collection (str):   collection name which is looked for in DB
+        db (object):   name of db object
+
+    Returns:
+        None
+
+    """
+    try:
+        descriptor_update = {
+            "_admin.usageState": "NOT_IN_USE",
+        }
+
+        if detect_descriptor_usage(descriptor, db_collection, db):
+            descriptor_update = {
+                "_admin.usageState": "IN_USE",
+            }
+
+        db.set_one(db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update)
+
+    except (DbException, KeyError, NBIBadArgumentsException) as error:
+        raise EngineException(f"Error occured while updating the descriptor usage state: {error}")
+
+
 def get_iterable(input_var):
     """
     Returns an iterable, in case input_var is None it just returns an empty tuple