Revert ""Remove unused lines of code""

[osm/N2VC.git] / n2vc / tests / unit / test_juju_observer.py

diff --git a/n2vc/tests/unit/test_juju_observer.py b/n2vc/tests/unit/test_juju_observer.py
new file mode 100644 (file)
index 0000000..f40824e
--- /dev/null
+++ b/n2vc/tests/unit/test_juju_observer.py
@@ -0,0 +1,159 @@
+# Copyright 2020 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+#     Unless required by applicable law or agreed to in writing, software
+#     distributed under the License is distributed on an "AS IS" BASIS,
+#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#     See the License for the specific language governing permissions and
+#     limitations under the License.
+
+
+import asyncio
+from unittest import mock
+from unittest.mock import Mock
+
+import asynctest
+
+from n2vc.exceptions import N2VCTimeoutException
+from n2vc.juju_observer import JujuModelObserver, _Entity
+
+
+class FakeObject:
+    def __init__(self):
+        self.complete = True
+
+
+class JujuModelObserverTest(asynctest.TestCase):
+    def setUp(self):
+        self.n2vc = Mock()
+        self.model = Mock()
+        self.juju_observer = JujuModelObserver(n2vc=self.n2vc, model=self.model)
+        self.loop = asyncio.new_event_loop()
+
+    def test_wait_no_retries(self):
+        obj = FakeObject()
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        result = self.loop.run_until_complete(
+            self.juju_observer._wait_for_entity(
+                entity=entity,
+                field_to_check="complete",
+                final_states_list=[True],
+                progress_timeout=None,
+                total_timeout=None,
+            )
+        )
+        self.assertEqual(result, 0)
+
+    @mock.patch("n2vc.juju_observer.asyncio.wait_for")
+    def test_wait_default_values(self, wait_for):
+        wait_for.return_value = asyncio.Future()
+        wait_for.return_value.set_result(None)
+        obj = FakeObject()
+        obj.complete = False
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        with self.assertRaises(N2VCTimeoutException):
+            self.loop.run_until_complete(
+                self.juju_observer._wait_for_entity(
+                    entity=entity,
+                    field_to_check="complete",
+                    final_states_list=[True],
+                    progress_timeout=None,
+                    total_timeout=None,
+                )
+            )
+        wait_for.assert_called_once_with(fut=mock.ANY, timeout=3600.0)
+
+    @mock.patch("n2vc.juju_observer.asyncio.wait_for")
+    def test_wait_default_progress(self, wait_for):
+        wait_for.return_value = asyncio.Future()
+        wait_for.return_value.set_result(None)
+        obj = FakeObject()
+        obj.complete = False
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        with self.assertRaises(N2VCTimeoutException):
+            self.loop.run_until_complete(
+                self.juju_observer._wait_for_entity(
+                    entity=entity,
+                    field_to_check="complete",
+                    final_states_list=[True],
+                    progress_timeout=4000,
+                    total_timeout=None,
+                )
+            )
+        wait_for.assert_called_once_with(fut=mock.ANY, timeout=3600.0)
+
+    @mock.patch("n2vc.juju_observer.asyncio.wait_for")
+    def test_wait_default_total(self, wait_for):
+        wait_for.return_value = asyncio.Future()
+        wait_for.return_value.set_result(None)
+        obj = FakeObject()
+        obj.complete = False
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        with self.assertRaises(N2VCTimeoutException):
+            self.loop.run_until_complete(
+                self.juju_observer._wait_for_entity(
+                    entity=entity,
+                    field_to_check="complete",
+                    final_states_list=[True],
+                    progress_timeout=None,
+                    total_timeout=4000.0,
+                )
+            )
+        wait_for.assert_called_once_with(fut=mock.ANY, timeout=3600.0)
+
+    @mock.patch("n2vc.juju_observer.asyncio.wait_for")
+    def test_wait_total_less_than_progress_timeout(self, wait_for):
+        wait_for.return_value = asyncio.Future()
+        wait_for.return_value.set_result(None)
+        obj = FakeObject()
+        obj.complete = False
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        with self.assertRaises(N2VCTimeoutException):
+            self.loop.run_until_complete(
+                self.juju_observer._wait_for_entity(
+                    entity=entity,
+                    field_to_check="complete",
+                    final_states_list=[True],
+                    progress_timeout=4500.0,
+                    total_timeout=3000.0,
+                )
+            )
+        wait_for.assert_called_once_with(fut=mock.ANY, timeout=3000.0)
+
+    @mock.patch("n2vc.juju_observer.asyncio.wait_for")
+    def test_wait_progress_less_than_total_timeout(self, wait_for):
+        wait_for.return_value = asyncio.Future()
+        wait_for.return_value.set_result(None)
+        obj = FakeObject()
+        obj.complete = False
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        with self.assertRaises(N2VCTimeoutException):
+            self.loop.run_until_complete(
+                self.juju_observer._wait_for_entity(
+                    entity=entity,
+                    field_to_check="complete",
+                    final_states_list=[True],
+                    progress_timeout=1500.0,
+                    total_timeout=3000.0,
+                )
+            )
+        wait_for.assert_called_once_with(fut=mock.ANY, timeout=1500.0)
+
+    def test_wait_negative_timeout(self):
+        obj = FakeObject()
+        entity = _Entity(entity_id="eid-1", entity_type="fake", obj=obj, db_dict={})
+        with self.assertRaises(N2VCTimeoutException):
+            self.loop.run_until_complete(
+                self.juju_observer._wait_for_entity(
+                    entity=entity,
+                    field_to_check="complete",
+                    final_states_list=[True],
+                    progress_timeout=None,
+                    total_timeout=-1000,
+                )
+            )