import asynctest
import asyncio
-from unittest import mock
+from unittest import mock, TestCase
from unittest.mock import Mock
-from n2vc.juju_watcher import JujuModelWatcher
-from n2vc.utils import EntityType
+from n2vc.juju_watcher import JujuModelWatcher, entity_ready, status
from n2vc.exceptions import EntityInvalidException
from .utils import FakeN2VC, AsyncMock, Deltas, FakeWatcher
+from juju.application import Application
+from juju.model import Model
+from juju.annotation import Annotation
+from juju.machine import Machine
+from juju.action import Action
class JujuWatcherTest(asynctest.TestCase):
def test_get_status(self):
tests = Deltas
for test in tests:
- (status, message, vca_status) = JujuModelWatcher.get_status(
- test.delta, test.entity.type
- )
+ (status, message, vca_status) = JujuModelWatcher.get_status(test.delta)
self.assertEqual(status, test.entity_status.status)
self.assertEqual(message, test.entity_status.message)
self.assertEqual(vca_status, test.entity_status.vca_status)
self.n2vc.last_written_values = None
@mock.patch("n2vc.juju_watcher.asyncio.wait")
- @mock.patch("n2vc.juju_watcher.EntityType.get_entity")
- def test_wait_for(self, get_entity, wait):
+ def test_wait_for(self, wait):
wait.return_value = asyncio.Future()
wait.return_value.set_result(None)
- get_entity.return_value = EntityType.MACHINE
machine = AsyncMock()
self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
@mock.patch("n2vc.juju_watcher.asyncio.wait")
- @mock.patch("n2vc.juju_watcher.EntityType.get_entity")
- def test_wait_for_exception(self, get_entity, wait):
+ def test_wait_for_exception(self, wait):
wait.return_value = asyncio.Future()
wait.return_value.set_result(None)
wait.side_effect = Exception("error")
- get_entity.return_value = EntityType.MACHINE
machine = AsyncMock()
with self.assertRaises(Exception):
def test_wait_for_invalid_entity_exception(self):
with self.assertRaises(EntityInvalidException):
self.loop.run_until_complete(
- JujuModelWatcher.wait_for(self.model, AsyncMock(), total_timeout=0)
+ JujuModelWatcher.wait_for(
+ self.model,
+ Annotation(0, self.model),
+ total_timeout=None,
+ progress_timeout=None,
+ )
)
+
+
+class EntityReadyTest(TestCase):
+ @mock.patch("juju.application.Application.units")
+ def setUp(self, mock_units):
+ self.model = Model()
+ self.model._connector = mock.MagicMock()
+
+ def test_invalid_entity(self):
+ with self.assertRaises(EntityInvalidException):
+ entity_ready(Annotation(0, self.model))
+
+ @mock.patch("juju.machine.Machine.agent_status")
+ def test_machine_entity(self, mock_machine_agent_status):
+ entity = Machine(0, self.model)
+ self.assertEqual(entity.entity_type, "machine")
+ self.assertTrue(isinstance(entity_ready(entity), bool))
+
+ @mock.patch("juju.action.Action.status")
+ def test_action_entity(self, mock_action_status):
+ entity = Action(0, self.model)
+ self.assertEqual(entity.entity_type, "action")
+ self.assertTrue(isinstance(entity_ready(entity), bool))
+
+ @mock.patch("juju.application.Application.status")
+ def test_application_entity(self, mock_application_status):
+ entity = Application(0, self.model)
+ self.assertEqual(entity.entity_type, "application")
+ self.assertTrue(isinstance(entity_ready(entity), bool))
+
+
+class StatusTest(TestCase):
+ def setUp(self):
+ self.model = Model()
+ self.model._connector = mock.MagicMock()
+
+ @mock.patch("n2vc.juju_watcher.derive_status")
+ def test_invalid_entity(self, mock_derive_status):
+ application = mock.MagicMock()
+ mock_derive_status.return_value = "active"
+
+ class FakeUnit:
+ @property
+ def workload_status(self):
+ return "active"
+
+ application.units = [FakeUnit()]
+ value = status(application)
+ mock_derive_status.assert_called_once()
+ self.assertTrue(isinstance(value, str))