tests/integration/test_unit.py

   1 import pytest
   2
   3 from .. import base
   4
   5
   6 @base.bootstrapped
   7 @pytest.mark.asyncio
   8 async def test_run(event_loop):
   9     from juju.action import Action
  10
  11     async with base.CleanModel() as model:
  12         app = await model.deploy(
  13             'ubuntu-0',
  14             application_name='ubuntu',
  15             series='trusty',
  16             channel='stable',
  17         )
  18
  19         for unit in app.units:
  20             action = await unit.run('unit-get public-address')
  21             assert isinstance(action, Action)
  22             assert 'Stdout' in action.results
  23             break
  24
  25
  26 @base.bootstrapped
  27 @pytest.mark.asyncio
  28 async def test_run_action(event_loop):
  29     async def run_action(unit):
  30         # unit.run() returns a juju.action.Action instance
  31         action = await unit.run_action('add-repo', repo='myrepo')
  32         # wait for the action to complete
  33         return await action.wait()
  34
  35     async with base.CleanModel() as model:
  36         app = await model.deploy(
  37             'git',
  38             application_name='git',
  39             series='trusty',
  40             channel='stable',
  41         )
  42
  43         for unit in app.units:
  44             action = await run_action(unit)
  45             assert action.results == {'dir': '/var/git/myrepo.git'}
  46             break