self.log = log or logging.getLogger("Libjuju")
self.db = db
db_endpoints = self._get_api_endpoints_db()
- self.endpoints = db_endpoints or [endpoint]
- if db_endpoints is None:
+ self.endpoints = None
+ if (db_endpoints and endpoint not in db_endpoints) or not db_endpoints:
+ self.endpoints = [endpoint]
self._update_api_endpoints_db(self.endpoints)
+ else:
+ self.endpoints = db_endpoints
self.api_proxy = api_proxy
self.username = username
self.password = password
self.models = set()
self.log.debug("Libjuju initialized!")
- self.health_check_task = self.loop.create_task(self.health_check())
+ self.health_check_task = self._create_health_check_task()
+
+ def _create_health_check_task(self):
+ return self.loop.create_task(self.health_check())
async def get_controller(self, timeout: float = 5.0) -> Controller:
"""
:param: controller: Controller that will be disconnected
"""
- await controller.disconnect()
+ if controller:
+ await controller.disconnect()
async def add_model(self, model_name: str, cloud_name: str, credential_name=None):
"""
return machine_id
+ async def deploy(
+ self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600
+ ):
+ """
+ Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
+
+ :param: uri: Path or Charm Store uri in which the charm or bundle can be found
+ :param: model_name: Model name
+ :param: wait: Indicates whether to wait or not until all applications are active
+ :param: timeout: Time in seconds to wait until all applications are active
+ """
+ controller = await self.get_controller()
+ model = await self.get_model(controller, model_name)
+ try:
+ await model.deploy(uri)
+ if wait:
+ await JujuModelWatcher.wait_for_model(model, timeout=timeout)
+ self.log.debug("All units active in model {}".format(model_name))
+ finally:
+ await self.disconnect_model(model)
+ await self.disconnect_controller(controller)
+
async def deploy_charm(
self,
application_name: str,
:param: interval: Time in seconds between checks
"""
+ controller = None
while True:
try:
controller = await self.get_controller()
auth_type = "certificate"
else:
raise JujuInvalidK8sConfiguration("authentication method not supported")
- return client.CloudCredential(auth_type=auth_type, attrs=attrs,)
+ return client.CloudCredential(auth_type=auth_type, attrs=attrs)
async def add_cloud(
self,