(self)
| 93 | assert exc_info.value.stream is stream |
| 94 | |
| 95 | def test_write(self): |
| 96 | data = io.BytesIO() |
| 97 | stream = messaging.JsonIOStream(data, data, "data") |
| 98 | for message in self.MESSAGES: |
| 99 | stream.write_json(message) |
| 100 | data = data.getvalue() |
| 101 | assert data == self.SERIALIZED_MESSAGES |
| 102 | |
| 103 | |
| 104 | class TestJsonMemoryStream(object): |
nothing calls this directly
no test coverage detected