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