← Back to branch summary

~ubuntu-branches/ubuntu/precise/landscape-client/precise

« back to all changes in this revision

Viewing changes to landscape/lib/tests/test_twisted_util.py

  • Committer: Package Import Robot
  • Author(s): Andreas Hasenack
  • Date: 2012-03-19 09:33:34 UTC
  • mto: This revision was merged to the branch mainline in revision 41.
  • Revision ID: package-import@ubuntu.com-20120319093334-oxjttz163vvfgq8s
Tags: upstream-12.04
Import upstream version 12.04

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/lib/tests/test_twisted_util.py
 
1
import os
 
2
 
 
3
from landscape.tests.helpers import LandscapeTest
 
4
from landscape.lib.twisted_util import spawn_process
 
5
from landscape.lib.fs import create_file
 
6
 
 
7
 
 
8
class SpawnProcessTest(LandscapeTest):
 
9
 
 
10
    def setUp(self):
 
11
        super(SpawnProcessTest, self).setUp()
 
12
        self.command = self.makeFile("#!/bin/sh\necho -n $@")
 
13
        os.chmod(self.command, 0755)
 
14
 
 
15
    def test_spawn_process_return_value(self):
 
16
        """
 
17
        The process is executed and returns the expected exit code.
 
18
        """
 
19
        create_file(self.command, "#!/bin/sh\nexit 2")
 
20
 
 
21
        def callback((out, err, code)):
 
22
            self.assertEqual(out, "")
 
23
            self.assertEqual(err, "")
 
24
            self.assertEqual(code, 2)
 
25
 
 
26
        result = spawn_process(self.command)
 
27
        result.addCallback(callback)
 
28
        return result
 
29
 
 
30
    def test_spawn_process_output(self):
 
31
        """
 
32
        The process returns the expected standard output.
 
33
        """
 
34
        def callback((out, err, code)):
 
35
            self.assertEqual(out, "a b")
 
36
            self.assertEqual(err, "")
 
37
            self.assertEqual(code, 0)
 
38
 
 
39
        result = spawn_process(self.command, args=("a", "b"))
 
40
        result.addCallback(callback)
 
41
        return result
 
42
 
 
43
    def test_spawn_process_error(self):
 
44
        """
 
45
        The process returns the expected standard error.
 
46
        """
 
47
        create_file(self.command, "#!/bin/sh\necho -n $@ >&2")
 
48
 
 
49
        def callback((out, err, code)):
 
50
            self.assertEqual(out, "")
 
51
            self.assertEqual(err, "a b")
 
52
            self.assertEqual(code, 0)
 
53
 
 
54
        result = spawn_process(self.command, args=("a", "b"))
 
55
        result.addCallback(callback)
 
56
        return result
 
57
 
 
58
    def test_spawn_process_callback(self):
 
59
        """
 
60
        If a callback for process output is provieded, it is called for every
 
61
        line of output.
 
62
        """
 
63
        create_file(self.command, "#!/bin/sh\n/bin/echo -ne $@")
 
64
        param = r"some text\nanother line\nok, last one\n"
 
65
        expected = ["some text", "another line", "ok, last one"]
 
66
        lines = []
 
67
 
 
68
        def line_received(line):
 
69
            lines.append(line)
 
70
 
 
71
        def callback((out, err, code)):
 
72
            self.assertEqual(expected, lines)
 
73
 
 
74
        result = spawn_process(self.command, args=(param,),
 
75
                               line_received=line_received)
 
76
        result.addCallback(callback)
 
77
        return result
 
78
 
 
79
    def test_spawn_process_callback_multiple_newlines(self):
 
80
        """
 
81
        If output ends with more than one newline, empty lines are preserved.
 
82
        """
 
83
        create_file(self.command, "#!/bin/sh\n/bin/echo -ne $@")
 
84
        param = r"some text\nanother line\n\n\n"
 
85
        expected = ["some text", "another line", "", ""]
 
86
        lines = []
 
87
 
 
88
        def line_received(line):
 
89
            lines.append(line)
 
90
 
 
91
        def callback((out, err, code)):
 
92
            self.assertEqual(expected, lines)
 
93
 
 
94
        result = spawn_process(self.command, args=(param,),
 
95
                               line_received=line_received)
 
96
        result.addCallback(callback)
 
97
        return result
 
98
 
 
99
    def test_spawn_process_callback_no_newline(self):
 
100
        """
 
101
        If output ends without a newline, the line is still passed to the
 
102
        callback.
 
103
        """
 
104
        create_file(self.command, "#!/bin/sh\n/bin/echo -ne $@")
 
105
        param = r"some text\nanother line\nok, last one"
 
106
        expected = ["some text", "another line", "ok, last one"]
 
107
        lines = []
 
108
 
 
109
        def line_received(line):
 
110
            lines.append(line)
 
111
 
 
112
        def callback((out, err, code)):
 
113
            self.assertEqual(expected, lines)
 
114
 
 
115
        result = spawn_process(self.command, args=(param,),
 
116
                               line_received=line_received)
 
117
        result.addCallback(callback)
 
118
        return result
 
119
 
 
120
    def test_spawn_process_with_stdin(self):
 
121
        """
 
122
        Optionally C{spawn_process} accepts a C{stdin} argument.
 
123
        """
 
124
        create_file(self.command, "#!/bin/sh\n/bin/cat")
 
125
 
 
126
        def callback((out, err, code)):
 
127
            self.assertEqual("hello", out)
 
128
 
 
129
        result = spawn_process(self.command, stdin="hello")
 
130
        result.addCallback(callback)
 
131
        return result