import asyncio
import asynctest
import tempfile
-from unittest import mock
+from unittest.mock import Mock, patch
import juju
import kubernetes
from juju.errors import JujuAPIError
import logging
-from .utils import FakeN2VC, FakeMachine, FakeApplication
+
+from n2vc.definitions import Offer, RelationEndpoint
+from .utils import (
+ FakeApplication,
+ FakeMachine,
+ FakeManualMachine,
+ FakeUnit,
+)
from n2vc.libjuju import Libjuju
from n2vc.exceptions import (
JujuControllerFailedConnecting,
JujuError,
)
from n2vc.k8s_juju_conn import generate_rbac_id
+from n2vc.tests.unit.utils import AsyncMock
+from n2vc.vca.connection import Connection
+from n2vc.vca.connection_data import ConnectionData
+
+
+cacert = """-----BEGIN CERTIFICATE-----
+SOMECERT
+-----END CERTIFICATE-----"""
+@asynctest.mock.patch("n2vc.libjuju.Controller")
class LibjujuTestCase(asynctest.TestCase):
- @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
- @asynctest.mock.patch("juju.client.connector.Connector.connect")
- @asynctest.mock.patch("juju.controller.Controller.connection")
- @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
+ @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
def setUp(
self,
- mock__get_api_endpoints_db=None,
- mock_connection=None,
- mock_connect=None,
- mock_update_endpoints=None,
- ):
- loop = asyncio.get_event_loop()
- n2vc = FakeN2VC()
- mock__get_api_endpoints_db.return_value = ["127.0.0.1:17070"]
- endpoints = "127.0.0.1:17070"
- username = "admin"
- password = "secret"
- cacert = """
- -----BEGIN CERTIFICATE-----
- SOMECERT
- -----END CERTIFICATE-----"""
- self.libjuju = Libjuju(
- endpoints,
- "192.168.0.155:17070",
- username,
- password,
- cacert,
- loop,
- log=None,
- db={"get_one": []},
- n2vc=n2vc,
- )
- logging.disable(logging.CRITICAL)
- 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):
+ mock_base64_to_cacert=None,
+ ):
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,
+ self.db = Mock()
+ mock_base64_to_cacert.return_value = cacert
+ Connection._load_vca_connection_data = Mock()
+ vca_connection = Connection(AsyncMock())
+ vca_connection._data = ConnectionData(
+ **{
+ "endpoints": ["1.2.3.4:17070"],
+ "user": "user",
+ "secret": "secret",
+ "cacert": "cacert",
+ "pubkey": "pubkey",
+ "lxd-cloud": "cloud",
+ "lxd-credentials": "credentials",
+ "k8s-cloud": "k8s_cloud",
+ "k8s-credentials": "k8s_credentials",
+ "model-config": {},
+ "api-proxy": "api_proxy",
+ }
)
- 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,
- )
- 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,
- )
- mock_update_endpoints.assert_called_once_with([self.endpoint])
- mock__get_api_endpoints_db.assert_called_once()
+ logging.disable(logging.CRITICAL)
+ self.libjuju = Libjuju(vca_connection, self.loop)
+ self.loop.run_until_complete(self.libjuju.disconnect())
@asynctest.mock.patch("juju.controller.Controller.connect")
"juju.controller.Controller.api_endpoints",
new_callable=asynctest.CoroutineMock(return_value=["127.0.0.1:17070"]),
)
-@asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
class GetControllerTest(LibjujuTestCase):
def setUp(self):
super(GetControllerTest, self).setUp()
- def test_diff_endpoint(
- self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
- ):
+ def test_diff_endpoint(self, mock_api_endpoints, mock_connect):
self.libjuju.endpoints = []
controller = self.loop.run_until_complete(self.libjuju.get_controller())
- mock__update_api_endpoints_db.assert_called_once_with(["127.0.0.1:17070"])
self.assertIsInstance(controller, juju.controller.Controller)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
def test_exception(
self,
mock_disconnect_controller,
- mock__update_api_endpoints_db,
mock_api_endpoints,
mock_connect,
):
self.libjuju.endpoints = []
- mock__update_api_endpoints_db.side_effect = Exception()
+
+ mock_connect.side_effect = Exception()
controller = None
with self.assertRaises(JujuControllerFailedConnecting):
controller = self.loop.run_until_complete(self.libjuju.get_controller())
self.assertIsNone(controller)
- mock_disconnect_controller.assert_called_once()
+ mock_disconnect_controller.assert_called()
- def test_same_endpoint_get_controller(
- self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
- ):
+ def test_same_endpoint_get_controller(self, mock_api_endpoints, mock_connect):
self.libjuju.endpoints = ["127.0.0.1:17070"]
controller = self.loop.run_until_complete(self.libjuju.get_controller())
- mock__update_api_endpoints_db.assert_not_called()
self.assertIsInstance(controller, juju.controller.Controller)
mock_model_exists.return_value = True
# This should not raise an exception
- self.loop.run_until_complete(
- self.libjuju.add_model("existing_model", "cloud")
- )
+ self.loop.run_until_complete(self.libjuju.add_model("existing_model", "cloud"))
mock_disconnect_controller.assert_called()
mock_get_controller.return_value = juju.controller.Controller()
self.loop.run_until_complete(
- self.libjuju.add_model("nonexisting_model", "cloud")
+ self.libjuju.add_model("nonexisting_model", Mock())
)
mock_add_model.assert_called_once()
executed_actions = self.loop.run_until_complete(
self.libjuju.get_executed_actions("model")
)
- expected_result = [{'id': 'id', 'action': 'action_name',
- 'status': 'status', 'output': 'completed'}]
+ expected_result = [
+ {
+ "id": "id",
+ "action": "action_name",
+ "status": "status",
+ "output": "completed",
+ }
+ ]
self.assertListEqual(expected_result, executed_actions)
self.assertIsInstance(executed_actions, list)
mock_get_model.return_value = None
with self.assertRaises(JujuError):
self.loop.run_until_complete(
- self.libjuju.get_application_configs("model", "app"))
+ self.libjuju.get_application_configs("model", "app")
+ )
mock_get_controller.assert_called_once()
mock_disconnect_controller.assert_called_once()
mock_get_controller,
):
mock_get_application.return_value = FakeApplication()
- application_configs = self.loop.run_until_complete(self.libjuju
- .get_application_configs("model", "app"))
+ application_configs = self.loop.run_until_complete(
+ self.libjuju.get_application_configs("model", "app")
+ )
self.assertEqual(application_configs, ["app_config"])
mock_disconnect_controller.assert_called()
mock_disconnect_model.assert_called()
- def test_succesful_exec(
+ def test_successful_exec(
self,
mock_get_action_status,
mock_get_action_output,
super(DestroyApplicationTest, self).setUp()
def test_success(
- self,
- mock_get_controller,
- mock_get_model,
- mock_disconnect_controller,
- mock_get_application,
- mock_disconnect_model,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
mock_get_application.return_value = FakeApplication()
mock_get_model.return_value = None
mock_disconnect_model.assert_called_once()
def test_no_application(
- self,
- mock_get_controller,
- mock_get_model,
- mock_disconnect_controller,
- mock_get_application,
- mock_disconnect_model,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
mock_get_model.return_value = None
mock_get_application.return_value = None
mock_get_application.assert_called()
def test_exception(
- self,
- mock_get_controller,
- mock_get_model,
- mock_disconnect_controller,
- mock_get_application,
- mock_disconnect_model,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
mock_get_application.return_value = FakeApplication
mock_get_model.return_value = None
mock_get_controller.return_value = juju.controller.Controller()
mock_list_offers.side_effect = Exception()
with self.assertRaises(Exception):
- self.loop.run_until_complete(self.libjuju.list_offers("model"))
+ self.loop.run_until_complete(self.libjuju._list_offers("model"))
mock_disconnect_controller.assert_called_once()
def test_empty_list(
mock_get_controller,
):
mock_get_controller.return_value = juju.controller.Controller()
- mock_list_offers.return_value = []
- offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
+ offer_results = Mock()
+ offer_results.results = []
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
self.assertEqual(offers, [])
mock_disconnect_controller.assert_called_once()
mock_get_controller,
):
mock_get_controller.return_value = juju.controller.Controller()
- mock_list_offers.return_value = ["offer"]
- offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
- self.assertEqual(offers, ["offer"])
+ offer = Mock()
+ offer_results = Mock()
+ offer_results.results = [offer]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
+ self.assertEqual(offers, [offer])
+ mock_disconnect_controller.assert_called_once()
+
+ def test_matching_offer_name(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ offer_1 = Mock()
+ offer_1.offer_name = "offer1"
+ offer_2 = Mock()
+ offer_2.offer_name = "offer2"
+ offer_results = Mock()
+ offer_results.results = [offer_1, offer_2]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(
+ self.libjuju._list_offers("model", offer_name="offer2")
+ )
+ self.assertEqual(offers, [offer_2])
+ mock_disconnect_controller.assert_called_once()
+
+ def test_not_matching_offer_name(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ offer_1 = Mock()
+ offer_1.offer_name = "offer1"
+ offer_2 = Mock()
+ offer_2.offer_name = "offer2"
+ offer_results = Mock()
+ offer_results.results = [offer_1, offer_2]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(
+ self.libjuju._list_offers("model", offer_name="offer3")
+ )
+ self.assertEqual(offers, [])
mock_disconnect_controller.assert_called_once()
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("juju.controller.Controller.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._list_offers")
+@asynctest.mock.patch("juju.model.Model.create_offer")
+class OfferTest(LibjujuTestCase):
+ def setUp(self):
+ super(OfferTest, self).setUp()
+
+ def test_offer(
+ self,
+ mock_create_offer,
+ mock__list_offers,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ controller = juju.controller.Controller()
+ model = juju.model.Model()
+ mock_get_controller.return_value = controller
+ mock_get_model.return_value = model
+ endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
+ self.loop.run_until_complete(self.libjuju.offer(endpoint))
+ mock_create_offer.assert_called_with(
+ "app-name:endpoint", offer_name="app-name-endpoint"
+ )
+ mock_disconnect_model.assert_called_once_with(model)
+ mock_disconnect_controller.assert_called_once_with(controller)
+
+ def test_offer_exception(
+ self,
+ mock_create_offer,
+ mock__list_offers,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ controller = juju.controller.Controller()
+ model = juju.model.Model()
+ mock_get_controller.return_value = controller
+ mock_get_model.return_value = model
+ mock__list_offers.return_value = []
+ endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.offer(endpoint))
+ mock_create_offer.assert_called_with(
+ "app-name:endpoint", offer_name="app-name-endpoint"
+ )
+ mock_disconnect_model.assert_called_once_with(model)
+ mock_disconnect_controller.assert_called_once_with(controller)
+
+
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
@asynctest.mock.patch("juju.controller.Controller.get_model")
@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
@asynctest.mock.patch("juju.model.Model.consume")
class ConsumeTest(LibjujuTestCase):
def setUp(self):
+ self.offer_url = "admin/model.offer_name"
super(ConsumeTest, self).setUp()
+ self.provider_libjuju = self.libjuju
def test_consume(
self,
mock_get_model,
mock_get_controller,
):
- mock_get_controller.return_value = juju.controller.Controller()
+ self_controller = juju.controller.Controller()
+ provider_controller = juju.controller.Controller()
+ mock_get_controller.side_effect = [self_controller, provider_controller]
mock_get_model.return_value = juju.model.Model()
- self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name"))
- mock_consume.assert_called_once()
+ self.loop.run_until_complete(
+ self.libjuju.consume(
+ "model_name",
+ Offer(self.offer_url, vca_id="vca-id"),
+ self.provider_libjuju,
+ )
+ )
+ mock_consume.assert_called_once_with(
+ "admin/model.offer_name",
+ application_alias="offer_name-model-vca-id",
+ controller=provider_controller,
+ )
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_parsing_error_exception(
self,
with self.assertRaises(juju.offerendpoints.ParseError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_juju_error_exception(
self,
with self.assertRaises(juju.errors.JujuError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_juju_api_error_exception(
self,
with self.assertRaises(juju.errors.JujuAPIError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
mock_configuration.key_file = None
self.token = None
self.cert_data = None
- with mock.patch.object(self.libjuju.log, "debug") as mock_debug:
+ with patch.object(self.libjuju.log, "debug") as mock_debug:
credential = self.libjuju.get_k8s_cloud_credential(
mock_configuration,
self.cert_data,
"Cannot set both token and user/pass",
)
self.assertTrue(exception_raised)
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+@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")
+class ScaleApplicationTest(LibjujuTestCase):
+ def setUp(self):
+ super(ScaleApplicationTest, self).setUp()
+
+ @asynctest.mock.patch("asyncio.sleep")
+ def test_scale_application(
+ self,
+ mock_sleep,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_application,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = FakeApplication()
+ self.loop.run_until_complete(self.libjuju.scale_application("model", "app", 2))
+ mock_wait_for_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_application(
+ self,
+ mock_wait_for,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_application,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_application.return_value = None
+ mock_get_model.return_value = juju.model.Model()
+ with self.assertRaises(JujuApplicationNotFound):
+ self.loop.run_until_complete(
+ self.libjuju.scale_application("model", "app", 2)
+ )
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
+
+ def test_exception(
+ self,
+ mock_wait_for,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_application,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = None
+ mock_get_application.return_value = FakeApplication()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.scale_application("model", "app", 2, total_timeout=0)
+ )
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+class GetUnitNumberTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetUnitNumberTest, self).setUp()
+
+ def test_successful_get_unit_number(
+ self,
+ mock_get_applications,
+ ):
+ mock_get_applications.return_value = FakeApplication()
+ model = juju.model.Model()
+ result = self.libjuju._get_application_count(model, "app")
+ self.assertEqual(result, 2)
+
+ def test_non_existing_application(
+ self,
+ mock_get_applications,
+ ):
+ mock_get_applications.return_value = None
+ model = juju.model.Model()
+ result = self.libjuju._get_application_count(model, "app")
+ self.assertEqual(result, None)
+
+
+@asynctest.mock.patch("juju.model.Model.machines", new_callable=asynctest.PropertyMock)
+class GetMachineInfoTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetMachineInfoTest, self).setUp()
+
+ def test_successful(
+ self,
+ mock_machines,
+ ):
+ machine_id = "existing_machine"
+ model = juju.model.Model()
+ mock_machines.return_value = {"existing_machine": FakeManualMachine()}
+ machine, series = self.libjuju._get_machine_info(
+ machine_id=machine_id,
+ model=model,
+ )
+ self.assertIsNotNone(machine, series)
+
+ def test_exception(
+ self,
+ mock_machines,
+ ):
+ machine_id = "not_existing_machine"
+ machine = series = None
+ model = juju.model.Model()
+ mock_machines.return_value = {"existing_machine": FakeManualMachine()}
+ with self.assertRaises(JujuMachineNotFound):
+ machine, series = self.libjuju._get_machine_info(
+ machine_id=machine_id,
+ model=model,
+ )
+ self.assertIsNone(machine, series)
+
+
+class GetUnitTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetUnitTest, self).setUp()
+
+ def test_successful(self):
+ result = self.libjuju._get_unit(FakeApplication(), "existing_machine_id")
+ self.assertIsInstance(result, FakeUnit)
+
+ def test_return_none(self):
+ result = self.libjuju._get_unit(FakeApplication(), "not_existing_machine_id")
+ self.assertIsNone(result)
+
+
+@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.libjuju.Libjuju._get_application")
+class CheckApplicationExists(LibjujuTestCase):
+ def setUp(self):
+ super(CheckApplicationExists, self).setUp()
+
+ def test_successful(
+ self,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = FakeApplication()
+ result = self.loop.run_until_complete(
+ self.libjuju.check_application_exists(
+ "model",
+ "app",
+ )
+ )
+ self.assertEqual(result, True)
+
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_application(
+ self,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = None
+ result = self.loop.run_until_complete(
+ self.libjuju.check_application_exists(
+ "model",
+ "app",
+ )
+ )
+ self.assertEqual(result, False)
+
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ 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")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_machine_info")
+class AddUnitTest(LibjujuTestCase):
+ def setUp(self):
+ super(AddUnitTest, self).setUp()
+
+ @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
+ @asynctest.mock.patch("asyncio.sleep")
+ def test_successful(
+ self,
+ mock_sleep,
+ mock_wait_for,
+ mock_get_machine_info,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = FakeApplication()
+ mock_get_machine_info.return_value = FakeMachine(), "series"
+ self.loop.run_until_complete(
+ self.libjuju.add_unit(
+ "existing_app",
+ "model",
+ "machine",
+ )
+ )
+
+ mock_wait_for.assert_called_once()
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_app(
+ self,
+ mock_get_machine_info,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = None
+ with self.assertRaises(JujuApplicationNotFound):
+ self.loop.run_until_complete(
+ self.libjuju.add_unit(
+ "existing_app",
+ "model",
+ "machine",
+ )
+ )
+
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_machine(
+ self,
+ mock_get_machine_info,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = FakeApplication()
+ mock_get_machine_info.side_effect = JujuMachineNotFound()
+ with self.assertRaises(JujuMachineNotFound):
+ self.loop.run_until_complete(
+ self.libjuju.add_unit(
+ "existing_app",
+ "model",
+ "machine",
+ )
+ )
+
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ 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")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_unit")
+class DestroyUnitTest(LibjujuTestCase):
+ def setUp(self):
+ super(DestroyUnitTest, self).setUp()
+
+ @asynctest.mock.patch("asyncio.sleep")
+ def test_successful(
+ self,
+ mock_sleep,
+ mock_get_unit,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = FakeApplication()
+
+ self.loop.run_until_complete(
+ self.libjuju.destroy_unit("app", "model", "machine", 0)
+ )
+
+ mock_get_unit.assert_called()
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_app(
+ self,
+ mock_get_unit,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = None
+
+ with self.assertRaises(JujuApplicationNotFound):
+ self.loop.run_until_complete(
+ self.libjuju.destroy_unit("app", "model", "machine")
+ )
+
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_unit(
+ self,
+ mock_get_unit,
+ mock_get_application,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+ mock_get_application.return_value = FakeApplication()
+ mock_get_unit.return_value = None
+
+ with self.assertRaises(JujuError):
+ self.loop.run_until_complete(
+ self.libjuju.destroy_unit("app", "model", "machine")
+ )
+
+ mock_get_unit.assert_called_once()
+ mock_get_application.assert_called_once()
+ mock_get_controller.assert_called_once()
+ mock_get_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()