def test_pipe(self):
        reader, writer = os.pipe()
        try:
            loop = EventLoop()
            nb_reads = [0]
            nb_writes = [0]
            
            duration = .1
            interval = .0095
            expected_nb = 10
            assert expected_nb == duration // interval
            # Avoid a race condition between the loop stop and the last read.
            assert expected_nb < duration / interval
            
            start = time.time()
            loop.add_timer(duration)(loop.stop)
            
            @loop.add_timer(interval, repeat=True)
            def write_something():
                assert os.write(writer, 'foo') == 3
                nb_writes[0] += 1

            @loop.block_reader(reader)
            def incoming(data):
                assert data == 'foo'
                nb_reads[0] += 1
            
            loop.run()
            assert round(time.time() - start, 2) == duration
            assert nb_writes[0] == expected_nb, nb_writes
            assert nb_reads[0] == expected_nb, nb_reads
        finally:
            os.close(reader)
            os.close(writer)
dataset/ETHPy150Open SimonSapin/snippets/event_loop/tests.py/TestEventLoop.test_pipe/original