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 from enum import Enum
  19 from juju.machine import Machine
  20 from juju.application import Application
  21 from juju.action import Action
  22 from juju.unit import Unit
  23 from n2vc.exceptions import N2VCInvalidCertificate
  24
  25
  26 def base64_to_cacert(b64string):
  27     """Convert the base64-encoded string containing the VCA CACERT.
  28
  29     The input string....
  30
  31     """
  32     try:
  33         cacert = base64.b64decode(b64string).decode("utf-8")
  34
  35         cacert = re.sub(r"\\n", r"\n", cacert,)
  36     except binascii.Error as e:
  37         raise N2VCInvalidCertificate(message="Invalid CA Certificate: {}".format(e))
  38
  39     return cacert
  40
  41
  42 class N2VCDeploymentStatus(Enum):
  43     PENDING = "pending"
  44     RUNNING = "running"
  45     COMPLETED = "completed"
  46     FAILED = "failed"
  47     UNKNOWN = "unknown"
  48
  49
  50 class Dict(dict):
  51     """
  52     Dict class that allows to access the keys like attributes
  53     """
  54
  55     def __getattribute__(self, name):
  56         if name in self:
  57             return self[name]
  58
  59
  60 class EntityType(Enum):
  61     MACHINE = Machine
  62     APPLICATION = Application
  63     ACTION = Action
  64     UNIT = Unit
  65
  66     @classmethod
  67     def has_value(cls, value):
  68         return value in cls._value2member_map_  # pylint: disable=E1101
  69
  70     @classmethod
  71     def get_entity(cls, value):
  72         return (
  73             cls._value2member_map_[value]  # pylint: disable=E1101
  74             if value in cls._value2member_map_  # pylint: disable=E1101
  75             else None  # pylint: disable=E1101
  76         )
  77
  78     @classmethod
  79     def get_entity_from_delta(cls, delta_entity: str):
  80         """
  81         Get Value from delta entity
  82
  83         :param: delta_entity: Possible values are "machine", "application", "unit", "action"
  84         """
  85         for v in cls._value2member_map_:  # pylint: disable=E1101
  86             if v.__name__.lower() == delta_entity:
  87                 return cls.get_entity(v)
  88
  89
  90 JujuStatusToOSM = {
  91     "machine": {
  92         "pending": N2VCDeploymentStatus.PENDING,
  93         "started": N2VCDeploymentStatus.COMPLETED,
  94     },
  95     "application": {
  96         "waiting": N2VCDeploymentStatus.RUNNING,
  97         "maintenance": N2VCDeploymentStatus.RUNNING,
  98         "blocked": N2VCDeploymentStatus.RUNNING,
  99         "error": N2VCDeploymentStatus.FAILED,
 100         "active": N2VCDeploymentStatus.COMPLETED,
 101     },
 102     "action": {
 103         "pending": N2VCDeploymentStatus.PENDING,
 104         "running": N2VCDeploymentStatus.RUNNING,
 105         "completed": N2VCDeploymentStatus.COMPLETED,
 106     },
 107     "unit": {
 108         "waiting": N2VCDeploymentStatus.RUNNING,
 109         "maintenance": N2VCDeploymentStatus.RUNNING,
 110         "blocked": N2VCDeploymentStatus.RUNNING,
 111         "error": N2VCDeploymentStatus.FAILED,
 112         "active": N2VCDeploymentStatus.COMPLETED,
 113     },
 114 }
 115
 116 DB_DATA = Dict(
 117     {
 118         "api_endpoints": Dict(
 119             {"table": "admin", "filter": {"_id": "juju"}, "key": "api_endpoints"}
 120         )
 121     }
 122 )