bb349699faa796a51e5cd81b1d2d0491bd491aed
[osm/N2VC.git] / tests / integration / test_unit.py
1 import asyncio
2 from tempfile import NamedTemporaryFile
3
4 import pytest
5
6 from .. import base
7
8
9 @base.bootstrapped
10 @pytest.mark.asyncio
11 async def test_run(event_loop):
12 from juju.action import Action
13
14 async with base.CleanModel() as model:
15 app = await model.deploy(
16 'ubuntu-0',
17 application_name='ubuntu',
18 series='trusty',
19 channel='stable',
20 )
21
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
26 break
27
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'
32 break
33
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'
38 break
39
40
41 @base.bootstrapped
42 @pytest.mark.asyncio
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()
49
50 async with base.CleanModel() as model:
51 app = await model.deploy(
52 'git',
53 application_name='git',
54 series='trusty',
55 channel='stable',
56 )
57
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'
65 break
66
67
68 @base.bootstrapped
69 @pytest.mark.asyncio
70 async def test_scp(event_loop):
71 # ensure that asyncio.subprocess will work;
72 try:
73 asyncio.get_child_watcher().attach_loop(event_loop)
74 except RuntimeError:
75 pytest.skip('test_scp will always fail outside of MainThread')
76 async with base.CleanModel() as model:
77 app = await model.deploy('ubuntu')
78
79 await asyncio.wait_for(
80 model.block_until(lambda: app.units),
81 timeout=60)
82 unit = app.units[0]
83 await asyncio.wait_for(
84 model.block_until(lambda: unit.machine),
85 timeout=60)
86 machine = unit.machine
87 await asyncio.wait_for(
88 model.block_until(lambda: (machine.status == 'running' and
89 machine.agent_status == 'started')),
90 timeout=480)
91
92 with NamedTemporaryFile() as f:
93 f.write(b'testcontents')
94 f.flush()
95 await unit.scp_to(f.name, 'testfile')
96
97 with NamedTemporaryFile() as f:
98 await unit.scp_from('testfile', f.name)
99 assert f.read() == b'testcontents'