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
31 async def test_run_action(event_loop
):
32 async def run_action(unit
):
33 # unit.run() returns a juju.action.Action instance
34 action
= await unit
.run_action('add-repo', repo
='myrepo')
35 # wait for the action to complete
36 return await action
.wait()
38 async with base
.CleanModel() as model
:
39 app
= await model
.deploy(
41 application_name
='git',
46 for unit
in app
.units
:
47 action
= await run_action(unit
)
48 assert action
.results
== {'dir': '/var/git/myrepo.git'}
54 async def test_scp(event_loop
):
55 # ensure that asyncio.subprocess will work;
57 asyncio
.get_child_watcher().attach_loop(event_loop
)
59 pytest
.skip('test_scp will always fail outside of MainThread')
60 async with base
.CleanModel() as model
:
61 app
= await model
.deploy('ubuntu')
63 await asyncio
.wait_for(
64 model
.block_until(lambda: app
.units
),
67 await asyncio
.wait_for(
68 model
.block_until(lambda: unit
.machine
),
70 machine
= unit
.machine
71 await asyncio
.wait_for(
72 model
.block_until(lambda: (machine
.status
== 'running' and
73 machine
.agent_status
== 'started')),
76 with
NamedTemporaryFile() as f
:
77 f
.write(b
'testcontents')
79 await unit
.scp_to(f
.name
, 'testfile')
81 with
NamedTemporaryFile() as f
:
82 await unit
.scp_from('testfile', f
.name
)
83 assert f
.read() == b
'testcontents'