Function test_fixture_reader_thread3

tests_python/test_fixtures.py:94–111

(_dummy_socket)

def test_fixture_reader_thread3(_dummy_socket):
    sock = _DummySocket()

    reader_thread = ReaderThread(sock)
    reader_thread.start()

    msg = "aaaaaaabbbbbbbccccccc"
    http = "Content-Length: %s\r\n\r\n%s" % (len(msg), msg)
    http *= 2
    initial = http
    for i in range(1, len(http)):
        while http:
            sock.put(http[:i])
            http = http[i:]

        assert reader_thread.get_next_message("check 1: %s" % i) == msg
        assert reader_thread.get_next_message("check 2: %s" % i) == msg
        http = initial