modules/libjuju/tests/integration/test_application.py

   1 import asyncio
   2 import pytest
   3
   4 from .. import base
   5
   6 MB = 1
   7
   8
   9 @base.bootstrapped
  10 @pytest.mark.asyncio
  11 async def test_action(event_loop):
  12     async with base.CleanModel() as model:
  13         ubuntu_app = await model.deploy(
  14             'mysql',
  15             application_name='mysql',
  16             series='trusty',
  17             channel='stable',
  18             config={
  19                 'tuning-level': 'safest',
  20             },
  21             constraints={
  22                 'mem': 256 * MB,
  23             },
  24         )
  25
  26         # update and check app config
  27         await ubuntu_app.set_config({'tuning-level': 'fast'})
  28         config = await ubuntu_app.get_config()
  29         assert config['tuning-level']['value'] == 'fast'
  30
  31         # update and check app constraints
  32         await ubuntu_app.set_constraints({'mem': 512 * MB})
  33         constraints = await ubuntu_app.get_constraints()
  34         assert constraints['mem'] == 512 * MB
  35
  36
  37 @base.bootstrapped
  38 @pytest.mark.asyncio
  39 async def test_add_units(event_loop):
  40     from juju.unit import Unit
  41
  42     async with base.CleanModel() as model:
  43         app = await model.deploy(
  44             'ubuntu-0',
  45             application_name='ubuntu',
  46             series='trusty',
  47             channel='stable',
  48         )
  49         units = await app.add_units(count=2)
  50
  51         assert len(units) == 2
  52         for unit in units:
  53             assert isinstance(unit, Unit)
  54
  55
  56 @base.bootstrapped
  57 @pytest.mark.asyncio
  58 async def test_upgrade_charm(event_loop):
  59     async with base.CleanModel() as model:
  60         app = await model.deploy('ubuntu-0')
  61         assert app.data['charm-url'] == 'cs:ubuntu-0'
  62         await app.upgrade_charm()
  63         assert app.data['charm-url'].startswith('cs:ubuntu-')
  64         assert app.data['charm-url'] != 'cs:ubuntu-0'
  65
  66
  67 @base.bootstrapped
  68 @pytest.mark.asyncio
  69 async def test_upgrade_charm_channel(event_loop):
  70     async with base.CleanModel() as model:
  71         app = await model.deploy('ubuntu-0')
  72         assert app.data['charm-url'] == 'cs:ubuntu-0'
  73         await app.upgrade_charm(channel='stable')
  74         assert app.data['charm-url'].startswith('cs:ubuntu-')
  75         assert app.data['charm-url'] != 'cs:ubuntu-0'
  76
  77
  78 @base.bootstrapped
  79 @pytest.mark.asyncio
  80 async def test_upgrade_charm_revision(event_loop):
  81     async with base.CleanModel() as model:
  82         app = await model.deploy('ubuntu-0')
  83         assert app.data['charm-url'] == 'cs:ubuntu-0'
  84         await app.upgrade_charm(revision=8)
  85         assert app.data['charm-url'] == 'cs:ubuntu-8'
  86
  87
  88 @base.bootstrapped
  89 @pytest.mark.asyncio
  90 async def test_upgrade_charm_switch(event_loop):
  91     async with base.CleanModel() as model:
  92         app = await model.deploy('ubuntu-0')
  93         assert app.data['charm-url'] == 'cs:ubuntu-0'
  94         await app.upgrade_charm(switch='ubuntu-8')
  95         assert app.data['charm-url'] == 'cs:ubuntu-8'
  96
  97
  98 @base.bootstrapped
  99 @pytest.mark.asyncio
 100 async def test_upgrade_charm_resource(event_loop):
 101     async with base.CleanModel() as model:
 102         app = await model.deploy('cs:~cynerva/upgrade-charm-resource-test-1')
 103
 104         def units_ready():
 105             if not app.units:
 106                 return False
 107             unit = app.units[0]
 108             return unit.workload_status == 'active' and \
 109                 unit.agent_status == 'idle'
 110
 111         await asyncio.wait_for(model.block_until(units_ready), timeout=480)
 112         unit = app.units[0]
 113         expected_message = 'I have no resource.'
 114         assert unit.workload_status_message == expected_message
 115
 116         await app.upgrade_charm(revision=2)
 117         await asyncio.wait_for(
 118             model.block_until(
 119                 lambda: unit.workload_status_message != 'I have no resource.'
 120             ),
 121             timeout=60
 122         )
 123         expected_message = 'My resource: I am the resource.'
 124         assert app.units[0].workload_status_message == expected_message