n2vc/utils.py

   1 # Copyright 2020 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import base64
  16 import re
  17 import binascii
  18 import yaml
  19 import string
  20 import secrets
  21 from enum import Enum
  22 from juju.machine import Machine
  23 from juju.application import Application
  24 from juju.action import Action
  25 from juju.unit import Unit
  26 from n2vc.exceptions import N2VCInvalidCertificate
  27 from typing import Tuple
  28
  29
  30 def base64_to_cacert(b64string):
  31     """Convert the base64-encoded string containing the VCA CACERT.
  32
  33     The input string....
  34
  35     """
  36     try:
  37         cacert = base64.b64decode(b64string).decode("utf-8")
  38
  39         cacert = re.sub(
  40             r"\\n",
  41             r"\n",
  42             cacert,
  43         )
  44     except binascii.Error as e:
  45         raise N2VCInvalidCertificate(message="Invalid CA Certificate: {}".format(e))
  46
  47     return cacert
  48
  49
  50 class N2VCDeploymentStatus(Enum):
  51     PENDING = "pending"
  52     RUNNING = "running"
  53     COMPLETED = "completed"
  54     FAILED = "failed"
  55     UNKNOWN = "unknown"
  56
  57
  58 class Dict(dict):
  59     """
  60     Dict class that allows to access the keys like attributes
  61     """
  62
  63     def __getattribute__(self, name):
  64         if name in self:
  65             return self[name]
  66
  67
  68 class EntityType(Enum):
  69     MACHINE = Machine
  70     APPLICATION = Application
  71     ACTION = Action
  72     UNIT = Unit
  73
  74     @classmethod
  75     def has_value(cls, value):
  76         return value in cls._value2member_map_  # pylint: disable=E1101
  77
  78     @classmethod
  79     def get_entity(cls, value):
  80         return (
  81             cls._value2member_map_[value]  # pylint: disable=E1101
  82             if value in cls._value2member_map_  # pylint: disable=E1101
  83             else None  # pylint: disable=E1101
  84         )
  85
  86     @classmethod
  87     def get_entity_from_delta(cls, delta_entity: str):
  88         """
  89         Get Value from delta entity
  90
  91         :param: delta_entity: Possible values are "machine", "application", "unit", "action"
  92         """
  93         for v in cls._value2member_map_:  # pylint: disable=E1101
  94             if v.__name__.lower() == delta_entity:
  95                 return cls.get_entity(v)
  96
  97
  98 JujuStatusToOSM = {
  99     "machine": {
 100         "pending": N2VCDeploymentStatus.PENDING,
 101         "started": N2VCDeploymentStatus.COMPLETED,
 102     },
 103     "application": {
 104         "waiting": N2VCDeploymentStatus.RUNNING,
 105         "maintenance": N2VCDeploymentStatus.RUNNING,
 106         "blocked": N2VCDeploymentStatus.RUNNING,
 107         "error": N2VCDeploymentStatus.FAILED,
 108         "active": N2VCDeploymentStatus.COMPLETED,
 109     },
 110     "action": {
 111         "pending": N2VCDeploymentStatus.PENDING,
 112         "running": N2VCDeploymentStatus.RUNNING,
 113         "completed": N2VCDeploymentStatus.COMPLETED,
 114     },
 115     "unit": {
 116         "waiting": N2VCDeploymentStatus.RUNNING,
 117         "maintenance": N2VCDeploymentStatus.RUNNING,
 118         "blocked": N2VCDeploymentStatus.RUNNING,
 119         "error": N2VCDeploymentStatus.FAILED,
 120         "active": N2VCDeploymentStatus.COMPLETED,
 121     },
 122 }
 123
 124
 125 def obj_to_yaml(obj: object) -> str:
 126     """
 127     Converts object to yaml format
 128     :return: yaml data
 129     """
 130     # dump to yaml
 131     dump_text = yaml.dump(obj, default_flow_style=False, indent=2)
 132     # split lines
 133     lines = dump_text.splitlines()
 134     # remove !!python/object tags
 135     yaml_text = ""
 136     for line in lines:
 137         index = line.find("!!python/object")
 138         if index >= 0:
 139             line = line[:index]
 140         yaml_text += line + "\n"
 141     return yaml_text
 142
 143
 144 def obj_to_dict(obj: object) -> dict:
 145     """
 146     Converts object to dictionary format
 147     :return: dict data
 148     """
 149     # convert obj to yaml
 150     yaml_text = obj_to_yaml(obj)
 151     # parse to dict
 152     return yaml.load(yaml_text, Loader=yaml.Loader)
 153
 154
 155 def get_ee_id_components(ee_id: str) -> Tuple[str, str, str]:
 156     """
 157     Get model, application and machine components from an execution environment id
 158     :param ee_id:
 159     :return: model_name, application_name, machine_id
 160     """
 161     parts = ee_id.split(".")
 162     if len(parts) != 3:
 163         raise Exception("invalid ee id.")
 164     model_name = parts[0]
 165     application_name = parts[1]
 166     machine_id = parts[2]
 167     return model_name, application_name, machine_id
 168
 169
 170 def generate_random_alfanum_string(size: int) -> str:
 171     """
 172     Generate random alfa-numeric string with a size given by argument
 173     :param size:
 174     :return: random generated string
 175     """
 176
 177     return "".join(
 178         secrets.choice(string.ascii_letters + string.digits) for i in range(size)
 179     )