Method test_thousand

Lib/test/_test_multiprocessing.py:2491–2505

(self)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    def test_thousand(self):
        if self.TYPE == 'manager':
            self.skipTest('test not appropriate for {}'.format(self.TYPE))
        passes = 1000
        lock = self.Lock()
        conn, child_conn = self.Pipe(False)
        for j in range(self.N):
            p = self.Process(target=self._test_thousand_f,
                           args=(self.barrier, passes, child_conn, lock))
            p.start()
            self.addCleanup(p.join)

        for i in range(passes):
            for j in range(self.N):
                self.assertEqual(conn.recv(), i)

#
#