tests/integration/test_model.py

   1 import asyncio
   2 from concurrent.futures import ThreadPoolExecutor
   3 from pathlib import Path
   4 import pytest
   5
   6 from .. import base
   7 from juju.model import Model
   8 from juju.client.client import ConfigValue
   9
  10 MB = 1
  11 GB = 1024
  12 SSH_KEY = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCsYMJGNGG74HAJha3n2CFmWYsOOaORnJK6VqNy86pj0MIpvRXBzFzVy09uPQ66GOQhTEoJHEqE77VMui7+62AcMXT+GG7cFHcnU8XVQsGM6UirCcNyWNysfiEMoAdZScJf/GvoY87tMEszhZIUV37z8PUBx6twIqMdr31W1J0IaPa+sV6FEDadeLaNTvancDcHK1zuKsL39jzAg7+LYjKJfEfrsQP+lj/EQcjtKqlhVS5kzsJVfx8ZEd0xhW5G7N6bCdKNalS8mKCMaBXJpijNQ82AiyqCIDCRrre2To0/i7pTjRiL0U9f9mV3S4NJaQaokR050w/ZLySFf6F7joJT mathijs@Qrama-Mathijs'  # noqa
  13
  14
  15 @base.bootstrapped
  16 @pytest.mark.asyncio
  17 async def test_deploy_local_bundle(event_loop):
  18     from pathlib import Path
  19     tests_dir = Path(__file__).absolute().parent.parent
  20     bundle_path = tests_dir / 'bundle'
  21
  22     async with base.CleanModel() as model:
  23         await model.deploy(str(bundle_path))
  24
  25         for app in ('wordpress', 'mysql'):
  26             assert app in model.applications
  27
  28
  29 @base.bootstrapped
  30 @pytest.mark.asyncio
  31 async def test_deploy_bundle(event_loop):
  32     async with base.CleanModel() as model:
  33         await model.deploy('bundle/wiki-simple')
  34
  35         for app in ('wiki', 'mysql'):
  36             assert app in model.applications
  37
  38
  39 @base.bootstrapped
  40 @pytest.mark.asyncio
  41 async def test_deploy_channels_revs(event_loop):
  42     async with base.CleanModel() as model:
  43         charm = 'cs:~johnsca/libjuju-test'
  44         stable = await model.deploy(charm, 'a1')
  45         edge = await model.deploy(charm, 'a2', channel='edge')
  46         rev = await model.deploy(charm+'-2', 'a3')
  47
  48         assert [a.charm_url for a in (stable, edge, rev)] == [
  49             'cs:~johnsca/libjuju-test-1',
  50             'cs:~johnsca/libjuju-test-2',
  51             'cs:~johnsca/libjuju-test-2',
  52         ]
  53
  54
  55 @base.bootstrapped
  56 @pytest.mark.asyncio
  57 async def test_add_machine(event_loop):
  58     from juju.machine import Machine
  59
  60     async with base.CleanModel() as model:
  61         # add a new default machine
  62         machine1 = await model.add_machine()
  63
  64         # add a machine with constraints, disks, and series
  65         machine2 = await model.add_machine(
  66             constraints={
  67                 'mem': 256 * MB,
  68             },
  69             disks=[{
  70                 'pool': 'rootfs',
  71                 'size': 10 * GB,
  72                 'count': 1,
  73             }],
  74             series='xenial',
  75         )
  76
  77         # add a lxd container to machine2
  78         machine3 = await model.add_machine(
  79             'lxd:{}'.format(machine2.id))
  80
  81         for m in (machine1, machine2, machine3):
  82             assert isinstance(m, Machine)
  83
  84         assert len(model.machines) == 3
  85
  86         await machine3.destroy(force=True)
  87         await machine2.destroy(force=True)
  88         res = await machine1.destroy(force=True)
  89
  90         assert res is None
  91         assert len(model.machines) == 0
  92
  93
  94 @base.bootstrapped
  95 @pytest.mark.asyncio
  96 async def test_relate(event_loop):
  97     from juju.relation import Relation
  98
  99     async with base.CleanModel() as model:
 100         await model.deploy(
 101             'ubuntu',
 102             application_name='ubuntu',
 103             series='trusty',
 104             channel='stable',
 105         )
 106         await model.deploy(
 107             'nrpe',
 108             application_name='nrpe',
 109             series='trusty',
 110             channel='stable',
 111             # subordinates must be deployed without units
 112             num_units=0,
 113         )
 114         my_relation = await model.add_relation(
 115             'ubuntu',
 116             'nrpe',
 117         )
 118
 119         assert isinstance(my_relation, Relation)
 120
 121
 122 async def _deploy_in_loop(new_loop, model_name):
 123     new_model = Model(new_loop)
 124     await new_model.connect_model(model_name)
 125     try:
 126         await new_model.deploy('cs:xenial/ubuntu')
 127         assert 'ubuntu' in new_model.applications
 128     finally:
 129         await new_model.disconnect()
 130
 131
 132 @base.bootstrapped
 133 @pytest.mark.asyncio
 134 async def test_explicit_loop(event_loop):
 135     async with base.CleanModel() as model:
 136         model_name = model.info.name
 137         new_loop = asyncio.new_event_loop()
 138         new_loop.run_until_complete(
 139             _deploy_in_loop(new_loop, model_name))
 140         await model._wait_for_new('application', 'ubuntu')
 141         assert 'ubuntu' in model.applications
 142
 143
 144 @base.bootstrapped
 145 @pytest.mark.asyncio
 146 async def test_explicit_loop_threaded(event_loop):
 147     async with base.CleanModel() as model:
 148         model_name = model.info.name
 149         new_loop = asyncio.new_event_loop()
 150         with ThreadPoolExecutor(1) as executor:
 151             f = executor.submit(
 152                 new_loop.run_until_complete,
 153                 _deploy_in_loop(new_loop, model_name))
 154             f.result()
 155         await model._wait_for_new('application', 'ubuntu')
 156         assert 'ubuntu' in model.applications
 157
 158
 159 @base.bootstrapped
 160 @pytest.mark.asyncio
 161 async def test_store_resources_charm(event_loop):
 162     async with base.CleanModel() as model:
 163         ghost = await model.deploy('cs:ghost-19')
 164         assert 'ghost' in model.applications
 165         terminal_statuses = ('active', 'error', 'blocked')
 166         await model.block_until(
 167             lambda: (
 168                 len(ghost.units) > 0 and
 169                 ghost.units[0].workload_status in terminal_statuses)
 170             )
 171         # ghost will go in to blocked (or error, for older
 172         # charm revs) if the resource is missing
 173         assert ghost.units[0].workload_status == 'active'
 174
 175
 176 @base.bootstrapped
 177 @pytest.mark.asyncio
 178 async def test_store_resources_bundle(event_loop):
 179     async with base.CleanModel() as model:
 180         bundle = str(Path(__file__).parent / 'bundle')
 181         await model.deploy(bundle)
 182         assert 'ghost' in model.applications
 183         ghost = model.applications['ghost']
 184         terminal_statuses = ('active', 'error', 'blocked')
 185         await model.block_until(
 186             lambda: (
 187                 len(ghost.units) > 0 and
 188                 ghost.units[0].workload_status in terminal_statuses)
 189             )
 190         # ghost will go in to blocked (or error, for older
 191         # charm revs) if the resource is missing
 192         assert ghost.units[0].workload_status == 'active'
 193
 194
 195 @base.bootstrapped
 196 @pytest.mark.asyncio
 197 async def test_ssh_key(event_loop):
 198     async with base.CleanModel() as model:
 199         await model.add_ssh_key('admin', SSH_KEY)
 200         result = await model.get_ssh_key(True)
 201         result = result.serialize()['results'][0].serialize()['result']
 202         assert SSH_KEY in result
 203         await model.remove_ssh_key('admin', SSH_KEY)
 204         result = await model.get_ssh_key(True)
 205         result = result.serialize()['results'][0].serialize()['result']
 206         assert result is None
 207
 208
 209 @base.bootstrapped
 210 @pytest.mark.asyncio
 211 async def test_get_machines(event_loop):
 212     async with base.CleanModel() as model:
 213         result = await model.get_machines()
 214         assert isinstance(result, list)
 215
 216
 217 @base.bootstrapped
 218 @pytest.mark.asyncio
 219 async def test_watcher_reconnect(event_loop):
 220     async with base.CleanModel() as model:
 221         await model.connection.ws.close()
 222         await asyncio.sleep(0.1)
 223         assert model.connection.is_open
 224
 225
 226 @base.bootstrapped
 227 @pytest.mark.asyncio
 228 async def test_config(event_loop):
 229     async with base.CleanModel() as model:
 230         await model.set_config({
 231             'extra-info': 'booyah',
 232             'test-mode': ConfigValue(value=True),
 233         })
 234         result = await model.get_config()
 235         assert 'extra-info' in result
 236         assert result['extra-info'].source == 'model'
 237         assert result['extra-info'].value == 'booyah'
 238
 239 # @base.bootstrapped
 240 # @pytest.mark.asyncio
 241 # async def test_grant(event_loop)
 242 #    async with base.CleanController() as controller:
 243 #        await controller.add_user('test-model-grant')
 244 #        await controller.grant('test-model-grant', 'superuser')
 245 #    async with base.CleanModel() as model:
 246 #        await model.grant('test-model-grant', 'admin')
 247 #        assert model.get_user('test-model-grant')['access'] == 'admin'
 248 #        await model.grant('test-model-grant', 'login')
 249 #        assert model.get_user('test-model-grant')['access'] == 'login'