Method test_wmi_query_threads

Lib/test/test_wmi.py:83–89

(self)

            pass

    def test_wmi_query_threads(self):
        from concurrent.futures import ThreadPoolExecutor
        query = "SELECT ProcessId FROM Win32_Process WHERE ProcessId < 1000"
        with ThreadPoolExecutor(4) as pool:
            task = [pool.submit(wmi_exec_query, query) for _ in range(32)]
            for t in task:
                self.assertRegex(t.result(), "ProcessId=")