blob: 8cb31fd7cfbc5f33c5a2e6de4d6216afe617e5da (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
"""Tests for handling of signals within cloud init."""
import subprocess
import sys
import time
from StringIO import StringIO
from mocker import MockerTestCase
class TestSignal(MockerTestCase):
def test_signal_output(self):
# This is done since nose/unittest is actually setting up
# output capturing, signal handling itself, and its easier
# to just call out to cloudinit with a loop and see what the result is
run_what = [sys.executable,
'-c',
('import time; from cloudinit import signal_handler;'
'signal_handler.attach_handlers(); time.sleep(120)')]
pc_info = subprocess.Popen(run_what,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE)
# Let it start up
time.sleep(0.5)
dead = None
while dead is None:
pc_info.terminate()
# Ok not dead yet. try again
time.sleep(0.5)
dead = pc_info.poll()
outputs = StringIO()
if pc_info.stdout:
outputs.write(pc_info.stdout.read())
if pc_info.stderr:
outputs.write(pc_info.stderr.read())
val = outputs.getvalue()
# Check some of the outputs that should of happened
self.assertEquals(1, pc_info.wait())
self.assertTrue(len(val) != 0)
self.assertTrue(val.find("terminated") != -1)
|
