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         for test in tests:
  49             with self.assertRaises(asyncio.TimeoutError):
  50                 allwatcher.return_value.delta_to_return = [test.delta]
  51                 self.loop.run_until_complete(
  52                     JujuModelWatcher.model_watcher(
  53                         self.model,
  54                         test.filter.entity_id,
  55                         test.filter.entity_type,
  56                         timeout=0,
  57                         db_dict={"something"},
  58                         n2vc=self.n2vc,
  59                     )
  60                 )
  61
  62             self.assertEqual(self.n2vc.last_written_values, test.db.data)
  63             self.n2vc.last_written_values = None
  64
  65     @mock.patch("n2vc.juju_watcher.asyncio.wait")
  66     def test_wait_for(self, wait):
  67         wait.return_value = asyncio.Future()
  68         wait.return_value.set_result(None)
  69
  70         machine = AsyncMock()
  71         self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
  72
  73     @mock.patch("n2vc.juju_watcher.asyncio.wait")
  74     def test_wait_for_exception(self, wait):
  75         wait.return_value = asyncio.Future()
  76         wait.return_value.set_result(None)
  77         wait.side_effect = Exception("error")
  78
  79         machine = AsyncMock()
  80         with self.assertRaises(Exception):
  81             self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
  82
  83     def test_wait_for_invalid_entity_exception(self):
  84         with self.assertRaises(EntityInvalidException):
  85             self.loop.run_until_complete(
  86                 JujuModelWatcher.wait_for(
  87                     self.model,
  88                     Annotation(0, self.model),
  89                     total_timeout=None,
  90                     progress_timeout=None,
  91                 )
  92             )
  93
  94
  95 class EntityReadyTest(TestCase):
  96     @mock.patch("juju.application.Application.units")
  97     def setUp(self, mock_units):
  98         self.model = Model()
  99         self.model._connector = mock.MagicMock()
 100
 101     def test_invalid_entity(self):
 102         with self.assertRaises(EntityInvalidException):
 103             entity_ready(Annotation(0, self.model))
 104
 105     @mock.patch("juju.machine.Machine.agent_status")
 106     def test_machine_entity(self, mock_machine_agent_status):
 107         entity = Machine(0, self.model)
 108         self.assertEqual(entity.entity_type, "machine")
 109         self.assertTrue(isinstance(entity_ready(entity), bool))
 110
 111     @mock.patch("juju.action.Action.status")
 112     def test_action_entity(self, mock_action_status):
 113         entity = Action(0, self.model)
 114         self.assertEqual(entity.entity_type, "action")
 115         self.assertTrue(isinstance(entity_ready(entity), bool))
 116
 117     @mock.patch("juju.application.Application.status")
 118     def test_application_entity(self, mock_application_status):
 119         entity = Application(0, self.model)
 120         self.assertEqual(entity.entity_type, "application")
 121         self.assertTrue(isinstance(entity_ready(entity), bool))
 122
 123
 124 class StatusTest(TestCase):
 125     def setUp(self):
 126         self.model = Model()
 127         self.model._connector = mock.MagicMock()
 128
 129     @mock.patch("n2vc.juju_watcher.derive_status")
 130     def test_invalid_entity(self, mock_derive_status):
 131         application = mock.MagicMock()
 132         mock_derive_status.return_value = "active"
 133
 134         class FakeUnit:
 135             @property
 136             def workload_status(self):
 137                 return "active"
 138
 139         application.units = [FakeUnit()]
 140         value = status(application)
 141         mock_derive_status.assert_called_once()
 142         self.assertTrue(isinstance(value, str))
 143
 144
 145 class WaitForModelTest(asynctest.TestCase):
 146     @asynctest.mock.patch("juju.client.connector.Connector.connect")
 147     def setUp(self, mock_connect=None):
 148         self.loop = asyncio.new_event_loop()
 149         self.model = Model()
 150
 151     @asynctest.mock.patch("juju.model.Model.block_until")
 152     def test_wait_for_model(self, mock_block_until):
 153         self.loop.run_until_complete(
 154             JujuModelWatcher.wait_for_model(self.model, timeout=None)
 155         )
 156         mock_block_until.assert_called()
 157
 158     @asynctest.mock.patch("asyncio.ensure_future")
 159     @asynctest.mock.patch("asyncio.wait")
 160     def test_wait_for_model_exception(self, mock_wait, mock_ensure_future):
 161         task = Mock()
 162         mock_ensure_future.return_value = task
 163         mock_wait.side_effect = Exception
 164         with self.assertRaises(Exception):
 165             self.loop.run_until_complete(
 166                 JujuModelWatcher.wait_for_model(self.model, timeout=None)
 167             )
 168         task.cancel.assert_called()