diff --git a/ci/rtd-requirements.txt b/ci/rtd-requirements.txt
index 67898ba..ea59c6b 100644
--- a/ci/rtd-requirements.txt
+++ b/ci/rtd-requirements.txt
@@ -2,3 +2,6 @@
 sphinx >= 1.6.1
 sphinx_rtd_theme
 sphinxcontrib-trio
+# Workaround for this weird issue:
+# https://travis-ci.org/python-trio/pytest-trio/jobs/407495415
+attrs >= 17.4.0
diff --git a/pytest_trio/__init__.py b/pytest_trio/__init__.py
index ffd3b89..5fbd91f 100644
--- a/pytest_trio/__init__.py
+++ b/pytest_trio/__init__.py
@@ -1,3 +1,6 @@
 """Top-level package for pytest-trio."""
 
 from ._version import __version__
+from .plugin import trio_fixture
+
+__all__ = ["trio_fixture"]
diff --git a/pytest_trio/_tests/helpers.py b/pytest_trio/_tests/helpers.py
new file mode 100644
index 0000000..5ae5f9e
--- /dev/null
+++ b/pytest_trio/_tests/helpers.py
@@ -0,0 +1,15 @@
+import pytest
+
+
+def enable_trio_mode_via_pytest_ini(testdir):
+    testdir.makefile(".ini", pytest="[pytest]\ntrio_mode = true\n")
+
+
+def enable_trio_mode_via_conftest_py(testdir):
+    testdir.makeconftest("from pytest_trio.enable_trio_mode import *")
+
+
+enable_trio_mode = pytest.mark.parametrize(
+    "enable_trio_mode",
+    [enable_trio_mode_via_pytest_ini, enable_trio_mode_via_conftest_py]
+)
diff --git a/pytest_trio/_tests/test_async_yield_fixture.py b/pytest_trio/_tests/test_async_yield_fixture.py
index b46dd06..4810456 100644
--- a/pytest_trio/_tests/test_async_yield_fixture.py
+++ b/pytest_trio/_tests/test_async_yield_fixture.py
@@ -317,39 +317,43 @@ def test_async_yield_fixture_crashed_teardown_allow_other_teardowns(
         import trio
         from async_generator import async_generator, yield_
 
-        events = []
+        setup_events = set()
+        teardown_events = set()
 
         @pytest.fixture
         @async_generator
         async def good_fixture():
             async with trio.open_nursery() as nursery:
-                events.append('good_fixture setup')
+                setup_events.add('good_fixture setup')
                 await yield_(None)
-                events.append('good_fixture teardown')
+                teardown_events.add('good_fixture teardown')
 
         @pytest.fixture
         @async_generator
         async def bad_fixture():
             async with trio.open_nursery() as nursery:
-                events.append('bad_fixture setup')
+                setup_events.add('bad_fixture setup')
                 await yield_(None)
-                events.append('bad_fixture teardown')
+                teardown_events.add('bad_fixture teardown')
                 raise RuntimeError('Crash during fixture teardown')
 
         def test_before():
-            assert not events
+            assert not setup_events
+            assert not teardown_events
 
         @pytest.mark.trio
         async def test_actual_test(bad_fixture, good_fixture):
             pass
 
         def test_after():
-            assert events == [
+            assert setup_events == {
                 'good_fixture setup',
                 'bad_fixture setup',
+            }
+            assert teardown_events == {
                 'bad_fixture teardown',
                 'good_fixture teardown',
-            ]
+            }
     """
         )
     )
diff --git a/pytest_trio/_tests/test_basic.py b/pytest_trio/_tests/test_basic.py
index 6ae233f..d442ee6 100644
--- a/pytest_trio/_tests/test_basic.py
+++ b/pytest_trio/_tests/test_basic.py
@@ -21,7 +21,7 @@ def test_check_async_test_called():
     """
     )
 
-    result = testdir.runpytest()
+    result = testdir.runpytest("-s")
 
     result.assert_outcomes(passed=2)
 
diff --git a/pytest_trio/_tests/test_contextvars.py b/pytest_trio/_tests/test_contextvars.py
new file mode 100644
index 0000000..33bf147
--- /dev/null
+++ b/pytest_trio/_tests/test_contextvars.py
@@ -0,0 +1,37 @@
+import pytest
+from pytest_trio import trio_fixture
+
+import contextvars
+
+cv = contextvars.ContextVar("cv", default=None)
+
+
+@trio_fixture
+def cv_checker():
+    assert cv.get() is None
+    yield
+    assert cv.get() is None
+
+
+@trio_fixture
+def cv_setter(cv_checker):
+    assert cv.get() is None
+    token = cv.set("cv_setter")
+    yield
+    assert cv.get() == "cv_setter2"
+    cv.reset(token)
+    assert cv.get() is None
+
+
+@trio_fixture
+def cv_setter2(cv_setter):
+    assert cv.get() == "cv_setter"
+    # Intentionally leak, so can check that this is visible back in cv_setter
+    cv.set("cv_setter2")
+    yield
+    assert cv.get() == "cv_setter2"
+
+
+@pytest.mark.trio
+async def test_contextvars(cv_setter2):
+    assert cv.get() == "cv_setter2"
diff --git a/pytest_trio/_tests/test_fixture_mistakes.py b/pytest_trio/_tests/test_fixture_mistakes.py
new file mode 100644
index 0000000..5c6873c
--- /dev/null
+++ b/pytest_trio/_tests/test_fixture_mistakes.py
@@ -0,0 +1,119 @@
+import pytest
+from pytest_trio import trio_fixture
+
+from .helpers import enable_trio_mode
+
+
+def test_trio_fixture_with_non_trio_test(testdir):
+    testdir.makepyfile(
+        """
+        import trio
+        from pytest_trio import trio_fixture
+        import pytest
+
+        @trio_fixture
+        def trio_time():
+            return trio.current_time()
+
+        @pytest.fixture
+        def indirect_trio_time(trio_time):
+            return trio_time + 1
+
+        @pytest.mark.trio
+        async def test_async(mock_clock, trio_time, indirect_trio_time):
+            assert trio_time == 0
+            assert indirect_trio_time == 1
+
+        def test_sync(trio_time):
+            pass
+
+        def test_sync_indirect(indirect_trio_time):
+            pass
+        """
+    )
+
+    result = testdir.runpytest()
+
+    result.assert_outcomes(passed=1, error=2)
+    result.stdout.fnmatch_lines(
+        ["*: Trio fixtures can only be used by Trio tests*"]
+    )
+
+
+def test_trio_fixture_with_wrong_scope_without_trio_mode(testdir):
+    # There's a trick here: when you have a non-function-scope fixture, it's
+    # not instantiated for any particular function (obviously). So... when our
+    # pytest_fixture_setup hook tries to check for marks, it can't normally
+    # see @pytest.mark.trio. So... it's actually almost impossible to have an
+    # async fixture get treated as a Trio fixture *and* have it be
+    # non-function-scope. But, class-scoped fixtures can see marks on the
+    # class, so this is one way (the only way?) it can happen:
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.fixture(scope="class")
+        async def async_class_fixture():
+            pass
+
+        @pytest.mark.trio
+        class TestFoo:
+            async def test_foo(self, async_class_fixture):
+                pass
+        """
+    )
+
+    result = testdir.runpytest()
+
+    result.assert_outcomes(error=1)
+    result.stdout.fnmatch_lines(["*: Trio fixtures must be function-scope*"])
+
+
+@enable_trio_mode
+def test_trio_fixture_with_wrong_scope_in_trio_mode(testdir, enable_trio_mode):
+    enable_trio_mode(testdir)
+
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.fixture(scope="session")
+        async def async_session_fixture():
+            pass
+
+
+        async def test_whatever(async_session_fixture):
+            pass
+        """
+    )
+
+    result = testdir.runpytest()
+
+    result.assert_outcomes(error=1)
+    result.stdout.fnmatch_lines(["*: Trio fixtures must be function-scope*"])
+
+
+@enable_trio_mode
+def test_async_fixture_with_sync_test_in_trio_mode(testdir, enable_trio_mode):
+    enable_trio_mode(testdir)
+
+    testdir.makepyfile(
+        """
+        import pytest
+
+        @pytest.fixture
+        async def async_fixture():
+            pass
+
+
+        def test_whatever(async_fixture):
+            pass
+        """
+    )
+
+    result = testdir.runpytest()
+
+    result.assert_outcomes(error=1)
+    result.stdout.fnmatch_lines(
+        ["*: Trio fixtures can only be used by Trio tests*"]
+    )
diff --git a/pytest_trio/_tests/test_fixture_names.py b/pytest_trio/_tests/test_fixture_names.py
new file mode 100644
index 0000000..f22f64e
--- /dev/null
+++ b/pytest_trio/_tests/test_fixture_names.py
@@ -0,0 +1,18 @@
+import pytest
+from pytest_trio import trio_fixture
+import trio
+
+
+@trio_fixture
+def fixture_with_unique_name(nursery):
+    nursery.start_soon(trio.sleep_forever)
+
+
+@pytest.mark.trio
+async def test_fixture_names(fixture_with_unique_name):
+    # This might be a bit fragile ... if we rearrange the nursery hierarchy
+    # somehow so it breaks, then we can make it more robust.
+    task = trio.hazmat.current_task()
+    assert task.name == "<test 'test_fixture_names'>"
+    sibling_names = {task.name for task in task.parent_nursery.child_tasks}
+    assert "<fixture 'fixture_with_unique_name'>" in sibling_names
diff --git a/pytest_trio/_tests/test_nursery_fixture.py b/pytest_trio/_tests/test_fixture_nursery.py
similarity index 100%
rename from pytest_trio/_tests/test_nursery_fixture.py
rename to pytest_trio/_tests/test_fixture_nursery.py
diff --git a/pytest_trio/_tests/test_fixture_ordering.py b/pytest_trio/_tests/test_fixture_ordering.py
new file mode 100644
index 0000000..352ca6a
--- /dev/null
+++ b/pytest_trio/_tests/test_fixture_ordering.py
@@ -0,0 +1,286 @@
+import pytest
+
+
+# Tests that:
+# - leaf_fix gets set up first and torn down last
+# - the two fix_concurrent_{1,2} fixtures run their setup/teardown code
+#   at the same time -- their execution can be interleaved.
+def test_fixture_basic_ordering(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+        from pytest_trio import trio_fixture
+        from trio.testing import Sequencer
+        from async_generator import async_generator, yield_
+
+        setup_events = []
+        teardown_events = []
+
+        @trio_fixture
+        def seq():
+            return Sequencer()
+
+        @pytest.fixture
+        @async_generator
+        async def leaf_fix():
+            setup_events.append("leaf_fix setup")
+            await yield_()
+            teardown_events.append("leaf_fix teardown")
+
+            assert teardown_events == [
+                "fix_concurrent_1 teardown 1",
+                "fix_concurrent_2 teardown 1",
+                "fix_concurrent_1 teardown 2",
+                "fix_concurrent_2 teardown 2",
+                "leaf_fix teardown",
+            ]
+
+        @pytest.fixture
+        @async_generator
+        async def fix_concurrent_1(leaf_fix, seq):
+            async with seq(0):
+                setup_events.append("fix_concurrent_1 setup 1")
+            async with seq(2):
+                setup_events.append("fix_concurrent_1 setup 2")
+            await yield_()
+            async with seq(4):
+                teardown_events.append("fix_concurrent_1 teardown 1")
+            async with seq(6):
+                teardown_events.append("fix_concurrent_1 teardown 2")
+
+        @pytest.fixture
+        @async_generator
+        async def fix_concurrent_2(leaf_fix, seq):
+            async with seq(1):
+                setup_events.append("fix_concurrent_2 setup 1")
+            async with seq(3):
+                setup_events.append("fix_concurrent_2 setup 2")
+            await yield_()
+            async with seq(5):
+                teardown_events.append("fix_concurrent_2 teardown 1")
+            async with seq(7):
+                teardown_events.append("fix_concurrent_2 teardown 2")
+
+        @pytest.mark.trio
+        async def test_root(fix_concurrent_1, fix_concurrent_2):
+            assert setup_events == [
+                "leaf_fix setup",
+                "fix_concurrent_1 setup 1",
+                "fix_concurrent_2 setup 1",
+                "fix_concurrent_1 setup 2",
+                "fix_concurrent_2 setup 2",
+            ]
+            assert teardown_events == []
+
+        """
+    )
+
+    result = testdir.runpytest()
+    result.assert_outcomes(passed=1)
+
+
+def test_nursery_fixture_teardown_ordering(testdir):
+    testdir.makepyfile(
+        """
+        import pytest
+        from pytest_trio import trio_fixture
+        import trio
+        from trio.testing import wait_all_tasks_blocked
+
+        events = []
+
+        async def record_cancel(msg):
+            try:
+                await trio.sleep_forever()
+            finally:
+                events.append(msg)
+
+        @pytest.fixture
+        def fix0():
+            yield
+            assert events == [
+                "test",
+                "test cancel",
+                "fix2 teardown",
+                "fix2 cancel",
+                "fix1 teardown",
+                "fix1 cancel",
+            ]
+
+        @trio_fixture
+        def fix1(nursery):
+            nursery.start_soon(record_cancel, "fix1 cancel")
+            yield
+            events.append("fix1 teardown")
+
+        @trio_fixture
+        def fix2(fix1, nursery):
+            nursery.start_soon(record_cancel, "fix2 cancel")
+            yield
+            events.append("fix2 teardown")
+
+        @pytest.mark.trio
+        async def test_root(fix2, nursery):
+            nursery.start_soon(record_cancel, "test cancel")
+            await wait_all_tasks_blocked()
+            events.append("test")
+        """
+    )
+
+    result = testdir.runpytest()
+    result.assert_outcomes(passed=1)
+
+
+def test_error_collection(testdir):
+    # We want to make sure that pytest ultimately reports all the different
+    # exceptions. We call .upper() on all the exceptions so that we have
+    # tokens to look for in the output corresponding to each exception, where
+    # those tokens don't appear at all the source (so we can't get a false
+    # positive due to pytest printing out the source file).
+
+    # We sleep at the beginning of all the fixtures b/c currently if any
+    # fixture crashes, we skip setting up unrelated fixtures whose setup
+    # hasn't even started yet. Maybe we shouldn't? But for now the sleeps make
+    # sure that all the fixtures have started before any of them start
+    # crashing.
+    testdir.makepyfile(
+        """
+        import pytest
+        from pytest_trio import trio_fixture
+        import trio
+        from async_generator import async_generator, yield_
+
+        test_started = False
+
+        @trio_fixture
+        async def crash_nongen():
+            await trio.sleep(2)
+            raise RuntimeError("crash_nongen".upper())
+
+        @trio_fixture
+        @async_generator
+        async def crash_early_agen():
+            await trio.sleep(2)
+            raise RuntimeError("crash_early_agen".upper())
+            await yield_()
+
+        @trio_fixture
+        @async_generator
+        async def crash_late_agen():
+            await yield_()
+            raise RuntimeError("crash_late_agen".upper())
+
+        async def crash(when, token):
+            with trio.open_cancel_scope(shield=True):
+                await trio.sleep(when)
+                raise RuntimeError(token.upper())
+
+        @trio_fixture
+        def crash_background(nursery):
+            nursery.start_soon(crash, 1, "crash_background_early")
+            nursery.start_soon(crash, 3, "crash_background_late")
+
+        @pytest.mark.trio
+        async def test_all_the_crashes(
+            autojump_clock,
+            crash_nongen, crash_early_agen, crash_late_agen, crash_background,
+        ):
+            global test_started
+            test_started = True
+
+        def test_followup():
+            assert not test_started
+
+        """
+    )
+
+    result = testdir.runpytest()
+    result.assert_outcomes(passed=1, failed=1)
+    result.stdout.fnmatch_lines_random(
+        [
+            "*CRASH_NONGEN*",
+            "*CRASH_EARLY_AGEN*",
+            "*CRASH_LATE_AGEN*",
+            "*CRASH_BACKGROUND_EARLY*",
+            "*CRASH_BACKGROUND_LATE*",
+        ]
+    )
+
+
+@pytest.mark.parametrize("bgmode", ["nursery fixture", "manual nursery"])
+def test_background_crash_cancellation_propagation(bgmode, testdir):
+    crashyfix_using_nursery_fixture = """
+        @trio_fixture
+        def crashyfix(nursery):
+            nursery.start_soon(crashy)
+            yield
+            # We should be cancelled here
+            teardown_deadlines["crashyfix"] = trio.current_effective_deadline()
+        """
+
+    crashyfix_using_manual_nursery = """
+        @trio_fixture
+        @async_generator
+        async def crashyfix():
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(crashy)
+                await yield_()
+                # We should be cancelled here
+                teardown_deadlines["crashyfix"] = trio.current_effective_deadline()
+        """
+
+    if bgmode == "nursery fixture":
+        crashyfix = crashyfix_using_nursery_fixture
+    else:
+        crashyfix = crashyfix_using_manual_nursery
+
+    testdir.makepyfile(
+        """
+        import pytest
+        from pytest_trio import trio_fixture
+        import trio
+        from async_generator import async_generator, yield_
+
+        teardown_deadlines = {}
+        final_time = None
+
+        async def crashy():
+            await trio.sleep(1)
+            raise RuntimeError
+
+        CRASHYFIX_HERE
+
+        @trio_fixture
+        def sidefix():
+            yield
+            # We should NOT be cancelled here
+            teardown_deadlines["sidefix"] = trio.current_effective_deadline()
+
+        @trio_fixture
+        def userfix(crashyfix):
+            yield
+            # Currently we should NOT be cancelled here... though maybe this
+            # should change?
+            teardown_deadlines["userfix"] = trio.current_effective_deadline()
+
+        @pytest.mark.trio
+        async def test_it(userfix, sidefix, autojump_clock):
+            try:
+                await trio.sleep_forever()
+            finally:
+                global final_time
+                final_time = trio.current_time()
+
+
+        def test_post():
+            assert teardown_deadlines == {
+                "crashyfix": -float("inf"),
+                "sidefix": float("inf"),
+                "userfix": float("inf"),
+            }
+            assert final_time == 1
+        """.replace("CRASHYFIX_HERE", crashyfix)
+    )
+
+    result = testdir.runpytest()
+    result.assert_outcomes(passed=1, failed=1)
diff --git a/pytest_trio/_tests/test_sync_fixture.py b/pytest_trio/_tests/test_sync_fixture.py
index 877da28..7a4cb0f 100644
--- a/pytest_trio/_tests/test_sync_fixture.py
+++ b/pytest_trio/_tests/test_sync_fixture.py
@@ -96,7 +96,8 @@ def test_sync_yield_fixture_crashed_teardown_allow_other_teardowns(testdir):
         import pytest
         import trio
 
-        events = []
+        setup_events = set()
+        teardown_events = set()
 
         @pytest.fixture
         async def force_async_fixture():
@@ -104,31 +105,34 @@ async def force_async_fixture():
 
         @pytest.fixture
         def good_fixture(force_async_fixture):
-            events.append('good_fixture setup')
+            setup_events.add('good_fixture setup')
             yield
-            events.append('good_fixture teardown')
+            teardown_events.add('good_fixture teardown')
 
         @pytest.fixture
         def bad_fixture(force_async_fixture):
-            events.append('bad_fixture setup')
+            setup_events.add('bad_fixture setup')
             yield
-            events.append('bad_fixture teardown')
+            teardown_events.add('bad_fixture teardown')
             raise RuntimeError('Crash during fixture teardown')
 
         def test_before():
-            assert not events
+            assert not setup_events
+            assert not teardown_events
 
         @pytest.mark.trio
         async def test_actual_test(bad_fixture, good_fixture):
             pass
 
         def test_after():
-            assert events == [
+            assert setup_events == {
                 'good_fixture setup',
                 'bad_fixture setup',
+            }
+            assert teardown_events == {
                 'bad_fixture teardown',
                 'good_fixture teardown',
-            ]
+            }
     """
     )
 
diff --git a/pytest_trio/_tests/test_trio_asyncio.py b/pytest_trio/_tests/test_trio_asyncio.py
new file mode 100644
index 0000000..d899bb8
--- /dev/null
+++ b/pytest_trio/_tests/test_trio_asyncio.py
@@ -0,0 +1,53 @@
+import pytest
+import sys
+import asyncio
+from async_generator import async_generator, yield_
+
+if sys.version_info < (3, 6):
+    pytestmark = pytest.mark.skip(
+        reason="trio-asyncio doesn't seem to work on 3.5"
+    )
+else:
+    import trio_asyncio
+
+
+async def use_run_asyncio():
+    await trio_asyncio.run_asyncio(asyncio.sleep, 0)
+
+
+@pytest.fixture()
+@async_generator
+async def asyncio_loop():
+    async with trio_asyncio.open_loop() as loop:
+        await yield_(loop)
+
+
+@pytest.fixture()
+@async_generator
+async def asyncio_fixture_with_fixtured_loop(asyncio_loop):
+    await use_run_asyncio()
+    await yield_()
+
+
+@pytest.fixture()
+@async_generator
+async def asyncio_fixture_own_loop():
+    async with trio_asyncio.open_loop():
+        await use_run_asyncio()
+        await yield_()
+
+
+@pytest.mark.trio
+async def test_no_fixture():
+    async with trio_asyncio.open_loop():
+        await use_run_asyncio()
+
+
+@pytest.mark.trio
+async def test_half_fixtured_asyncpg_conn(asyncio_fixture_own_loop):
+    await use_run_asyncio()
+
+
+@pytest.mark.trio
+async def test_fixtured_asyncpg_conn(asyncio_fixture_with_fixtured_loop):
+    await use_run_asyncio()
diff --git a/pytest_trio/_tests/test_trio_mode.py b/pytest_trio/_tests/test_trio_mode.py
index 764f5da..efc66a8 100644
--- a/pytest_trio/_tests/test_trio_mode.py
+++ b/pytest_trio/_tests/test_trio_mode.py
@@ -1,5 +1,7 @@
 import pytest
 
+from .helpers import enable_trio_mode
+
 test_text = """
 import pytest
 import trio
@@ -26,19 +28,11 @@ async def test_hypothesis_fail(b):
 """
 
 
-def test_trio_mode_pytest_ini(testdir):
-    testdir.makepyfile(test_text)
-
-    testdir.makefile(".ini", pytest="[pytest]\ntrio_mode = true\n")
-
-    result = testdir.runpytest()
-    result.assert_outcomes(passed=2, failed=2)
+@enable_trio_mode
+def test_trio_mode(testdir, enable_trio_mode):
+    enable_trio_mode(testdir)
 
-
-def test_trio_mode_conftest(testdir):
     testdir.makepyfile(test_text)
 
-    testdir.makeconftest("from pytest_trio.enable_trio_mode import *")
-
     result = testdir.runpytest()
     result.assert_outcomes(passed=2, failed=2)
diff --git a/pytest_trio/enable_trio_mode.py b/pytest_trio/enable_trio_mode.py
index 6f05933..0c13614 100644
--- a/pytest_trio/enable_trio_mode.py
+++ b/pytest_trio/enable_trio_mode.py
@@ -1,7 +1,11 @@
-__all__ = ["pytest_collection_modifyitems"]
+__all__ = ["pytest_collection_modifyitems", "pytest_fixture_setup"]
 
-from .plugin import automark
+from .plugin import automark, handle_fixture
 
 
 def pytest_collection_modifyitems(items):
     automark(items)
+
+
+def pytest_fixture_setup(fixturedef, request):
+    return handle_fixture(fixturedef, request, force_trio_mode=True)
diff --git a/pytest_trio/plugin.py b/pytest_trio/plugin.py
index 59d79e0..498593b 100644
--- a/pytest_trio/plugin.py
+++ b/pytest_trio/plugin.py
@@ -1,12 +1,15 @@
 """pytest-trio implementation."""
 import sys
 from traceback import format_exception
+from collections.abc import Coroutine, Generator
 from inspect import iscoroutinefunction, isgeneratorfunction
+import contextvars
 import pytest
 import trio
 from trio.testing import MockClock, trio_test
 from async_generator import (
-    async_generator, yield_, asynccontextmanager, isasyncgenfunction
+    async_generator, yield_, asynccontextmanager, isasyncgen,
+    isasyncgenfunction
 )
 
 ################################################################
@@ -46,211 +49,297 @@ def pytest_exception_interact(node, call, report):
 
 
 ################################################################
-# Core support for running tests and constructing fixtures
+# Core support for trio fixtures and trio tests
 ################################################################
 
-
-def _trio_test_runner_factory(item, testfunc=None):
-    testfunc = testfunc or item.function
-
-    if getattr(testfunc, '_trio_test_runner_wrapped', False):
-        # We have already wrapped this, perhaps because we combined Hypothesis
-        # with pytest.mark.parametrize
-        return testfunc
-
-    if not iscoroutinefunction(testfunc):
-        pytest.fail(
-            'test function `%r` is marked trio but is not async' % item
-        )
-
-    @trio_test
-    async def _bootstrap_fixture_and_run_test(**kwargs):
-        __tracebackhide__ = True
-        user_exc = None
-        # Open the nursery exposed as fixture
-        async with trio.open_nursery() as nursery:
-            item._trio_nursery = nursery
-            try:
-                async with _setup_async_fixtures_in(kwargs) as resolved_kwargs:
-                    try:
-                        await testfunc(**resolved_kwargs)
-                    except BaseException as exc:
-                        # Regular pytest fixture don't have access to the test
-                        # exception in there teardown, we mimic this behavior
-                        # here.
-                        user_exc = exc
-            except BaseException as exc:
-                # If we are here, the exception comes from the fixtures setup
-                # or teardown
-                if user_exc:
-                    raise exc from user_exc
-                else:
-                    raise exc
-            finally:
-                # No matter what the nursery fixture should be closed when
-                # test is over
-                nursery.cancel_scope.cancel()
-
-        # Finally re-raise or original exception coming from the test if
-        # needed
-        if user_exc:
-            raise user_exc
-
-    _bootstrap_fixture_and_run_test._trio_test_runner_wrapped = True
-    return _bootstrap_fixture_and_run_test
-
-
-@asynccontextmanager
-@async_generator
-async def _setup_async_fixtures_in(deps):
-    __tracebackhide__ = True
-
-    need_resolved_deps_stack = [
-        (k, v) for k, v in deps.items() if isinstance(v, BaseAsyncFixture)
-    ]
-    if not ORDERED_DICTS:
-        # Make the fixture resolution order determinist
-        need_resolved_deps_stack = sorted(need_resolved_deps_stack)
-
-    if not need_resolved_deps_stack:
-        await yield_(deps)
-        return
-
-    @asynccontextmanager
-    @async_generator
-    async def _recursive_setup(deps_stack):
-        __tracebackhide__ = True
-        name, dep = deps_stack.pop()
-        async with dep.setup() as resolved:
-            if not deps_stack:
-                await yield_([(name, resolved)])
-            else:
-                async with _recursive_setup(deps_stack
-                                            ) as remains_deps_stack_resolved:
-                    await yield_(
-                        remains_deps_stack_resolved + [(name, resolved)]
-                    )
-
-    async with _recursive_setup(need_resolved_deps_stack
-                                ) as resolved_deps_stack:
-        await yield_({**deps, **dict(resolved_deps_stack)})
-
-
-class BaseAsyncFixture:
+# This is more complicated than you might expect.
+
+# The first complication is that all of pytest's machinery for setting up,
+# running a test, and then tearing it down again is synchronous. But we want
+# to have async setup, async tests, and async teardown.
+#
+# Our trick: from pytest's point of view, trio fixtures return an unevaluated
+# placeholder value, a TrioFixture object. This contains all the information
+# needed to do the actual setup/teardown, but doesn't actually perform these
+# operations.
+#
+# Then, pytest runs what it thinks of as "the test", we enter trio, and use
+# our own logic to setup the trio fixtures, run the actual test, and then tear
+# down the trio fixtures. This works pretty well, though it has some
+# limitations:
+# - trio fixtures have to be test-scoped
+# - normally pytest considers a fixture crash to be an ERROR, but when a trio
+#   fixture crashes, it gets classified as a FAIL.
+
+# The other major complication is that we really want to allow trio fixtures
+# to yield inside a nursery. (See gh-55 for more discussion.) And then while
+# the fixture function is suspended, a task inside that nursery might crash.
+#
+# Why is this a problem? Two reasons. First, a technical one: Trio's cancel
+# scope machinery assumes that it can inject a Cancelled exception into any
+# code inside the cancel scope, and that exception will eventually make its
+# way back to the 'with' block.
+#
+# A fixture that yields inside a nursery violates this rule: the cancel scope
+# remains "active" from when the fixture yields until when it's reentered, but
+# if a Cancelled exception is raised during this time, then it *won't* go into
+# the fixture. (And we can't throw it in there either, because that's just not
+# how pytest fixtures work. Whoops.)
+#
+# And second, our setup/test/teardown process needs to account for the
+# possibility that any fixture's background task might crash at any moment,
+# and do something sensible with it.
+#
+# You should think of fixtures as a dependency graph: each fixtures *uses*
+# zero or more other fixtures, and is *used by* zero or more other fixtures. A
+# fixture should be setup before any of the fixtures it's used by are setup,
+# and it should be torn down as soon as all of the fixtures that use it are
+# torn down. And at root of this dependency graph, we have the test itself,
+# which is just like a fixture except that instead of having a separate setup
+# and teardown phase, it runs straight through.
+#
+# To implement this, we isolate each fixture into its own task: this makes
+# sure that crashes in one can't trigger implicit cancellation in another.
+# Then we use trio.Event objects to implement the ordering described above.
+#
+# If a fixture crashes, whether during setup, teardown, or in a background
+# task at any other point, then we mark the whole test run as "crashed". When
+# a run is "crashed", two things happen: (1) if any fixtures or the test
+# itself haven't started yet, then we don't start them. (2) if the test is
+# running, we cancel it. That's all. In particular, if a fixture has a
+# background crash, we don't propagate that to any other fixtures, we still
+# follow the normal teardown sequence, and so on – but since the test is
+# cancelled, the teardown sequence should start immediately.
+
+canary = contextvars.ContextVar("pytest-trio canary")
+
+
+class TrioTestContext:
+    def __init__(self):
+        self.crashed = False
+        self.test_cancel_scope = None
+        self.error_list = []
+
+    def crash(self, exc):
+        if exc is not None:
+            self.error_list.append(exc)
+        self.crashed = True
+        if self.test_cancel_scope is not None:
+            self.test_cancel_scope.cancel()
+
+
+class TrioFixture:
     """
     Represent a fixture that need to be run in a trio context to be resolved.
+
+    The name is actually a misnomer, because we use it to represent the actual
+    test itself as well, since the test is basically just a fixture with no
+    dependents and no teardown.
     """
 
-    def __init__(self, fixturedef, deps={}):
-        self.fixturedef = fixturedef
-        self.deps = deps
-        self.setup_done = False
-        self.result = None
+    def __init__(self, name, func, pytest_kwargs, is_test=False):
+        self.name = name
+        self._func = func
+        self._pytest_kwargs = pytest_kwargs
+        self._is_test = is_test
+        self._teardown_done = trio.Event()
+
+        # These attrs are all accessed from other objects:
+        # Downstream users read this value.
+        self.fixture_value = None
+        # This event notifies downstream users that we're done setting up.
+        # Invariant: if this is set, then either fixture_value is usable *or*
+        # test_ctx.crashed is True.
+        self.setup_done = trio.Event()
+        # Downstream users *modify* this value, by adding their _teardown_done
+        # events to it, so we know who we need to wait for before tearing
+        # down.
+        self.user_done_events = set()
+
+    def register_and_collect_dependencies(self):
+        # Returns the set of all TrioFixtures that this fixture depends on,
+        # directly or indirectly, and sets up all their user_done_events.
+        deps = set()
+        deps.add(self)
+        for value in self._pytest_kwargs.values():
+            if isinstance(value, TrioFixture):
+                value.user_done_events.add(self._teardown_done)
+                deps.update(value.register_and_collect_dependencies())
+        return deps
 
     @asynccontextmanager
     @async_generator
-    async def setup(self):
+    async def _fixture_manager(self, test_ctx):
         __tracebackhide__ = True
-        if self.setup_done:
-            await yield_(self.result)
-        else:
-            async with _setup_async_fixtures_in(self.deps) as resolved_deps:
-                async with self._setup(resolved_deps) as self.result:
-                    self.setup_done = True
-                    await yield_(self.result)
-
-    async def _setup(self):
-        raise NotImplementedError()
-
-
-class AsyncYieldFixture(BaseAsyncFixture):
-    """
-    Async generator fixture.
-    """
+        try:
+            async with trio.open_nursery() as nursery_fixture:
+                try:
+                    await yield_(nursery_fixture)
+                finally:
+                    nursery_fixture.cancel_scope.cancel()
+        except BaseException as exc:
+            test_ctx.crash(exc)
+        finally:
+            self.setup_done.set()
+            self._teardown_done.set()
 
-    @asynccontextmanager
-    @async_generator
-    async def _setup(self, resolved_deps):
+    async def run(self, test_ctx, contextvars_ctx):
         __tracebackhide__ = True
-        agen = self.fixturedef.func(**resolved_deps)
 
-        try:
-            await yield_(await agen.asend(None))
-        finally:
-            try:
-                await agen.asend(None)
-            except StopAsyncIteration:
-                pass
+        # This is a gross hack. I guess Trio should provide a context=
+        # argument to start_soon/start?
+        task = trio.hazmat.current_task()
+        assert canary not in task.context
+        task.context = contextvars_ctx
+        # Force a yield so we pick up the new context
+        await trio.sleep(0)
+        # Check that it worked, since technically trio doesn't *guarantee*
+        # that sleep(0) will actually yield.
+        assert canary.get() == "in correct context"
+
+        # This 'with' block handles the nursery fixture lifetime, the
+        # teardone_done event, and crashing the context if there's an
+        # unhandled exception.
+        async with self._fixture_manager(test_ctx) as nursery_fixture:
+            # Resolve our kwargs
+            resolved_kwargs = {}
+            for name, value in self._pytest_kwargs.items():
+                if isinstance(value, TrioFixture):
+                    await value.setup_done.wait()
+                    if value.fixture_value is NURSERY_FIXTURE_PLACEHOLDER:
+                        resolved_kwargs[name] = nursery_fixture
+                    else:
+                        resolved_kwargs[name] = value.fixture_value
+                else:
+                    resolved_kwargs[name] = value
+
+            # If something's already crashed before we're ready to start, then
+            # there's no point in even setting up.
+            if test_ctx.crashed:
+                return
+
+            # Run actual fixture setup step
+            if self._is_test:
+                # Tests are exactly like fixtures, except that they (1) have
+                # to be regular async functions, (2) if there's a crash, we
+                # should cancel them.
+                assert not self.user_done_events
+                func_value = None
+                with trio.open_cancel_scope() as cancel_scope:
+                    test_ctx.test_cancel_scope = cancel_scope
+                    assert not test_ctx.crashed
+                    await self._func(**resolved_kwargs)
             else:
-                raise RuntimeError('Only one yield in fixture is allowed')
+                func_value = self._func(**resolved_kwargs)
+                if isinstance(func_value, Coroutine):
+                    self.fixture_value = await func_value
+                elif isasyncgen(func_value):
+                    self.fixture_value = await func_value.asend(None)
+                elif isinstance(func_value, Generator):
+                    self.fixture_value = func_value.send(None)
+                else:
+                    # Regular synchronous function
+                    self.fixture_value = func_value
+
+            # Notify our users that self.fixture_value is ready
+            self.setup_done.set()
+
+            # Wait for users to be finished
+            #
+            # At this point we're in a very strange state: if the fixture
+            # yielded inside a nursery or cancel scope, then we are still
+            # "inside" that scope even though its with block is not on the
+            # stack. In particular this means that if they get cancelled, then
+            # our waiting might get a Cancelled error, that we cannot really
+            # deal with – it should get thrown back into the fixture
+            # generator, but pytest fixture generators don't work that way:
+            #   https://github.com/python-trio/pytest-trio/issues/55
+            # And besides, we can't start tearing down until all our users
+            # have finished.
+            #
+            # So if we get an exception here, we crash the context (which
+            # cancels the test and starts the cleanup process), save any
+            # exception that *isn't* Cancelled (because if its Cancelled then
+            # we can't route it to the right place, and anyway the teardown
+            # code will get it again if it matters), and then use a shield to
+            # keep waiting for the teardown to finish without having to worry
+            # about cancellation.
+            try:
+                for event in self.user_done_events:
+                    await event.wait()
+            except BaseException as exc:
+                assert isinstance(exc, trio.Cancelled)
+                test_ctx.crash(None)
+                with trio.open_cancel_scope(shield=True):
+                    for event in self.user_done_events:
+                        await event.wait()
+
+            # Do our teardown
+            if isasyncgen(func_value):
+                try:
+                    await func_value.asend(None)
+                except StopAsyncIteration:
+                    pass
+                else:
+                    raise RuntimeError("too many yields in fixture")
+            elif isinstance(func_value, Generator):
+                try:
+                    func_value.send(None)
+                except StopIteration:
+                    pass
+                else:
+                    raise RuntimeError("too many yields in fixture")
 
 
-class SyncFixtureWithAsyncDeps(BaseAsyncFixture):
-    """
-    Synchronous function fixture with asynchronous dependencies fixtures.
-    """
+def _trio_test_runner_factory(item, testfunc=None):
+    testfunc = testfunc or item.function
 
-    @asynccontextmanager
-    @async_generator
-    async def _setup(self, resolved_deps):
+    if getattr(testfunc, '_trio_test_runner_wrapped', False):
+        # We have already wrapped this, perhaps because we combined Hypothesis
+        # with pytest.mark.parametrize
+        return testfunc
+
+    if not iscoroutinefunction(testfunc):
+        pytest.fail(
+            'test function `%r` is marked trio but is not async' % item
+        )
+
+    @trio_test
+    async def _bootstrap_fixtures_and_run_test(**kwargs):
         __tracebackhide__ = True
-        await yield_(self.fixturedef.func(**resolved_deps))
 
+        test_ctx = TrioTestContext()
+        test = TrioFixture(
+            "<test {!r}>".format(testfunc.__name__),
+            testfunc,
+            kwargs,
+            is_test=True
+        )
 
-class SyncYieldFixtureWithAsyncDeps(BaseAsyncFixture):
-    """
-    Synchronous generator fixture with asynchronous dependencies fixtures.
-    """
+        contextvars_ctx = contextvars.copy_context()
+        contextvars_ctx.run(canary.set, "in correct context")
 
-    @asynccontextmanager
-    @async_generator
-    async def _setup(self, resolved_deps):
-        __tracebackhide__ = True
-        gen = self.fixturedef.func(**resolved_deps)
+        async with trio.open_nursery() as nursery:
+            for fixture in test.register_and_collect_dependencies():
+                nursery.start_soon(
+                    fixture.run, test_ctx, contextvars_ctx, name=fixture.name
+                )
 
-        try:
-            await yield_(gen.send(None))
-        finally:
-            try:
-                gen.send(None)
-            except StopIteration:
-                pass
-            else:
-                raise RuntimeError('Only one yield in fixture is allowed')
+        if test_ctx.error_list:
+            raise trio.MultiError(test_ctx.error_list)
 
+    _bootstrap_fixtures_and_run_test._trio_test_runner_wrapped = True
+    return _bootstrap_fixtures_and_run_test
 
-class AsyncFixture(BaseAsyncFixture):
-    """
-    Regular async fixture (i.e. coroutine).
-    """
 
-    @asynccontextmanager
-    @async_generator
-    async def _setup(self, resolved_deps):
-        __tracebackhide__ = True
-        await yield_(await self.fixturedef.func(**resolved_deps))
-
-
-def _install_async_fixture_if_needed(fixturedef, request):
-    asyncfix = None
-    deps = {dep: request.getfixturevalue(dep) for dep in fixturedef.argnames}
-    if iscoroutinefunction(fixturedef.func):
-        asyncfix = AsyncFixture(fixturedef, deps)
-    elif isasyncgenfunction(fixturedef.func):
-        asyncfix = AsyncYieldFixture(fixturedef, deps)
-    elif any(isinstance(dep, BaseAsyncFixture) for dep in deps.values()):
-        if isgeneratorfunction(fixturedef.func):
-            asyncfix = SyncYieldFixtureWithAsyncDeps(fixturedef, deps)
-        else:
-            asyncfix = SyncFixtureWithAsyncDeps(fixturedef, deps)
-    if asyncfix:
-        fixturedef.cached_result = (asyncfix, request.param_index, None)
-        return asyncfix
+################################################################
+# Hooking up the test/fixture machinery to pytest
+################################################################
 
 
 @pytest.hookimpl(hookwrapper=True)
 def pytest_runtest_call(item):
-    if 'trio' in item.keywords:
+    if item.get_closest_marker("trio") is not None:
         if hasattr(item.obj, 'hypothesis'):
             # If it's a Hypothesis test, we go in a layer.
             item.obj.hypothesis.inner_test = _trio_test_runner_factory(
@@ -267,10 +356,52 @@ def pytest_runtest_call(item):
     yield
 
 
-@pytest.hookimpl()
+# It's intentionally impossible to use this to create a non-function-scoped
+# fixture (since that would require exposing a way to pass scope= to
+# pytest.fixture).
+def trio_fixture(func):
+    func._force_trio_fixture = True
+    return pytest.fixture(func)
+
+
+def _is_trio_fixture(func, coerce_async, kwargs):
+    if getattr(func, "_force_trio_fixture", False):
+        return True
+    if (coerce_async and
+        (iscoroutinefunction(func) or isasyncgenfunction(func))):
+        return True
+    if any(isinstance(value, TrioFixture) for value in kwargs.values()):
+        return True
+    return False
+
+
+def handle_fixture(fixturedef, request, force_trio_mode):
+    is_trio_test = (request.node.get_closest_marker("trio") is not None)
+    if force_trio_mode:
+        is_trio_mode = True
+    else:
+        is_trio_mode = request.node.config.getini("trio_mode")
+    coerce_async = (is_trio_test or is_trio_mode)
+    kwargs = {
+        name: request.getfixturevalue(name)
+        for name in fixturedef.argnames
+    }
+    if _is_trio_fixture(fixturedef.func, coerce_async, kwargs):
+        if request.scope != "function":
+            raise RuntimeError("Trio fixtures must be function-scope")
+        if not is_trio_test:
+            raise RuntimeError("Trio fixtures can only be used by Trio tests")
+        fixture = TrioFixture(
+            "<fixture {!r}>".format(fixturedef.argname),
+            fixturedef.func,
+            kwargs,
+        )
+        fixturedef.cached_result = (fixture, request.param_index, None)
+        return fixture
+
+
 def pytest_fixture_setup(fixturedef, request):
-    if 'trio' in request.keywords:
-        return _install_async_fixture_if_needed(fixturedef, request)
+    return handle_fixture(fixturedef, request, force_trio_mode=False)
 
 
 ################################################################
@@ -298,6 +429,10 @@ def pytest_collection_modifyitems(config, items):
 ################################################################
 
 
+class NURSERY_FIXTURE_PLACEHOLDER:
+    pass
+
+
 @pytest.fixture
 def mock_clock():
     return MockClock()
@@ -308,6 +443,6 @@ def autojump_clock():
     return MockClock(autojump_threshold=0)
 
 
-@pytest.fixture
-async def nursery(request):
-    return request.node._trio_nursery
+@trio_fixture
+def nursery(request):
+    return NURSERY_FIXTURE_PLACEHOLDER
diff --git a/setup.py b/setup.py
index 03a577e..6b25497 100644
--- a/setup.py
+++ b/setup.py
@@ -18,6 +18,8 @@
     install_requires=[
         "trio",
         "async_generator >= 1.9",
+        # For node.get_closest_marker
+        "pytest >= 3.6"
     ],
     keywords=[
         'async',
diff --git a/test-requirements.txt b/test-requirements.txt
index ffdc000..d2ac8c8 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -1,3 +1,4 @@
-pytest
+pytest !=3.7.0, !=3.7.1  # https://github.com/python-trio/pytest-trio/pull/50#issuecomment-413124393
 pytest-cov
 hypothesis>=3.64
+trio-asyncio