},
"verticalscale": {
"METHODS": ("POST",),
- "ROLE_PERMISSION": "ns_instances:id:verticalscale:"
- },
+ "ROLE_PERMISSION": "ns_instances:id:verticalscale:",
+ },
},
},
"ns_lcm_op_occs": {
},
"vnflcm": {
"v1": {
- "vnf_instances": {"METHODS": ("GET", "POST"),
- "ROLE_PERMISSION": "vnflcm_instances:",
- "<ID>": {"METHODS": ("GET", "DELETE"),
- "ROLE_PERMISSION": "vnflcm_instances:id:",
- "scale": {"METHODS": ("POST",),
- "ROLE_PERMISSION": "vnflcm_instances:id:scale:"
- },
- "terminate": {"METHODS": ("POST",),
- "ROLE_PERMISSION": "vnflcm_instances:id:terminate:"
- },
- "instantiate": {"METHODS": ("POST",),
- "ROLE_PERMISSION": "vnflcm_instances:id:instantiate:"
- },
- }
- },
- "vnf_lcm_op_occs": {"METHODS": ("GET",),
- "ROLE_PERMISSION": "vnf_instances:opps:",
- "<ID>": {"METHODS": ("GET",),
- "ROLE_PERMISSION": "vnf_instances:opps:id:"
- },
- },
- "subscriptions": {"METHODS": ("GET", "POST"),
- "ROLE_PERMISSION": "vnflcm_subscriptions:",
- "<ID>": {"METHODS": ("GET", "DELETE"),
- "ROLE_PERMISSION": "vnflcm_subscriptions:id:"
- }
- },
+ "vnf_instances": {
+ "METHODS": ("GET", "POST"),
+ "ROLE_PERMISSION": "vnflcm_instances:",
+ "<ID>": {
+ "METHODS": ("GET", "DELETE"),
+ "ROLE_PERMISSION": "vnflcm_instances:id:",
+ "scale": {
+ "METHODS": ("POST",),
+ "ROLE_PERMISSION": "vnflcm_instances:id:scale:",
+ },
+ "terminate": {
+ "METHODS": ("POST",),
+ "ROLE_PERMISSION": "vnflcm_instances:id:terminate:",
+ },
+ "instantiate": {
+ "METHODS": ("POST",),
+ "ROLE_PERMISSION": "vnflcm_instances:id:instantiate:",
+ },
+ },
+ },
+ "vnf_lcm_op_occs": {
+ "METHODS": ("GET",),
+ "ROLE_PERMISSION": "vnf_instances:opps:",
+ "<ID>": {
+ "METHODS": ("GET",),
+ "ROLE_PERMISSION": "vnf_instances:opps:id:",
+ },
+ },
+ "subscriptions": {
+ "METHODS": ("GET", "POST"),
+ "ROLE_PERMISSION": "vnflcm_subscriptions:",
+ "<ID>": {
+ "METHODS": ("GET", "DELETE"),
+ "ROLE_PERMISSION": "vnflcm_subscriptions:id:",
+ },
+ },
}
},
"nst": {
},
"nsfm": {
"v1": {
- "alarms": {"METHODS": ("GET", "PATCH"),
- "ROLE_PERMISSION": "alarms:",
- "<ID>": {"METHODS": ("GET", "PATCH"),
- "ROLE_PERMISSION": "alarms:id:",
- },
- }
+ "alarms": {
+ "METHODS": ("GET", "PATCH"),
+ "ROLE_PERMISSION": "alarms:",
+ "<ID>": {
+ "METHODS": ("GET", "PATCH"),
+ "ROLE_PERMISSION": "alarms:id:",
+ },
+ }
},
},
}
self.engine = Engine(self.authenticator)
def _format_in(self, kwargs):
+ error_text = "" # error_text must be initialized outside try
try:
indata = None
if cherrypy.request.body.length:
# NS Fault Management
@cherrypy.expose
- def nsfm(self, version=None, topic=None, uuid=None, project_name=None, ns_id=None, *args, **kwargs):
- if topic == 'alarms':
+ def nsfm(
+ self,
+ version=None,
+ topic=None,
+ uuid=None,
+ project_name=None,
+ ns_id=None,
+ *args,
+ **kwargs
+ ):
+ if topic == "alarms":
try:
method = cherrypy.request.method
- role_permission = self._check_valid_url_method(method, "nsfm", version, topic, None, None, *args)
- query_string_operations = self._extract_query_string_operations(kwargs, method)
+ role_permission = self._check_valid_url_method(
+ method, "nsfm", version, topic, None, None, *args
+ )
+ query_string_operations = self._extract_query_string_operations(
+ kwargs, method
+ )
- self.authenticator.authorize(role_permission, query_string_operations, None)
+ self.authenticator.authorize(
+ role_permission, query_string_operations, None
+ )
# to handle get request
- if cherrypy.request.method == 'GET':
+ if cherrypy.request.method == "GET":
# if request is on basis of uuid
- if uuid and uuid != 'None':
+ if uuid and uuid != "None":
try:
alarm = self.engine.db.get_one("alarms", {"uuid": uuid})
- alarm_action = self.engine.db.get_one("alarms_action", {"uuid": uuid})
+ alarm_action = self.engine.db.get_one(
+ "alarms_action", {"uuid": uuid}
+ )
alarm.update(alarm_action)
- vnf = self.engine.db.get_one("vnfrs", {"nsr-id-ref": alarm["tags"]["ns_id"]})
+ vnf = self.engine.db.get_one(
+ "vnfrs", {"nsr-id-ref": alarm["tags"]["ns_id"]}
+ )
alarm["vnf-id"] = vnf["_id"]
return self._format_out(str(alarm))
except Exception:
return self._format_out("Please provide valid alarm uuid")
- elif ns_id and ns_id != 'None':
+ elif ns_id and ns_id != "None":
# if request is on basis of ns_id
try:
- alarms = self.engine.db.get_list("alarms", {"tags.ns_id": ns_id})
+ alarms = self.engine.db.get_list(
+ "alarms", {"tags.ns_id": ns_id}
+ )
for alarm in alarms:
- alarm_action = self.engine.db.get_one("alarms_action", {"uuid": alarm['uuid']})
+ alarm_action = self.engine.db.get_one(
+ "alarms_action", {"uuid": alarm["uuid"]}
+ )
alarm.update(alarm_action)
return self._format_out(str(alarms))
except Exception:
return self._format_out("Please provide valid ns id")
else:
# to return only alarm which are related to given project
- project = self.engine.db.get_one("projects", {"name": project_name})
- project_id = project.get('_id')
- ns_list = self.engine.db.get_list("nsrs", {"_admin.projects_read": project_id})
+ project = self.engine.db.get_one(
+ "projects", {"name": project_name}
+ )
+ project_id = project.get("_id")
+ ns_list = self.engine.db.get_list(
+ "nsrs", {"_admin.projects_read": project_id}
+ )
ns_ids = []
for ns in ns_list:
ns_ids.append(ns.get("_id"))
alarms = self.engine.db.get_list("alarms")
- alarm_list = [alarm for alarm in alarms if alarm["tags"]["ns_id"] in ns_ids]
+ alarm_list = [
+ alarm
+ for alarm in alarms
+ if alarm["tags"]["ns_id"] in ns_ids
+ ]
for alrm in alarm_list:
- action = self.engine.db.get_one("alarms_action", {"uuid": alrm.get("uuid")})
+ action = self.engine.db.get_one(
+ "alarms_action", {"uuid": alrm.get("uuid")}
+ )
alrm.update(action)
return self._format_out(str(alarm_list))
# to handle patch request for alarm update
- elif cherrypy.request.method == 'PATCH':
+ elif cherrypy.request.method == "PATCH":
data = yaml.load(cherrypy.request.body, Loader=yaml.SafeLoader)
try:
# check if uuid is valid
return self._format_out("Please provide valid alarm uuid.")
if data.get("is_enable") is not None:
if data.get("is_enable"):
- alarm_status = 'ok'
+ alarm_status = "ok"
else:
- alarm_status = 'disabled'
- self.engine.db.set_one("alarms", {"uuid": data.get("uuid")},
- {"alarm_status": alarm_status})
+ alarm_status = "disabled"
+ self.engine.db.set_one(
+ "alarms",
+ {"uuid": data.get("uuid")},
+ {"alarm_status": alarm_status},
+ )
else:
- self.engine.db.set_one("alarms", {"uuid": data.get("uuid")},
- {"threshold": data.get("threshold")})
+ self.engine.db.set_one(
+ "alarms",
+ {"uuid": data.get("uuid")},
+ {"threshold": data.get("threshold")},
+ )
return self._format_out("Alarm updated")
except Exception as e:
- cherrypy.response.status = e.http_code.value
- if isinstance(e, (NbiException, EngineException, DbException, FsException, MsgException, AuthException,
- ValidationError, AuthconnException)):
+ if isinstance(
+ e,
+ (
+ NbiException,
+ EngineException,
+ DbException,
+ FsException,
+ MsgException,
+ AuthException,
+ ValidationError,
+ AuthconnException,
+ ),
+ ):
http_code_value = cherrypy.response.status = e.http_code.value
http_code_name = e.http_code.name
cherrypy.log("Exception {}".format(e))
else:
- http_code_value = cherrypy.response.status = HTTPStatus.BAD_REQUEST.value # INTERNAL_SERVER_ERROR
+ http_code_value = (
+ cherrypy.response.status
+ ) = HTTPStatus.BAD_REQUEST.value # INTERNAL_SERVER_ERROR
cherrypy.log("CRITICAL: Exception {}".format(e), traceback=True)
http_code_name = HTTPStatus.BAD_REQUEST.name
problem_details = {
outdata = token_info = self.authenticator.new_token(
token_info, indata, cherrypy.request.remote
)
- cherrypy.session["Authorization"] = outdata["_id"]
+ cherrypy.session["Authorization"] = outdata["_id"] # pylint: disable=E1101
self._set_location_header("admin", "v1", "tokens", outdata["_id"])
# for logging
self._format_login(token_info)
# password expiry check
if self.authenticator.check_password_expiry(outdata):
- outdata = {"id": outdata["id"],
- "message": "change_password",
- "user_id": outdata["user_id"]
- }
+ outdata = {
+ "id": outdata["id"],
+ "message": "change_password",
+ "user_id": outdata["user_id"],
+ }
# cherrypy.response.cookie["Authorization"] = outdata["id"]
# cherrypy.response.cookie["Authorization"]['expires'] = 3600
elif method == "DELETE":
token_id = token_info["_id"]
outdata = self.authenticator.del_token(token_id)
token_info = None
- cherrypy.session["Authorization"] = "logout"
+ cherrypy.session["Authorization"] = "logout" # pylint: disable=E1101
# cherrypy.response.cookie["Authorization"] = token_id
# cherrypy.response.cookie["Authorization"]['expires'] = 0
else:
elif args and args[0] == "init":
try:
# self.engine.load_dbase(cherrypy.request.app.config)
- self.engine.create_admin()
+ pid = self.authenticator.create_admin_project()
+ self.authenticator.create_admin_user(pid)
return "Done. User 'admin', password 'admin' created"
except Exception:
cherrypy.response.status = HTTPStatus.FORBIDDEN.value
+ " headers: {}\n".format(cherrypy.request.headers)
+ " path_info: {}\n".format(cherrypy.request.path_info)
+ " query_string: {}\n".format(cherrypy.request.query_string)
- + " session: {}\n".format(cherrypy.session)
+ + " session: {}\n".format(cherrypy.session) # pylint: disable=E1101
+ " cookie: {}\n".format(cherrypy.request.cookie)
+ " method: {}\n".format(cherrypy.request.method)
- + " session: {}\n".format(cherrypy.session.get("fieldname"))
+ + " session: {}\n".format(
+ cherrypy.session.get("fieldname") # pylint: disable=E1101
+ )
+ " body:\n"
)
return_text += " length: {}\n".format(cherrypy.request.body.length)
filter_q = None
if "vcaStatusRefresh" in kwargs:
filter_q = {"vcaStatusRefresh": kwargs["vcaStatusRefresh"]}
- outdata = self.engine.get_item(engine_session, engine_topic, _id, filter_q, True)
+ outdata = self.engine.get_item(
+ engine_session, engine_topic, _id, filter_q, True
+ )
elif method == "POST":
cherrypy.response.status = HTTPStatus.CREATED.value
elif topic == "vnf_instances" and item:
indata["lcmOperationType"] = item
indata["vnfInstanceId"] = _id
- _id, _ = self.engine.new_item(rollback, engine_session, "vnflcmops", indata, kwargs)
- self._set_location_header(main_topic, version, "vnf_lcm_op_occs", _id)
+ _id, _ = self.engine.new_item(
+ rollback, engine_session, "vnflcmops", indata, kwargs
+ )
+ self._set_location_header(
+ main_topic, version, "vnf_lcm_op_occs", _id
+ )
outdata = {"id": _id}
cherrypy.response.status = HTTPStatus.ACCEPTED.value
else:
self.engine.db.del_list(
rollback_item["topic"],
rollback_item["filter"],
- fail_on_empty=False,
)
else:
self.engine.db.del_one(
except ValueError as e:
cherrypy.log.error("Ignoring environ '{}': " + str(e))
except Exception as e:
- cherrypy.log.warn("skipping environ '{}' on exception '{}'".format(k, e))
+ cherrypy.log(
+ "WARNING: skipping environ '{}' on exception '{}'".format(k, e)
+ )
if update_dict:
cherrypy.config.update(update_dict)
projects / osm / NBI.git / blobdiff