diff options
Diffstat (limited to 'src/onapsdk/utils/headers_creator.py')
-rw-r--r-- | src/onapsdk/utils/headers_creator.py | 245 |
1 files changed, 245 insertions, 0 deletions
diff --git a/src/onapsdk/utils/headers_creator.py b/src/onapsdk/utils/headers_creator.py new file mode 100644 index 0000000..adb0609 --- /dev/null +++ b/src/onapsdk/utils/headers_creator.py @@ -0,0 +1,245 @@ +"""Header creator package.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from typing import Dict +from uuid import uuid4 +import base64 +import hashlib + +from onapsdk.configuration import settings + + +def headers_sdc_creator(base_header: Dict[str, str], + user: str = "cs0008", + authorization: str = None): + """ + Create the right headers for SDC creator type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to cs0008 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_tester(base_header: Dict[str, str], + user: str = "jm0007", + authorization: str = None): + """ + Create the right headers for SDC tester type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to jm0007 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_governor(base_header: Dict[str, str], + user: str = "gv0001", + authorization: str = None): + """ + Create the right headers for SDC governor type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to gv0001 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_operator(base_header: Dict[str, str], + user: str = "op0001", + authorization: str = None): + """ + Create the right headers for SDC operator type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to op0001 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_generic(base_header: Dict[str, str], + user: str, + authorization: str = None): + """ + Create the right headers for SDC generic type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str): the user to use. + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["USER_ID"] = user + headers["Authorization"] = authorization or settings.SDC_AUTH + headers["X-ECOMP-InstanceID"] = "onapsdk" + return headers + + +def headers_aai_creator(base_header: Dict[str, str]): + """ + Create the right headers for AAI creator type. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["x-fromappid"] = "AAI" + headers["x-transactionid"] = "0a3f6713-ba96-4971-a6f8-c2da85a3176e" + headers["authorization"] = settings.AAI_AUTH + return headers + + +def headers_so_creator(base_header: Dict[str, str]): + """ + Create the right headers for SO creator type. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["x-fromappid"] = "AAI" + headers["x-transactionid"] = str(uuid4()) + headers["authorization"] = settings.SO_AUTH + headers["cache-control"] = "no-cache" + return headers + +def headers_so_catelog_db_creator(base_header: Dict[str, str]): + """ + Create the right headers for SO creator type. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["x-fromappid"] = "AAI" + headers["x-transactionid"] = str(uuid4()) + headers["authorization"] = settings.SO_CAT_DB_AUTH + headers["cache-control"] = "no-cache" + return headers + +def headers_msb_creator(base_header: Dict[str, str]): + """ + Create the right headers for MSB. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["cache-control"] = "no-cache" + return headers + + +def headers_sdnc_creator(base_header: Dict[str, str]): + """ + Create the right headers for SDNC. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["authorization"] = settings.SDNC_AUTH + headers["x-transactionid"] = str(uuid4()) + headers["x-fromappid"] = "API client" + return headers + + +def headers_sdc_artifact_upload(base_header: Dict[str, str], data: str): + """ + Create the right headers for sdc artifact upload. + + Args: + base_header (Dict[str, str]): the base header to use + data (str): payload data used to create an md5 content header + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["Accept"] = "application/json, text/plain, */*" + headers["Accept-Encoding"] = "gzip, deflate, br" + headers["Content-Type"] = "application/json; charset=UTF-8" + md5_content = hashlib.md5(data.encode('UTF-8')).hexdigest() + content = base64.b64encode(md5_content.encode('ascii')).decode('UTF-8') + headers["Content-MD5"] = content + return headers + +def headers_clamp_creator(base_header: Dict[str, str]): + """ + Create the right headers for CLAMP generic type. + + base_header (Dict[str, str]): the base header to use + data (str): payload data used to create an md5 content header + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["Authorization"] = settings.CLAMP_AUTH + headers["X-ECOMP-InstanceID"] = "onapsdk" + return headers |