Test poll timeout behavior.
(self)
| 82 | server.close() |
| 83 | |
| 84 | def test_poll_timeout(self): |
| 85 | """Test poll timeout behavior.""" |
| 86 | server = ZeroMqQueue( |
| 87 | address=None, |
| 88 | socket_type=zmq.PAIR, |
| 89 | is_server=True, |
| 90 | is_async=False, |
| 91 | name="test_poll_server", |
| 92 | use_hmac_encryption=False, |
| 93 | ) |
| 94 | |
| 95 | try: |
| 96 | # Poll should timeout when no data available |
| 97 | start = time.time() |
| 98 | result = server.poll(timeout=1) |
| 99 | elapsed = time.time() - start |
| 100 | assert result is False |
| 101 | assert elapsed >= 1.0 |
| 102 | assert elapsed < 1.5 # Allow some margin |
| 103 | finally: |
| 104 | server.close() |
| 105 | |
| 106 | def test_poll_with_data(self): |
| 107 | """Test poll returns True when data is available.""" |
nothing calls this directly
no test coverage detected