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