1604c311cb5f403b61e61cf9d94cb4a109f34ed9
4 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 async with base
.CleanModel() as model
:
56 app
= await model
.deploy('ubuntu')
58 await asyncio
.wait_for(
59 model
.block_until(lambda: app
.units
),
62 await asyncio
.wait_for(
63 model
.block_until(lambda: unit
.machine
),
65 machine
= unit
.machine
66 await asyncio
.wait_for(
67 model
.block_until(lambda: (machine
.status
== 'running' and
68 machine
.agent_status
== 'started')),
71 with
NamedTemporaryFile() as f
:
72 f
.write(b
'testcontents')
74 await unit
.scp_to(f
.name
, 'testfile')
76 with
NamedTemporaryFile() as f
:
77 await unit
.scp_from('testfile', f
.name
)
78 assert f
.read() == b
'testcontents'