← 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: Aaron Bentley
  • Date: 2009-11-03 15:45:56 UTC
  • mto: (4634.97.2 2.0)
  • mto: This revision was merged to the branch mainline in revision 4798.
  • Revision ID: aaron@aaronbentley.com-20091103154556-e953dmegqbinyokq
Improve patch binary section handling.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_breakin.py
41
41
 
42
42
    def setUp(self):
43
43
        super(TestBreakin, self).setUp()
44
 
        self.requireFeature(tests.BreakinFeature)
 
44
        if breakin.determine_signal() is None:
 
45
            raise tests.TestSkipped('this platform is missing SIGQUIT'
 
46
                                    ' or SIGBREAK')
45
47
        if sys.platform == 'win32':
 
48
            # Windows doesn't have os.kill, and we catch the SIGBREAK signal.
 
49
            # We trigger SIGBREAK via a Console api so we need ctypes to access
 
50
            # the function
 
51
            if not have_ctypes:
 
52
                raise tests.UnavailableFeature('ctypes')
46
53
            self._send_signal = self._send_signal_win32
47
54
        else:
48
55
            self._send_signal = self._send_signal_via_kill
54
61
            sig_num = signal.SIGKILL
55
62
        else:
56
63
            raise ValueError("unknown signal type: %s" % (sig_type,))
57
 
        try:
58
 
            os.kill(pid, sig_num)
59
 
        except OSError, e:
60
 
            if e.errno != errno.ESRCH:
61
 
                raise
 
64
        os.kill(pid, sig_num)
62
65
 
63
66
    def _send_signal_win32(self, pid, sig_type):
64
67
        """Send a 'signal' on Windows.
107
110
            raise tests.TestNotApplicable(
108
111
                '%s raises a popup on OSX' % self.id())
109
112
 
110
 
    def _wait_for_process(self, pid, sig=None, count=100):
 
113
    def _wait_for_process(self, pid, sig=None):
111
114
        # We don't know quite how long waiting for the process 'pid' will take,
112
115
        # but if it's more than 10s then it's probably not going to work.
113
 
        for i in range(count):
 
116
        for i in range(100):
 
117
            time.sleep(0.1)
114
118
            if sig is not None:
115
119
                self._send_signal(pid, sig)
116
120
            # Use WNOHANG to ensure we don't get blocked, doing so, we may
123
127
                #       instead. Most notably, the WNOHANG isn't allowed, so
124
128
                #       this can hang indefinitely.
125
129
                pid_killed, returncode = os.waitpid(pid, opts)
126
 
                if pid_killed != 0 and returncode != 0:
 
130
                if (pid_killed, returncode) != (0, 0):
127
131
                    if sig is not None:
128
132
                        # high bit in low byte says if core was dumped; we
129
133
                        # don't care
135
139
                    return True, None
136
140
                else:
137
141
                    raise
138
 
            if i + 1 != count:
139
 
                time.sleep(0.1)
140
142
 
141
143
        return False, None
142
144
 
159
161
        # os.read(proc.stderr.fileno())?
160
162
        err = proc.stderr.readline()
161
163
        self.assertContainsRe(err, r'entering debugger')
162
 
        # Try to shutdown cleanly;
163
164
        # Now that the debugger is entered, we can ask him to quit
164
165
        proc.stdin.write("q\n")
165
 
        # But we don't really care if it doesn't.
166
 
        dead, sig = self._wait_for_process(proc.pid, count=3)
 
166
        # We wait a bit to let the child process handles our query and avoid
 
167
        # triggering deadlocks leading to hangs on multi-core hosts...
 
168
        dead, sig = self._wait_for_process(proc.pid)
167
169
        if not dead:
168
 
            # The process didn't finish, let's kill it.
169
 
            dead, sig = self._wait_for_process(proc.pid, 'kill', count=10)
170
 
            if not dead:
171
 
                # process isn't gone, user will have to hunt it down and kill
172
 
                # it.
173
 
                self.fail("subprocess %d wasn't terminated by repeated SIGKILL" %
 
170
            # The process didn't finish, let's kill it before reporting failure
 
171
            dead, sig = self._wait_for_process(proc.pid, 'kill')
 
172
            if dead:
 
173
                raise tests.KnownFailure(
 
174
                    "subprocess wasn't terminated, it had to be killed")
 
175
            else:
 
176
                self.fail("subprocess %d wasn't terminated by repeated SIGKILL",
174
177
                          proc.pid)
175
178
 
176
179
    def test_breakin_harder(self):