1 # Copyright 2020 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
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
30 def base64_to_cacert(b64string
):
31 """Convert the base64-encoded string containing the VCA CACERT.
37 cacert
= base64
.b64decode(b64string
).decode("utf-8")
44 except binascii
.Error
as e
:
45 raise N2VCInvalidCertificate(message
="Invalid CA Certificate: {}".format(e
))
50 class N2VCDeploymentStatus(Enum
):
53 COMPLETED
= "completed"
60 Dict class that allows to access the keys like attributes
63 def __getattribute__(self
, name
):
68 class EntityType(Enum
):
70 APPLICATION
= Application
75 def has_value(cls
, value
):
76 return value
in cls
._value
2member
_map
_ # pylint: disable=E1101
79 def get_entity(cls
, value
):
81 cls
._value
2member
_map
_[value
] # pylint: disable=E1101
82 if value
in cls
._value
2member
_map
_ # pylint: disable=E1101
83 else None # pylint: disable=E1101
87 def get_entity_from_delta(cls
, delta_entity
: str):
89 Get Value from delta entity
91 :param: delta_entity: Possible values are "machine", "application", "unit", "action"
93 for v
in cls
._value
2member
_map
_: # pylint: disable=E1101
94 if v
.__name
__.lower() == delta_entity
:
95 return cls
.get_entity(v
)
100 "pending": N2VCDeploymentStatus
.PENDING
,
101 "started": N2VCDeploymentStatus
.COMPLETED
,
104 "waiting": N2VCDeploymentStatus
.RUNNING
,
105 "maintenance": N2VCDeploymentStatus
.RUNNING
,
106 "blocked": N2VCDeploymentStatus
.RUNNING
,
107 "error": N2VCDeploymentStatus
.FAILED
,
108 "active": N2VCDeploymentStatus
.COMPLETED
,
111 "pending": N2VCDeploymentStatus
.PENDING
,
112 "running": N2VCDeploymentStatus
.RUNNING
,
113 "completed": N2VCDeploymentStatus
.COMPLETED
,
116 "waiting": N2VCDeploymentStatus
.RUNNING
,
117 "maintenance": N2VCDeploymentStatus
.RUNNING
,
118 "blocked": N2VCDeploymentStatus
.RUNNING
,
119 "error": N2VCDeploymentStatus
.FAILED
,
120 "active": N2VCDeploymentStatus
.COMPLETED
,
125 def obj_to_yaml(obj
: object) -> str:
127 Converts object to yaml format
131 dump_text
= yaml
.dump(obj
, default_flow_style
=False, indent
=2)
133 lines
= dump_text
.splitlines()
134 # remove !!python/object tags
137 index
= line
.find("!!python/object")
140 yaml_text
+= line
+ "\n"
144 def obj_to_dict(obj
: object) -> dict:
146 Converts object to dictionary format
149 # convert obj to yaml
150 yaml_text
= obj_to_yaml(obj
)
152 return yaml
.load(yaml_text
, Loader
=yaml
.SafeLoader
)
155 def get_ee_id_components(ee_id
: str) -> Tuple
[str, str, str]:
157 Get model, application and machine components from an execution environment id
159 :return: model_name, application_name, machine_id
161 parts
= ee_id
.split(".")
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
170 def generate_random_alfanum_string(size
: int) -> str:
172 Generate random alfa-numeric string with a size given by argument
174 :return: random generated string
178 secrets
.choice(string
.ascii_letters
+ string
.digits
) for i
in range(size
)