| Total Complexity | 6 |
| Total Lines | 30 |
| Duplicated Lines | 0 % |
| Changes | 1 | ||
| Bugs | 0 | Features | 0 |
| 1 | import unittest |
||
| 7 | class TestKafkaPython(unittest.TestCase): |
||
| 8 | def setUp(self): |
||
| 9 | self.host = "myhost" |
||
| 10 | self.port = 1234 |
||
| 11 | self.group = "mygroup" |
||
| 12 | self.topic = "mytopic" |
||
| 13 | self.reconnect_wait_time = 0.01 |
||
| 14 | self.reader = self.create_reader() |
||
| 15 | |||
| 16 | def create_reader(self): |
||
| 17 | reader = kafka_python.Reader(self.host, |
||
| 18 | self.port, |
||
| 19 | self.group, |
||
| 20 | self.topic) |
||
| 21 | reader.consumer = mock.MagicMock() |
||
| 22 | return reader |
||
| 23 | |||
| 24 | def sample_messages(self, payload, count): |
||
| 25 | return count * [Message(0, 0, None, payload)], count * [payload] |
||
| 26 | |||
| 27 | def test_handle_read(self): |
||
| 28 | sample_messages, extracted_messages = self.sample_messages("hello", 3) |
||
| 29 | self.reader.consumer.__iter__.return_value = sample_messages |
||
| 30 | self.reader._connect = mock.MagicMock() |
||
| 31 | received_messages = list(self.reader._handle_read()) |
||
| 32 | self.assertEqual(received_messages, extracted_messages) |
||
| 33 | |||
| 34 | def receive_messages(self): |
||
| 35 | for message in self.reader.read(): |
||
| 36 | yield message |
||
| 37 |