Feature 9952: Distributed Proxy Charms
[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 for test in tests:
49 with self.assertRaises(asyncio.TimeoutError):
50 allwatcher.return_value.delta_to_return = [test.delta]
51 self.loop.run_until_complete(
52 JujuModelWatcher.model_watcher(
53 self.model,
54 test.filter.entity_id,
55 test.filter.entity_type,
56 timeout=0,
57 db_dict={"something"},
58 n2vc=self.n2vc,
59 )
60 )
61
62 self.assertEqual(self.n2vc.last_written_values, test.db.data)
63 self.n2vc.last_written_values = None
64
65 @mock.patch("n2vc.juju_watcher.asyncio.wait")
66 def test_wait_for(self, wait):
67 wait.return_value = asyncio.Future()
68 wait.return_value.set_result(None)
69
70 machine = AsyncMock()
71 self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
72
73 @mock.patch("n2vc.juju_watcher.asyncio.wait")
74 def test_wait_for_exception(self, wait):
75 wait.return_value = asyncio.Future()
76 wait.return_value.set_result(None)
77 wait.side_effect = Exception("error")
78
79 machine = AsyncMock()
80 with self.assertRaises(Exception):
81 self.loop.run_until_complete(JujuModelWatcher.wait_for(self.model, machine))
82
83 def test_wait_for_invalid_entity_exception(self):
84 with self.assertRaises(EntityInvalidException):
85 self.loop.run_until_complete(
86 JujuModelWatcher.wait_for(
87 self.model,
88 Annotation(0, self.model),
89 total_timeout=None,
90 progress_timeout=None,
91 )
92 )
93
94
95 class EntityReadyTest(TestCase):
96 @mock.patch("juju.application.Application.units")
97 def setUp(self, mock_units):
98 self.model = Model()
99 self.model._connector = mock.MagicMock()
100
101 def test_invalid_entity(self):
102 with self.assertRaises(EntityInvalidException):
103 entity_ready(Annotation(0, self.model))
104
105 @mock.patch("juju.machine.Machine.agent_status")
106 def test_machine_entity(self, mock_machine_agent_status):
107 entity = Machine(0, self.model)
108 self.assertEqual(entity.entity_type, "machine")
109 self.assertTrue(isinstance(entity_ready(entity), bool))
110
111 @mock.patch("juju.action.Action.status")
112 def test_action_entity(self, mock_action_status):
113 entity = Action(0, self.model)
114 self.assertEqual(entity.entity_type, "action")
115 self.assertTrue(isinstance(entity_ready(entity), bool))
116
117 @mock.patch("juju.application.Application.status")
118 def test_application_entity(self, mock_application_status):
119 entity = Application(0, self.model)
120 self.assertEqual(entity.entity_type, "application")
121 self.assertTrue(isinstance(entity_ready(entity), bool))
122
123
124 class StatusTest(TestCase):
125 def setUp(self):
126 self.model = Model()
127 self.model._connector = mock.MagicMock()
128
129 @mock.patch("n2vc.juju_watcher.derive_status")
130 def test_invalid_entity(self, mock_derive_status):
131 application = mock.MagicMock()
132 mock_derive_status.return_value = "active"
133
134 class FakeUnit:
135 @property
136 def workload_status(self):
137 return "active"
138
139 application.units = [FakeUnit()]
140 value = status(application)
141 mock_derive_status.assert_called_once()
142 self.assertTrue(isinstance(value, str))
143
144
145 class WaitForModelTest(asynctest.TestCase):
146 @asynctest.mock.patch("juju.client.connector.Connector.connect")
147 def setUp(self, mock_connect=None):
148 self.loop = asyncio.new_event_loop()
149 self.model = Model()
150
151 @asynctest.mock.patch("juju.model.Model.block_until")
152 def test_wait_for_model(self, mock_block_until):
153 self.loop.run_until_complete(
154 JujuModelWatcher.wait_for_model(self.model, timeout=None)
155 )
156 mock_block_until.assert_called()
157
158 @asynctest.mock.patch("asyncio.ensure_future")
159 @asynctest.mock.patch("asyncio.wait")
160 def test_wait_for_model_exception(self, mock_wait, mock_ensure_future):
161 task = Mock()
162 mock_ensure_future.return_value = task
163 mock_wait.side_effect = Exception
164 with self.assertRaises(Exception):
165 self.loop.run_until_complete(
166 JujuModelWatcher.wait_for_model(self.model, timeout=None)
167 )
168 task.cancel.assert_called()