← Back to branch summary

~ubuntu-branches/ubuntu/lucid/enigmail/lucid-security

« back to all changes in this revision

Viewing changes to testing/mozbase/mozprocess/tests/mozprocess1.py

  • Committer: Package Import Robot
  • Author(s): Chris Coulson
  • Date: 2012-11-12 16:36:01 UTC
  • mfrom: (0.11.6)
  • Revision ID: package-import@ubuntu.com-20121112163601-8c5yisdr8sxei4rv
Tags: 2:1.4.6-0ubuntu0.10.04.1
https://launchpad.net/bugs/1080212
* New upstream release v1.4.6
  - see LP: #1080212 for USN information
* Drop unneeded patches
  - remove debian/patches/correct-version-number.diff
  - remove debian/patches/dont_register_cids_multiple_times.diff
  - update debian/patches/series
* Support building in an objdir
  - update debian/rules

Show diffs side-by-side

added added

removed removed

Lines of Context:
testing/mozbase/mozprocess/tests/mozprocess1.py
 
1
#!/usr/bin/env python
 
2
 
 
3
# This Source Code Form is subject to the terms of the Mozilla Public
 
4
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
 
5
# You can obtain one at http://mozilla.org/MPL/2.0/.
 
6
 
 
7
import os
 
8
import subprocess
 
9
import sys
 
10
import unittest
 
11
from time import sleep
 
12
 
 
13
from mozprocess import processhandler
 
14
 
 
15
here = os.path.dirname(os.path.abspath(__file__))
 
16
 
 
17
def make_proclaunch(aDir):
 
18
    """
 
19
        Makes the proclaunch executable.
 
20
        Params:
 
21
            aDir - the directory in which to issue the make commands
 
22
        Returns:
 
23
            the path to the proclaunch executable that is generated
 
24
    """
 
25
    # Ideally make should take care of this, but since it doesn't - on windows,
 
26
    # anyway, let's just call out both targets explicitly.
 
27
    p = subprocess.call(["make", "-C", "iniparser"], cwd=aDir)
 
28
    p = subprocess.call(["make"], cwd=aDir)
 
29
    if sys.platform == "win32":
 
30
        exepath = os.path.join(aDir, "proclaunch.exe")
 
31
    else:
 
32
        exepath = os.path.join(aDir, "proclaunch")
 
33
    return exepath
 
34
 
 
35
def check_for_process(processName):
 
36
    """
 
37
        Use to determine if process of the given name is still running.
 
38
 
 
39
        Returns:
 
40
        detected -- True if process is detected to exist, False otherwise
 
41
        output -- if process exists, stdout of the process, '' otherwise
 
42
    """
 
43
    output = ''
 
44
    if sys.platform == "win32":
 
45
        # On windows we use tasklist
 
46
        p1 = subprocess.Popen(["tasklist"], stdout=subprocess.PIPE)
 
47
        output = p1.communicate()[0]
 
48
        detected = False
 
49
        for line in output.splitlines():
 
50
            if processName in line:
 
51
                detected = True
 
52
                break
 
53
    else:
 
54
        p1 = subprocess.Popen(["ps", "-ef"], stdout=subprocess.PIPE)
 
55
        p2 = subprocess.Popen(["grep", processName], stdin=p1.stdout, stdout=subprocess.PIPE)
 
56
        p1.stdout.close()
 
57
        output = p2.communicate()[0]
 
58
        detected = False
 
59
        for line in output.splitlines():
 
60
            if "grep %s" % processName in line:
 
61
                continue
 
62
            elif processName in line and not 'defunct' in line: 
 
63
                detected = True
 
64
                break
 
65
 
 
66
    return detected, output
 
67
 
 
68
 
 
69
class ProcTest1(unittest.TestCase):
 
70
 
 
71
    def __init__(self, *args, **kwargs):
 
72
 
 
73
        # Ideally, I'd use setUpClass but that only exists in 2.7.
 
74
        # So, we'll do this make step now.
 
75
        self.proclaunch = make_proclaunch(here)
 
76
        unittest.TestCase.__init__(self, *args, **kwargs)
 
77
 
 
78
    def test_process_normal_finish(self):
 
79
        """Process is started, runs to completion while we wait for it"""
 
80
 
 
81
        p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"],
 
82
                                          cwd=here)
 
83
        p.run()
 
84
        p.processOutput()
 
85
        p.waitForFinish()
 
86
 
 
87
        detected, output = check_for_process(self.proclaunch)
 
88
        self.determine_status(detected,
 
89
                              output,
 
90
                              p.proc.returncode,
 
91
                              p.didTimeout)
 
92
 
 
93
    def test_process_waittimeout(self):
 
94
        """ Process is started, runs but we time out waiting on it
 
95
            to complete
 
96
        """
 
97
        p = processhandler.ProcessHandler([self.proclaunch, "process_waittimeout.ini"],
 
98
                                          cwd=here)
 
99
        p.run()
 
100
        p.processOutput(timeout=10) 
 
101
        p.waitForFinish()
 
102
 
 
103
        detected, output = check_for_process(self.proclaunch)
 
104
        self.determine_status(detected,
 
105
                              output,
 
106
                              p.proc.returncode,
 
107
                              p.didTimeout,
 
108
                              False,
 
109
                              ['returncode', 'didtimeout'])
 
110
 
 
111
    def test_process_kill(self):
 
112
        """ Process is started, we kill it
 
113
        """
 
114
        p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"],
 
115
                                          cwd=here)
 
116
        p.run()
 
117
        p.processOutput()
 
118
        p.kill()
 
119
 
 
120
        detected, output = check_for_process(self.proclaunch)
 
121
        self.determine_status(detected,
 
122
                              output,
 
123
                              p.proc.returncode,
 
124
                              p.didTimeout)
 
125
 
 
126
    def determine_status(self,
 
127
                         detected=False,
 
128
                         output='',
 
129
                         returncode=0,
 
130
                         didtimeout=False,
 
131
                         isalive=False,
 
132
                         expectedfail=[]):
 
133
        """
 
134
        Use to determine if the situation has failed.
 
135
        Parameters:
 
136
            detected -- value from check_for_process to determine if the process is detected
 
137
            output -- string of data from detected process, can be ''
 
138
            returncode -- return code from process, defaults to 0
 
139
            didtimeout -- True if process timed out, defaults to False
 
140
            isalive -- Use True to indicate we pass if the process exists; however, by default
 
141
                       the test will pass if the process does not exist (isalive == False)
 
142
            expectedfail -- Defaults to [], used to indicate a list of fields that are expected to fail
 
143
        """
 
144
        if 'returncode' in expectedfail:
 
145
            self.assertTrue(returncode, "Detected an unexpected return code of: %s" % returncode)
 
146
        elif not isalive:
 
147
            self.assertTrue(returncode == 0, "Detected non-zero return code of: %d" % returncode)
 
148
 
 
149
        if 'didtimeout' in expectedfail:
 
150
            self.assertTrue(didtimeout, "Detected that process didn't time out")
 
151
        else:
 
152
            self.assertTrue(not didtimeout, "Detected that process timed out")
 
153
 
 
154
        if isalive:
 
155
            self.assertTrue(detected, "Detected process is not running, process output: %s" % output)
 
156
        else:
 
157
            self.assertTrue(not detected, "Detected process is still running, process output: %s" % output)
 
158
 
 
159
if __name__ == '__main__':
 
160
    unittest.main()