from osm_common.msgbase import MsgException
from osm_ng_ro import version as ro_version, version_date as ro_version_date
import osm_ng_ro.html_out as html
from osm_common.msgbase import MsgException
from osm_ng_ro import version as ro_version, version_date as ro_version_date
import osm_ng_ro.html_out as html
from osm_ng_ro.ns import Ns, NsException
from osm_ng_ro.validation import ValidationError
from osm_ng_ro.vim_admin import VimAdminThread
from osm_ng_ro.ns import Ns, NsException
from osm_ng_ro.validation import ValidationError
from osm_ng_ro.vim_admin import VimAdminThread
cherrypy.request.headers.pop("Content-File-MD5", None)
elif "application/yaml" in cherrypy.request.headers["Content-Type"]:
error_text = "Invalid yaml format "
cherrypy.request.headers.pop("Content-File-MD5", None)
elif "application/yaml" in cherrypy.request.headers["Content-Type"]:
error_text = "Invalid yaml format "
cherrypy.request.headers.pop("Content-File-MD5", None)
elif (
"application/binary" in cherrypy.request.headers["Content-Type"]
cherrypy.request.headers.pop("Content-File-MD5", None)
elif (
"application/binary" in cherrypy.request.headers["Content-Type"]
else:
# raise cherrypy.HTTPError(HTTPStatus.Not_Acceptable,
# "Only 'Content-Type' of type 'application/json' or
# 'application/yaml' for input format are available")
error_text = "Invalid yaml format "
else:
# raise cherrypy.HTTPError(HTTPStatus.Not_Acceptable,
# "Only 'Content-Type' of type 'application/json' or
# 'application/yaml' for input format are available")
error_text = "Invalid yaml format "
cherrypy.request.headers.pop("Content-File-MD5", None)
else:
error_text = "Invalid yaml format "
cherrypy.request.headers.pop("Content-File-MD5", None)
else:
error_text = "Invalid yaml format "
- kwargs[k] = yaml.load(v, Loader=yaml.SafeLoader)
- except Exception:
- pass
+ kwargs[k] = yaml.safe_load(v)
+ except Exception as yaml_error:
+ logging.exception(
+ f"{yaml_error} occured while parsing the yaml"
+ )
- v[index] = yaml.load(v[index], Loader=yaml.SafeLoader)
- except Exception:
- pass
+ v[index] = yaml.safe_load(v[index])
+ except Exception as error:
+ logging.exception(
+ f"{error} occured while parsing the yaml"
+ )
return ",".join(folders) + " folders deleted\n"
elif args and args[0] == "login":
if not cherrypy.request.headers.get("Authorization"):
return ",".join(folders) + " folders deleted\n"
elif args and args[0] == "login":
if not cherrypy.request.headers.get("Authorization"):
- cherrypy.response.headers[
- "WWW-Authenticate"
- ] = 'Basic realm="Access to OSM site", charset="UTF-8"'
+ cherrypy.response.headers["WWW-Authenticate"] = (
+ 'Basic realm="Access to OSM site", charset="UTF-8"'
+ )
cherrypy.response.status = HTTPStatus.UNAUTHORIZED.value
elif args and args[0] == "login2":
if not cherrypy.request.headers.get("Authorization"):
cherrypy.response.status = HTTPStatus.UNAUTHORIZED.value
elif args and args[0] == "login2":
if not cherrypy.request.headers.get("Authorization"):
for k, v in to_send.items():
self.ns.msg.write(main_topic, k, v)
return_text += " {}: {}\n".format(k, v)
elif cherrypy.request.method == "GET":
for k, v in kwargs.items():
for k, v in to_send.items():
self.ns.msg.write(main_topic, k, v)
return_text += " {}: {}\n".format(k, v)
elif cherrypy.request.method == "GET":
for k, v in kwargs.items():
- self.ns.msg.write(
- main_topic, k, yaml.load(v, Loader=yaml.SafeLoader)
- )
- return_text += " {}: {}\n".format(
- k, yaml.load(v, Loader=yaml.SafeLoader)
- )
+ self.ns.msg.write(main_topic, k, yaml.safe_load(v))
+ return_text += " {}: {}\n".format(k, yaml.safe_load(v))
- 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
cherrypy.log("CRITICAL: Exception {}".format(e), traceback=True)
http_code_name = HTTPStatus.BAD_REQUEST.name
# # start subscriptions thread:
vim_admin_thread = VimAdminThread(config=engine_config, engine=ro_server.ns)
vim_admin_thread.start()
# # start subscriptions thread:
vim_admin_thread = VimAdminThread(config=engine_config, engine=ro_server.ns)
vim_admin_thread.start()
# # Do not capture except SubscriptionException
# backend = engine_config["authentication"]["backend"]
# # Do not capture except SubscriptionException
# backend = engine_config["authentication"]["backend"]