projects
/
osm
/
NBI.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fix Bug 2229 Set fixed IP address for VDU through VNFD and the instantiation params
[osm/NBI.git]
/
osm_nbi
/
base_topic.py
diff --git
a/osm_nbi/base_topic.py
b/osm_nbi/base_topic.py
index
591d71a
..
820bd8e
100644
(file)
--- a/
osm_nbi/base_topic.py
+++ b/
osm_nbi/base_topic.py
@@
-29,6
+29,7
@@
class EngineException(Exception):
self.http_code = http_code
super(Exception, self).__init__(message)
self.http_code = http_code
super(Exception, self).__init__(message)
+
class NBIBadArgumentsException(Exception):
"""
Bad argument values exception
class NBIBadArgumentsException(Exception):
"""
Bad argument values exception
@@
-40,9
+41,8
@@
class NBIBadArgumentsException(Exception):
self.bad_args = bad_args
def __str__(self):
self.bad_args = bad_args
def __str__(self):
- return "{}, Bad arguments: {}".format(
- self.message, self.bad_args
- )
+ return "{}, Bad arguments: {}".format(self.message, self.bad_args)
+
def deep_get(target_dict, key_list):
"""
def deep_get(target_dict, key_list):
"""
@@
-59,9
+59,7
@@
def deep_get(target_dict, key_list):
return target_dict
return target_dict
-def detect_descriptor_usage(
- descriptor: dict, db_collection: str, db: object
-) -> bool:
+def detect_descriptor_usage(descriptor: dict, db_collection: str, db: object) -> bool:
"""Detect the descriptor usage state.
Args:
"""Detect the descriptor usage state.
Args:
@@
-88,7
+86,9
@@
def detect_descriptor_usage(
}
if db_collection not in search_dict:
}
if db_collection not in search_dict:
- raise NBIBadArgumentsException("db_collection should be equal to vnfds or nsds", "db_collection")
+ raise NBIBadArgumentsException(
+ "db_collection should be equal to vnfds or nsds", "db_collection"
+ )
record_list = db.get_list(
search_dict[db_collection][0],
record_list = db.get_list(
search_dict[db_collection][0],
@@
-99,7
+99,9
@@
def detect_descriptor_usage(
return True
except (DbException, KeyError, NBIBadArgumentsException) as error:
return True
except (DbException, KeyError, NBIBadArgumentsException) as error:
- raise EngineException(f"Error occured while detecting the descriptor usage: {error}")
+ raise EngineException(
+ f"Error occured while detecting the descriptor usage: {error}"
+ )
def update_descriptor_usage_state(
def update_descriptor_usage_state(
@@
-126,10
+128,14
@@
def update_descriptor_usage_state(
"_admin.usageState": "IN_USE",
}
"_admin.usageState": "IN_USE",
}
- db.set_one(db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update)
+ db.set_one(
+ db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
+ )
except (DbException, KeyError, NBIBadArgumentsException) as error:
except (DbException, KeyError, NBIBadArgumentsException) as error:
- raise EngineException(f"Error occured while updating the descriptor usage state: {error}")
+ raise EngineException(
+ f"Error occured while updating the descriptor usage state: {error}"
+ )
def get_iterable(input_var):
def get_iterable(input_var):