diff --git a/README.rst b/README.rst
index 58ff88c..ae4b9d4 100644
--- a/README.rst
+++ b/README.rst
@@ -83,6 +83,22 @@ Waiting for deferreds in fixtures
       return pytest_twisted.blockon(d)
 
 
+async/await fixtures
+====================
+``async``/``await`` fixtures can be used along with ``yield`` for normal
+pytest fixture semantics of setup, value, and teardown.  At present only
+function scope is supported::
+
+  @pytest_twisted.async_fixture
+  async def foo():
+      d1, d2 = defer.Deferred(), defer.Deferred()
+      reactor.callLater(0.01, d1.callback, 42)
+      reactor.callLater(0.02, d2.callback, 37)
+      value = await d1
+      yield value
+      await d2
+
+
 The twisted greenlet
 ====================
 Some libraries (e.g. corotwine) need to know the greenlet, which is
diff --git a/pytest_twisted.py b/pytest_twisted.py
index c79d483..1046cd2 100644
--- a/pytest_twisted.py
+++ b/pytest_twisted.py
@@ -15,6 +15,30 @@ class WrongReactorAlreadyInstalledError(Exception):
     pass
 
 
+class UnrecognizedCoroutineMarkError(Exception):
+    @classmethod
+    def from_mark(cls, mark):
+        return cls(
+            'Coroutine wrapper mark not recognized: {}'.format(repr(mark)),
+        )
+
+
+class AsyncGeneratorFixtureDidNotStopError(Exception):
+    @classmethod
+    def from_generator(cls, generator):
+        return cls(
+            'async fixture did not stop: {}'.format(generator),
+        )
+
+
+class AsyncFixtureUnsupportedScopeError(Exception):
+    @classmethod
+    def from_scope(cls, scope):
+        return cls(
+            'Unsupported scope used for async fixture: {}'.format(scope)
+        )
+
+
 class _config:
     external_reactor = False
 
@@ -105,16 +129,95 @@ def stop_twisted_greenlet():
         _instances.gr_twisted.switch()
 
 
+class _CoroutineWrapper:
+    def __init__(self, coroutine, mark):
+        self.coroutine = coroutine
+        self.mark = mark
+
+
+def _marked_async_fixture(mark):
+    @functools.wraps(pytest.fixture)
+    def fixture(*args, **kwargs):
+        try:
+            scope = args[0]
+        except IndexError:
+            scope = kwargs.get('scope', 'function')
+
+        if scope != 'function':
+            raise AsyncFixtureUnsupportedScopeError.from_scope(scope=scope)
+
+        def marker(f):
+            @functools.wraps(f)
+            def w(*args, **kwargs):
+                return _CoroutineWrapper(
+                    coroutine=f(*args, **kwargs),
+                    mark=mark,
+                )
+
+            return w
+
+        def decorator(f):
+            result = pytest.fixture(*args, **kwargs)(marker(f))
+
+            return result
+
+        return decorator
+
+    return fixture
+
+
+async_fixture = _marked_async_fixture('async_fixture')
+async_yield_fixture = _marked_async_fixture('async_yield_fixture')
+
+
+@defer.inlineCallbacks
 def _pytest_pyfunc_call(pyfuncitem):
     testfunction = pyfuncitem.obj
+    async_generators = []
     funcargs = pyfuncitem.funcargs
     if hasattr(pyfuncitem, "_fixtureinfo"):
         testargs = {}
         for arg in pyfuncitem._fixtureinfo.argnames:
-            testargs[arg] = funcargs[arg]
+            if isinstance(funcargs[arg], _CoroutineWrapper):
+                wrapper = funcargs[arg]
+
+                if wrapper.mark == 'async_fixture':
+                    arg_value = yield defer.ensureDeferred(
+                        wrapper.coroutine
+                    )
+                elif wrapper.mark == 'async_yield_fixture':
+                    async_generators.append((arg, wrapper))
+                    arg_value = yield defer.ensureDeferred(
+                        wrapper.coroutine.__anext__(),
+                    )
+                else:
+                    raise UnrecognizedCoroutineMarkError.from_mark(
+                        mark=wrapper.mark,
+                    )
+            else:
+                arg_value = funcargs[arg]
+
+            testargs[arg] = arg_value
     else:
         testargs = funcargs
-    return testfunction(**testargs)
+    result = yield testfunction(**testargs)
+
+    async_generator_deferreds = [
+        (arg, defer.ensureDeferred(g.coroutine.__anext__()))
+        for arg, g in reversed(async_generators)
+    ]
+
+    for arg, d in async_generator_deferreds:
+        try:
+            yield d
+        except StopAsyncIteration:
+            continue
+        else:
+            raise AsyncGeneratorFixtureDidNotStopError.from_generator(
+                generator=arg,
+            )
+
+    defer.returnValue(result)
 
 
 def pytest_pyfunc_call(pyfuncitem):
diff --git a/testing/test_basic.py b/testing/test_basic.py
index d45a64d..1c8ccf3 100755
--- a/testing/test_basic.py
+++ b/testing/test_basic.py
@@ -4,8 +4,12 @@
 import pytest
 
 
+# https://docs.python.org/3/whatsnew/3.5.html#pep-492-coroutines-with-async-and-await-syntax
 ASYNC_AWAIT = sys.version_info >= (3, 5)
 
+# https://docs.python.org/3/whatsnew/3.6.html#pep-525-asynchronous-generators
+ASYNC_GENERATORS = sys.version_info >= (3, 6)
+
 
 def assert_outcomes(run_result, outcomes):
     formatted_output = format_run_result_output_for_assert(run_result)
@@ -47,6 +51,13 @@ def skip_if_no_async_await():
     )
 
 
+def skip_if_no_async_generators():
+    return pytest.mark.skipif(
+        not ASYNC_GENERATORS,
+        reason="async generators not support on Python <3.6",
+    )
+
+
 @pytest.fixture
 def cmd_opts(request):
     reactor = request.config.getoption("reactor", "default")
@@ -303,6 +314,157 @@ async def test_succeed(foo):
     assert_outcomes(rr, {"passed": 2, "failed": 1})
 
 
+@skip_if_no_async_await()
+def test_async_fixture(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet import reactor, defer
+    import pytest
+    import pytest_twisted
+
+    @pytest_twisted.async_fixture(scope="function", params=["fs", "imap", "web"])
+    @pytest.mark.redgreenblue
+    async def foo(request):
+        d1, d2 = defer.Deferred(), defer.Deferred()
+        reactor.callLater(0.01, d1.callback, 1)
+        reactor.callLater(0.02, d2.callback, request.param)
+        await d1
+        return d2,
+
+    @pytest_twisted.inlineCallbacks
+    def test_succeed_blue(foo):
+        x = yield foo[0]
+        if x == "web":
+            raise RuntimeError("baz")
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", "-v", *cmd_opts)
+    assert_outcomes(rr, {"passed": 2, "failed": 1})
+
+
+@skip_if_no_async_generators()
+def test_async_yield_fixture_concurrent_teardown(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet import reactor, defer
+    import pytest
+    import pytest_twisted
+
+
+    here = defer.Deferred()
+    there = defer.Deferred()
+
+    @pytest_twisted.async_yield_fixture()
+    async def this():
+        yield 42
+
+        there.callback(None)
+        reactor.callLater(5, here.cancel)
+        await here
+
+    @pytest_twisted.async_yield_fixture()
+    async def that():
+        yield 37
+
+        here.callback(None)
+        reactor.callLater(5, there.cancel)
+        await there
+
+    def test_succeed(this, that):
+        pass
+    """
+    testdir.makepyfile(test_file)
+    # TODO: add a timeout, failure just hangs indefinitely for now
+    # https://github.com/pytest-dev/pytest/issues/4073
+    rr = testdir.run(sys.executable, "-m", "pytest", "-v", *cmd_opts)
+    assert_outcomes(rr, {"passed": 1})
+
+
+@skip_if_no_async_generators()
+def test_async_yield_fixture(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet import reactor, defer
+    import pytest
+    import pytest_twisted
+
+    @pytest_twisted.async_yield_fixture(
+        scope="function",
+        params=["fs", "imap", "web", "gopher", "archie"],
+    )
+    async def foo(request):
+        d1, d2 = defer.Deferred(), defer.Deferred()
+        reactor.callLater(0.01, d1.callback, 1)
+        reactor.callLater(0.02, d2.callback, request.param)
+        await d1
+
+        # Twisted doesn't allow calling back with a Deferred as a value.
+        # This deferred is being wrapped up in a tuple to sneak through.
+        # https://github.com/twisted/twisted/blob/c0f1394c7bfb04d97c725a353a1f678fa6a1c602/src/twisted/internet/defer.py#L459
+        yield d2,
+
+        if request.param == "gopher":
+            raise RuntimeError("gaz")
+
+        if request.param == "archie":
+            yield 42
+
+    @pytest_twisted.inlineCallbacks
+    def test_succeed(foo):
+        x = yield foo[0]
+        if x == "web":
+            raise RuntimeError("baz")
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", "-v", *cmd_opts)
+    assert_outcomes(rr, {"passed": 2, "failed": 3})
+
+
+@skip_if_no_async_generators()
+def test_async_yield_fixture_function_scope(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet import reactor, defer
+    import pytest
+    import pytest_twisted
+
+    check_me = 0
+
+    @pytest_twisted.async_yield_fixture(scope="function")
+    async def foo():
+        global check_me
+
+        if check_me != 0:
+            raise Exception('check_me already modified before fixture run')
+
+        check_me = 1
+
+        yield 42
+
+        if check_me != 2:
+            raise Exception(
+                'check_me not updated properly: {}'.format(check_me),
+            )
+
+        check_me = 0
+
+    def test_first(foo):
+        global check_me
+
+        assert check_me == 1
+        assert foo == 42
+
+        check_me = 2
+
+    def test_second(foo):
+        global check_me
+
+        assert check_me == 1
+        assert foo == 42
+
+        check_me = 2
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", "-v", *cmd_opts)
+    assert_outcomes(rr, {"passed": 2})
+
+
 def test_blockon_in_hook(testdir, cmd_opts, request):
     skip_if_reactor_not(request, "default")
     conftest_file = """