2 from tempfile
import NamedTemporaryFile
11 async def test_run(event_loop
):
12 from juju
.action
import Action
14 async with base
.CleanModel() as model
:
15 app
= await model
.deploy(
17 application_name
='ubuntu',
22 for unit
in app
.units
:
23 action
= await unit
.run('unit-get public-address')
24 assert isinstance(action
, Action
)
25 assert 'Stdout' in action
.results
28 for unit
in app
.units
:
29 action
= await unit
.run('sleep 1', timeout
=0.5)
30 assert isinstance(action
, Action
)
31 assert action
.status
== 'failed'
34 for unit
in app
.units
:
35 action
= await unit
.run('sleep 0.5', timeout
=2)
36 assert isinstance(action
, Action
)
37 assert action
.status
== 'completed'
43 async def test_run_action(event_loop
):
44 async def run_action(unit
):
45 # unit.run() returns a juju.action.Action instance
46 action
= await unit
.run_action('add-repo', repo
='myrepo')
47 # wait for the action to complete
48 return await action
.wait()
50 async with base
.CleanModel() as model
:
51 app
= await model
.deploy(
53 application_name
='git',
58 for unit
in app
.units
:
59 action
= await run_action(unit
)
60 assert action
.results
== {'dir': '/var/git/myrepo.git'}
61 out
= await model
.get_action_output(action
.entity_id
, wait
=5)
62 assert out
== {'dir': '/var/git/myrepo.git'}
63 status
= await model
.get_action_status(uuid_or_prefix
=action
.entity_id
)
64 assert status
[action
.entity_id
] == 'completed'
70 async def test_scp(event_loop
):
71 # ensure that asyncio.subprocess will work;
73 asyncio
.get_child_watcher().attach_loop(event_loop
)
75 pytest
.skip('test_scp will always fail outside of MainThread')
76 async with base
.CleanModel() as model
:
77 app
= await model
.deploy('ubuntu')
79 await asyncio
.wait_for(
80 model
.block_until(lambda: app
.units
),
83 await asyncio
.wait_for(
84 model
.block_until(lambda: unit
.machine
),
86 machine
= unit
.machine
87 await asyncio
.wait_for(
88 model
.block_until(lambda: (machine
.status
== 'running' and
89 machine
.agent_status
== 'started')),
92 with
NamedTemporaryFile() as f
:
93 f
.write(b
'testcontents')
95 await unit
.scp_to(f
.name
, 'testfile')
97 with
NamedTemporaryFile() as f
:
98 await unit
.scp_from('testfile', f
.name
)
99 assert f
.read() == b
'testcontents'