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.
18 from unittest
import mock
19 from unittest
.mock
import Mock
20 from n2vc
.juju_watcher
import JujuModelWatcher
21 from n2vc
.utils
import EntityType
22 from n2vc
.exceptions
import EntityInvalidException
23 from .utils
import FakeN2VC
, AsyncMock
, Deltas
, FakeWatcher
26 class JujuWatcherTest(asynctest
.TestCase
):
28 self
.n2vc
= FakeN2VC()
30 self
.loop
= asyncio
.new_event_loop()
32 def test_get_status(self
):
35 (status
, message
, vca_status
) = JujuModelWatcher
.get_status(
36 test
.delta
, test
.entity
.type
38 self
.assertEqual(status
, test
.entity_status
.status
)
39 self
.assertEqual(message
, test
.entity_status
.message
)
40 self
.assertEqual(vca_status
, test
.entity_status
.vca_status
)
42 @mock.patch("n2vc.juju_watcher.client.AllWatcherFacade.from_connection")
43 def test_model_watcher(self
, allwatcher
):
45 allwatcher
.return_value
= FakeWatcher()
47 with self
.assertRaises(asyncio
.TimeoutError
):
48 allwatcher
.return_value
.delta_to_return
= [test
.delta
]
49 self
.loop
.run_until_complete(
50 JujuModelWatcher
.model_watcher(
52 test
.filter.entity_id
,
53 test
.filter.entity_type
,
55 db_dict
={"something"},
60 self
.assertEqual(self
.n2vc
.last_written_values
, test
.db
.data
)
61 self
.n2vc
.last_written_values
= None
63 @mock.patch("n2vc.juju_watcher.asyncio.wait")
64 @mock.patch("n2vc.juju_watcher.EntityType.get_entity")
65 def test_wait_for(self
, get_entity
, wait
):
66 wait
.return_value
= asyncio
.Future()
67 wait
.return_value
.set_result(None)
68 get_entity
.return_value
= EntityType
.MACHINE
71 self
.loop
.run_until_complete(JujuModelWatcher
.wait_for(self
.model
, machine
))
73 @mock.patch("n2vc.juju_watcher.asyncio.wait")
74 @mock.patch("n2vc.juju_watcher.EntityType.get_entity")
75 def test_wait_for_exception(self
, get_entity
, wait
):
76 wait
.return_value
= asyncio
.Future()
77 wait
.return_value
.set_result(None)
78 wait
.side_effect
= Exception("error")
79 get_entity
.return_value
= EntityType
.MACHINE
82 with self
.assertRaises(Exception):
83 self
.loop
.run_until_complete(JujuModelWatcher
.wait_for(self
.model
, machine
))
85 def test_wait_for_invalid_entity_exception(self
):
86 with self
.assertRaises(EntityInvalidException
):
87 self
.loop
.run_until_complete(
88 JujuModelWatcher
.wait_for(self
.model
, AsyncMock(), total_timeout
=0)