8957ae16b6330847b31624d9eb4cba19d6eae562
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 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)),
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
44 async def test_scp(event_loop
):
45 # ensure that asyncio.subprocess will work;
47 asyncio
.get_child_watcher().attach_loop(event_loop
)
49 pytest
.skip('test_scp will always fail outside of MainThread')
50 async with base
.CleanModel() as model
:
51 await model
.add_machine()
52 await asyncio
.wait_for(
53 model
.block_until(lambda: model
.machines
),
55 machine
= model
.machines
['0']
56 await asyncio
.wait_for(
57 model
.block_until(lambda: (machine
.status
== 'running' and
58 machine
.agent_status
== 'started')),
61 with
NamedTemporaryFile() as f
:
62 f
.write(b
'testcontents')
64 await machine
.scp_to(f
.name
, 'testfile')
66 with
NamedTemporaryFile() as f
:
67 await machine
.scp_from('testfile', f
.name
)
68 assert f
.read() == b
'testcontents'