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
  19 from unittest.mock import Mock
  20 from n2vc.juju_watcher import JujuModelWatcher
  21 from n2vc.utils import EntityType
  22 from n2vc.exceptions import EntityInvalidException
  23 from .utils import FakeN2VC, AsyncMock, Deltas, FakeWatcher
  24
  25
  26 class JujuWatcherTest(asynctest.TestCase):
  27     def setUp(self):
  28         self.n2vc = FakeN2VC()
  29         self.model = Mock()
  30         self.loop = asyncio.new_event_loop()
  31
  32     def test_get_status(self):
  33         tests = Deltas
  34         for test in tests:
  35             (status, message, vca_status) = JujuModelWatcher.get_status(
  36                 test.delta, test.entity.type
  37             )
  38             self.assertEqual(status, test.entity_status.status)
  39             self.assertEqual(message, test.entity_status.message)
  40             self.assertEqual(vca_status, test.entity_status.vca_status)
  41
  42     @mock.patch("n2vc.juju_watcher.client.AllWatcherFacade.from_connection")
  43     def test_model_watcher(self, allwatcher):
  44         tests = Deltas
  45         allwatcher.return_value = FakeWatcher()
  46         for test in tests:
  47             with self.assertRaises(asyncio.TimeoutError):
  48                 allwatcher.return_value.delta_to_return = [test.delta]
  49                 self.loop.run_until_complete(
  50                     JujuModelWatcher.model_watcher(
  51                         self.model,
  52                         test.filter.entity_id,
  53                         test.filter.entity_type,
  54                         timeout=0,
  55                         db_dict={"something"},
  56                         n2vc=self.n2vc,
  57                     )
  58                 )
  59
  60             self.assertEqual(self.n2vc.last_written_values, test.db.data)
  61             self.n2vc.last_written_values = None
  62
  63     @mock.patch("n2vc.juju_watcher.asyncio.wait")
  64     @mock.patch("n2vc.juju_watcher.EntityType.get_entity")
  65     def test_wait_for(self, get_entity, wait):
  66         wait.return_value = asyncio.Future()
  67         wait.return_value.set_result(None)
  68         get_entity.return_value = EntityType.MACHINE
  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     @mock.patch("n2vc.juju_watcher.EntityType.get_entity")
  75     def test_wait_for_exception(self, get_entity, wait):
  76         wait.return_value = asyncio.Future()
  77         wait.return_value.set_result(None)
  78         wait.side_effect = Exception("error")
  79         get_entity.return_value = EntityType.MACHINE
  80
  81         machine = AsyncMock()
  82         with self.assertRaises(Exception):
  83             self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
  84
  85     def test_wait_for_invalid_entity_exception(self):
  86         with self.assertRaises(EntityInvalidException):
  87             self.loop.run_until_complete(
  88                 JujuModelWatcher.wait_for(self.model, AsyncMock(), total_timeout=0)
  89             )