7 from juju
.client
.jujudata
import FileJujuData
10 def _make_delta(entity
, type_
, data
=None):
11 from juju
.client
.client
import Delta
12 from juju
.delta
import get_entity_delta
14 delta
= Delta([entity
, type_
, data
])
15 return get_entity_delta(delta
)
18 class TestObserver(unittest
.TestCase
):
19 def _make_observer(self
, *args
):
20 from juju
.model
import _Observer
21 return _Observer(*args
)
23 def test_cares_about_id(self
):
26 o
= self
._make
_observer
(
27 None, None, None, id_
, None)
30 'application', 'change', dict(name
=id_
))
32 self
.assertTrue(o
.cares_about(delta
))
34 def test_cares_about_type(self
):
37 o
= self
._make
_observer
(
38 None, type_
, None, None, None)
41 type_
, 'change', dict(name
='foo'))
43 self
.assertTrue(o
.cares_about(delta
))
45 def test_cares_about_action(self
):
48 o
= self
._make
_observer
(
49 None, None, action
, None, None)
52 'application', action
, dict(name
='foo'))
54 self
.assertTrue(o
.cares_about(delta
))
56 def test_cares_about_predicate(self
):
58 return delta
.data
.get('fizz') == 'bang'
60 o
= self
._make
_observer
(
61 None, None, None, None, predicate
)
64 'application', 'change', dict(fizz
='bang'))
66 self
.assertTrue(o
.cares_about(delta
))
69 class TestModelState(unittest
.TestCase
):
70 def test_apply_delta(self
):
71 from juju
.model
import Model
72 from juju
.application
import Application
75 model
._connector
= mock
.MagicMock()
76 delta
= _make_delta('application', 'add', dict(name
='foo'))
79 prev
, new
= model
.state
.apply_delta(delta
)
81 len(model
.state
.state
[delta
.entity
][delta
.get_id()]), 1)
82 self
.assertIsNone(prev
)
83 self
.assertIsInstance(new
, Application
)
87 prev
, new
= model
.state
.apply_delta(delta
)
88 # length of the entity history deque is now 3:
89 # - 1 for the first delta
90 # - 1 for the second delta
91 # - 1 for the None sentinel appended after the 'remove'
93 len(model
.state
.state
[delta
.entity
][delta
.get_id()]), 3)
94 self
.assertIsInstance(new
, Application
)
95 # new object is falsy because its data is None
97 self
.assertIsInstance(prev
, Application
)
101 def test_get_series():
102 from juju
.model
import Model
106 'supported-series': {
114 assert model
._get
_series
('cs:trusty/ubuntu', entity
) == 'trusty'
115 assert model
._get
_series
('xenial/ubuntu', entity
) == 'xenial'
116 assert model
._get
_series
('~foo/xenial/ubuntu', entity
) == 'xenial'
117 assert model
._get
_series
('~foo/ubuntu', entity
) == 'xenial'
118 assert model
._get
_series
('ubuntu', entity
) == 'xenial'
119 assert model
._get
_series
('cs:ubuntu', entity
) == 'xenial'
122 class TestContextManager(asynctest
.TestCase
):
123 @asynctest.patch('juju.model.Model.disconnect')
124 @asynctest.patch('juju.model.Model.connect')
125 async def test_normal_use(self
, mock_connect
, mock_disconnect
):
126 from juju
.model
import Model
128 async with
Model() as model
:
129 self
.assertTrue(isinstance(model
, Model
))
131 self
.assertTrue(mock_connect
.called
)
132 self
.assertTrue(mock_disconnect
.called
)
134 @asynctest.patch('juju.model.Model.disconnect')
135 @asynctest.patch('juju.model.Model.connect')
136 async def test_exception(self
, mock_connect
, mock_disconnect
):
137 from juju
.model
import Model
139 class SomeException(Exception):
142 with self
.assertRaises(SomeException
):
144 raise SomeException()
146 self
.assertTrue(mock_connect
.called
)
147 self
.assertTrue(mock_disconnect
.called
)
149 async def test_no_current_connection(self
):
150 from juju
.model
import Model
151 from juju
.errors
import JujuConnectionError
153 class NoControllerJujuData(FileJujuData
):
154 def current_controller(self
):
157 with self
.assertRaises(JujuConnectionError
):
158 async with
Model(jujudata
=NoControllerJujuData()):
162 class TestModelConnect(asynctest
.TestCase
):
163 @asynctest.patch('juju.client.connector.Connector.connect_model')
164 @asynctest.patch('juju.model.Model._after_connect')
165 async def test_model_connect_no_args(self
, mock_after_connect
, mock_connect_model
):
166 from juju
.model
import Model
169 mock_connect_model
.assert_called_once_with(None)
171 @asynctest.patch('juju.client.connector.Connector.connect_model')
172 @asynctest.patch('juju.model.Model._after_connect')
173 async def test_model_connect_with_model_name(self
, mock_after_connect
, mock_connect_model
):
174 from juju
.model
import Model
176 await m
.connect(model_name
='foo')
177 mock_connect_model
.assert_called_once_with('foo')
179 @asynctest.patch('juju.client.connector.Connector.connect_model')
180 @asynctest.patch('juju.model.Model._after_connect')
181 async def test_model_connect_with_endpoint_but_no_uuid(
186 from juju
.model
import Model
188 with self
.assertRaises(ValueError):
189 await m
.connect(endpoint
='0.1.2.3:4566')
190 self
.assertEqual(mock_connect_model
.call_count
, 0)
192 @asynctest.patch('juju.client.connector.Connector.connect')
193 @asynctest.patch('juju.model.Model._after_connect')
194 async def test_model_connect_with_endpoint_and_uuid(
199 from juju
.model
import Model
201 await m
.connect(endpoint
='0.1.2.3:4566', uuid
='some-uuid')
202 mock_connect
.assert_called_once_with(endpoint
='0.1.2.3:4566', uuid
='some-uuid')