|
3 | 3 |
|
4 | 4 | import time |
5 | 5 | import unittest |
6 | | -from test.support import import_helper, requires_resource, LOOPBACK_TIMEOUT |
| 6 | +from test import support |
| 7 | +from test.support import import_helper |
7 | 8 |
|
8 | 9 |
|
9 | 10 | # Do this first so test will be skipped if module doesn't exist |
|
12 | 13 |
|
13 | 14 | def wmi_exec_query(query): |
14 | 15 | # gh-112278: WMI maybe slow response when first call. |
15 | | - try: |
16 | | - return _wmi.exec_query(query) |
17 | | - except BrokenPipeError: |
18 | | - pass |
19 | | - except WindowsError as e: |
20 | | - if e.winerror != 258: |
21 | | - raise |
22 | | - time.sleep(LOOPBACK_TIMEOUT) |
23 | | - return _wmi.exec_query(query) |
| 16 | + for _ in support.sleeping_retry(support.LONG_TIMEOUT): |
| 17 | + try: |
| 18 | + return _wmi.exec_query(query) |
| 19 | + except BrokenPipeError: |
| 20 | + pass |
| 21 | + # retry on pipe error |
| 22 | + except WindowsError as exc: |
| 23 | + if exc.winerror != 258: |
| 24 | + raise |
| 25 | + # retry on timeout |
24 | 26 |
|
25 | 27 |
|
26 | 28 | class WmiTests(unittest.TestCase): |
@@ -58,7 +60,7 @@ def test_wmi_query_not_select(self): |
58 | 60 | with self.assertRaises(ValueError): |
59 | 61 | wmi_exec_query("not select, just in case someone tries something") |
60 | 62 |
|
61 | | - @requires_resource('cpu') |
| 63 | + @support.requires_resource('cpu') |
62 | 64 | def test_wmi_query_overflow(self): |
63 | 65 | # Ensure very big queries fail |
64 | 66 | # Test multiple times to ensure consistency |
|
0 commit comments