222d881a6b5c16090fe6f3517e045009c9b72dc4
7 def _make_delta(entity
, type_
, data
=None):
8 from juju
.client
.client
import Delta
9 from juju
.delta
import get_entity_delta
11 delta
= Delta([entity
, type_
, data
])
12 return get_entity_delta(delta
)
15 class TestObserver(unittest
.TestCase
):
16 def _make_observer(self
, *args
):
17 from juju
.model
import _Observer
18 return _Observer(*args
)
20 def test_cares_about_id(self
):
23 o
= self
._make
_observer
(
24 None, None, None, id_
, None)
27 'application', 'change', dict(name
=id_
))
29 self
.assertTrue(o
.cares_about(delta
))
31 def test_cares_about_type(self
):
34 o
= self
._make
_observer
(
35 None, type_
, None, None, None)
38 type_
, 'change', dict(name
='foo'))
40 self
.assertTrue(o
.cares_about(delta
))
42 def test_cares_about_action(self
):
45 o
= self
._make
_observer
(
46 None, None, action
, None, None)
49 'application', action
, dict(name
='foo'))
51 self
.assertTrue(o
.cares_about(delta
))
53 def test_cares_about_predicate(self
):
55 return delta
.data
.get('fizz') == 'bang'
57 o
= self
._make
_observer
(
58 None, None, None, None, predicate
)
61 'application', 'change', dict(fizz
='bang'))
63 self
.assertTrue(o
.cares_about(delta
))
66 class TestModelState(unittest
.TestCase
):
67 def test_apply_delta(self
):
68 from juju
.model
import Model
69 from juju
.application
import Application
71 loop
= mock
.MagicMock()
72 model
= Model(loop
=loop
)
73 delta
= _make_delta('application', 'add', dict(name
='foo'))
76 prev
, new
= model
.state
.apply_delta(delta
)
78 len(model
.state
.state
[delta
.entity
][delta
.get_id()]), 1)
79 self
.assertIsNone(prev
)
80 self
.assertIsInstance(new
, Application
)
84 prev
, new
= model
.state
.apply_delta(delta
)
85 # length of the entity history deque is now 3:
86 # - 1 for the first delta
87 # - 1 for the second delta
88 # - 1 for the None sentinel appended after the 'remove'
90 len(model
.state
.state
[delta
.entity
][delta
.get_id()]), 3)
91 self
.assertIsInstance(new
, Application
)
92 # new object is falsy because its data is None
94 self
.assertIsInstance(prev
, Application
)
98 def test_get_series():
99 from juju
.model
import Model
103 'supported-series': {
111 assert model
._get
_series
('cs:trusty/ubuntu', entity
) == 'trusty'
112 assert model
._get
_series
('xenial/ubuntu', entity
) == 'xenial'
113 assert model
._get
_series
('~foo/xenial/ubuntu', entity
) == 'xenial'
114 assert model
._get
_series
('~foo/ubuntu', entity
) == 'xenial'
115 assert model
._get
_series
('ubuntu', entity
) == 'xenial'
116 assert model
._get
_series
('cs:ubuntu', entity
) == 'xenial'
119 class TestContextManager(asynctest
.TestCase
):
120 @asynctest.patch('juju.model.Model.disconnect')
121 @asynctest.patch('juju.model.Model.connect_current')
122 async def test_normal_use(self
, mock_connect
, mock_disconnect
):
123 from juju
.model
import Model
125 async with
Model() as model
:
126 self
.assertTrue(isinstance(model
, Model
))
128 self
.assertTrue(mock_connect
.called
)
129 self
.assertTrue(mock_disconnect
.called
)
131 @asynctest.patch('juju.model.Model.disconnect')
132 @asynctest.patch('juju.model.Model.connect_current')
133 async def test_exception(self
, mock_connect
, mock_disconnect
):
134 from juju
.model
import Model
136 class SomeException(Exception):
139 with self
.assertRaises(SomeException
):
141 raise SomeException()
143 self
.assertTrue(mock_connect
.called
)
144 self
.assertTrue(mock_disconnect
.called
)
146 @asynctest.patch('juju.client.connection.JujuData.current_controller')
147 async def test_no_current_connection(self
, mock_current_controller
):
148 from juju
.model
import Model
149 from juju
.errors
import JujuConnectionError
151 mock_current_controller
.return_value
= ""
153 with self
.assertRaises(JujuConnectionError
):