n2vc/tests/unit/test_juju_watcher.py

   1 # Copyright 2020 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import asynctest
  16 import asyncio
  17
  18 from unittest import mock, TestCase
  19 from unittest.mock import Mock
  20 from n2vc.juju_watcher import JujuModelWatcher, entity_ready, status
  21 from n2vc.exceptions import EntityInvalidException
  22 from .utils import FakeN2VC, AsyncMock, Deltas, FakeWatcher
  23 from juju.application import Application
  24 from juju.model import Model
  25 from juju.annotation import Annotation
  26 from juju.machine import Machine
  27 from juju.action import Action
  28
  29
  30 class JujuWatcherTest(asynctest.TestCase):
  31     def setUp(self):
  32         self.n2vc = FakeN2VC()
  33         self.model = Mock()
  34         self.loop = asyncio.new_event_loop()
  35
  36     def test_get_status(self):
  37         tests = Deltas
  38         for test in tests:
  39             (status, message, vca_status) = JujuModelWatcher.get_status(test.delta)
  40             self.assertEqual(status, test.entity_status.status)
  41             self.assertEqual(message, test.entity_status.message)
  42             self.assertEqual(vca_status, test.entity_status.vca_status)
  43
  44     @mock.patch("n2vc.juju_watcher.client.AllWatcherFacade.from_connection")
  45     def test_model_watcher(self, allwatcher):
  46         tests = Deltas
  47         allwatcher.return_value = FakeWatcher()
  48         n2vc = AsyncMock()
  49         for test in tests:
  50             with self.assertRaises(asyncio.TimeoutError):
  51                 allwatcher.return_value.delta_to_return = [test.delta]
  52                 self.loop.run_until_complete(
  53                     JujuModelWatcher.model_watcher(
  54                         self.model,
  55                         test.filter.entity_id,
  56                         test.filter.entity_type,
  57                         timeout=0,
  58                         db_dict={"something"},
  59                         n2vc=n2vc,
  60                         vca_id=None,
  61                     )
  62                 )
  63
  64             n2vc.write_app_status_to_db.assert_called()
  65
  66     @mock.patch("n2vc.juju_watcher.asyncio.wait")
  67     def test_wait_for(self, wait):
  68         wait.return_value = asyncio.Future()
  69         wait.return_value.set_result(None)
  70
  71         machine = AsyncMock()
  72         self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
  73
  74     @mock.patch("n2vc.juju_watcher.asyncio.wait")
  75     def test_wait_for_exception(self, wait):
  76         wait.return_value = asyncio.Future()
  77         wait.return_value.set_result(None)
  78         wait.side_effect = Exception("error")
  79
  80         machine = AsyncMock()
  81         with self.assertRaises(Exception):
  82             self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
  83
  84     def test_wait_for_invalid_entity_exception(self):
  85         with self.assertRaises(EntityInvalidException):
  86             self.loop.run_until_complete(
  87                 JujuModelWatcher.wait_for(
  88                     self.model,
  89                     Annotation(0, self.model),
  90                     total_timeout=None,
  91                     progress_timeout=None,
  92                 )
  93             )
  94
  95
  96 class EntityReadyTest(TestCase):
  97     @mock.patch("juju.application.Application.units")
  98     def setUp(self, mock_units):
  99         self.model = Model()
 100         self.model._connector = mock.MagicMock()
 101
 102     def test_invalid_entity(self):
 103         with self.assertRaises(EntityInvalidException):
 104             entity_ready(Annotation(0, self.model))
 105
 106     @mock.patch("juju.machine.Machine.agent_status")
 107     def test_machine_entity(self, mock_machine_agent_status):
 108         entity = Machine(0, self.model)
 109         self.assertEqual(entity.entity_type, "machine")
 110         self.assertTrue(isinstance(entity_ready(entity), bool))
 111
 112     @mock.patch("juju.action.Action.status")
 113     def test_action_entity(self, mock_action_status):
 114         entity = Action(0, self.model)
 115         self.assertEqual(entity.entity_type, "action")
 116         self.assertTrue(isinstance(entity_ready(entity), bool))
 117
 118     @mock.patch("juju.application.Application.status")
 119     def test_application_entity(self, mock_application_status):
 120         entity = Application(0, self.model)
 121         self.assertEqual(entity.entity_type, "application")
 122         self.assertTrue(isinstance(entity_ready(entity), bool))
 123
 124
 125 class StatusTest(TestCase):
 126     def setUp(self):
 127         self.model = Model()
 128         self.model._connector = mock.MagicMock()
 129
 130     @mock.patch("n2vc.juju_watcher.derive_status")
 131     def test_invalid_entity(self, mock_derive_status):
 132         application = mock.MagicMock()
 133         mock_derive_status.return_value = "active"
 134
 135         class FakeUnit:
 136             @property
 137             def workload_status(self):
 138                 return "active"
 139
 140         application.units = [FakeUnit()]
 141         value = status(application)
 142         mock_derive_status.assert_called_once()
 143         self.assertTrue(isinstance(value, str))
 144
 145
 146 @asynctest.mock.patch("asyncio.sleep")
 147 class WaitForModelTest(asynctest.TestCase):
 148     @asynctest.mock.patch("juju.client.connector.Connector.connect")
 149     def setUp(self, mock_connect=None):
 150         self.loop = asyncio.new_event_loop()
 151         self.model = Model()
 152
 153     @asynctest.mock.patch("juju.model.Model.block_until")
 154     def test_wait_for_model(self, mock_block_until, mock_sleep):
 155         self.loop.run_until_complete(
 156             JujuModelWatcher.wait_for_model(self.model, timeout=None)
 157         )
 158         mock_block_until.assert_called()
 159
 160     @asynctest.mock.patch("asyncio.ensure_future")
 161     @asynctest.mock.patch("asyncio.wait")
 162     def test_wait_for_model_exception(self, mock_wait, mock_ensure_future, mock_sleep):
 163         task = Mock()
 164         mock_ensure_future.return_value = task
 165         mock_wait.side_effect = Exception
 166         with self.assertRaises(Exception):
 167             self.loop.run_until_complete(
 168                 JujuModelWatcher.wait_for_model(self.model, timeout=None)
 169             )
 170         task.cancel.assert_called()