Fix subprocess handling when forbid_global_loop.

README.rst

@@ -169,6 +169,9 @@ Changelog
 0.6.0 (UNRELEASED)
 ~~~~~~~~~~~~~~~~~~
 - ``pytestmark`` now works on both module and class level.
+- Using ``forbid_global_loop`` now allows tests to use ``asyncio`` 
+  subprocesses.
+  `#36 <https://github.com/pytest-dev/pytest-asyncio/issues/36>`_
 
 0.5.0 (2016-09-07)
 ~~~~~~~~~~~~~~~~~~

pytest_asyncio/plugin.py

@@ -1,3 +1,4 @@
+"""pytest-asyncio implementation."""
 import asyncio
 import inspect
 import socket
@@ -10,9 +11,18 @@
 from _pytest.python import transfer_markers
 
 
-class ForbiddenEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
-    """An event loop policy that raises errors on any operation."""
-    pass
+class ForbiddenEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+    """An event loop policy that raises errors on most operations.
+
+    Operations involving child watchers are permitted."""
+
+    def get_event_loop(self):
+        """Not allowed."""
+        raise NotImplementedError
+
+    def set_event_loop(self, _):
+        """Not allowed."""
+        raise NotImplementedError
 
 
 def _is_coroutine(obj):
@@ -21,6 +31,7 @@ def _is_coroutine(obj):
 
 
 def pytest_configure(config):
+    """Inject documentation."""
     config.addinivalue_line("markers",
                             "asyncio: "
                             "mark the test as a coroutine, it will be "
@@ -65,6 +76,7 @@ def pytest_fixture_setup(fixturedef, request):
             policy = asyncio.get_event_loop_policy()
             if forbid_global_loop:
                 asyncio.set_event_loop_policy(ForbiddenEventLoopPolicy())
+                asyncio.get_child_watcher().attach_loop(loop)
                 fixturedef.addfinalizer(lambda: asyncio.set_event_loop_policy(policy))
             else:
                 policy.set_event_loop(loop)

test_requirements.txt

@@ -1 +1,2 @@
 coverage==4.1
+tox==2.5.0

tests/test_subprocess.py

@@ -0,0 +1,26 @@
+"""Tests for using subprocesses in tests."""
+import sys
+import asyncio
+import asyncio.subprocess
+
+import pytest
+
+
+@pytest.mark.asyncio(forbid_global_loop=False)
+@asyncio.coroutine
+def test_subprocess(event_loop):
+    """Starting a subprocess should be possible."""
+    proc = yield from asyncio.subprocess.create_subprocess_exec(
+        sys.executable, '--version', stdout=asyncio.subprocess.PIPE,
+        loop=event_loop)
+    yield from proc.communicate()
+
+
+@pytest.mark.asyncio(forbid_global_loop=True)
+@asyncio.coroutine
+def test_subprocess_forbid(event_loop):
+    """Starting a subprocess should be possible."""
+    proc = yield from asyncio.subprocess.create_subprocess_exec(
+        sys.executable, '--version', stdout=asyncio.subprocess.PIPE,
+        loop=event_loop)
+    yield from proc.communicate()