1 # Copyright 2020 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 from unittest
import mock
18 from unittest
.mock
import Mock
22 from n2vc
.exceptions
import N2VCTimeoutException
23 from n2vc
.juju_observer
import JujuModelObserver
, _Entity
31 class JujuModelObserverTest(asynctest
.TestCase
):
35 self
.juju_observer
= JujuModelObserver(n2vc
=self
.n2vc
, model
=self
.model
)
36 self
.loop
= asyncio
.new_event_loop()
38 def test_wait_no_retries(self
):
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
(
44 field_to_check
="complete",
45 final_states_list
=[True],
46 progress_timeout
=None,
50 self
.assertEqual(result
, 0)
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)
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
(
63 field_to_check
="complete",
64 final_states_list
=[True],
65 progress_timeout
=None,
69 wait_for
.assert_called_once_with(fut
=mock
.ANY
, timeout
=3600.0)
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)
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
(
82 field_to_check
="complete",
83 final_states_list
=[True],
84 progress_timeout
=4000,
88 wait_for
.assert_called_once_with(fut
=mock
.ANY
, timeout
=3600.0)
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)
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
(
101 field_to_check
="complete",
102 final_states_list
=[True],
103 progress_timeout
=None,
104 total_timeout
=4000.0,
107 wait_for
.assert_called_once_with(fut
=mock
.ANY
, timeout
=3600.0)
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)
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
(
120 field_to_check
="complete",
121 final_states_list
=[True],
122 progress_timeout
=4500.0,
123 total_timeout
=3000.0,
126 wait_for
.assert_called_once_with(fut
=mock
.ANY
, timeout
=3000.0)
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)
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
(
139 field_to_check
="complete",
140 final_states_list
=[True],
141 progress_timeout
=1500.0,
142 total_timeout
=3000.0,
145 wait_for
.assert_called_once_with(fut
=mock
.ANY
, timeout
=1500.0)
147 def test_wait_negative_timeout(self
):
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
(
154 field_to_check
="complete",
155 final_states_list
=[True],
156 progress_timeout
=None,