diff options
Diffstat (limited to 'src/onapsdk/clamp/clamp_element.py')
-rw-r--r-- | src/onapsdk/clamp/clamp_element.py | 79 |
1 files changed, 79 insertions, 0 deletions
diff --git a/src/onapsdk/clamp/clamp_element.py b/src/onapsdk/clamp/clamp_element.py new file mode 100644 index 0000000..843db42 --- /dev/null +++ b/src/onapsdk/clamp/clamp_element.py @@ -0,0 +1,79 @@ +"""Clamp module.""" +# 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 onapsdk.configuration import settings +from onapsdk.onap_service import OnapService as Onap +from onapsdk.sdc.service import Service +from onapsdk.exceptions import ResourceNotFound +from onapsdk.utils.headers_creator import headers_clamp_creator + + +class Clamp(Onap): + """Mother Class of all CLAMP elements.""" + + #class variable + _base_url = settings.CLAMP_URL + name: str = "CLAMP" + headers = headers_clamp_creator(Onap.headers) + + @classmethod + def base_url(cls) -> str: + """Give back the base url of Clamp.""" + return f"{cls._base_url}/restservices/clds/v2" + + @classmethod + def check_loop_template(cls, service: Service) -> str: + """ + Return loop template name if exists. + + Args: + service (Service): the distributed sdc service with tca blueprint artifact + + Raises: + ResourceNotFound: Template not found. + + Returns: + if required template exists in CLAMP or not + + """ + url = f"{cls.base_url()}/templates/" + for template in cls.send_message_json('GET', + 'Get Loop Templates', + url): + if template["modelService"]["serviceDetails"]["name"] == service.name: + return template["name"] + raise ResourceNotFound("Template not found.") + + @classmethod + def check_policies(cls, policy_name: str, req_policies: int = 30) -> bool: + """ + Ensure that a policy is stored in CLAMP. + + Args: + policy_name (str): policy acronym + req_policies (int): number of required policies in CLAMP + + Returns: + if required policy exists in CLAMP or not + + """ + url = f"{cls.base_url()}/policyToscaModels/" + policies = cls.send_message_json('GET', + 'Get stocked policies', + url) + exist_policy = False + for policy in policies: + if policy["policyAcronym"] == policy_name: + exist_policy = True + return (len(policies) >= req_policies) and exist_policy |