modules/libjuju/tests/unit/test_model.py

   1 import unittest
   2
   3 import mock
   4 import asynctest
   5
   6
   7 def _make_delta(entity, type_, data=None):
   8     from juju.client.client import Delta
   9     from juju.delta import get_entity_delta
  10
  11     delta = Delta([entity, type_, data])
  12     return get_entity_delta(delta)
  13
  14
  15 class TestObserver(unittest.TestCase):
  16     def _make_observer(self, *args):
  17         from juju.model import _Observer
  18         return _Observer(*args)
  19
  20     def test_cares_about_id(self):
  21         id_ = 'foo'
  22
  23         o = self._make_observer(
  24             None, None, None, id_, None)
  25
  26         delta = _make_delta(
  27             'application', 'change', dict(name=id_))
  28
  29         self.assertTrue(o.cares_about(delta))
  30
  31     def test_cares_about_type(self):
  32         type_ = 'application'
  33
  34         o = self._make_observer(
  35             None, type_, None, None, None)
  36
  37         delta = _make_delta(
  38             type_, 'change', dict(name='foo'))
  39
  40         self.assertTrue(o.cares_about(delta))
  41
  42     def test_cares_about_action(self):
  43         action = 'change'
  44
  45         o = self._make_observer(
  46             None, None, action, None, None)
  47
  48         delta = _make_delta(
  49             'application', action, dict(name='foo'))
  50
  51         self.assertTrue(o.cares_about(delta))
  52
  53     def test_cares_about_predicate(self):
  54         def predicate(delta):
  55             return delta.data.get('fizz') == 'bang'
  56
  57         o = self._make_observer(
  58             None, None, None, None, predicate)
  59
  60         delta = _make_delta(
  61             'application', 'change', dict(fizz='bang'))
  62
  63         self.assertTrue(o.cares_about(delta))
  64
  65
  66 class TestModelState(unittest.TestCase):
  67     def test_apply_delta(self):
  68         from juju.model import Model
  69         from juju.application import Application
  70
  71         loop = mock.MagicMock()
  72         model = Model(loop=loop)
  73         delta = _make_delta('application', 'add', dict(name='foo'))
  74
  75         # test add
  76         prev, new = model.state.apply_delta(delta)
  77         self.assertEqual(
  78             len(model.state.state[delta.entity][delta.get_id()]), 1)
  79         self.assertIsNone(prev)
  80         self.assertIsInstance(new, Application)
  81
  82         # test remove
  83         delta.type = 'remove'
  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'
  89         self.assertEqual(
  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
  93         self.assertFalse(new)
  94         self.assertIsInstance(prev, Application)
  95         self.assertTrue(prev)
  96
  97
  98 def test_get_series():
  99     from juju.model import Model
 100     model = Model()
 101     entity = {
 102         'Meta': {
 103             'supported-series': {
 104                 'SupportedSeries': [
 105                     'xenial',
 106                     'trusty',
 107                 ],
 108             },
 109         },
 110     }
 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'
 117
 118
 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
 124
 125         async with Model() as model:
 126             self.assertTrue(isinstance(model, Model))
 127
 128         self.assertTrue(mock_connect.called)
 129         self.assertTrue(mock_disconnect.called)
 130
 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
 135
 136         class SomeException(Exception):
 137             pass
 138
 139         with self.assertRaises(SomeException):
 140             async with Model():
 141                 raise SomeException()
 142
 143         self.assertTrue(mock_connect.called)
 144         self.assertTrue(mock_disconnect.called)
 145
 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
 150
 151         mock_current_controller.return_value = ""
 152
 153         with self.assertRaises(JujuConnectionError):
 154             async with Model():
 155                 pass