tests/integration/test_model.py

   1 import asyncio
   2 from concurrent.futures import ThreadPoolExecutor
   3 import pytest
   4
   5 from .. import base
   6 from juju.model import Model
   7
   8 MB = 1
   9 GB = 1024
  10
  11
  12 @base.bootstrapped
  13 @pytest.mark.asyncio
  14 async def test_deploy_local_bundle(event_loop):
  15     from pathlib import Path
  16     tests_dir = Path(__file__).absolute().parent.parent
  17     bundle_path = tests_dir / 'bundle'
  18
  19     async with base.CleanModel() as model:
  20         await model.deploy(str(bundle_path))
  21
  22         for app in ('wordpress', 'mysql'):
  23             assert app in model.applications
  24
  25
  26 @base.bootstrapped
  27 @pytest.mark.asyncio
  28 async def test_deploy_bundle(event_loop):
  29     async with base.CleanModel() as model:
  30         await model.deploy('bundle/wiki-simple')
  31
  32         for app in ('wiki', 'mysql'):
  33             assert app in model.applications
  34
  35
  36 @base.bootstrapped
  37 @pytest.mark.asyncio
  38 async def test_add_machine(event_loop):
  39     from juju.machine import Machine
  40
  41     async with base.CleanModel() as model:
  42         # add a new default machine
  43         machine1 = await model.add_machine()
  44
  45         # add a machine with constraints, disks, and series
  46         machine2 = await model.add_machine(
  47             constraints={
  48                 'mem': 256 * MB,
  49             },
  50             disks=[{
  51                 'pool': 'rootfs',
  52                 'size': 10 * GB,
  53                 'count': 1,
  54             }],
  55             series='xenial',
  56         )
  57
  58         # add a lxd container to machine2
  59         machine3 = await model.add_machine(
  60             'lxd:{}'.format(machine2.id))
  61
  62         for m in (machine1, machine2, machine3):
  63             assert isinstance(m, Machine)
  64
  65         assert len(model.machines) == 3
  66
  67         await machine3.destroy(force=True)
  68         await machine2.destroy(force=True)
  69         res = await machine1.destroy(force=True)
  70
  71         assert res is None
  72         assert len(model.machines) == 0
  73
  74
  75 @base.bootstrapped
  76 @pytest.mark.asyncio
  77 async def test_relate(event_loop):
  78     from juju.relation import Relation
  79
  80     async with base.CleanModel() as model:
  81         await model.deploy(
  82             'ubuntu',
  83             application_name='ubuntu',
  84             series='trusty',
  85             channel='stable',
  86         )
  87         await model.deploy(
  88             'nrpe',
  89             application_name='nrpe',
  90             series='trusty',
  91             channel='stable',
  92             # subordinates must be deployed without units
  93             num_units=0,
  94         )
  95         my_relation = await model.add_relation(
  96             'ubuntu',
  97             'nrpe',
  98         )
  99
 100         assert isinstance(my_relation, Relation)
 101
 102
 103 async def _deploy_in_loop(new_loop, model_name):
 104     new_model = Model(new_loop)
 105     await new_model.connect_model(model_name)
 106     try:
 107         await new_model.deploy('cs:xenial/ubuntu')
 108         assert 'ubuntu' in new_model.applications
 109     finally:
 110         await new_model.disconnect()
 111
 112
 113 @base.bootstrapped
 114 @pytest.mark.asyncio
 115 async def test_explicit_loop(event_loop):
 116     async with base.CleanModel() as model:
 117         model_name = model.info.name
 118         new_loop = asyncio.new_event_loop()
 119         new_loop.run_until_complete(
 120             _deploy_in_loop(new_loop, model_name))
 121         await model._wait_for_new('application', 'ubuntu')
 122         assert 'ubuntu' in model.applications
 123
 124
 125 @base.bootstrapped
 126 @pytest.mark.asyncio
 127 async def test_explicit_loop_threaded(event_loop):
 128     async with base.CleanModel() as model:
 129         model_name = model.info.name
 130         new_loop = asyncio.new_event_loop()
 131         with ThreadPoolExecutor(1) as executor:
 132             f = executor.submit(
 133                 new_loop.run_until_complete,
 134                 _deploy_in_loop(new_loop, model_name))
 135             f.result()
 136         await model._wait_for_new('application', 'ubuntu')
 137         assert 'ubuntu' in model.applications