modules/libjuju/tests/integration/test_application.py

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