modules/libjuju/tests/integration/test_machine.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_status(event_loop):
  12     async with base.CleanModel() as model:
  13         await model.deploy(
  14             'ubuntu-0',
  15             application_name='ubuntu',
  16             series='trusty',
  17             channel='stable',
  18         )
  19
  20         await asyncio.wait_for(
  21             model.block_until(lambda: len(model.machines)),
  22             timeout=240)
  23         machine = model.machines['0']
  24
  25         assert machine.status in ('allocating', 'pending')
  26         assert machine.agent_status == 'pending'
  27         assert not machine.agent_version
  28
  29         await asyncio.wait_for(
  30             model.block_until(lambda: (machine.status == 'running' and
  31                                        machine.agent_status == 'started' and
  32                                        machine.agent_version is not None)),
  33             timeout=480)
  34
  35         assert machine.status == 'running'
  36         # there is some inconsistency in the message case between providers
  37         assert machine.status_message.lower() == 'running'
  38         assert machine.agent_status == 'started'
  39         assert machine.agent_version.major >= 2
  40
  41
  42 @base.bootstrapped
  43 @pytest.mark.asyncio
  44 async def test_scp(event_loop):
  45     async with base.CleanModel() as model:
  46         await model.add_machine()
  47         await asyncio.wait_for(
  48             model.block_until(lambda: model.machines),
  49             timeout=240)
  50         machine = model.machines['0']
  51         await asyncio.wait_for(
  52             model.block_until(lambda: (machine.status == 'running' and
  53                                        machine.agent_status == 'started')),
  54             timeout=480)
  55
  56         with NamedTemporaryFile() as f:
  57             f.write(b'testcontents')
  58             f.flush()
  59             await machine.scp_to(f.name, 'testfile')
  60
  61         with NamedTemporaryFile() as f:
  62             await machine.scp_from('testfile', f.name)
  63             assert f.read() == b'testcontents'