move GenericWatcherTests into the other unix tests

python · gvanrossum · Oct 8, 2022 · Jun 23, 2022 · Jun 23, 2022 · Jun 23, 2022
commit 9607d8e09870a4e5731a4505c94b2b755fb9a7f9
@@ -702,75 +702,73 @@ class SubprocessPidfdWatcherTests(SubprocessWatcherMixin,
                                       test_utils.TestCase):
         Watcher = unix_events.PidfdChildWatcher
 
-else:
-    # Windows
-    class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase):
 
-        def setUp(self):
-            super().setUp()
-            self.loop = asyncio.ProactorEventLoop()
-            self.set_event_loop(self.loop)
-
-
-class GenericWatcherTests(test_utils.TestCase):
-
-    def test_create_subprocess_fails_with_inactive_watcher(self):
-        watcher = mock.create_autospec(
-            asyncio.AbstractChildWatcher,
-            **{"__enter__.return_value.is_active.return_value": False}
-        )
+    class GenericWatcherTests(test_utils.TestCase):
 
-        async def execute():
-            asyncio.set_child_watcher(watcher)
+        def test_create_subprocess_fails_with_inactive_watcher(self):
+            watcher = mock.create_autospec(
+                asyncio.AbstractChildWatcher,
+                **{"__enter__.return_value.is_active.return_value": False}
+            )
 
-            with self.assertRaises(RuntimeError):
-                await subprocess.create_subprocess_exec(
-                    os_helper.FakePath(sys.executable), '-c', 'pass')
+            async def execute():
+                asyncio.set_child_watcher(watcher)
 
-            watcher.add_child_handler.assert_not_called()
+                with self.assertRaises(RuntimeError):
+                    await subprocess.create_subprocess_exec(
+                        os_helper.FakePath(sys.executable), '-c', 'pass')
 
-        self.assertIsNone(asyncio.run(execute()))
-        self.assertListEqual(watcher.mock_calls, [
-            mock.call.__enter__(),
-            mock.call.__enter__().is_active(),
-            mock.call.__exit__(RuntimeError, mock.ANY, mock.ANY),
-        ])
+                watcher.add_child_handler.assert_not_called()
 
+            self.assertIsNone(asyncio.run(execute()))
+            self.assertListEqual(watcher.mock_calls, [
+                mock.call.__enter__(),
+                mock.call.__enter__().is_active(),
+                mock.call.__exit__(RuntimeError, mock.ANY, mock.ANY),
+            ])
 
-    def has_pidfd_support():
-        if not hasattr(os, 'pidfd_open'):
-            return False
-        try:
-            os.close(os.pidfd_open(os.getpid()))
-        except OSError:
-            return False
-        return True
 
-    @unittest.skipUnless(
-        has_pidfd_support(),
-        "operating system does not support pidfds",
-    )
-    def test_create_subprocess_with_pidfd(self):
-        async def in_thread():
-            proc = await asyncio.create_subprocess_exec(
-                *PROGRAM_CAT,
-                stdin=subprocess.PIPE,
-                stdout=subprocess.PIPE,
-            )
-            stdout, stderr = await proc.communicate(b"some data")
-            return proc.returncode, stdout
+        def has_pidfd_support():
+            if not hasattr(os, 'pidfd_open'):
+                return False
+            try:
+                os.close(os.pidfd_open(os.getpid()))
+            except OSError:
+                return False
+            return True
+
+        @unittest.skipUnless(
+            has_pidfd_support(),
+            "operating system does not support pidfds",
+        )
+        def test_create_subprocess_with_pidfd(self):
+            async def in_thread():
+                proc = await asyncio.create_subprocess_exec(
+                    *PROGRAM_CAT,
+                    stdin=subprocess.PIPE,
+                    stdout=subprocess.PIPE,
+                )
+                stdout, stderr = await proc.communicate(b"some data")
+                return proc.returncode, stdout
 
-        async def main():
-            return await asyncio.to_thread(asyncio.run, in_thread())
+            async def main():
+                return await asyncio.to_thread(asyncio.run, in_thread())
 
-        asyncio.set_child_watcher(asyncio.PidfdChildWatcher())
-        try:
-            returncode, stdout = asyncio.run(main())
-            self.assertEqual(returncode, 0)
-            self.assertEqual(stdout, b'some data')
-        finally:
-            asyncio.set_child_watcher(None)
+            asyncio.set_child_watcher(asyncio.PidfdChildWatcher())
+            try:
+                returncode, stdout = asyncio.run(main())
+                self.assertEqual(returncode, 0)
+                self.assertEqual(stdout, b'some data')
+            finally:
+                asyncio.set_child_watcher(None)
+else:
+    # Windows
+    class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase):
 
+        def setUp(self):
+            super().setUp()
+            self.loop = asyncio.ProactorEventLoop()
+            self.set_event_loop(self.loop)
 
 if __name__ == '__main__':
     unittest.main()