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
   9
  10 def _make_delta(entity, type_, data=None):
  11     from juju.client.client import Delta
  12     from juju.delta import get_entity_delta
  13
  14     delta = Delta([entity, type_, data])
  15     return get_entity_delta(delta)
  16
  17
  18 class TestObserver(unittest.TestCase):
  19     def _make_observer(self, *args):
  20         from juju.model import _Observer
  21         return _Observer(*args)
  22
  23     def test_cares_about_id(self):
  24         id_ = 'foo'
  25
  26         o = self._make_observer(
  27             None, None, None, id_, None)
  28
  29         delta = _make_delta(
  30             'application', 'change', dict(name=id_))
  31
  32         self.assertTrue(o.cares_about(delta))
  33
  34     def test_cares_about_type(self):
  35         type_ = 'application'
  36
  37         o = self._make_observer(
  38             None, type_, None, None, None)
  39
  40         delta = _make_delta(
  41             type_, 'change', dict(name='foo'))
  42
  43         self.assertTrue(o.cares_about(delta))
  44
  45     def test_cares_about_action(self):
  46         action = 'change'
  47
  48         o = self._make_observer(
  49             None, None, action, None, None)
  50
  51         delta = _make_delta(
  52             'application', action, dict(name='foo'))
  53
  54         self.assertTrue(o.cares_about(delta))
  55
  56     def test_cares_about_predicate(self):
  57         def predicate(delta):
  58             return delta.data.get('fizz') == 'bang'
  59
  60         o = self._make_observer(
  61             None, None, None, None, predicate)
  62
  63         delta = _make_delta(
  64             'application', 'change', dict(fizz='bang'))
  65
  66         self.assertTrue(o.cares_about(delta))
  67
  68
  69 class TestModelState(unittest.TestCase):
  70     def test_apply_delta(self):
  71         from juju.model import Model
  72         from juju.application import Application
  73
  74         model = Model()
  75         model._connector = mock.MagicMock()
  76         delta = _make_delta('application', 'add', dict(name='foo'))
  77
  78         # test add
  79         prev, new = model.state.apply_delta(delta)
  80         self.assertEqual(
  81             len(model.state.state[delta.entity][delta.get_id()]), 1)
  82         self.assertIsNone(prev)
  83         self.assertIsInstance(new, Application)
  84
  85         # test remove
  86         delta.type = 'remove'
  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'
  92         self.assertEqual(
  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
  96         self.assertFalse(new)
  97         self.assertIsInstance(prev, Application)
  98         self.assertTrue(prev)
  99
 100
 101 def test_get_series():
 102     from juju.model import Model
 103     model = Model()
 104     entity = {
 105         'Meta': {
 106             'supported-series': {
 107                 'SupportedSeries': [
 108                     'xenial',
 109                     'trusty',
 110                 ],
 111             },
 112         },
 113     }
 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'
 120
 121
 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
 127
 128         async with Model() as model:
 129             self.assertTrue(isinstance(model, Model))
 130
 131         self.assertTrue(mock_connect.called)
 132         self.assertTrue(mock_disconnect.called)
 133
 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
 138
 139         class SomeException(Exception):
 140             pass
 141
 142         with self.assertRaises(SomeException):
 143             async with Model():
 144                 raise SomeException()
 145
 146         self.assertTrue(mock_connect.called)
 147         self.assertTrue(mock_disconnect.called)
 148
 149     async def test_no_current_connection(self):
 150         from juju.model import Model
 151         from juju.errors import JujuConnectionError
 152
 153         class NoControllerJujuData(FileJujuData):
 154             def current_controller(self):
 155                 return ""
 156
 157         with self.assertRaises(JujuConnectionError):
 158             async with Model(jujudata=NoControllerJujuData()):
 159                 pass
 160
 161
 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
 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     @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
 175         m = Model()
 176         await m.connect(model_name='foo')
 177         mock_connect_model.assert_called_once_with('foo')
 178
 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(
 182         self,
 183         mock_after_connect,
 184         mock_connect_model,
 185     ):
 186         from juju.model import Model
 187         m = 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)
 191
 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(
 195         self,
 196         mock_after_connect,
 197         mock_connect,
 198     ):
 199         from juju.model import Model
 200         m = 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')