Set up test fixtures
(self)
| 32 | """Test the core RequestBatcher functionality""" |
| 33 | |
| 34 | def setUp(self): |
| 35 | """Set up test fixtures""" |
| 36 | self.batcher = RequestBatcher(max_batch_size=4, max_wait_ms=100) |
| 37 | self.test_responses = [] |
| 38 | |
| 39 | def mock_processor(requests): |
| 40 | """Mock batch processor that returns simple responses""" |
| 41 | responses = [] |
| 42 | for i, req in enumerate(requests): |
| 43 | responses.append({ |
| 44 | "id": f"test-{i}", |
| 45 | "object": "chat.completion", |
| 46 | "choices": [{ |
| 47 | "index": 0, |
| 48 | "message": { |
| 49 | "role": "assistant", |
| 50 | "content": f"Response to request {i}" |
| 51 | }, |
| 52 | "finish_reason": "stop" |
| 53 | }], |
| 54 | "usage": {"completion_tokens": 10, "total_tokens": 20} |
| 55 | }) |
| 56 | return responses |
| 57 | |
| 58 | self.batcher.set_processor(mock_processor) |
| 59 | |
| 60 | def tearDown(self): |
| 61 | """Clean up after tests""" |
nothing calls this directly
no test coverage detected