modules/libjuju/tests/integration/test_unit.py

   1 import asyncio
   2 import pytest
   3
   4 from tempfile import NamedTemporaryFile
   5
   6 from .. import base
   7
   8
   9 @base.bootstrapped
  10 @pytest.mark.asyncio
  11 async def test_run(event_loop):
  12     from juju.action import Action
  13
  14     async with base.CleanModel() as model:
  15         app = await model.deploy(
  16             'ubuntu-0',
  17             application_name='ubuntu',
  18             series='trusty',
  19             channel='stable',
  20         )
  21
  22         for unit in app.units:
  23             action = await unit.run('unit-get public-address')
  24             assert isinstance(action, Action)
  25             assert 'Stdout' in action.results
  26             break
  27
  28
  29 @base.bootstrapped
  30 @pytest.mark.asyncio
  31 async def test_run_action(event_loop):
  32     async def run_action(unit):
  33         # unit.run() returns a juju.action.Action instance
  34         action = await unit.run_action('add-repo', repo='myrepo')
  35         # wait for the action to complete
  36         return await action.wait()
  37
  38     async with base.CleanModel() as model:
  39         app = await model.deploy(
  40             'git',
  41             application_name='git',
  42             series='trusty',
  43             channel='stable',
  44         )
  45
  46         for unit in app.units:
  47             action = await run_action(unit)
  48             assert action.results == {'dir': '/var/git/myrepo.git'}
  49             break
  50
  51
  52 @base.bootstrapped
  53 @pytest.mark.asyncio
  54 async def test_scp(event_loop):
  55     async with base.CleanModel() as model:
  56         app = await model.deploy('ubuntu')
  57
  58         await asyncio.wait_for(
  59             model.block_until(lambda: app.units),
  60             timeout=60)
  61         unit = app.units[0]
  62         await asyncio.wait_for(
  63             model.block_until(lambda: unit.machine),
  64             timeout=60)
  65         machine = unit.machine
  66         await asyncio.wait_for(
  67             model.block_until(lambda: (machine.status == 'running' and
  68                                        machine.agent_status == 'started')),
  69             timeout=480)
  70
  71         with NamedTemporaryFile() as f:
  72             f.write(b'testcontents')
  73             f.flush()
  74             await unit.scp_to(f.name, 'testfile')
  75
  76         with NamedTemporaryFile() as f:
  77             await unit.scp_from('testfile', f.name)
  78             assert f.read() == b'testcontents'