modules/libjuju/tests/unit/test_model.py

   1 import unittest
   2
   3 import mock
   4
   5 import asynctest
   6
   7 from juju.client.jujudata import FileJujuData
   8 from juju.model import Model
   9
  10
  11 def _make_delta(entity, type_, data=None):
  12     from juju.client.client import Delta
  13     from juju.delta import get_entity_delta
  14
  15     delta = Delta([entity, type_, data])
  16     return get_entity_delta(delta)
  17
  18
  19 class TestObserver(unittest.TestCase):
  20     def _make_observer(self, *args):
  21         from juju.model import _Observer
  22         return _Observer(*args)
  23
  24     def test_cares_about_id(self):
  25         id_ = 'foo'
  26
  27         o = self._make_observer(
  28             None, None, None, id_, None)
  29
  30         delta = _make_delta(
  31             'application', 'change', dict(name=id_))
  32
  33         self.assertTrue(o.cares_about(delta))
  34
  35     def test_cares_about_type(self):
  36         type_ = 'application'
  37
  38         o = self._make_observer(
  39             None, type_, None, None, None)
  40
  41         delta = _make_delta(
  42             type_, 'change', dict(name='foo'))
  43
  44         self.assertTrue(o.cares_about(delta))
  45
  46     def test_cares_about_action(self):
  47         action = 'change'
  48
  49         o = self._make_observer(
  50             None, None, action, None, None)
  51
  52         delta = _make_delta(
  53             'application', action, dict(name='foo'))
  54
  55         self.assertTrue(o.cares_about(delta))
  56
  57     def test_cares_about_predicate(self):
  58         def predicate(delta):
  59             return delta.data.get('fizz') == 'bang'
  60
  61         o = self._make_observer(
  62             None, None, None, None, predicate)
  63
  64         delta = _make_delta(
  65             'application', 'change', dict(fizz='bang'))
  66
  67         self.assertTrue(o.cares_about(delta))
  68
  69
  70 class TestModelState(unittest.TestCase):
  71     def test_apply_delta(self):
  72         from juju.model import Model
  73         from juju.application import Application
  74
  75         model = Model()
  76         model._connector = mock.MagicMock()
  77         delta = _make_delta('application', 'add', dict(name='foo'))
  78
  79         # test add
  80         prev, new = model.state.apply_delta(delta)
  81         self.assertEqual(
  82             len(model.state.state[delta.entity][delta.get_id()]), 1)
  83         self.assertIsNone(prev)
  84         self.assertIsInstance(new, Application)
  85
  86         # test remove
  87         delta.type = 'remove'
  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'
  93         self.assertEqual(
  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
  97         self.assertFalse(new)
  98         self.assertIsInstance(prev, Application)
  99         self.assertTrue(prev)
 100
 101
 102 def test_get_series():
 103     from juju.model import Model
 104     model = Model()
 105     entity = {
 106         'Meta': {
 107             'supported-series': {
 108                 'SupportedSeries': [
 109                     'xenial',
 110                     'trusty',
 111                 ],
 112             },
 113         },
 114     }
 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'
 121
 122
 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
 128
 129         async with Model() as model:
 130             self.assertTrue(isinstance(model, Model))
 131
 132         self.assertTrue(mock_connect.called)
 133         self.assertTrue(mock_disconnect.called)
 134
 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
 139
 140         class SomeException(Exception):
 141             pass
 142
 143         with self.assertRaises(SomeException):
 144             async with Model():
 145                 raise SomeException()
 146
 147         self.assertTrue(mock_connect.called)
 148         self.assertTrue(mock_disconnect.called)
 149
 150     async def test_no_current_connection(self):
 151         from juju.model import Model
 152         from juju.errors import JujuConnectionError
 153
 154         class NoControllerJujuData(FileJujuData):
 155             def current_controller(self):
 156                 return ""
 157
 158         with self.assertRaises(JujuConnectionError):
 159             async with Model(jujudata=NoControllerJujuData()):
 160                 pass
 161
 162
 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, _):
 167         m = Model()
 168         await m.connect()
 169         mock_connect_model.assert_called_once_with(None)
 170
 171     @asynctest.patch('juju.client.connector.Connector.connect_model')
 172     async def test_with_model_name(self, mock_connect_model, _):
 173         m = Model()
 174         await m.connect(model_name='foo')
 175         mock_connect_model.assert_called_once_with('foo')
 176
 177     @asynctest.patch('juju.client.connector.Connector.connect_model')
 178     async def test_with_endpoint_but_no_uuid(self, mock_connect_model, _):
 179         m = 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)
 183
 184     @asynctest.patch('juju.client.connector.Connector.connect')
 185     async def test_with_endpoint_and_uuid_no_auth(self, mock_connect, _):
 186         m = Model()
 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)
 190
 191     @asynctest.patch('juju.client.connector.Connector.connect')
 192     async def test_with_endpoint_and_uuid_with_userpass(self, mock_connect, _):
 193         m = Model()
 194         with self.assertRaises(TypeError):
 195             await m.connect(endpoint='0.1.2.3:4566',
 196                             uuid='some-uuid',
 197                             username='user')
 198         await m.connect(endpoint='0.1.2.3:4566',
 199                         uuid='some-uuid',
 200                         username='user',
 201                         password='pass')
 202         mock_connect.assert_called_once_with(endpoint='0.1.2.3:4566',
 203                                              uuid='some-uuid',
 204                                              username='user',
 205                                              password='pass')
 206
 207     @asynctest.patch('juju.client.connector.Connector.connect')
 208     async def test_with_endpoint_and_uuid_with_bakery(self, mock_connect, _):
 209         m = Model()
 210         await m.connect(endpoint='0.1.2.3:4566',
 211                         uuid='some-uuid',
 212                         bakery_client='bakery')
 213         mock_connect.assert_called_once_with(endpoint='0.1.2.3:4566',
 214                                              uuid='some-uuid',
 215                                              bakery_client='bakery')
 216
 217     @asynctest.patch('juju.client.connector.Connector.connect')
 218     async def test_with_endpoint_and_uuid_with_macaroon(self, mock_connect, _):
 219         m = Model()
 220         with self.assertRaises(TypeError):
 221             await m.connect(endpoint='0.1.2.3:4566',
 222                             uuid='some-uuid',
 223                             username='user')
 224         await m.connect(endpoint='0.1.2.3:4566',
 225                         uuid='some-uuid',
 226                         macaroons=['macaroon'])
 227         mock_connect.assert_called_with(endpoint='0.1.2.3:4566',
 228                                         uuid='some-uuid',
 229                                         macaroons=['macaroon'])
 230         await m.connect(endpoint='0.1.2.3:4566',
 231                         uuid='some-uuid',
 232                         bakery_client='bakery',
 233                         macaroons=['macaroon'])
 234         mock_connect.assert_called_with(endpoint='0.1.2.3:4566',
 235                                         uuid='some-uuid',
 236                                         bakery_client='bakery',
 237                                         macaroons=['macaroon'])
 238
 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, _):
 242         m = 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',
 251                                              uuid='uuid',
 252                                              username='user',
 253                                              password='pass')
 254         await m.connect('endpoint', 'uuid', 'user', 'pass', 'cacert', 'bakery',
 255                         'macaroons', 'loop', 'max_frame_size')
 256         mock_connect.assert_called_with(endpoint='endpoint',
 257                                         uuid='uuid',
 258                                         username='user',
 259                                         password='pass',
 260                                         cacert='cacert',
 261                                         bakery_client='bakery',
 262                                         macaroons='macaroons',
 263                                         loop='loop',
 264                                         max_frame_size='max_frame_size')