Method _test_wait_close

src/tests/observable_test.py:483–492

(self, source, observable, postclose_callback=None)

        self.assertTrue(observer2.closed)

    def _test_wait_close(self, source, observable, postclose_callback=None):
        thread = threading.Thread(target=self.close_delayed, args=[source], daemon=True)
        thread.start()

        observable.wait_close(timeout=0.2)

        if postclose_callback is not None:
            postclose_callback()

        self.assertTrue(observable.closed)

    def test_wait_close(self):
        observable = self.create_observable()