iostream_test.py 文件源码

def test_read_until_regex_max_bytes_inline(self):
        server, client = self.make_iostream_pair()
        client.set_close_callback(lambda: self.stop("closed"))
        try:
            # Similar to the error case in the previous test, but the
            # server writes first so client reads are satisfied
            # inline.  For consistency with the out-of-line case, we
            # do not raise the error synchronously.
            server.write(b"123456")
            with ExpectLog(gen_log, "Unsatisfiable read"):
                client.read_until_regex(b"def", self.stop, max_bytes=5)
                data = self.wait()
            self.assertEqual(data, "closed")
        finally:
            server.close()
            client.close()