X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_juju_observer.py;fp=n2vc%2Ftests%2Funit%2Ftest_juju_observer.py;h=f40824e20063c0ceff1cfdd9fab48ae1dbffa187;hp=0000000000000000000000000000000000000000;hb=0a8c9afacca5010da381e6382b01e31dd4e59d23;hpb=32b3812aaf09482ed9eb2b4492525bfccad3d0d2 diff --git a/n2vc/tests/unit/test_juju_observer.py b/n2vc/tests/unit/test_juju_observer.py new file mode 100644 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, + ) + )