Reformat files according to new black validation
[osm/N2VC.git] / 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, TestCase
19 from unittest.mock import Mock
20 from n2vc.juju_watcher import JujuModelWatcher, entity_ready, status
21 from n2vc.exceptions import EntityInvalidException
22 from .utils import FakeN2VC, AsyncMock, Deltas, FakeWatcher
23 from juju.application import Application
24 from juju.model import Model
25 from juju.annotation import Annotation
26 from juju.machine import Machine
27 from juju.action import Action
28
29
30 class JujuWatcherTest(asynctest.TestCase):
31 def setUp(self):
32 self.n2vc = FakeN2VC()
33 self.model = Mock()
34 self.loop = asyncio.new_event_loop()
35
36 def test_get_status(self):
37 tests = Deltas
38 for test in tests:
39 (status, message, vca_status) = JujuModelWatcher.get_status(test.delta)
40 self.assertEqual(status, test.entity_status.status)
41 self.assertEqual(message, test.entity_status.message)
42 self.assertEqual(vca_status, test.entity_status.vca_status)
43
44 @mock.patch("n2vc.juju_watcher.client.AllWatcherFacade.from_connection")
45 def test_model_watcher(self, allwatcher):
46 tests = Deltas
47 allwatcher.return_value = FakeWatcher()
48 n2vc = AsyncMock()
49 for test in tests:
50 with self.assertRaises(asyncio.TimeoutError):
51 allwatcher.return_value.delta_to_return = [test.delta]
52 self.loop.run_until_complete(
53 JujuModelWatcher.model_watcher(
54 self.model,
55 test.filter.entity_id,
56 test.filter.entity_type,
57 timeout=0,
58 db_dict={"something"},
59 n2vc=n2vc,
60 vca_id=None,
61 )
62 )
63
64 n2vc.write_app_status_to_db.assert_called()
65
66 @mock.patch("n2vc.juju_watcher.asyncio.wait")
67 def test_wait_for(self, wait):
68 wait.return_value = asyncio.Future()
69 wait.return_value.set_result(None)
70
71 machine = AsyncMock()
72 self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
73
74 @mock.patch("n2vc.juju_watcher.asyncio.wait")
75 def test_wait_for_exception(self, wait):
76 wait.return_value = asyncio.Future()
77 wait.return_value.set_result(None)
78 wait.side_effect = Exception("error")
79
80 machine = AsyncMock()
81 with self.assertRaises(Exception):
82 self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
83
84 def test_wait_for_invalid_entity_exception(self):
85 with self.assertRaises(EntityInvalidException):
86 self.loop.run_until_complete(
87 JujuModelWatcher.wait_for(
88 self.model,
89 Annotation(0, self.model),
90 total_timeout=None,
91 progress_timeout=None,
92 )
93 )
94
95
96 class EntityReadyTest(TestCase):
97 @mock.patch("juju.application.Application.units")
98 def setUp(self, mock_units):
99 self.model = Model()
100 self.model._connector = mock.MagicMock()
101
102 def test_invalid_entity(self):
103 with self.assertRaises(EntityInvalidException):
104 entity_ready(Annotation(0, self.model))
105
106 @mock.patch("juju.machine.Machine.agent_status")
107 def test_machine_entity(self, mock_machine_agent_status):
108 entity = Machine(0, self.model)
109 self.assertEqual(entity.entity_type, "machine")
110 self.assertTrue(isinstance(entity_ready(entity), bool))
111
112 @mock.patch("juju.action.Action.status")
113 def test_action_entity(self, mock_action_status):
114 entity = Action(0, self.model)
115 self.assertEqual(entity.entity_type, "action")
116 self.assertTrue(isinstance(entity_ready(entity), bool))
117
118 @mock.patch("juju.application.Application.status")
119 def test_application_entity(self, mock_application_status):
120 entity = Application(0, self.model)
121 self.assertEqual(entity.entity_type, "application")
122 self.assertTrue(isinstance(entity_ready(entity), bool))
123
124
125 class StatusTest(TestCase):
126 def setUp(self):
127 self.model = Model()
128 self.model._connector = mock.MagicMock()
129
130 @mock.patch("n2vc.juju_watcher.derive_status")
131 def test_invalid_entity(self, mock_derive_status):
132 application = mock.MagicMock()
133 mock_derive_status.return_value = "active"
134
135 class FakeUnit:
136 @property
137 def workload_status(self):
138 return "active"
139
140 application.units = [FakeUnit()]
141 value = status(application)
142 mock_derive_status.assert_called_once()
143 self.assertTrue(isinstance(value, str))
144
145
146 @asynctest.mock.patch("asyncio.sleep")
147 class WaitForModelTest(asynctest.TestCase):
148 @asynctest.mock.patch("juju.client.connector.Connector.connect")
149 def setUp(self, mock_connect=None):
150 self.loop = asyncio.new_event_loop()
151 self.model = Model()
152
153 @asynctest.mock.patch("juju.model.Model.block_until")
154 def test_wait_for_model(self, mock_block_until, mock_sleep):
155 self.loop.run_until_complete(
156 JujuModelWatcher.wait_for_model(self.model, timeout=None)
157 )
158 mock_block_until.assert_called()
159
160 @asynctest.mock.patch("asyncio.ensure_future")
161 @asynctest.mock.patch("asyncio.wait")
162 def test_wait_for_model_exception(self, mock_wait, mock_ensure_future, mock_sleep):
163 task = Mock()
164 mock_ensure_future.return_value = task
165 mock_wait.side_effect = Exception
166 with self.assertRaises(Exception):
167 self.loop.run_until_complete(
168 JujuModelWatcher.wait_for_model(self.model, timeout=None)
169 )
170 task.cancel.assert_called()