← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/blackbox/test_breakin.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-03-06 06:48:25 UTC
  • mfrom: (4070.8.6 debug-config)
  • Revision ID: pqm@pqm.ubuntu.com-20090306064825-kbpwggw21dygeix6
(mbp) debug_flags configuration option

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_breakin.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Blackbox tests for debugger breakin"""
18
18
 
19
 
import errno
20
19
import os
21
20
import signal
22
21
import subprocess
23
22
import sys
24
23
import time
25
24
 
26
 
from bzrlib import (
27
 
    errors,
28
 
    tests,
29
 
    )
 
25
from bzrlib import tests
30
26
 
31
27
 
32
28
class TestBreakin(tests.TestCase):
48
44
            raise tests.TestNotApplicable(
49
45
                '%s raises a popup on OSX' % self.id())
50
46
 
51
 
    def _wait_for_process(self, pid, sig=None):
52
 
        # We don't know quite how long waiting for the process 'pid' will take,
53
 
        # but if it's more than 10s then it's probably not going to work.
54
 
        for i in range(100):
55
 
            time.sleep(0.1)
56
 
            if sig is not None:
57
 
                os.kill(pid, sig)
58
 
            # Use WNOHANG to ensure we don't get blocked, doing so, we may
59
 
            # leave the process continue after *we* die...
60
 
            try:
61
 
                # note: waitpid is different on win32, but this test only runs
62
 
                # on unix
63
 
                pid_killed, returncode = os.waitpid(pid, os.WNOHANG)
64
 
                if (pid_killed, returncode) != (0, 0):
65
 
                    if sig is not None:
66
 
                        # high bit in low byte says if core was dumped; we
67
 
                        # don't care
68
 
                        status, sig = (returncode >> 8, returncode & 0x7f)
69
 
                        return True, sig
70
 
            except OSError, e:
71
 
                if e.errno in (errno.ECHILD, errno.ESRCH):
72
 
                    # The process doesn't exist anymore
73
 
                    return True, None
74
 
                else:
75
 
                    raise
76
 
 
77
 
        return False, None
78
 
 
79
47
    # port 0 means to allocate any port
80
48
    _test_process_args = ['serve', '--port', 'localhost:0']
81
49
 
89
57
        proc.stderr.readline()
90
58
        # first sigquit pops into debugger
91
59
        os.kill(proc.pid, signal.SIGQUIT)
92
 
        # Wait for the debugger to acknowledge the signal reception
93
 
        err = proc.stderr.readline()
94
 
        self.assertContainsRe(err, r'entering debugger')
95
 
        # Now that the debugger is entered, we can ask him to quit
96
60
        proc.stdin.write("q\n")
97
 
        # We wait a bit to let the child process handles our query and avoid
98
 
        # triggering deadlocks leading to hangs on multi-core hosts...
99
 
        dead, sig = self._wait_for_process(proc.pid)
100
 
        if not dead:
101
 
            # The process didn't finish, let's kill it before reporting failure
102
 
            dead, sig = self._wait_for_process(proc.pid, signal.SIGKILL)
103
 
            if dead:
104
 
                raise tests.KnownFailure(
105
 
                    "subprocess wasn't terminated, it had to be killed")
106
 
            else:
107
 
                self.fail("subprocess %d wasn't terminated by repeated SIGKILL",
108
 
                          proc.pid)
 
61
        time.sleep(.5)
 
62
        err = proc.stderr.readline()
 
63
        self.assertContainsRe(err, r'entering debugger')
109
64
 
110
65
    def test_breakin_harder(self):
111
 
        """SIGQUITting twice ends the process."""
112
66
        self._dont_SIGQUIT_on_darwin()
113
67
        proc = self.start_bzr_subprocess(self._test_process_args,
114
68
                env_changes=dict(BZR_SIGQUIT_PDB=None))
116
70
        proc.stderr.readline()
117
71
        # break into the debugger
118
72
        os.kill(proc.pid, signal.SIGQUIT)
119
 
        # Wait for the debugger to acknowledge the signal reception (since we
120
 
        # want to send a second signal, we ensure it doesn't get lost by
121
 
        # validating the first get received and produce its effect).
122
 
        err = proc.stderr.readline()
123
 
        self.assertContainsRe(err, r'entering debugger')
124
 
        dead, sig = self._wait_for_process(proc.pid, signal.SIGQUIT)
125
 
        self.assertTrue((dead and sig == signal.SIGQUIT),
126
 
                        msg="subprocess wasn't terminated by repeated SIGQUIT")
 
73
        # now send a second sigquit, which should cause it to exit.  That
 
74
        # won't happen until the original signal has been noticed by the
 
75
        # child and it's run its signal handler.  We don't know quite how long
 
76
        # this will take, but if it's more than 10s then it's probably not
 
77
        # going to work.
 
78
        for i in range(100):
 
79
            time.sleep(0.1)
 
80
            os.kill(proc.pid, signal.SIGQUIT)
 
81
            # note: waitpid is different on win32, but this test only runs on
 
82
            # unix
 
83
            r = os.waitpid(proc.pid, os.WNOHANG)
 
84
            if r != (0, 0):
 
85
                # high bit says if core was dumped; we don't care
 
86
                self.assertEquals(r[1] & 0x7f, signal.SIGQUIT)
 
87
                break
 
88
        else:
 
89
            self.fail("subprocess wasn't terminated by repeated SIGQUIT")
127
90
 
128
91
    def test_breakin_disabled(self):
129
92
        self._dont_SIGQUIT_on_darwin()