(_dummy_socket)
| 53 | |
| 54 | |
| 55 | def test_fixture_reader_thread1(_dummy_socket): |
| 56 | sock = _dummy_socket |
| 57 | |
| 58 | reader_thread = ReaderThread(sock) |
| 59 | reader_thread.start() |
| 60 | |
| 61 | json_part = json.dumps({"key": "val"}) |
| 62 | json_part = json_part.replace(":", ":\n") |
| 63 | msg = json_part |
| 64 | |
| 65 | msg = ("Content-Length: %s\r\n\r\n%s" % (len(msg), msg)).encode("utf-8") |
| 66 | # Check that receiving 2 messages at a time we're able to properly deal |
| 67 | # with each one. |
| 68 | sock.put(msg + msg) |
| 69 | |
| 70 | assert reader_thread.get_next_message("check 1") == json_part |
| 71 | assert reader_thread.get_next_message("check 2") == json_part |
| 72 | |
| 73 | |
| 74 | def test_fixture_reader_thread2(_dummy_socket): |
nothing calls this directly
no test coverage detected