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')),
  32             timeout=480)
  33
  34         assert machine.status == 'running'
  35         # there is some inconsistency in the message case between providers
  36         assert machine.status_message.lower() == 'running'
  37         assert machine.agent_status == 'started'
  38         assert machine.agent_version.major >= 2
  39
  40
  41 @base.bootstrapped
  42 @pytest.mark.asyncio
  43 async def test_scp(event_loop):
  44     async with base.CleanModel() as model:
  45         await model.add_machine()
  46         await asyncio.wait_for(
  47             model.block_until(lambda: model.machines),
  48             timeout=240)
  49         machine = model.machines['0']
  50         await asyncio.wait_for(
  51             model.block_until(lambda: (machine.status == 'running' and
  52                                        machine.agent_status == 'started')),
  53             timeout=480)
  54
  55         with NamedTemporaryFile() as f:
  56             f.write(b'testcontents')
  57             f.flush()
  58             await machine.scp_to(f.name, 'testfile')
  59
  60         with NamedTemporaryFile() as f:
  61             await machine.scp_from('testfile', f.name)
  62             assert f.read() == b'testcontents'