7 from juju
.client
.jujudata
import FileJujuData
8 from juju
.model
import Model
11 def _make_delta(entity
, type_
, data
=None):
12 from juju
.client
.client
import Delta
13 from juju
.delta
import get_entity_delta
15 delta
= Delta([entity
, type_
, data
])
16 return get_entity_delta(delta
)
19 class TestObserver(unittest
.TestCase
):
20 def _make_observer(self
, *args
):
21 from juju
.model
import _Observer
22 return _Observer(*args
)
24 def test_cares_about_id(self
):
27 o
= self
._make
_observer
(
28 None, None, None, id_
, None)
31 'application', 'change', dict(name
=id_
))
33 self
.assertTrue(o
.cares_about(delta
))
35 def test_cares_about_type(self
):
38 o
= self
._make
_observer
(
39 None, type_
, None, None, None)
42 type_
, 'change', dict(name
='foo'))
44 self
.assertTrue(o
.cares_about(delta
))
46 def test_cares_about_action(self
):
49 o
= self
._make
_observer
(
50 None, None, action
, None, None)
53 'application', action
, dict(name
='foo'))
55 self
.assertTrue(o
.cares_about(delta
))
57 def test_cares_about_predicate(self
):
59 return delta
.data
.get('fizz') == 'bang'
61 o
= self
._make
_observer
(
62 None, None, None, None, predicate
)
65 'application', 'change', dict(fizz
='bang'))
67 self
.assertTrue(o
.cares_about(delta
))
70 class TestModelState(unittest
.TestCase
):
71 def test_apply_delta(self
):
72 from juju
.model
import Model
73 from juju
.application
import Application
76 model
._connector
= mock
.MagicMock()
77 delta
= _make_delta('application', 'add', dict(name
='foo'))
80 prev
, new
= model
.state
.apply_delta(delta
)
82 len(model
.state
.state
[delta
.entity
][delta
.get_id()]), 1)
83 self
.assertIsNone(prev
)
84 self
.assertIsInstance(new
, Application
)
88 prev
, new
= model
.state
.apply_delta(delta
)
89 # length of the entity history deque is now 3:
90 # - 1 for the first delta
91 # - 1 for the second delta
92 # - 1 for the None sentinel appended after the 'remove'
94 len(model
.state
.state
[delta
.entity
][delta
.get_id()]), 3)
95 self
.assertIsInstance(new
, Application
)
96 # new object is falsy because its data is None
98 self
.assertIsInstance(prev
, Application
)
102 def test_get_series():
103 from juju
.model
import Model
107 'supported-series': {
115 assert model
._get
_series
('cs:trusty/ubuntu', entity
) == 'trusty'
116 assert model
._get
_series
('xenial/ubuntu', entity
) == 'xenial'
117 assert model
._get
_series
('~foo/xenial/ubuntu', entity
) == 'xenial'
118 assert model
._get
_series
('~foo/ubuntu', entity
) == 'xenial'
119 assert model
._get
_series
('ubuntu', entity
) == 'xenial'
120 assert model
._get
_series
('cs:ubuntu', entity
) == 'xenial'
123 class TestContextManager(asynctest
.TestCase
):
124 @asynctest.patch('juju.model.Model.disconnect')
125 @asynctest.patch('juju.model.Model.connect')
126 async def test_normal_use(self
, mock_connect
, mock_disconnect
):
127 from juju
.model
import Model
129 async with
Model() as model
:
130 self
.assertTrue(isinstance(model
, Model
))
132 self
.assertTrue(mock_connect
.called
)
133 self
.assertTrue(mock_disconnect
.called
)
135 @asynctest.patch('juju.model.Model.disconnect')
136 @asynctest.patch('juju.model.Model.connect')
137 async def test_exception(self
, mock_connect
, mock_disconnect
):
138 from juju
.model
import Model
140 class SomeException(Exception):
143 with self
.assertRaises(SomeException
):
145 raise SomeException()
147 self
.assertTrue(mock_connect
.called
)
148 self
.assertTrue(mock_disconnect
.called
)
150 async def test_no_current_connection(self
):
151 from juju
.model
import Model
152 from juju
.errors
import JujuConnectionError
154 class NoControllerJujuData(FileJujuData
):
155 def current_controller(self
):
158 with self
.assertRaises(JujuConnectionError
):
159 async with
Model(jujudata
=NoControllerJujuData()):
163 @asynctest.patch('juju.model.Model._after_connect')
164 class TestModelConnect(asynctest
.TestCase
):
165 @asynctest.patch('juju.client.connector.Connector.connect_model')
166 async def test_no_args(self
, mock_connect_model
, _
):
169 mock_connect_model
.assert_called_once_with(None)
171 @asynctest.patch('juju.client.connector.Connector.connect_model')
172 async def test_with_model_name(self
, mock_connect_model
, _
):
174 await m
.connect(model_name
='foo')
175 mock_connect_model
.assert_called_once_with('foo')
177 @asynctest.patch('juju.client.connector.Connector.connect_model')
178 async def test_with_endpoint_but_no_uuid(self
, mock_connect_model
, _
):
180 with self
.assertRaises(TypeError):
181 await m
.connect(endpoint
='0.1.2.3:4566')
182 self
.assertEqual(mock_connect_model
.call_count
, 0)
184 @asynctest.patch('juju.client.connector.Connector.connect')
185 async def test_with_endpoint_and_uuid_no_auth(self
, mock_connect
, _
):
187 with self
.assertRaises(TypeError):
188 await m
.connect(endpoint
='0.1.2.3:4566', uuid
='some-uuid')
189 self
.assertEqual(mock_connect
.call_count
, 0)
191 @asynctest.patch('juju.client.connector.Connector.connect')
192 async def test_with_endpoint_and_uuid_with_userpass(self
, mock_connect
, _
):
194 with self
.assertRaises(TypeError):
195 await m
.connect(endpoint
='0.1.2.3:4566',
198 await m
.connect(endpoint
='0.1.2.3:4566',
202 mock_connect
.assert_called_once_with(endpoint
='0.1.2.3:4566',
207 @asynctest.patch('juju.client.connector.Connector.connect')
208 async def test_with_endpoint_and_uuid_with_bakery(self
, mock_connect
, _
):
210 await m
.connect(endpoint
='0.1.2.3:4566',
212 bakery_client
='bakery')
213 mock_connect
.assert_called_once_with(endpoint
='0.1.2.3:4566',
215 bakery_client
='bakery')
217 @asynctest.patch('juju.client.connector.Connector.connect')
218 async def test_with_endpoint_and_uuid_with_macaroon(self
, mock_connect
, _
):
220 with self
.assertRaises(TypeError):
221 await m
.connect(endpoint
='0.1.2.3:4566',
224 await m
.connect(endpoint
='0.1.2.3:4566',
226 macaroons
=['macaroon'])
227 mock_connect
.assert_called_with(endpoint
='0.1.2.3:4566',
229 macaroons
=['macaroon'])
230 await m
.connect(endpoint
='0.1.2.3:4566',
232 bakery_client
='bakery',
233 macaroons
=['macaroon'])
234 mock_connect
.assert_called_with(endpoint
='0.1.2.3:4566',
236 bakery_client
='bakery',
237 macaroons
=['macaroon'])
239 @asynctest.patch('juju.client.connector.Connector.connect_model')
240 @asynctest.patch('juju.client.connector.Connector.connect')
241 async def test_with_posargs(self
, mock_connect
, mock_connect_model
, _
):
243 await m
.connect('foo')
244 mock_connect_model
.assert_called_once_with('foo')
245 with self
.assertRaises(TypeError):
246 await m
.connect('endpoint', 'uuid')
247 with self
.assertRaises(TypeError):
248 await m
.connect('endpoint', 'uuid', 'user')
249 await m
.connect('endpoint', 'uuid', 'user', 'pass')
250 mock_connect
.assert_called_once_with(endpoint
='endpoint',
254 await m
.connect('endpoint', 'uuid', 'user', 'pass', 'cacert', 'bakery',
255 'macaroons', 'loop', 'max_frame_size')
256 mock_connect
.assert_called_with(endpoint
='endpoint',
261 bakery_client
='bakery',
262 macaroons
='macaroons',
264 max_frame_size
='max_frame_size')