import time
-from juju.errors import JujuAPIError
+import juju.errors
from juju.model import Model
from juju.machine import Machine
from juju.application import Application
cloud_name=cloud.name,
credential_name=cloud.credential_name,
)
- except JujuAPIError as e:
+ except juju.errors.JujuAPIError as e:
if "already exists" in e.message:
pass
else:
"""
return await controller.get_model(model_name)
- async def model_exists(self, model_name: str, controller: Controller = None) -> bool:
+ async def model_exists(
+ self, model_name: str, controller: Controller = None
+ ) -> bool:
"""
Check if model exists
db_dict: dict = None,
progress_timeout: float = None,
total_timeout: float = None,
- series: str = "xenial",
+ series: str = "bionic",
wait: bool = True,
) -> (Machine, bool):
"""
return application
async def scale_application(
- self,
- model_name: str,
- application_name: str,
- scale: int = 1,
- total_timeout: float = None,
+ self,
+ model_name: str,
+ application_name: str,
+ scale: int = 1,
+ total_timeout: float = None,
):
"""
Scale application (K8s)
await application.scale(scale=scale)
# Wait until application is scaled in model
self.log.debug(
- "Waiting for application {} to be scaled in model {}...".format
- (
+ "Waiting for application {} to be scaled in model {}...".format(
application_name, model_name
)
)
# wait until application unit count and scale count are equal.
# Because there is a delay before scaling triggers in Juju model.
if application_scale == scale:
- await JujuModelWatcher.wait_for_model(model=model, timeout=total_timeout)
+ await JujuModelWatcher.wait_for_model(
+ model=model, timeout=total_timeout
+ )
self.log.debug(
"Application {} is scaled in model {}".format(
application_name, model_name
db_dict: dict = None,
progress_timeout: float = None,
total_timeout: float = None,
- **kwargs
+ **kwargs,
):
"""Execute action
# Add relation
try:
await model.add_relation(endpoint_1, endpoint_2)
- except JujuAPIError as e:
+ except juju.errors.JujuAPIError as e:
if "not found" in e.message:
self.log.warning("Relation not found: {}".format(e.message))
return
controller = await self.get_controller()
try:
await controller.remove_cloud(name)
+ except juju.errors.JujuError as e:
+ if len(e.errors) == 1 and f'cloud "{name}" not found' == e.errors[0]:
+ self.log.warning(f"Cloud {name} not found, so it could not be deleted.")
+ else:
+ raise e
finally:
await self.disconnect_controller(controller)