(self)
| 108 | ] |
| 109 | |
| 110 | def test_read(self): |
| 111 | stream = JsonMemoryStream(self.MESSAGES, []) |
| 112 | for expected_message in self.MESSAGES: |
| 113 | message = stream.read_json() |
| 114 | assert message == expected_message |
| 115 | with pytest.raises(messaging.NoMoreMessages) as exc_info: |
| 116 | stream.read_json() |
| 117 | assert exc_info.value.stream is stream |
| 118 | |
| 119 | def test_write(self): |
| 120 | messages = [] |
nothing calls this directly
no test coverage detected