(self, expected, admin_client)
| 79 | assert self.thread_count == threading.active_count() |
| 80 | |
| 81 | def _expect(self, expected, admin_client): |
| 82 | events = {} |
| 83 | while expected: |
| 84 | data = admin_client.receive(timeout=5) |
| 85 | if data[0] in expected: |
| 86 | if expected[data[0]] == 1: |
| 87 | events[data[0]] = data[1] |
| 88 | del expected[data[0]] |
| 89 | else: |
| 90 | expected[data[0]] -= 1 |
| 91 | return events |
| 92 | |
| 93 | def test_missing_auth(self): |
| 94 | sio = socketio.Server(async_mode='threading') |
no test coverage detected