loop.run_until_complete(self.libjuju.disconnect())
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
+class LibjujuInitTestCase(asynctest.TestCase):
+ def setUp(self):
+ self.loop = asyncio.get_event_loop()
+ self.n2vc = FakeN2VC()
+ self.endpoint = "192.168.100.100:17070"
+ self.username = "admin"
+ self.password = "secret"
+ self.cacert = """
+ -----BEGIN CERTIFICATE-----
+ SOMECERT
+ -----END CERTIFICATE-----"""
+
+ def test_endpoint_not_in_db(
+ self,
+ mock__get_api_endpoints_db,
+ mock_update_endpoints,
+ mock_create_health_check_task,
+ ):
+ mock__get_api_endpoints_db.return_value = ["another_ip"]
+ Libjuju(
+ self.endpoint,
+ "192.168.0.155:17070",
+ self.username,
+ self.password,
+ self.cacert,
+ self.loop,
+ log=None,
+ db={"get_one": []},
+ n2vc=self.n2vc,
+ apt_mirror="192.168.0.100",
+ enable_os_upgrade=True,
+ )
+ mock_update_endpoints.assert_called_once_with([self.endpoint])
+ mock__get_api_endpoints_db.assert_called_once()
+
+ def test_endpoint_in_db(
+ self,
+ mock__get_api_endpoints_db,
+ mock_update_endpoints,
+ mock_create_health_check_task,
+ ):
+ mock__get_api_endpoints_db.return_value = [self.endpoint, "another_ip"]
+ Libjuju(
+ self.endpoint,
+ "192.168.0.155:17070",
+ self.username,
+ self.password,
+ self.cacert,
+ self.loop,
+ log=None,
+ db={"get_one": []},
+ n2vc=self.n2vc,
+ apt_mirror="192.168.0.100",
+ enable_os_upgrade=True,
+ )
+ mock_update_endpoints.assert_not_called()
+ mock__get_api_endpoints_db.assert_called_once()
+
+ def test_no_db_endpoints(
+ self,
+ mock__get_api_endpoints_db,
+ mock_update_endpoints,
+ mock_create_health_check_task,
+ ):
+ mock__get_api_endpoints_db.return_value = None
+ Libjuju(
+ self.endpoint,
+ "192.168.0.155:17070",
+ self.username,
+ self.password,
+ self.cacert,
+ self.loop,
+ log=None,
+ db={"get_one": []},
+ n2vc=self.n2vc,
+ apt_mirror="192.168.0.100",
+ enable_os_upgrade=True,
+ )
+ mock_update_endpoints.assert_called_once_with([self.endpoint])
+ mock__get_api_endpoints_db.assert_called_once()
+
+
@asynctest.mock.patch("juju.controller.Controller.connect")
@asynctest.mock.patch(
"juju.controller.Controller.api_endpoints",
# TODO test provision machine
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
+@asynctest.mock.patch("juju.model.Model.deploy")
+class DeployTest(LibjujuTestCase):
+ def setUp(self):
+ super(DeployTest, self).setUp()
+
+ def test_deploy(
+ self,
+ mock_deploy,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ self.loop.run_until_complete(
+ self.libjuju.deploy("cs:osm", "model", wait=True, timeout=0)
+ )
+ mock_deploy.assert_called_once()
+ mock_wait_for_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_deploy_no_wait(
+ self,
+ mock_deploy,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ self.loop.run_until_complete(
+ self.libjuju.deploy("cs:osm", "model", wait=False, timeout=0)
+ )
+ mock_deploy.assert_called_once()
+ mock_wait_for_model.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_deploy_exception(
+ self,
+ mock_deploy,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_deploy.side_effect = Exception()
+ mock_get_model.return_value = juju.model.Model()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.deploy("cs:osm", "model"))
+ mock_deploy.assert_called_once()
+ mock_wait_for_model.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")