tests/integration/test_unit.py

   1 import asyncio
   2 from tempfile import NamedTemporaryFile
   3
   4 import pytest
   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         for unit in app.units:
  29             action = await unit.run('sleep 1', timeout=0.5)
  30             assert isinstance(action, Action)
  31             assert action.status == 'failed'
  32             break
  33
  34         for unit in app.units:
  35             action = await unit.run('sleep 0.5', timeout=2)
  36             assert isinstance(action, Action)
  37             assert action.status == 'completed'
  38             break
  39
  40
  41 @base.bootstrapped
  42 @pytest.mark.asyncio
  43 async def test_run_action(event_loop):
  44     async def run_action(unit):
  45         # unit.run() returns a juju.action.Action instance
  46         action = await unit.run_action('add-repo', repo='myrepo')
  47         # wait for the action to complete
  48         return await action.wait()
  49
  50     async with base.CleanModel() as model:
  51         app = await model.deploy(
  52             'git',
  53             application_name='git',
  54             series='trusty',
  55             channel='stable',
  56         )
  57
  58         for unit in app.units:
  59             action = await run_action(unit)
  60             assert action.results == {'dir': '/var/git/myrepo.git'}
  61             out = await model.get_action_output(action.entity_id, wait=5)
  62             assert out == {'dir': '/var/git/myrepo.git'}
  63             status = await model.get_action_status(uuid_or_prefix=action.entity_id)
  64             assert status[action.entity_id] == 'completed'
  65             break
  66
  67
  68 @base.bootstrapped
  69 @pytest.mark.asyncio
  70 async def test_scp(event_loop):
  71     # ensure that asyncio.subprocess will work;
  72     try:
  73         asyncio.get_child_watcher().attach_loop(event_loop)
  74     except RuntimeError:
  75         pytest.skip('test_scp will always fail outside of MainThread')
  76     async with base.CleanModel() as model:
  77         app = await model.deploy('ubuntu')
  78
  79         await asyncio.wait_for(
  80             model.block_until(lambda: app.units),
  81             timeout=60)
  82         unit = app.units[0]
  83         await asyncio.wait_for(
  84             model.block_until(lambda: unit.machine),
  85             timeout=60)
  86         machine = unit.machine
  87         await asyncio.wait_for(
  88             model.block_until(lambda: (machine.status == 'running' and
  89                                        machine.agent_status == 'started')),
  90             timeout=480)
  91
  92         with NamedTemporaryFile() as f:
  93             f.write(b'testcontents')
  94             f.flush()
  95             await unit.scp_to(f.name, 'testfile')
  96
  97         with NamedTemporaryFile() as f:
  98             await unit.scp_from('testfile', f.name)
  99             assert f.read() == b'testcontents'