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 from enum import Enum
  20 from juju.machine import Machine
  21 from juju.application import Application
  22 from juju.action import Action
  23 from juju.unit import Unit
  24 from n2vc.exceptions import N2VCInvalidCertificate
  25 from typing import Tuple
  26
  27
  28 def base64_to_cacert(b64string):
  29     """Convert the base64-encoded string containing the VCA CACERT.
  30
  31     The input string....
  32
  33     """
  34     try:
  35         cacert = base64.b64decode(b64string).decode("utf-8")
  36
  37         cacert = re.sub(
  38             r"\\n",
  39             r"\n",
  40             cacert,
  41         )
  42     except binascii.Error as e:
  43         raise N2VCInvalidCertificate(message="Invalid CA Certificate: {}".format(e))
  44
  45     return cacert
  46
  47
  48 class N2VCDeploymentStatus(Enum):
  49     PENDING = "pending"
  50     RUNNING = "running"
  51     COMPLETED = "completed"
  52     FAILED = "failed"
  53     UNKNOWN = "unknown"
  54
  55
  56 class Dict(dict):
  57     """
  58     Dict class that allows to access the keys like attributes
  59     """
  60
  61     def __getattribute__(self, name):
  62         if name in self:
  63             return self[name]
  64
  65
  66 class EntityType(Enum):
  67     MACHINE = Machine
  68     APPLICATION = Application
  69     ACTION = Action
  70     UNIT = Unit
  71
  72     @classmethod
  73     def has_value(cls, value):
  74         return value in cls._value2member_map_  # pylint: disable=E1101
  75
  76     @classmethod
  77     def get_entity(cls, value):
  78         return (
  79             cls._value2member_map_[value]  # pylint: disable=E1101
  80             if value in cls._value2member_map_  # pylint: disable=E1101
  81             else None  # pylint: disable=E1101
  82         )
  83
  84     @classmethod
  85     def get_entity_from_delta(cls, delta_entity: str):
  86         """
  87         Get Value from delta entity
  88
  89         :param: delta_entity: Possible values are "machine", "application", "unit", "action"
  90         """
  91         for v in cls._value2member_map_:  # pylint: disable=E1101
  92             if v.__name__.lower() == delta_entity:
  93                 return cls.get_entity(v)
  94
  95
  96 JujuStatusToOSM = {
  97     "machine": {
  98         "pending": N2VCDeploymentStatus.PENDING,
  99         "started": N2VCDeploymentStatus.COMPLETED,
 100     },
 101     "application": {
 102         "waiting": N2VCDeploymentStatus.RUNNING,
 103         "maintenance": N2VCDeploymentStatus.RUNNING,
 104         "blocked": N2VCDeploymentStatus.RUNNING,
 105         "error": N2VCDeploymentStatus.FAILED,
 106         "active": N2VCDeploymentStatus.COMPLETED,
 107     },
 108     "action": {
 109         "pending": N2VCDeploymentStatus.PENDING,
 110         "running": N2VCDeploymentStatus.RUNNING,
 111         "completed": N2VCDeploymentStatus.COMPLETED,
 112     },
 113     "unit": {
 114         "waiting": N2VCDeploymentStatus.RUNNING,
 115         "maintenance": N2VCDeploymentStatus.RUNNING,
 116         "blocked": N2VCDeploymentStatus.RUNNING,
 117         "error": N2VCDeploymentStatus.FAILED,
 118         "active": N2VCDeploymentStatus.COMPLETED,
 119     },
 120 }
 121
 122
 123 def obj_to_yaml(obj: object) -> str:
 124     """
 125     Converts object to yaml format
 126     :return: yaml data
 127     """
 128     # dump to yaml
 129     dump_text = yaml.dump(obj, default_flow_style=False, indent=2)
 130     # split lines
 131     lines = dump_text.splitlines()
 132     # remove !!python/object tags
 133     yaml_text = ""
 134     for line in lines:
 135         index = line.find("!!python/object")
 136         if index >= 0:
 137             line = line[:index]
 138         yaml_text += line + "\n"
 139     return yaml_text
 140
 141
 142 def obj_to_dict(obj: object) -> dict:
 143     """
 144     Converts object to dictionary format
 145     :return: dict data
 146     """
 147     # convert obj to yaml
 148     yaml_text = obj_to_yaml(obj)
 149     # parse to dict
 150     return yaml.load(yaml_text, Loader=yaml.Loader)
 151
 152
 153 def get_ee_id_components(ee_id: str) -> Tuple[str, str, str]:
 154     """
 155     Get model, application and machine components from an execution environment id
 156     :param ee_id:
 157     :return: model_name, application_name, machine_id
 158     """
 159     parts = ee_id.split(".")
 160     if len(parts) != 3:
 161         raise Exception("invalid ee id.")
 162     model_name = parts[0]
 163     application_name = parts[1]
 164     machine_id = parts[2]
 165     return model_name, application_name, machine_id