4 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')),
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
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
),
49 machine
= model
.machines
['0']
50 await asyncio
.wait_for(
51 model
.block_until(lambda: (machine
.status
== 'running' and
52 machine
.agent_status
== 'started')),
55 with
NamedTemporaryFile() as f
:
56 f
.write(b
'testcontents')
58 await machine
.scp_to(f
.name
, 'testfile')
60 with
NamedTemporaryFile() as f
:
61 await machine
.scp_from('testfile', f
.name
)
62 assert f
.read() == b
'testcontents'