2 from tempfile
import NamedTemporaryFile
11 async def test_status(event_loop
):
12 async with base
.CleanModel() as model
:
15 application_name
='ubuntu',
20 await asyncio
.wait_for(
21 model
.block_until(lambda: len(model
.machines
)),
23 machine
= model
.machines
['0']
25 assert machine
.status
in ('allocating', 'pending')
26 assert machine
.agent_status
== 'pending'
27 assert not machine
.agent_version
29 # there is some inconsistency in the capitalization of status_message
30 # between different providers
31 await asyncio
.wait_for(
33 lambda: (machine
.status
== 'running' and
34 machine
.status_message
.lower() == 'running' and
35 machine
.agent_status
== 'started')),
41 async def test_scp(event_loop
):
42 # ensure that asyncio.subprocess will work;
44 asyncio
.get_child_watcher().attach_loop(event_loop
)
46 pytest
.skip('test_scp will always fail outside of MainThread')
47 async with base
.CleanModel() as model
:
48 await model
.add_machine()
49 await asyncio
.wait_for(
50 model
.block_until(lambda: model
.machines
),
52 machine
= model
.machines
['0']
53 await asyncio
.wait_for(
54 model
.block_until(lambda: (machine
.status
== 'running' and
55 machine
.agent_status
== 'started')),
58 with
NamedTemporaryFile() as f
:
59 f
.write(b
'testcontents')
61 await machine
.scp_to(f
.name
, 'testfile', scp_opts
='-p')
63 with
NamedTemporaryFile() as f
:
64 await machine
.scp_from('testfile', f
.name
, scp_opts
='-p')
65 assert f
.read() == b
'testcontents'