Test the specified socket method. The method is run at most `count` times and must raise a TimeoutError within `timeout` + self.fuzz seconds.
(self, count, timeout, method, *args)
| 112 | raise NotImplementedError() |
| 113 | |
| 114 | def _sock_operation(self, count, timeout, method, *args): |
| 115 | """ |
| 116 | Test the specified socket method. |
| 117 | |
| 118 | The method is run at most `count` times and must raise a TimeoutError |
| 119 | within `timeout` + self.fuzz seconds. |
| 120 | """ |
| 121 | self.sock.settimeout(timeout) |
| 122 | method = getattr(self.sock, method) |
| 123 | for i in range(count): |
| 124 | t1 = time.monotonic() |
| 125 | try: |
| 126 | method(*args) |
| 127 | except TimeoutError as e: |
| 128 | delta = time.monotonic() - t1 |
| 129 | break |
| 130 | else: |
| 131 | self.fail('TimeoutError was not raised') |
| 132 | # These checks should account for timing unprecision |
| 133 | self.assertLess(delta, timeout + self.fuzz) |
| 134 | self.assertGreater(delta, timeout - 1.0) |
| 135 | |
| 136 | |
| 137 | class TCPTimeoutTestCase(TimeoutTestCase): |
no test coverage detected