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
29 def base64_to_cacert(b64string
):
30 """Convert the base64-encoded string containing the VCA CACERT.
36 cacert
= base64
.b64decode(b64string
).decode("utf-8")
43 except binascii
.Error
as e
:
44 raise N2VCInvalidCertificate(message
="Invalid CA Certificate: {}".format(e
))
49 class N2VCDeploymentStatus(Enum
):
52 COMPLETED
= "completed"
59 Dict class that allows to access the keys like attributes
62 def __getattribute__(self
, name
):
67 class EntityType(Enum
):
69 APPLICATION
= Application
74 def has_value(cls
, value
):
75 return value
in cls
._value
2member
_map
_ # pylint: disable=E1101
78 def get_entity(cls
, value
):
80 cls
._value
2member
_map
_[value
] # pylint: disable=E1101
81 if value
in cls
._value
2member
_map
_ # pylint: disable=E1101
82 else None # pylint: disable=E1101
86 def get_entity_from_delta(cls
, delta_entity
: str):
88 Get Value from delta entity
90 :param: delta_entity: Possible values are "machine", "application", "unit", "action"
92 for v
in cls
._value
2member
_map
_: # pylint: disable=E1101
93 if v
.__name
__.lower() == delta_entity
:
94 return cls
.get_entity(v
)
99 "pending": N2VCDeploymentStatus
.PENDING
,
100 "started": N2VCDeploymentStatus
.COMPLETED
,
103 "waiting": N2VCDeploymentStatus
.RUNNING
,
104 "maintenance": N2VCDeploymentStatus
.RUNNING
,
105 "blocked": N2VCDeploymentStatus
.RUNNING
,
106 "error": N2VCDeploymentStatus
.FAILED
,
107 "active": N2VCDeploymentStatus
.COMPLETED
,
110 "pending": N2VCDeploymentStatus
.PENDING
,
111 "running": N2VCDeploymentStatus
.RUNNING
,
112 "completed": N2VCDeploymentStatus
.COMPLETED
,
115 "waiting": N2VCDeploymentStatus
.RUNNING
,
116 "maintenance": N2VCDeploymentStatus
.RUNNING
,
117 "blocked": N2VCDeploymentStatus
.RUNNING
,
118 "error": N2VCDeploymentStatus
.FAILED
,
119 "active": N2VCDeploymentStatus
.COMPLETED
,
124 def obj_to_yaml(obj
: object) -> str:
126 Converts object to yaml format
130 dump_text
= yaml
.dump(obj
, default_flow_style
=False, indent
=2)
132 lines
= dump_text
.splitlines()
133 # remove !!python/object tags
136 index
= line
.find("!!python/object")
139 yaml_text
+= line
+ "\n"
143 def obj_to_dict(obj
: object) -> dict:
145 Converts object to dictionary format
148 # convert obj to yaml
149 yaml_text
= obj_to_yaml(obj
)
151 return yaml
.load(yaml_text
, Loader
=yaml
.Loader
)
154 def generate_random_alfanum_string(size
: int) -> str:
156 Generate random alfa-numeric string with a size given by argument
158 :return: random generated string
162 secrets
.choice(string
.ascii_letters
+ string
.digits
) for i
in range(size
)