smo-install/test/pythonsdk/src/oransdk/policy/policy.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2021-2022 AT&T Intellectual Property. All rights
   7 #                             reserved.
   8 # ================================================================================
   9 # Licensed under the Apache License, Version 2.0 (the "License");
  10 # you may not use this file except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 # http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 # ============LICENSE_END============================================
  21 # ===================================================================
  22 #
  23 ###
  24 """Onap Policy module."""
  25
  26 from dataclasses import dataclass
  27 from typing import Dict
  28 from onapsdk.onap_service import OnapService
  29 from oransdk.configuration import settings
  30
  31 @dataclass
  32 class PolicyType:
  33     """PolicyType dataclass."""
  34
  35     type: str
  36     version: str
  37
  38
  39 class OranPolicy(OnapService):
  40     """Onap Policy library."""
  41
  42     pap_url = settings.POLICY_PAP_URL
  43     api_url = settings.POLICY_API_URL
  44     header = {"Accept": "application/json", "Content-Type": "application/json"}
  45
  46     @classmethod
  47     def get_components_status(cls,
  48                               basic_auth: Dict[str, str]) -> Dict:
  49         """
  50         Get status of Policy component.
  51
  52         Args:
  53            basic_auth: (Dict[str, str]) for example:{ 'username': 'bob', 'password': 'secret' }
  54
  55         Returns:
  56            the status of the Policy component
  57
  58         """
  59         url = f"{cls.pap_url}/policy/pap/v1/components/healthcheck"
  60         status = cls.send_message_json('GET',
  61                                        'Get status of Policy component',
  62                                        url,
  63                                        basic_auth=basic_auth)
  64         return status
  65
  66     @classmethod
  67     def get_policy_status(cls,
  68                           basic_auth: Dict[str, str]) -> Dict:
  69         """
  70         Get status of all the policies.
  71
  72         Returns:
  73            the status of all the policies
  74
  75         """
  76         url = f"{cls.pap_url}/policy/pap/v1/policies/status"
  77         status = cls.send_message_json('GET',
  78                                        'Get status of all the policies',
  79                                        url,
  80                                        basic_auth=basic_auth)
  81         return status
  82
  83     @classmethod
  84     def get_policy(cls,
  85                    policy_type: PolicyType,
  86                    policy_name,
  87                    policy_version,
  88                    basic_auth: Dict[str, str]) -> Dict:
  89         """
  90         Get the policy.
  91
  92         Args:
  93            policy_type: the policy type
  94            policy_name: the policy name
  95            policy_version: the version of the policy
  96            basic_auth: (Dict[str, str]) for example:{ 'username': 'bob', 'password': 'secret' }
  97
  98         Returns:
  99            the policy
 100
 101         """
 102         url = f"{cls.api_url}/policy/api/v1/policytypes/{policy_type.type}/versions/"\
 103               + f"{policy_type.version}/policies/{policy_name}/versions/{policy_version}"
 104         policy = cls.send_message_json('GET',
 105                                        'Get the policy',
 106                                        url,
 107                                        basic_auth=basic_auth)
 108         return policy
 109
 110     @classmethod
 111     def create_policy(cls,
 112                       policy_type: PolicyType,
 113                       policy_data,
 114                       basic_auth: Dict[str, str]) -> None:
 115         """
 116         Create a policy.
 117
 118         Args:
 119            policy_type: the policy type
 120            type_version: the version of the policy type
 121            policy_data: the policy to be created, in binary format
 122
 123         """
 124         url = f"{cls.api_url}/policy/api/v1/policytypes/{policy_type.type}/"\
 125               + f"versions/{policy_type.version}/policies"
 126         cls.send_message('POST',
 127                          'Create Policy',
 128                          url,
 129                          data=policy_data,
 130                          headers=cls.header,
 131                          basic_auth=basic_auth)
 132
 133     @classmethod
 134     def deploy_policy(cls,
 135                       policy_data,
 136                       basic_auth: Dict[str, str]) -> None:
 137         """
 138         Deploy a policy.
 139
 140         Args:
 141            policy_data: the policy to be deployed, in binary format
 142
 143         """
 144         url = f"{cls.pap_url}/policy/pap/v1/pdps/policies"
 145         cls.send_message('POST',
 146                          'Deploy Policy',
 147                          url,
 148                          data=policy_data,
 149                          headers=cls.header,
 150                          basic_auth=basic_auth)