n2vc/tests/unit/test_juju_observer.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
  16 import asyncio
  17 from unittest import mock
  18 from unittest.mock import Mock
  19
  20 import asynctest
  21
  22 from n2vc.exceptions import N2VCTimeoutException
  23 from n2vc.juju_observer import JujuModelObserver, _Entity
  24
  25
  26 class FakeObject:
  27     def __init__(self):
  28         self.complete = True
  29
  30
  31 class JujuModelObserverTest(asynctest.TestCase):
  32     def setUp(self):
  33         self.n2vc = Mock()
  34         self.model = Mock()
  35         self.juju_observer = JujuModelObserver(n2vc=self.n2vc, model=self.model)
  36         self.loop = asyncio.new_event_loop()
  37
  38     def test_wait_no_retries(self):
  39         obj = FakeObject()
  40         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
  41         result = self.loop.run_until_complete(
  42             self.juju_observer._wait_for_entity(
  43                 entity=entity,
  44                 field_to_check="complete",
  45                 final_states_list=[True],
  46                 progress_timeout=None,
  47                 total_timeout=None,
  48             )
  49         )
  50         self.assertEqual(result, 0)
  51
  52     @mock.patch("n2vc.juju_observer.asyncio.wait_for")
  53     def test_wait_default_values(self, wait_for):
  54         wait_for.return_value = asyncio.Future()
  55         wait_for.return_value.set_result(None)
  56         obj = FakeObject()
  57         obj.complete = False
  58         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
  59         with self.assertRaises(N2VCTimeoutException):
  60             self.loop.run_until_complete(
  61                 self.juju_observer._wait_for_entity(
  62                     entity=entity,
  63                     field_to_check="complete",
  64                     final_states_list=[True],
  65                     progress_timeout=None,
  66                     total_timeout=None,
  67                 )
  68             )
  69         wait_for.assert_called_once_with(fut=mock.ANY, timeout=3600.0)
  70
  71     @mock.patch("n2vc.juju_observer.asyncio.wait_for")
  72     def test_wait_default_progress(self, wait_for):
  73         wait_for.return_value = asyncio.Future()
  74         wait_for.return_value.set_result(None)
  75         obj = FakeObject()
  76         obj.complete = False
  77         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
  78         with self.assertRaises(N2VCTimeoutException):
  79             self.loop.run_until_complete(
  80                 self.juju_observer._wait_for_entity(
  81                     entity=entity,
  82                     field_to_check="complete",
  83                     final_states_list=[True],
  84                     progress_timeout=4000,
  85                     total_timeout=None,
  86                 )
  87             )
  88         wait_for.assert_called_once_with(fut=mock.ANY, timeout=3600.0)
  89
  90     @mock.patch("n2vc.juju_observer.asyncio.wait_for")
  91     def test_wait_default_total(self, wait_for):
  92         wait_for.return_value = asyncio.Future()
  93         wait_for.return_value.set_result(None)
  94         obj = FakeObject()
  95         obj.complete = False
  96         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
  97         with self.assertRaises(N2VCTimeoutException):
  98             self.loop.run_until_complete(
  99                 self.juju_observer._wait_for_entity(
 100                     entity=entity,
 101                     field_to_check="complete",
 102                     final_states_list=[True],
 103                     progress_timeout=None,
 104                     total_timeout=4000.0,
 105                 )
 106             )
 107         wait_for.assert_called_once_with(fut=mock.ANY, timeout=3600.0)
 108
 109     @mock.patch("n2vc.juju_observer.asyncio.wait_for")
 110     def test_wait_total_less_than_progress_timeout(self, wait_for):
 111         wait_for.return_value = asyncio.Future()
 112         wait_for.return_value.set_result(None)
 113         obj = FakeObject()
 114         obj.complete = False
 115         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
 116         with self.assertRaises(N2VCTimeoutException):
 117             self.loop.run_until_complete(
 118                 self.juju_observer._wait_for_entity(
 119                     entity=entity,
 120                     field_to_check="complete",
 121                     final_states_list=[True],
 122                     progress_timeout=4500.0,
 123                     total_timeout=3000.0,
 124                 )
 125             )
 126         wait_for.assert_called_once_with(fut=mock.ANY, timeout=3000.0)
 127
 128     @mock.patch("n2vc.juju_observer.asyncio.wait_for")
 129     def test_wait_progress_less_than_total_timeout(self, wait_for):
 130         wait_for.return_value = asyncio.Future()
 131         wait_for.return_value.set_result(None)
 132         obj = FakeObject()
 133         obj.complete = False
 134         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
 135         with self.assertRaises(N2VCTimeoutException):
 136             self.loop.run_until_complete(
 137                 self.juju_observer._wait_for_entity(
 138                     entity=entity,
 139                     field_to_check="complete",
 140                     final_states_list=[True],
 141                     progress_timeout=1500.0,
 142                     total_timeout=3000.0,
 143                 )
 144             )
 145         wait_for.assert_called_once_with(fut=mock.ANY, timeout=1500.0)
 146
 147     def test_wait_negative_timeout(self):
 148         obj = FakeObject()
 149         entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
 150         with self.assertRaises(N2VCTimeoutException):
 151             self.loop.run_until_complete(
 152                 self.juju_observer._wait_for_entity(
 153                     entity=entity,
 154                     field_to_check="complete",
 155                     final_states_list=[True],
 156                     progress_timeout=None,
 157                     total_timeout=-1000,
 158                 )
 159             )