From 3e9fdb40d5170b228d9e3ce76cad0a365f393865 Mon Sep 17 00:00:00 2001
From: Juliya Smith <jules@apeworx.io>
Date: Tue, 5 Nov 2024 08:58:36 -0600
Subject: [PATCH 1/3] fix: test isolation

---
 docs/userguides/testing.md                    |  79 ++-
 src/ape/__init__.py                           |   6 +
 src/ape/exceptions.py                         |   2 +-
 src/ape/managers/chain.py                     |  10 +-
 src/ape/pytest/config.py                      |  12 +
 src/ape/pytest/fixtures.py                    | 642 ++++++++++++++++--
 src/ape/pytest/plugin.py                      |  20 +-
 src/ape/pytest/runners.py                     | 133 +++-
 src/ape/pytest/utils.py                       |  16 +
 src/ape/pytest/warnings.py                    |  20 +
 src/ape_test/provider.py                      |  19 +-
 tests/functional/test_exceptions.py           |  13 +
 tests/functional/test_fixtures.py             | 115 +++-
 tests/functional/test_plugins.py              |   2 +-
 tests/functional/test_project.py              |   6 +-
 tests/functional/test_provider.py             |  20 +
 tests/functional/test_test.py                 | 351 +++++++++-
 .../cli/projects/test/tests/conftest.py       |  11 +
 .../test/tests/test_fixture_isolation.py      | 166 +++++
 .../tests/test_fixture_isolation_session.py   |  24 +
 tests/integration/cli/test_test.py            |  14 +-
 21 files changed, 1573 insertions(+), 108 deletions(-)
 create mode 100644 src/ape/pytest/utils.py
 create mode 100644 src/ape/pytest/warnings.py
 create mode 100644 tests/integration/cli/projects/test/tests/conftest.py
 create mode 100644 tests/integration/cli/projects/test/tests/test_fixture_isolation_session.py

diff --git a/docs/userguides/testing.md b/docs/userguides/testing.md
index 439b662111..81e7e9c198 100644
--- a/docs/userguides/testing.md
+++ b/docs/userguides/testing.md
@@ -70,11 +70,6 @@ def test_authorization(my_contract, owner, not_owner):
         my_contract.authorized_method(sender=not_owner)
 ```
 
-```{note}
-Ape has built-in test and fixture isolation for all pytest scopes.
-To disable isolation add the `--disable-isolation` flag when running `ape test`
-```
-
 ## Fixtures
 
 Now that we have discussed the full flow of a test, let's dive deeper into the specific parts, starting with `pytest.fixtures`.
@@ -200,12 +195,10 @@ You also have access to the `project` you are testing. You will need this to dep
 ```python
 import pytest
 
-
 @pytest.fixture
 def owner(accounts):
     return accounts[0]
 
-
 @pytest.fixture
 def my_contract(project, owner):
     #           ^ use the 'project' fixture from the 'ape-test' plugin
@@ -226,6 +219,72 @@ def my_contract(Contract):
 
 It has the same interface as the [ChainManager](../methoddocs/managers.html#ape.managers.chain.ChainManager).
 
+## Isolation
+
+By default, tests run with chain-isolation.
+This means, at the start of each test, a snapshot is taken.
+After each test completes, the chain reverts to that snapshot from the beginning of the test.
+
+By default, every `pytest` fixture is `function` scoped, meaning it will be replayed each time it is requested (no result-caching).
+For example, if you deploy a contract in a function-scoped fixture, it will be re-deployed each time the fixture gets used in your tests.
+To only deploy once, you can use different scopes, such as `"session"`, `"package"`, `"module"`, or `"class"`, and you **must** use these fixtures right away, either via `autouse=True` or using them in the first collected tests.
+Otherwise, higher-scoped fixtures that arrive late in a Pytest session will cause the snapshotting system to have to rebase itself, which can be costly.
+For example, if you define a session scoped fixture that deploys a contract and makes transactions, the state changes from those transactions remain in subsequent tests, whether those tests use that fixture or not.
+However, if a new fixture of a session scope comes into play after module, package, or class scoped snapshots have already been taken, those lower-scoped fixtures are now invalid and have to re-run after the session fixture to ensure the session fixture remains in the session-snapshot.
+
+In the following example, the `my_contract` fixture gets deployed upon its first usage, which happens in the test `test_my_contract_0()`.
+During the test `test_something_else()`, it may not have been deployed yet, as it was not requested, and it is defined before the other tests.
+Then, during `test_my_contract_1()`, instead of deploying again, it uses the cached result from the session-scoped fixture and the chain still has it in its state because the fixture is session-scoped and runs before the test-isolation.
+
+```python
+import pytest
+
+@pytest.fixture(scope="session")
+def my_contract(accounts, project):
+    owner = accounts[0]
+    contract = project.MyContract.deploy(sender=owner)
+    # Can also do stateful transactions in a session-scoped fixture.
+    contract.initialize(sender=owner)
+    return contract
+
+def test_something_else():
+    ...
+
+def test_my_contract_0(my_contract):
+    my_contract.myMethod()
+
+def test_my_contract_1(my_contract):
+    my_contract.myMethod()
+```
+
+To disable isolation, run `ape test` with the `--disable-isolation` flag.
+When isolation is disabled, the blockchain's state persists as the tests run.
+This will be more performant and less complex, but will also cause non-deterministic results in your tests as each test inherits the state of whatever was run before it.
+
+This may be further complicated when running with other pytest plugins such as `pytest-xdist` or `pytest-split` which re-arranges the order that tests are executed in (not recommended to use these plugins together with ape until more proper integrations are developed).
+
+```shell
+ape test --disable-isolation
+```
+
+```{warning}
+Be mindful if, when, and how you define non-function scoped fixtures.
+Pytest activates fixtures in the order they are used.
+If a session scoped fixture comes into play after package, module, or class scoped fixtures, the isolation logic has to invalidate each of those scopes and replay them after the session scoped, which causes any benefits of package, module, or class scopes to be void.
+If you are using higher-scoped fixtures for parametrized fixtures with lower-scoped fixtures, each itertion of the parametried fixture invalidates the lower-level fixtures each time, rendering everything to behave as function scoped until the end of the parametrized fixtures first run-through.
+```
+
+If you are using chain-isolation and have a higher-scoped fixture that you know is for-sure not chain-altering, you can use `ape.fixture` and the `chain_isolation` flag, and it may improve performance:
+
+```python
+import ape
+from ape_tokens import tokens
+
+@ape.fixture(scope="session", chain_isolation=False, params=("WETH", "DAI", "BAT"))
+def token_addresses(request):
+    return tokens[request].address
+```
+
 ## Ape testing commands
 
 ```bash
@@ -345,7 +404,8 @@ You may also supply an `re.Pattern` object to assert on a message pattern, rathe
 import ape
 import re
 
-# Matches explicitly "foo" or "bar"
+# Matches
+# "foo" or "bar"
 with ape.reverts(re.compile(r"^(foo|bar)$")):
     ...
 ```
@@ -396,8 +456,9 @@ You may also supply an `re.Pattern` object to assert on a dev message pattern, r
 
 ```python
 import ape
+import re
 
-# Matches explictly "dev: foo" or "dev: bar"
+# Matches "dev: foo" or "dev: bar"
 with ape.reverts(dev_message=re.compile(r"^dev: (foo|bar)$")):
     ...
 ```
diff --git a/src/ape/__init__.py b/src/ape/__init__.py
index efb1345297..3372ba2ff2 100644
--- a/src/ape/__init__.py
+++ b/src/ape/__init__.py
@@ -14,6 +14,7 @@
     "config",
     "convert",
     "Contract",
+    "fixture",
     "networks",
     "project",
     "Project",  # So you can load other projects
@@ -30,6 +31,11 @@ def __getattr__(name: str):
 
         return RevertsContextManager
 
+    elif name == "fixture":
+        from ape.pytest.fixtures import fixture
+
+        return fixture
+
     else:
         from ape.utils.basemodel import ManagerAccessMixin as access
 
diff --git a/src/ape/exceptions.py b/src/ape/exceptions.py
index 14e4aed00a..aabdc51ae3 100644
--- a/src/ape/exceptions.py
+++ b/src/ape/exceptions.py
@@ -625,7 +625,7 @@ def __init__(self, snapshot_id: "SnapshotID"):
             # Is block hash
             snapshot_id = humanize_hash(cast(Hash32, snapshot_id))
 
-        super().__init__(f"Unknown snapshot ID '{str(snapshot_id)}'.")
+        super().__init__(f"Unknown snapshot ID '{snapshot_id}'.")
 
 
 class QueryEngineError(ApeException):
diff --git a/src/ape/managers/chain.py b/src/ape/managers/chain.py
index 4c14bad123..2211d4c261 100644
--- a/src/ape/managers/chain.py
+++ b/src/ape/managers/chain.py
@@ -28,6 +28,7 @@
 from ape.contracts import ContractContainer, ContractInstance
 from ape.exceptions import (
     APINotImplementedError,
+    BlockNotFoundError,
     ChainError,
     ContractNotFoundError,
     ConversionError,
@@ -75,10 +76,15 @@ def height(self) -> int:
         """
         The latest block number.
         """
-        if self.head.number is None:
+        try:
+            head = self.head
+        except BlockNotFoundError:
+            return 0
+
+        if head.number is None:
             raise ChainError("Latest block has no number.")
 
-        return self.head.number
+        return head.number
 
     @property
     def network_confirmations(self) -> int:
diff --git a/src/ape/pytest/config.py b/src/ape/pytest/config.py
index a60cf15fd9..1a83963a7d 100644
--- a/src/ape/pytest/config.py
+++ b/src/ape/pytest/config.py
@@ -27,6 +27,18 @@ class ConfigWrapper(ManagerAccessMixin):
 
     def __init__(self, pytest_config: "PytestConfig"):
         self.pytest_config = pytest_config
+        if not self.verbosity:
+            # Enable verbose output if stdout capture is disabled
+            self.verbosity = self.pytest_config.getoption("capture") == "no"
+        # else: user has already changes verbosity to an equal or higher level; avoid downgrading.
+
+    @property
+    def verbosity(self) -> int:
+        return self.pytest_config.option.verbose
+
+    @verbosity.setter
+    def verbosity(self, value):
+        self.pytest_config.option.verbose = value
 
     @cached_property
     def supports_tracing(self) -> bool:
diff --git a/src/ape/pytest/fixtures.py b/src/ape/pytest/fixtures.py
index 925a10d903..563278aad1 100644
--- a/src/ape/pytest/fixtures.py
+++ b/src/ape/pytest/fixtures.py
@@ -1,12 +1,17 @@
-from collections.abc import Iterator
+import inspect
+import re
+from collections import defaultdict
+from collections.abc import Iterable, Iterator, Mapping
+from dataclasses import dataclass, field
 from fnmatch import fnmatch
 from functools import cached_property
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, ClassVar, Optional
 
 import pytest
 from eth_utils import to_hex
+from rich import print as rich_print
 
-from ape.exceptions import BlockNotFoundError, ChainError
+from ape.exceptions import BlockNotFoundError, ChainError, ProviderNotConnectedError
 from ape.logging import logger
 from ape.utils.basemodel import ManagerAccessMixin
 from ape.utils.rpc import allow_disconnected
@@ -18,29 +23,423 @@
     from ape.managers.networks import NetworkManager
     from ape.managers.project import ProjectManager
     from ape.pytest.config import ConfigWrapper
+    from ape.pytest.utils import Scope
     from ape.types.vm import SnapshotID
 
 
-class PytestApeFixtures(ManagerAccessMixin):
-    # NOTE: Avoid including links, markdown, or rst in method-docs
-    # for fixtures, as they are used in output from the command
-    # `ape test -q --fixture` (`pytest -q --fixture`).
+@dataclass()
+class FixtureRebase:
+    return_scope: "Scope"
+    invalid_fixtures: dict["Scope", list[str]]
 
-    _supports_snapshot: bool = True
-    receipt_capture: "ReceiptCapture"
 
-    def __init__(self, config_wrapper: "ConfigWrapper", receipt_capture: "ReceiptCapture"):
+class FixtureManager(ManagerAccessMixin):
+    _builtin_fixtures: ClassVar[list] = []
+    _stateful_fixtures_cache: ClassVar[dict[str, bool]] = {}
+    _ISOLATION_FIXTURE_REGEX = re.compile(r"_(session|package|module|class|function)_isolation")
+
+    def __init__(self, config_wrapper: "ConfigWrapper", isolation_manager: "IsolationManager"):
         self.config_wrapper = config_wrapper
-        self.receipt_capture = receipt_capture
+        self.isolation_manager = isolation_manager
+        self._nodeid_to_fixture_map: dict[str, "FixtureMap"] = {}
+        self._fixture_name_to_info: dict[str, dict] = {}
+
+    @classmethod
+    def set_builtins(cls, fixture_map: "FixtureMap"):
+        cls._builtin_fixtures = [
+            n
+            for n, defs in fixture_map._arg2fixturedefs.items()
+            if any("pytest" in fixture.func.__module__ for fixture in defs)
+        ]
 
     @cached_property
-    def _track_transactions(self) -> bool:
+    def _ape_fixtures(self) -> tuple[str, ...]:
+        return tuple(
+            [
+                n
+                for n, itm in inspect.getmembers(PytestApeFixtures)
+                if callable(itm) and not n.startswith("_")
+            ]
+        )
+
+    @property
+    def builtin_fixtures(self) -> list[str]:
+        return self._builtin_fixtures
+
+    def is_builtin(self, name: str) -> bool:
+        return name in self.builtin_fixtures
+
+    @classmethod
+    def is_isolation(cls, name: str) -> bool:
+        return bool(re.match(cls._ISOLATION_FIXTURE_REGEX, name))
+
+    def is_ape(self, name: str) -> bool:
+        return name in self._ape_fixtures
+
+    def is_custom(self, name) -> bool:
+        return not self.is_builtin(name) and not self.is_ape(name) and not self.is_isolation(name)
+
+    def get_fixtures(self, item) -> "FixtureMap":
+        if isinstance(item, str):
+            # Cached map referenced where only have nodeid.
+            if fixture_map := self._get_cached_fixtures(item):
+                return fixture_map
+
+            raise KeyError(f"No item found with nodeid '{item}'.")
+
+        elif fixture_map := self._get_cached_fixtures(item.nodeid):
+            # Cached map.
+            return fixture_map
+
+        return self.cache_fixtures(item)
+
+    def cache_fixtures(self, item) -> "FixtureMap":
+        fixture_map = FixtureMap.from_test_item(item)
+        if not FixtureManager._builtin_fixtures:
+            FixtureManager.set_builtins(fixture_map)
+
+        self._nodeid_to_fixture_map[item.nodeid] = fixture_map
+        for scope, fixture_set in fixture_map.items():
+            for fixture_name in fixture_set:
+                if fixture_name not in self._fixture_name_to_info:
+                    self._fixture_name_to_info[fixture_name] = {"scope": scope}
+
+        return fixture_map
+
+    def get_fixture_scope(self, fixture_name: str) -> Optional["Scope"]:
+        return self._fixture_name_to_info.get(fixture_name, {}).get("scope")
+
+    def is_stateful(self, name: str) -> Optional[bool]:
+        if name in self._stateful_fixtures_cache:
+            # Used `@ape.fixture(chain_isolation=<bool>)
+            # Or we already calculated.
+            return self._stateful_fixtures_cache[name]
+
+        try:
+            is_auto_mine = self.provider.auto_mine
+        except (NotImplementedError, ProviderNotConnectedError):
+            # Assume it's on since it can't be turned off.
+            is_auto_mine = True
+
+        if not is_auto_mine:
+            # When auto-mine is disabled, it's unknown.
+            return None
+
+        elif not (info := self._fixture_name_to_info.get(name)):
+            # Statefulness not yet tracked. Unknown.
+            return None
+
+        setup_block = info.get("setup_block")
+        teardown_block = info.get("teardown_block")
+        if setup_block is None or teardown_block is None:
+            # Blocks no set. Unknown.
+            return None
+
+        # If the two are not equal, state has changed.
+        is_stateful = setup_block != teardown_block
+        self._stateful_fixtures_cache[name] = is_stateful
+
+        # Clear out blocks since they are no longer needed.
+        self._fixture_name_to_info[name] = {
+            k: v
+            for k, v in self._fixture_name_to_info[name].items()
+            if k not in ("setup_block", "teardown_block")
+        }
+
+        return is_stateful
+
+    def add_fixture_info(self, name: str, **info):
+        if name not in self._fixture_name_to_info:
+            self._fixture_name_to_info[name] = info
+        else:
+            self._fixture_name_to_info[name] = {
+                **self._fixture_name_to_info[name],
+                **info,
+            }
+
+    def _get_cached_fixtures(self, nodeid: str) -> Optional["FixtureMap"]:
+        return self._nodeid_to_fixture_map.get(nodeid)
+
+    def rebase(self, scope: "Scope", fixtures: "FixtureMap"):
+        if not (rebase := self._get_rebase(scope)):
+            # Rebase avoided: nothing would change.
+            return
+
+        from ape.pytest.warnings import warn_invalid_isolation
+
+        warn_invalid_isolation()
+        self.isolation_manager.restore(rebase.return_scope)
+
+        # Invalidate fixtures by clearing out their cached result.
+        invalidated = []
+        for invalid_scope, invalid_fixture_ls in rebase.invalid_fixtures.items():
+            for invalid_fixture in invalid_fixture_ls:
+                info_ls = fixtures.get_info(invalid_fixture)
+                for info in info_ls:
+                    if self.is_stateful(info.argname) is False:
+                        # It has been determined that this fixture is not stateful.
+                        continue
+
+                    info.cached_result = None
+                    invalidated.append(info.argname)
+
+            # Also, invalidate the corresponding isolation fixture.
+            if invalid_isolation_fixture_ls := fixtures.get_info(
+                invalid_scope.isolation_fixturename
+            ):
+                for invalid_isolation_fixture in invalid_isolation_fixture_ls:
+                    invalid_isolation_fixture.cached_result = None
+                    invalidated.append(invalid_isolation_fixture.argname)
+
+            if invalidated and self.config_wrapper.verbosity:
+                log = "rebase"
+                if rebase.return_scope is not None:
+                    log = f"{log} scope={rebase.return_scope}"
+
+                log = f"{log} invalidated-fixtures='{', '.join(invalidated)}'"
+                self.isolation_manager._records.append(log)
+
+    def _get_rebase(self, scope: "Scope") -> Optional[FixtureRebase]:
+        # Check for fixtures that are now invalid. For example, imagine a session
+        # fixture comes into play after the module snapshot has been set.
+        # Once we restore the module's state and move to the next module,
+        # that session fixture will no longer exist. To remedy this situation,
+        # we invalidate the lower-scoped fixtures and re-snapshot everything.
+        scope_to_revert = None
+        invalids = defaultdict(list)
+        for next_snapshot in self.isolation_manager.next_snapshots(scope):
+            if next_snapshot.identifier is None:
+                # Thankfully, we haven't reached this scope yet.
+                # In this case, things are running in a performant order.
+                continue
+
+            if scope_to_revert is None:
+                # Revert to the closest scope to use. For example, a new
+                # session comes in but we have already calculated a module
+                # and a class, revert to pre-module and invalidate the module
+                # and class fixtures.
+                scope_to_revert = next_snapshot.scope
+
+            # All stateful fixtures downward are "below scope"
+            fixtures = [f for f in next_snapshot.fixtures if self.is_stateful(f) is not False]
+            invalids[next_snapshot.scope].extend(fixtures)
+
+        invalids_dict = dict(invalids)
         return (
-            self.network_manager.provider is not None
-            and self.provider.is_connected
-            and (self.config_wrapper.track_gas or self.config_wrapper.track_coverage)
+            FixtureRebase(return_scope=scope_to_revert, invalid_fixtures=invalids_dict)
+            if scope_to_revert is not None and any(len(ls) > 0 for ls in invalids_dict.values())
+            else None
+        )
+
+
+class FixtureMap(dict["Scope", list[str]]):
+    def __init__(self, item):
+        from ape.pytest.utils import Scope
+
+        self._item = item
+        self._parametrized_names: Optional[list[str]] = None
+        super().__init__(
+            {
+                Scope.SESSION: [],
+                Scope.PACKAGE: [],
+                Scope.MODULE: [],
+                Scope.CLASS: [],
+                Scope.FUNCTION: [],
+            }
         )
 
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} {self._item.nodeid}>"
+
+    @classmethod
+    def from_test_item(cls, item) -> "FixtureMap":
+        obj = cls(item)
+        for name, info_ls in obj._arg2fixturedefs.items():
+            if not info_ls or name not in item.fixturenames:
+                continue
+
+            for info in info_ls:
+                obj[info.scope].append(name)
+
+        return obj
+
+    @property
+    def names(self) -> list[str]:
+        """
+        Outputs in correct order for item.fixturenames.
+        Also, injects isolation fixtures if needed.
+        """
+        from ape.pytest.utils import Scope
+
+        result = []
+        for scope, ls in self.items():
+            # NOTE: For function scoped, we always add the isolation fixture.
+            if not ls and scope is not Scope.FUNCTION:
+                continue
+
+            result.append(scope.isolation_fixturename)
+            result.extend(ls)
+
+        return result
+
+    @property
+    def parameters(self) -> list[str]:
+        """
+        Test-parameters (not fixtures!)
+        """
+        return [n for n in self._item.fixturenames if n not in self._arg2fixturedefs]
+
+    @property
+    def isolation(self) -> list[str]:
+        return [n.lstrip("_").split("_")[0] for n in self.names if FixtureManager.is_isolation(n)]
+
+    @property
+    def parametrized(self) -> dict[str, list]:
+        if self._parametrized_names is not None:
+            # We have already done this.
+            return {
+                n: ls for n, ls in self._arg2fixturedefs.items() if n in self._parametrized_names
+            }
+
+        # Calculating for first time.
+        self._parametrized_names = []
+        result: dict[str, list] = {}
+        for name, info_ls in self._arg2fixturedefs.items():
+            if name not in self._item.fixturenames or not any(info.params for info in info_ls):
+                continue
+
+            self._parametrized_names.append(name)
+            result[name] = info_ls
+
+        return result
+
+    @property
+    def _arg2fixturedefs(self) -> Mapping:
+        return self._item.session._fixturemanager._arg2fixturedefs
+
+    def __setitem__(self, key, value):
+        # NOTE: Not using singledispatchmethod because it requires
+        #  types at runtime.
+        if isinstance(key, Scope):
+            self.__setitem_scope(key, value)
+        elif isinstance(key, str):
+            self.__setitem_str(key, value)
+        elif isinstance(key, int):
+            self.__setitem_int(key, value)
+
+        raise NotImplementedError(type(key))
+
+    def __setitem_int(self, key: int, value: list[str]):
+        super().__setitem__(Scope(key), value)
+
+    def __setitem_str(self, key: str, value: list[str]):
+        for scope in Scope:
+            if f"{scope}" == key:
+                super().__setitem__(scope, value)
+                return
+
+        raise KeyError(key)
+
+    def __setitem_scope(self, key: "Scope", value: list[str]):
+        super().__setitem__(key, value)
+
+    def __getitem__(self, key):
+        # NOTE: Not using singledispatchmethod because it requires
+        #  types at runtime.
+        from ape.pytest.utils import Scope
+
+        if isinstance(key, Scope):
+            return self.__getitem_scope(key)
+        elif isinstance(key, str):
+            return self.__getitem_str(key)
+        elif isinstance(key, int):
+            return self.__getitem_int(key)
+
+        raise NotImplementedError(type(key))
+
+    def __getitem_int(self, key: int) -> list[str]:
+        from ape.pytest.utils import Scope
+
+        return super().__getitem__(Scope(key))
+
+    def __getitem_str(self, key: str) -> list[str]:
+        from ape.pytest.utils import Scope
+
+        for scope in Scope:
+            if f"{scope}" == key:
+                return super().__getitem__(scope)
+
+        raise KeyError(key)
+
+    def __getitem_scope(self, key: "Scope") -> list[str]:
+        return super().__getitem__(key)
+
+    def get_info(self, name: str) -> list:
+        """
+        Get fixture info.
+
+        Args:
+            name (str):
+
+        Returns:
+            list of info
+        """
+        if name not in self._arg2fixturedefs:
+            return []
+
+        return self._arg2fixturedefs[name]
+
+    def is_known(self, name: str) -> bool:
+        """
+        True when fixture-info is known for the given fixture name.
+        """
+        return name in self._arg2fixturedefs
+
+    def is_iterating(self, name: str) -> bool:
+        """
+        True when is a non-function scoped parametrized fixture that hasn't
+        fully iterated.
+        """
+        if name not in self.parametrized:
+            return False
+
+        elif not (info_ls := self.get_info(name)):
+            return False
+
+        for info in info_ls:
+            if not info.params:
+                continue
+
+            if not info.cached_result:
+                return True  # First iteration
+
+            elif len(info.cached_result) < 2:
+                continue  # ?
+
+            last_param_ran = info.cached_result[1]
+            last_param = info.params[-1]
+            if last_param_ran != last_param:
+                return True  # Is iterating.
+
+        return False
+
+    def apply_fixturenames(self):
+        """
+        Set the fixturenames on the test item in the order they should be used.
+        Carefully ignore non-fixtures, such as keys from parametrized tests.
+        """
+        self._item.fixturenames = [*self.names, *self.parameters]
+
+
+class PytestApeFixtures(ManagerAccessMixin):
+    # NOTE: Avoid including links, markdown, or rst in method-docs
+    # for fixtures, as they are used in output from the command
+    # `ape test -q --fixture` (`pytest -q --fixture`).
+
+    def __init__(self, config_wrapper: "ConfigWrapper", isolation_manager: "IsolationManager"):
+        self.config_wrapper = config_wrapper
+        self.isolation_manager = isolation_manager
+
     @pytest.fixture(scope="session")
     def accounts(self) -> list["TestAccountAPI"]:
         """
@@ -84,19 +483,125 @@ def Contract(self):
         """
         return self.chain_manager.contracts.instance_at
 
-    def _isolation(self) -> Iterator[None]:
+    @pytest.fixture(scope="session")
+    def _session_isolation(self) -> Iterator[None]:
+        from ape.pytest.utils import Scope
+
+        yield from self.isolation_manager.isolation(Scope.SESSION)
+
+    @pytest.fixture(scope="package")
+    def _package_isolation(self) -> Iterator[None]:
+        from ape.pytest.utils import Scope
+
+        yield from self.isolation_manager.isolation(Scope.PACKAGE)
+
+    @pytest.fixture(scope="module")
+    def _module_isolation(self) -> Iterator[None]:
+        from ape.pytest.utils import Scope
+
+        yield from self.isolation_manager.isolation(Scope.MODULE)
+
+    @pytest.fixture(scope="class")
+    def _class_isolation(self) -> Iterator[None]:
+        from ape.pytest.utils import Scope
+
+        yield from self.isolation_manager.isolation(Scope.CLASS)
+
+    @pytest.fixture(scope="function")
+    def _function_isolation(self) -> Iterator[None]:
+        from ape.pytest.utils import Scope
+
+        yield from self.isolation_manager.isolation(Scope.FUNCTION)
+
+
+@dataclass
+class Snapshot:
+    """
+    All the data necessary for accurately supporting isolation.
+    """
+
+    scope: "Scope"
+    """Corresponds to fixture scope."""
+
+    identifier: Optional["SnapshotID"] = None
+    """Snapshot ID taken before the peer-fixtures in the same scope."""
+
+    fixtures: list = field(default_factory=list)
+    """All peer fixtures, tracked so we know when new ones are added."""
+
+    def append_fixtures(self, fixtures: Iterable[str]):
+        for fixture in fixtures:
+            if fixture in self.fixtures:
+                continue
+
+            self.fixtures.append(fixture)
+
+
+class SnapshotRegistry(dict["Scope", Snapshot]):
+    def __init__(self):
+        from ape.pytest.utils import Scope
+
+        super().__init__(
+            {
+                Scope.SESSION: Snapshot(Scope.SESSION),
+                Scope.PACKAGE: Snapshot(Scope.PACKAGE),
+                Scope.MODULE: Snapshot(Scope.MODULE),
+                Scope.CLASS: Snapshot(Scope.CLASS),
+                Scope.FUNCTION: Snapshot(Scope.FUNCTION),
+            }
+        )
+
+    def get_snapshot_id(self, scope: "Scope") -> Optional["SnapshotID"]:
+        return self[scope].identifier
+
+    def set_snapshot_id(self, scope: "Scope", snapshot_id: "SnapshotID"):
+        self[scope].identifier = snapshot_id
+
+    def clear_snapshot_id(self, scope: "Scope"):
+        self[scope].identifier = None
+
+    def next_snapshots(self, scope: "Scope") -> Iterator[Snapshot]:
+        from ape.pytest.utils import Scope
+
+        for scope_value in range(scope + 1, Scope.FUNCTION + 1):
+            yield self[scope_value]  # type: ignore
+
+    def extend_fixtures(self, scope: "Scope", fixtures: Iterable[str]):
+        self[scope].fixtures.extend(fixtures)
+
+
+class IsolationManager(ManagerAccessMixin):
+    supported: bool = True
+    snapshots: SnapshotRegistry = SnapshotRegistry()
+
+    def __init__(self, config_wrapper: "ConfigWrapper", receipt_capture: "ReceiptCapture"):
+        self.config_wrapper = config_wrapper
+        self.receipt_capture = receipt_capture
+        self._records: list[str] = []
+
+    @cached_property
+    def _track_transactions(self) -> bool:
+        return (
+            self.network_manager.provider is not None
+            and self.provider.is_connected
+            and (self.config_wrapper.track_gas or self.config_wrapper.track_coverage)
+        )
+
+    def get_snapshot(self, scope: "Scope") -> Snapshot:
+        return self.snapshots[scope]
+
+    def extend_fixtures(self, scope: "Scope", fixtures: Iterable[str]):
+        self.snapshots.extend_fixtures(scope, fixtures)
+
+    def next_snapshots(self, scope: "Scope") -> Iterator[Snapshot]:
+        yield from self.snapshots.next_snapshots(scope)
+
+    def isolation(self, scope: "Scope") -> Iterator[None]:
         """
         Isolation logic used to implement isolation fixtures for each pytest scope.
         When tracing support is available, will also assist in capturing receipts.
         """
-        snapshot_id = None
-
-        if self._supports_snapshot:
-            try:
-                snapshot_id = self._snapshot()
-            except BlockNotFoundError:
-                self._supports_snapshot = False
-
+        self.set_snapshot(scope)
         if self._track_transactions:
             did_yield = False
             try:
@@ -108,22 +613,34 @@ def _isolation(self) -> Iterator[None]:
                 if not did_yield:
                     # Prevent double yielding.
                     yield
-
         else:
             yield
 
-        if snapshot_id is not None:
-            self._restore(snapshot_id)
+        # NOTE: self._supported may have gotten set to False
+        #   someplace else _after_ snapshotting succeeded.
+        if not self.supported:
+            return
 
-    # isolation fixtures
-    _session_isolation = pytest.fixture(_isolation, scope="session")
-    _package_isolation = pytest.fixture(_isolation, scope="package")
-    _module_isolation = pytest.fixture(_isolation, scope="module")
-    _class_isolation = pytest.fixture(_isolation, scope="class")
-    _function_isolation = pytest.fixture(_isolation, scope="function")
+        self.restore(scope)
+
+    def set_snapshot(self, scope: "Scope"):
+        # Also can be used to re-set snapshot.
+        if not self.supported:
+            return
+
+        if self.config_wrapper.verbosity:
+            self._records.append(f"snapshot-taken '{scope.name.upper()}'")
+
+        try:
+            snapshot_id = self.take_snapshot()
+        except Exception:
+            self.supported = False
+        else:
+            if snapshot_id is not None:
+                self.snapshots.set_snapshot_id(scope, snapshot_id)
 
     @allow_disconnected
-    def _snapshot(self) -> Optional["SnapshotID"]:
+    def take_snapshot(self) -> Optional["SnapshotID"]:
         try:
             return self.chain_manager.snapshot()
         except NotImplementedError:
@@ -132,14 +649,24 @@ def _snapshot(self) -> Optional["SnapshotID"]:
                 "Tests will not be completely isolated."
             )
             # To avoid trying again
-            self._supports_snapshot = False
+            self.supported = False
 
         return None
 
     @allow_disconnected
-    def _restore(self, snapshot_id: "SnapshotID"):
-        if snapshot_id not in self.chain_manager._snapshots[self.provider.chain_id]:
+    def restore(self, scope: "Scope"):
+        snapshot_id = self.snapshots.get_snapshot_id(scope)
+        if snapshot_id is None:
             return
+
+        elif snapshot_id not in self.chain_manager._snapshots[self.provider.chain_id]:
+            # Still clear out.
+            self.snapshots.clear_snapshot_id(scope)
+            return
+
+        if self.config_wrapper.verbosity:
+            self._records.append(f"restoring '{scope.name.upper()}'")
+
         try:
             self.chain_manager.restore(snapshot_id)
         except NotImplementedError:
@@ -148,11 +675,20 @@ def _restore(self, snapshot_id: "SnapshotID"):
                 "Tests will not be completely isolated."
             )
             # To avoid trying again
-            self._supports_snapshot = False
+            self.supported = False
+
+        self.snapshots.clear_snapshot_id(scope)
+
+    def show_records(self):
+        if not self._records:
+            return
+
+        records_str = "\n".join(self._records)
+        rich_print(f"\n{records_str}")
+        self._records = []
 
 
 class ReceiptCapture(ManagerAccessMixin):
-    config_wrapper: "ConfigWrapper"
     receipt_map: dict[str, dict[str, "ReceiptAPI"]] = {}
     enter_blocks: list[int] = []
 
@@ -250,3 +786,31 @@ def _exclude_from_gas_report(
                 return True
 
         return False
+
+
+def fixture(chain_isolation: Optional[bool], **kwargs):
+    """
+    A thin-wrapper around ``@pytest.fixture`` with extra capabilities.
+    Set ``chain_isolation`` to ``False`` to signal to Ape that this fixture's
+    cached result is the same regardless of block number and it does not
+    need to be invalidated during times or pytest-scoped based chain rebasing.
+
+    Usage example::
+
+        import ape
+        from ape_tokens import tokens
+
+        @ape.fixture(scope="session", chain_isolation=False, params=("WETH", "DAI", "BAT"))
+        def token_addresses(request):
+            return tokens[request].address
+
+    """
+
+    def decorator(fixture_function):
+        if chain_isolation is not None:
+            name = kwargs.get("name", fixture_function.__name__)
+            FixtureManager._stateful_fixtures_cache[name] = chain_isolation
+
+        return pytest.fixture(fixture_function, **kwargs)
+
+    return decorator
diff --git a/src/ape/pytest/plugin.py b/src/ape/pytest/plugin.py
index e23dd5bf32..c650e38407 100644
--- a/src/ape/pytest/plugin.py
+++ b/src/ape/pytest/plugin.py
@@ -79,7 +79,12 @@ def is_module(v):
 
     from ape.pytest.config import ConfigWrapper
     from ape.pytest.coverage import CoverageTracker
-    from ape.pytest.fixtures import PytestApeFixtures, ReceiptCapture
+    from ape.pytest.fixtures import (
+        FixtureManager,
+        IsolationManager,
+        PytestApeFixtures,
+        ReceiptCapture,
+    )
     from ape.pytest.gas import GasTracker
     from ape.pytest.runners import PytestApeRunner
     from ape.utils.basemodel import ManagerAccessMixin
@@ -87,16 +92,25 @@ def is_module(v):
     # Register the custom Ape test runner
     config_wrapper = ConfigWrapper(config)
     receipt_capture = ReceiptCapture(config_wrapper)
+    isolation_manager = IsolationManager(config_wrapper, receipt_capture)
+    fixture_manager = FixtureManager(config_wrapper, isolation_manager)
     gas_tracker = GasTracker(config_wrapper)
     coverage_tracker = CoverageTracker(config_wrapper)
-    runner = PytestApeRunner(config_wrapper, receipt_capture, gas_tracker, coverage_tracker)
+    runner = PytestApeRunner(
+        config_wrapper,
+        isolation_manager,
+        receipt_capture,
+        gas_tracker,
+        coverage_tracker,
+        fixture_manager=fixture_manager,
+    )
     config.pluginmanager.register(runner, "ape-test")
 
     # Inject runner for access to gas and coverage trackers.
     ManagerAccessMixin._test_runner = runner
 
     # Include custom fixtures for project, accounts etc.
-    fixtures = PytestApeFixtures(config_wrapper, receipt_capture)
+    fixtures = PytestApeFixtures(config_wrapper, isolation_manager)
     config.pluginmanager.register(fixtures, "ape-fixtures")
 
     # Add custom markers
diff --git a/src/ape/pytest/runners.py b/src/ape/pytest/runners.py
index a6528c53cb..d6dfea143f 100644
--- a/src/ape/pytest/runners.py
+++ b/src/ape/pytest/runners.py
@@ -4,9 +4,10 @@
 import click
 import pytest
 from _pytest._code.code import Traceback as PytestTraceback
+from _pytest.reports import TestReport
 from rich import print as rich_print
 
-from ape.exceptions import ConfigError
+from ape.exceptions import ConfigError, ProviderNotConnectedError
 from ape.logging import LogLevel
 from ape.utils.basemodel import ManagerAccessMixin
 
@@ -14,7 +15,7 @@
     from ape.api.networks import ProviderContextManager
     from ape.pytest.config import ConfigWrapper
     from ape.pytest.coverage import CoverageTracker
-    from ape.pytest.fixtures import ReceiptCapture
+    from ape.pytest.fixtures import FixtureManager, IsolationManager, ReceiptCapture
     from ape.pytest.gas import GasTracker
     from ape.types.coverage import CoverageReport
 
@@ -23,11 +24,14 @@ class PytestApeRunner(ManagerAccessMixin):
     def __init__(
         self,
         config_wrapper: "ConfigWrapper",
+        isolation_manager: "IsolationManager",
         receipt_capture: "ReceiptCapture",
         gas_tracker: "GasTracker",
         coverage_tracker: "CoverageTracker",
+        fixture_manager: Optional["FixtureManager"] = None,
     ):
         self.config_wrapper = config_wrapper
+        self.isolation_manager = isolation_manager
         self.receipt_capture = receipt_capture
         self._provider_is_connected = False
 
@@ -36,6 +40,17 @@ def __init__(
         self.gas_tracker = gas_tracker
         self.coverage_tracker = coverage_tracker
 
+        if fixture_manager is None:
+            from ape.pytest.fixtures import FixtureManager
+
+            self.fixture_manager = FixtureManager(config_wrapper, isolation_manager)
+
+        else:
+            self.fixture_manager = fixture_manager
+
+        self._initialized_fixtures: list[str] = []
+        self._finalized_fixtures: list[str] = []
+
     @property
     def _provider_context(self) -> "ProviderContextManager":
         return self.network_manager.parse_network_choice(self.config_wrapper.network)
@@ -144,39 +159,59 @@ def pytest_runtest_setup(self, item):
         https://docs.pytest.org/en/6.2.x/reference.html#pytest.hookspec.pytest_runtest_setup
         """
         if (
-            self.config_wrapper.isolation is False
+            not self.config_wrapper.isolation
             # doctests don't have fixturenames
             or (hasattr(pytest, "DoctestItem") and isinstance(item, pytest.DoctestItem))
             or "_function_isolation" in item.fixturenames  # prevent double injection
         ):
-            # isolation is disabled via cmdline option
+            # isolation is disabled via cmdline option or running doc-tests.
             return
 
-        fixture_map = item.session._fixturemanager._arg2fixturedefs
-        scopes = [
-            definition.scope
-            for name, definitions in fixture_map.items()
-            if name in item.fixturenames
-            for definition in definitions
-        ]
-
-        for scope in ["session", "package", "module", "class"]:
-            # iterate through scope levels and insert the isolation fixture
-            # prior to the first fixture with that scope
-            try:
-                idx = scopes.index(scope)  # will raise ValueError if `scope` not found
-                item.fixturenames.insert(idx, f"_{scope}_isolation")
-                scopes.insert(idx, scope)
-            except ValueError:
-                # intermediate scope isolations aren't filled in
+        if self.config_wrapper.isolation:
+            self._setup_isolation(item)
+
+    def _setup_isolation(self, item):
+        from ape.pytest.utils import Scope
+
+        fixtures = self.fixture_manager.get_fixtures(item)
+        for scope in (Scope.SESSION, Scope.PACKAGE, Scope.MODULE, Scope.CLASS):
+            if not (
+                custom_fixtures := [f for f in fixtures[scope] if self.fixture_manager.is_custom(f)]
+            ):
+                # Intermediate scope isolations aren't filled in, or only using
+                # built-in Ape fixtures.
                 continue
 
-        # insert function isolation by default
-        try:
-            item.fixturenames.insert(scopes.index("function"), "_function_isolation")
-        except ValueError:
-            # no fixtures with function scope, so append function isolation
-            item.fixturenames.append("_function_isolation")
+            snapshot = self.isolation_manager.get_snapshot(scope)
+
+            # Gather new fixtures. Also, be mindful of parametrized fixtures
+            # which strangely have the same name.
+            new_fixtures = []
+            for custom_fixture in custom_fixtures:
+                # Parametrized fixtures must always be considered new
+                # because of severe complications of using them.
+                is_custom = custom_fixture in fixtures.parametrized
+                is_iterating = is_custom and fixtures.is_iterating(custom_fixture)
+                is_new = custom_fixture not in snapshot.fixtures
+
+                # NOTE: Consider ``None`` to be stateful here to be safe.
+                stateful = self.fixture_manager.is_stateful(custom_fixture) is not False
+
+                if (is_new or is_iterating) and stateful:
+                    new_fixtures.append(custom_fixture)
+                    continue
+
+            # Rebase if there are new fixtures found of non-function scope.
+            # And there are stateful fixtures of lower scopes that need resetting.
+            may_need_rebase = bool(new_fixtures and snapshot.fixtures)
+            if may_need_rebase:
+                self.fixture_manager.rebase(scope, fixtures)
+
+            # Append these fixtures so we know when new ones arrive
+            # and need to trigger the invalidation logic above.
+            snapshot.append_fixtures(new_fixtures)
+
+        fixtures.apply_fixturenames()
 
     def pytest_sessionstart(self):
         """
@@ -209,6 +244,46 @@ def pytest_runtest_call(self, item):
         else:
             yield
 
+    def pytest_fixture_setup(self, fixturedef, request):
+        fixture_name = fixturedef.argname
+        if fixture_name in self._initialized_fixtures:
+            return
+
+        self._initialized_fixtures.append(fixture_name)
+        if self._track_fixture_blocks(fixture_name):
+            try:
+                block_number = self.chain_manager.blocks.height
+            except Exception:
+                pass
+            else:
+                self.fixture_manager.add_fixture_info(fixture_name, setup_block=block_number)
+
+    def pytest_fixture_post_finalizer(self, fixturedef, request):
+        fixture_name = fixturedef.argname
+        if fixture_name in self._finalized_fixtures:
+            return
+
+        self._finalized_fixtures.append(fixture_name)
+        if self._track_fixture_blocks(fixture_name):
+            try:
+                block_number = self.chain_manager.blocks.height
+            except ProviderNotConnectedError:
+                pass
+            else:
+                self.fixture_manager.add_fixture_info(fixture_name, teardown_block=block_number)
+
+    def _track_fixture_blocks(self, fixture_name: str) -> bool:
+        from ape.pytest.utils import Scope
+
+        if not self.fixture_manager.is_custom(fixture_name):
+            return False
+
+        scope = self.fixture_manager.get_fixture_scope(fixture_name)
+        if scope in (None, Scope.FUNCTION):
+            return False
+
+        return True
+
     @pytest.hookimpl(trylast=True, hookwrapper=True)
     def pytest_collection_finish(self, session):
         """
@@ -237,6 +312,10 @@ def _connect(self):
         self._provider_context.push_provider()
         self._provider_is_connected = True
 
+    def pytest_runtest_logreport(self, report: TestReport):
+        if self.config_wrapper.verbosity >= 3:
+            self.isolation_manager.show_records()
+
     def pytest_terminal_summary(self, terminalreporter):
         """
         Add a section to terminal summary reporting.
diff --git a/src/ape/pytest/utils.py b/src/ape/pytest/utils.py
new file mode 100644
index 0000000000..acd38dadca
--- /dev/null
+++ b/src/ape/pytest/utils.py
@@ -0,0 +1,16 @@
+from enum import Enum
+
+
+class Scope(int, Enum):
+    SESSION = 0
+    PACKAGE = 1
+    MODULE = 2
+    CLASS = 3
+    FUNCTION = 4
+
+    def __str__(self) -> str:
+        return self.name.lower()
+
+    @property
+    def isolation_fixturename(self) -> str:
+        return f"_{self}_isolation"
diff --git a/src/ape/pytest/warnings.py b/src/ape/pytest/warnings.py
new file mode 100644
index 0000000000..45d50847ae
--- /dev/null
+++ b/src/ape/pytest/warnings.py
@@ -0,0 +1,20 @@
+"""
+Warnings are great for Pytest because they always show up
+at the end of a test-run.
+"""
+
+import warnings
+
+
+class InvalidIsolationWarning(Warning):
+    """
+    Occurs when fixtures disrupt isolation causing performance degradation.
+    """
+
+
+def warn_invalid_isolation():
+    message = (
+        "Invalid isolation; Ensure session|package|module|class scoped fixtures "
+        "run earlier. Rebasing fixtures is costly."
+    )
+    warnings.warn(message, InvalidIsolationWarning)
diff --git a/src/ape_test/provider.py b/src/ape_test/provider.py
index f6c63e8060..db27231aac 100644
--- a/src/ape_test/provider.py
+++ b/src/ape_test/provider.py
@@ -308,13 +308,18 @@ def snapshot(self) -> "SnapshotID":
         return self.evm_backend.take_snapshot()
 
     def restore(self, snapshot_id: "SnapshotID"):
-        if snapshot_id:
-            current_hash = self._get_latest_block_rpc().get("hash")
-            if current_hash != snapshot_id:
-                try:
-                    return self.evm_backend.revert_to_snapshot(snapshot_id)
-                except HeaderNotFound:
-                    raise UnknownSnapshotError(snapshot_id)
+        # NOTE: Snapshot ID can be 0!
+        if snapshot_id is None:
+            return
+
+        current_hash = self._get_latest_block_rpc().get("hash")
+        if current_hash == snapshot_id:
+            return
+
+        try:
+            return self.evm_backend.revert_to_snapshot(snapshot_id)
+        except (HeaderNotFound, ValidationError):
+            raise UnknownSnapshotError(snapshot_id)
 
     def set_timestamp(self, new_timestamp: int):
         current_timestamp = self.evm_backend.get_block_by_number("pending")["timestamp"]
diff --git a/tests/functional/test_exceptions.py b/tests/functional/test_exceptions.py
index c74c63d88b..d6ccf200eb 100644
--- a/tests/functional/test_exceptions.py
+++ b/tests/functional/test_exceptions.py
@@ -11,6 +11,7 @@
     ContractNotFoundError,
     NetworkNotFoundError,
     TransactionError,
+    UnknownSnapshotError,
     handle_ape_exception,
 )
 from ape.types.trace import SourceTraceback
@@ -246,3 +247,15 @@ def test_fork_network(self):
             "Try installing an explorer plugin using \x1b[32mape plugins install etherscan"
             "\x1b[0m, or using a network with explorer support."
         )
+
+
+class TestUnknownSnapshotError:
+    def test_bytes(self):
+        snapshot_id = b"asdfasdfasdf"
+        err = UnknownSnapshotError(snapshot_id)
+        assert str(err) == "Unknown snapshot ID '6173..6466'."
+
+    @pytest.mark.parametrize("snapshot_id", (123, "123"))
+    def test_not_bytes(self, snapshot_id):
+        err = UnknownSnapshotError(snapshot_id)
+        assert str(err) == "Unknown snapshot ID '123'."
diff --git a/tests/functional/test_fixtures.py b/tests/functional/test_fixtures.py
index b27b3979f7..3ab36bdf86 100644
--- a/tests/functional/test_fixtures.py
+++ b/tests/functional/test_fixtures.py
@@ -1,7 +1,13 @@
 import pytest
 
 from ape.exceptions import BlockNotFoundError
-from ape.pytest.fixtures import PytestApeFixtures
+from ape.pytest.fixtures import IsolationManager, PytestApeFixtures
+from ape.pytest.utils import Scope
+
+
+@pytest.fixture
+def config_wrapper(mocker):
+    return mocker.MagicMock()
 
 
 @pytest.fixture
@@ -10,13 +16,41 @@ def receipt_capture(mocker):
 
 
 @pytest.fixture
-def fixtures(mocker, receipt_capture):
-    return PytestApeFixtures(mocker.MagicMock(), receipt_capture)
+def isolation_manager(config_wrapper, receipt_capture):
+    return IsolationManager(config_wrapper, receipt_capture)
+
+
+@pytest.fixture
+def fixtures(mocker, isolation_manager):
+    return PytestApeFixtures(mocker.MagicMock(), isolation_manager)
 
 
 @pytest.fixture
 def isolation(fixtures):
-    return fixtures._isolation()
+    return fixtures.isolation_manager.isolation(Scope.FUNCTION)
+
+
+@pytest.fixture
+def mock_evm(mocker):
+    return mocker.MagicMock()
+
+
+@pytest.fixture
+def use_mock_provider(networks, mock_provider, mock_evm):
+    orig_provider = networks.active_provider
+    mock_provider._web3.eth.get_block.side_effect = orig_provider._web3.eth.get_block
+    networks.active_provider = mock_provider
+    orig_backend = mock_provider._evm_backend
+
+    # Ensure functional isolation still uses snapshot.
+    mock_evm.take_snapshot.side_effect = orig_backend.take_snapshot
+
+    try:
+        mock_provider._evm_backend = mock_evm
+        yield mock_provider
+    finally:
+        mock_provider._evm_backend = orig_backend
+        networks.active_provider = orig_provider
 
 
 def test_isolation(isolation, receipt_capture):
@@ -32,13 +66,13 @@ def test_isolation(isolation, receipt_capture):
 
 
 def test_isolation_restore_not_implemented(mocker, networks, fixtures):
-    isolation = fixtures._isolation()
+    isolation = fixtures.isolation_manager.isolation(Scope.FUNCTION)
     mock_provider = mocker.MagicMock()
     mock_provider.restore.side_effect = NotImplementedError
     mock_provider.snapshot.return_value = 123
     orig_provider = networks.active_provider
     networks.active_provider = mock_provider
-    fixtures._supports_snapshot = True
+    fixtures.isolation_manager.supported = True
 
     try:
         _ = next(isolation)
@@ -47,9 +81,9 @@ def test_isolation_restore_not_implemented(mocker, networks, fixtures):
             _ = next(isolation)
 
         # Is false because of the not-implemented error side-effect.
-        assert fixtures._supports_snapshot is False
+        assert fixtures.isolation_manager.supported is False
 
-        isolation = fixtures._isolation()
+        isolation = fixtures.isolation_manager.isolation(Scope.FUNCTION)
         _ = next(isolation)
         # It does not call snapshot again.
         assert mock_provider.snapshot.call_count == 1
@@ -60,3 +94,68 @@ def test_isolation_restore_not_implemented(mocker, networks, fixtures):
 
     finally:
         networks.active_provider = orig_provider
+
+
+@pytest.mark.parametrize("snapshot_id", (0, 1, "123"))
+def test_isolation_snapshot_id_types(snapshot_id, use_mock_provider, fixtures, mock_evm):
+    mock_evm.take_snapshot.side_effect = lambda: snapshot_id
+    isolation_context = fixtures.isolation_manager.isolation(Scope.FUNCTION)
+    next(isolation_context)  # Enter.
+    assert mock_evm.take_snapshot.call_count == 1
+    assert mock_evm.revert_to_snapshot.call_count == 0
+    next(isolation_context, None)  # Exit.
+    mock_evm.revert_to_snapshot.assert_called_once_with(snapshot_id)
+
+
+def test_isolation_when_snapshot_fails_avoids_restore(use_mock_provider, fixtures, mock_evm):
+    mock_evm.take_snapshot.side_effect = NotImplementedError
+    isolation_context = fixtures.isolation_manager.isolation(Scope.FUNCTION)
+    next(isolation_context)  # Enter.
+    assert mock_evm.take_snapshot.call_count == 1
+    assert mock_evm.revert_to_snapshot.call_count == 0
+    next(isolation_context, None)  # Exit.
+    # It doesn't even try!
+    assert mock_evm.revert_to_snapshot.call_count == 0
+
+
+def test_isolation_restore_fails_avoids_snapshot_next_time(
+    networks, use_mock_provider, fixtures, mock_evm
+):
+    mock_evm.take_snapshot.return_value = 123
+    mock_evm.revert_to_snapshot.side_effect = NotImplementedError
+    isolation_context = fixtures.isolation_manager.isolation(Scope.FUNCTION)
+    next(isolation_context)  # Enter.
+    # Snapshot works, we get this far.
+    assert mock_evm.take_snapshot.call_count == 1
+    assert mock_evm.revert_to_snapshot.call_count == 0
+
+    # At this point, it is realized snapshotting is no-go.
+    mock_evm.take_snapshot.reset_mock()
+    next(isolation_context, None)  # Exit.
+    isolation_context = fixtures.isolation_manager.isolation(Scope.FUNCTION)
+    next(isolation_context)  # Enter again.
+    # This time, snapshotting is NOT attempted.
+    assert mock_evm.take_snapshot.call_count == 0
+
+
+def test_isolation_supported_flag_set_after_successful_snapshot(
+    use_mock_provider, fixtures, mock_evm
+):
+    """
+    Testing the unusual case where `.supported` was changed manually after
+    a successful snapshot and before the restore attempt.
+    """
+    mock_evm.take_snapshot.return_value = 123
+    isolation_context = fixtures.isolation_manager.isolation(Scope.FUNCTION)
+    next(isolation_context)  # Enter.
+    assert mock_evm.take_snapshot.call_count == 1
+    assert mock_evm.revert_to_snapshot.call_count == 0
+
+    # HACK: Change the flag manually to show it will avoid
+    #   the restore.
+    fixtures.isolation_manager.supported = False
+
+    next(isolation_context, None)  # Exit.
+    # Even though snapshotting worked, the flag was changed,
+    # and so the restore never gets attempted.
+    assert mock_evm.revert_to_snapshot.call_count == 0
diff --git a/tests/functional/test_plugins.py b/tests/functional/test_plugins.py
index 409929eda6..07cce2f81a 100644
--- a/tests/functional/test_plugins.py
+++ b/tests/functional/test_plugins.py
@@ -82,7 +82,7 @@ def package_names() -> set[str]:
 
 
 @pytest.fixture
-def plugin_metadata(package_names) -> PluginMetadataList:
+def plugin_metadata(package_names, plugin_test_env) -> PluginMetadataList:
     names = {x for x in package_names}
     names.remove("ape-installed")
     names.add(f"ape-installed==0.{ape_version.minor}.0")
diff --git a/tests/functional/test_project.py b/tests/functional/test_project.py
index b432fc1b94..7f11b500a8 100644
--- a/tests/functional/test_project.py
+++ b/tests/functional/test_project.py
@@ -180,7 +180,7 @@ def test_isolate_in_tempdir(project):
 
 def test_isolate_in_tempdir_does_not_alter_sources(project):
     # First, create a bad source.
-    with project.temp_config(contracts_folder="tests"):
+    with project.temp_config(contracts_folder="build"):
         new_src = project.contracts_folder / "newsource.json"
         new_src.write_text("this is not json, oops")
         project.sources.refresh()  # Only need to be called when run with other tests.
@@ -194,8 +194,8 @@ def test_isolate_in_tempdir_does_not_alter_sources(project):
             project.sources.refresh()
 
         # Ensure "newsource" did not persist in the in-memory manifest.
-        assert "tests/newsource.json" in actual, project.path
-        assert "tests/newsource.json" not in (project.manifest.sources or {})
+        assert "build/newsource.json" in actual
+        assert "build/newsource.json" not in (project.manifest.sources or {})
 
 
 def test_in_tempdir(project, tmp_project):
diff --git a/tests/functional/test_provider.py b/tests/functional/test_provider.py
index 6d0455e170..60f3f8b698 100644
--- a/tests/functional/test_provider.py
+++ b/tests/functional/test_provider.py
@@ -19,6 +19,7 @@
     ProviderError,
     TransactionError,
     TransactionNotFoundError,
+    UnknownSnapshotError,
 )
 from ape.types.events import LogFilter
 from ape.utils.testing import DEFAULT_TEST_ACCOUNT_BALANCE, DEFAULT_TEST_CHAIN_ID
@@ -591,6 +592,25 @@ def test_ipc_per_network(project, key):
         assert node.ipc_path == Path(ipc)
 
 
+def test_snapshot(eth_tester_provider):
+    snapshot = eth_tester_provider.snapshot()
+    assert snapshot
+
+
+def test_restore(eth_tester_provider, accounts):
+    account = accounts[0]
+    start_nonce = account.nonce
+    snapshot = eth_tester_provider.snapshot()
+    account.transfer(account, 0)
+    eth_tester_provider.restore(snapshot)
+    assert account.nonce == start_nonce
+
+
+def test_restore_zero(eth_tester_provider):
+    with pytest.raises(UnknownSnapshotError, match="Unknown snapshot ID '0'."):
+        eth_tester_provider.restore(0)
+
+
 def test_update_settings_invalidates_snapshots(eth_tester_provider, chain):
     snapshot = chain.snapshot()
     assert snapshot in chain._snapshots[eth_tester_provider.chain_id]
diff --git a/tests/functional/test_test.py b/tests/functional/test_test.py
index e063973e0e..413c08f30d 100644
--- a/tests/functional/test_test.py
+++ b/tests/functional/test_test.py
@@ -3,11 +3,52 @@
 import pytest
 
 from ape.exceptions import ConfigError
+from ape.pytest.fixtures import FixtureManager, FixtureMap, IsolationManager, SnapshotRegistry
 from ape.pytest.runners import PytestApeRunner
+from ape.pytest.utils import Scope
+from ape.pytest.warnings import InvalidIsolationWarning
 from ape_test import ApeTestConfig
 from ape_test._watch import run_with_observer
 
 
+@pytest.fixture
+def create_fixture_info(mocker):
+    def fn(name="my_fixture", scope=Scope.FUNCTION.value, params=None, cached_result=None):
+        info = mocker.MagicMock()
+        info.argname = name
+        info.scope = scope
+        info.params = params
+        info.cached_result = cached_result
+        return info
+
+    return fn
+
+
+@pytest.fixture
+def item(mocker, create_fixture_info):
+    # foo, bar, and baz are fixtures; param0 and param1 are test-params.
+    fixturenames = ["foo", "bar", "baz", "param0", "param1"]
+    mock = mocker.MagicMock()
+    mock.nodeid = "test_nodeid"
+    mock.fixturenames = fixturenames
+    mock.session._fixturemanager._arg2fixturedefs = {
+        "_session_isolation": [create_fixture_info("_session_isolation", Scope.SESSION.value)],
+        "_package_isolation": [create_fixture_info("_package_isolation", Scope.PACKAGE.value)],
+        "foo": [create_fixture_info("foo", Scope.SESSION.value, [1, 2, 3])],
+        "_module_isolation": [create_fixture_info("_module_isolation", Scope.MODULE.value)],
+        "bar": [create_fixture_info("bar", Scope.MODULE.value)],
+        "_class_isolation": [create_fixture_info("_class_isolation", Scope.CLASS.value)],
+        "baz": [create_fixture_info("baz", Scope.CLASS.value)],
+        "_function_isolation": [create_fixture_info("_function_isolation", Scope.FUNCTION.value)],
+    }
+    return mock
+
+
+@pytest.fixture
+def fixture_map(item):
+    return FixtureMap.from_test_item(item)
+
+
 class TestApeTestConfig:
     def test_balance_set_from_currency_str(self):
         curr_val = "10 Eth"
@@ -27,8 +68,9 @@ def test_connect_to_mainnet_by_default(mocker):
 
     cfg = mocker.MagicMock()
     cfg.network = "ethereum:mainnet:node"
-    runner = PytestApeRunner(cfg, mocker.MagicMock(), mocker.MagicMock(), mocker.MagicMock())
-
+    runner = PytestApeRunner(
+        cfg, mocker.MagicMock(), mocker.MagicMock(), mocker.MagicMock(), mocker.MagicMock()
+    )
     expected = (
         "Default network is mainnet; unable to run tests on mainnet. "
         "Please specify the network using the `--network` flag or "
@@ -38,6 +80,311 @@ def test_connect_to_mainnet_by_default(mocker):
         runner._connect()
 
 
+class TestFixtureManager:
+    @pytest.fixture
+    def fixture_manager(self, mocker):
+        config = mocker.MagicMock()
+        isolation = mocker.MagicMock()
+        return FixtureManager(config, isolation)
+
+    def test_ape_fixtures(self, fixture_manager):
+        actual = fixture_manager._ape_fixtures
+        assert "accounts" in actual
+        assert "project" in actual
+        assert "networks" in actual
+        assert "chain" in actual
+
+    def test_is_isolation(self, fixture_manager):
+        assert fixture_manager.is_isolation("_session_isolation")
+        assert fixture_manager.is_isolation("_package_isolation")
+        assert fixture_manager.is_isolation("_module_isolation")
+        assert fixture_manager.is_isolation("_class_isolation")
+        assert fixture_manager.is_isolation("_function_isolation")
+        assert not fixture_manager.is_isolation("my_custom_isolation")
+
+    def tes_is_ape(self, fixture_manager):
+        assert fixture_manager.is_ape("accounts")
+        assert not fixture_manager.is_ape("acct")
+
+    def test_is_custom(self, fixture_manager):
+        assert fixture_manager.is_custom("acct")
+        assert not fixture_manager.is_custom("accounts")
+        assert not fixture_manager.is_custom("_module_isolation")
+
+    def test_get_fixtures(self, fixture_manager, item):
+        with pytest.raises(KeyError):
+            assert fixture_manager.get_fixtures(item.nodeid) is None
+
+        actual = fixture_manager.get_fixtures(item)
+        assert isinstance(actual, FixtureMap)
+        # Now, can use nodeid.
+        assert fixture_manager.get_fixtures(item.nodeid) == actual
+
+    def test_is_stateful(self, fixture_manager, item):
+        fixture_manager.cache_fixtures(item)
+        assert fixture_manager.is_stateful("foo") is None  # Unknown.
+        fixture_manager.add_fixture_info("foo", setup_block=1, teardown_block=1)
+        assert fixture_manager.is_stateful("foo") is False
+        fixture_manager.add_fixture_info("bar", setup_block=1, teardown_block=2)
+        assert fixture_manager.is_stateful("bar") is True
+
+    def test_rebase(self, mocker, fixture_manager, fixture_map, create_fixture_info):
+        # We must have already started our module-scope isolation.
+        isolation_manager = IsolationManager(fixture_manager.config_wrapper, mocker.MagicMock())
+        isolation_manager.snapshots[Scope.MODULE].identifier = "123"
+        isolation_manager.snapshots[Scope.MODULE].fixtures = ["bar"]
+        fixture_manager.isolation_manager = isolation_manager
+
+        # New session fixture arrives, triggering a rebase.
+        fixture_map[Scope.SESSION].append("new_session_fixture")
+        fixture_map._item.fixturenames.append("new_session_fixture")
+        fixture_map._item.session._fixturemanager._arg2fixturedefs["new_session_fixture"] = [
+            create_fixture_info("new_session_fixture", Scope.SESSION)
+        ]
+
+        # Cache a module result so we can prove it gets cleared.
+        fixture_map._item.session._fixturemanager._arg2fixturedefs["bar"][0].cached_result = "CACHE"
+
+        # Show the module-isolation was cache and gets cleared as well.
+        # NOTE: Pytest caches yield-based fixtures as a tuple, even when yields None.
+        fixture_map._item.session._fixturemanager._arg2fixturedefs["_module_isolation"][
+            0
+        ].cached_result = (None, None, None)
+
+        expected = (
+            r"Invalid isolation; Ensure session|package|module|class scoped "
+            r"fixtures run earlier\. Rebasing fixtures is costly\."
+        )
+        with pytest.warns(InvalidIsolationWarning, match=expected):
+            fixture_manager.rebase(Scope.SESSION, fixture_map)
+
+        # Show that module-level fixtures are invalidated, including the isolation fixture.
+        for module_fixture_name in ("bar", "_module_isolation"):
+            assert (
+                fixture_map._item.session._fixturemanager._arg2fixturedefs[module_fixture_name][
+                    0
+                ].cached_result
+                is None
+            )
+        # Show that we have reverted our module-level snapshot.
+        assert isolation_manager.snapshots[Scope.MODULE].identifier is None
+
+
+class TestFixtureMap:
+    def test_from_test_item(self, item):
+        actual = FixtureMap.from_test_item(item)
+        assert actual[Scope.SESSION] == ["foo"]
+        assert actual[Scope.MODULE] == ["bar"]
+        assert actual[Scope.CLASS] == ["baz"]
+
+    def test_names(self, fixture_map):
+        """
+        Show that we have both the initialized fixtures as well
+        as the properly injected isolation fixtures. Order is
+        EXTREMELY important here! It determines the order in which
+        fixtures run; isolation should run before their sister fixtures.
+        Function isolation is expected even when not using other function-scoped
+        fixtures. Package isolation is missing because there are no
+        package-scoped fixtures being used.
+        """
+        actual = fixture_map.names
+        expected = [
+            "_session_isolation",
+            "foo",
+            "_module_isolation",
+            "bar",
+            "_class_isolation",
+            "baz",
+            "_function_isolation",
+        ]
+        assert actual == expected
+
+    def test_parameters(self, fixture_map):
+        actual = fixture_map.parameters
+        expected = ["param0", "param1"]
+        assert actual == expected
+
+    def test_isolation(self, fixture_map):
+        actual = fixture_map.isolation
+        expected = [
+            "session",
+            "module",
+            "class",
+            "function",
+        ]
+        assert actual == expected
+
+    def test_parametrized(self, fixture_map):
+        actual = fixture_map.parametrized
+        assert "foo" in actual
+        assert len(actual) == 1
+
+    def test_get_info(self, fixture_map):
+        actual = fixture_map.get_info("foo")
+        assert len(actual) == 1
+        assert actual[0].argname == "foo"
+        assert actual[0].scope == Scope.SESSION
+
+    def test_is_known(self, fixture_map):
+        assert fixture_map.is_known("foo")
+        assert not fixture_map.is_known("param0")
+
+    def test_is_iterating(self, fixture_map):
+        assert fixture_map.is_iterating("foo")
+        assert not fixture_map.is_iterating("baz")
+
+        # Iterate.
+        fixture_map._item.session._fixturemanager._arg2fixturedefs["foo"][0].cached_result = (
+            None,
+            1,
+            None,
+        )
+        assert fixture_map.is_iterating("foo")
+
+        # Complete.
+        fixture_map._item.session._fixturemanager._arg2fixturedefs["foo"][0].cached_result = (
+            None,
+            3,
+            None,
+        )
+        assert not fixture_map.is_iterating("foo")
+
+    def test_apply_fixturenames(self, fixture_map):
+        assert fixture_map._item.fixturenames == ["foo", "bar", "baz", "param0", "param1"]
+        fixture_map.apply_fixturenames()
+        assert fixture_map._item.fixturenames == [
+            "_session_isolation",
+            "foo",
+            "_module_isolation",
+            "bar",
+            "_class_isolation",
+            "baz",
+            "_function_isolation",
+            "param0",
+            "param1",
+        ]
+
+
+class TestSnapshotRegistry:
+    """
+    Note: Most isolation-based tests occur in `functional/test_fixtures.py`.
+    """
+
+    @pytest.fixture
+    def registry(self):
+        return SnapshotRegistry()
+
+    def test_get_snapshot_id(self, registry):
+        actual = registry.get_snapshot_id(Scope.SESSION)
+        assert actual is None
+
+    def test_next_snapshots(self, registry):
+        actual = [x for x in registry.next_snapshots(Scope.SESSION)]
+        assert actual[0].scope is Scope.PACKAGE
+        assert actual[1].scope is Scope.MODULE
+        assert actual[2].scope is Scope.CLASS
+        assert actual[3].scope is Scope.FUNCTION
+
+
+class TestIsolationManager:
+    """
+    Note: Most isolation-based tests occur in `functional/test_fixtures.py`.
+    """
+
+    @pytest.fixture
+    def isolation_manager(self, mocker):
+        config_wrapper = mocker.MagicMock()
+        receipt_capture = mocker.MagicMock()
+        return IsolationManager(config_wrapper, receipt_capture)
+
+    @pytest.fixture
+    def empty_snapshot_registry(self, isolation_manager):
+        snapshots = isolation_manager.snapshots
+        isolation_manager.snapshots = SnapshotRegistry()
+        yield
+        isolation_manager.snapshots = snapshots
+
+    def test_get_snapshot(self, isolation_manager):
+        actual = isolation_manager.get_snapshot(Scope.SESSION)
+        assert actual.scope is Scope.SESSION
+
+    def test_next_snapshots(self, isolation_manager):
+        actual = [x for x in isolation_manager.next_snapshots(Scope.SESSION)]
+        assert actual[0].scope is Scope.PACKAGE
+        assert actual[1].scope is Scope.MODULE
+        assert actual[2].scope is Scope.CLASS
+        assert actual[3].scope is Scope.FUNCTION
+
+    def test_isolate(
+        self, isolation_manager, owner, vyper_contract_instance, empty_snapshot_registry
+    ):
+        """
+        Low-level test simulating how pytest interacts with these yield-based
+        isolation fixtures.
+        """
+        start_number = vyper_contract_instance.myNumber()
+        session = isolation_manager.isolation(Scope.SESSION)
+        module = isolation_manager.isolation(Scope.MODULE)
+        function = isolation_manager.isolation(Scope.FUNCTION)
+
+        expected_session = 10_000_000
+        expected_module = 20_000_000
+        expected_test = 30_000_000
+
+        # Show we start off clear of snapshots.
+        assert all(
+            isolation_manager.snapshots[s].identifier is None for s in Scope
+        ), "Setup failed - snapshots not empty"
+
+        # Start session.
+        next(session)
+        assert isolation_manager.snapshots[Scope.SESSION].identifier is not None
+        vyper_contract_instance.setNumber(expected_session, sender=owner)
+
+        # Start module.
+        next(module)
+        vyper_contract_instance.setNumber(expected_module, sender=owner)
+
+        # Start test.
+        next(function)
+        vyper_contract_instance.setNumber(expected_test, sender=owner)
+        assert vyper_contract_instance.myNumber() == expected_test
+
+        # End test; back to module.
+        next(function, None)
+        assert vyper_contract_instance.myNumber() == expected_module, "Is not back at module."
+
+        # End module; back to session.
+        assert isolation_manager.snapshots[Scope.MODULE].identifier is not None
+        next(module, None)
+        assert vyper_contract_instance.myNumber() == expected_session, "Is not back at session."
+
+        # Start new module.
+        module = isolation_manager.isolation(Scope.MODULE)
+        next(module)
+        vyper_contract_instance.setNumber(expected_module, sender=owner)
+
+        # Start new test.
+        function = isolation_manager.isolation(Scope.FUNCTION)
+        next(function)
+        vyper_contract_instance.setNumber(expected_test, sender=owner)
+        assert vyper_contract_instance.myNumber() == expected_test
+
+        # End test.
+        next(function, None)
+        assert vyper_contract_instance.myNumber() == expected_module, "(2) Is not back at module."
+
+        # End module.
+        assert isolation_manager.snapshots[Scope.MODULE].identifier is not None
+        next(module, None)
+        assert isolation_manager.snapshots[Scope.MODULE].identifier is None
+        assert vyper_contract_instance.myNumber() == expected_session, "(2) Is not back at session."
+
+        # End session.
+        next(session, None)
+        assert vyper_contract_instance.myNumber() == start_number, "(2) Is not back pre-session."
+
+
 def test_watch(mocker):
     mock_event_handler = mocker.MagicMock()
     event_handler_patch = mocker.patch("ape_test._watch._create_event_handler")
diff --git a/tests/integration/cli/projects/test/tests/conftest.py b/tests/integration/cli/projects/test/tests/conftest.py
new file mode 100644
index 0000000000..07658f7d38
--- /dev/null
+++ b/tests/integration/cli/projects/test/tests/conftest.py
@@ -0,0 +1,11 @@
+import pytest
+
+
+@pytest.fixture(scope="session")
+def session_one(chain):
+    chain.mine(4)
+
+
+@pytest.fixture(scope="session")
+def session_two(chain):
+    chain.mine(2)
diff --git a/tests/integration/cli/projects/test/tests/test_fixture_isolation.py b/tests/integration/cli/projects/test/tests/test_fixture_isolation.py
index 576cf6561e..c218ddddce 100644
--- a/tests/integration/cli/projects/test/tests/test_fixture_isolation.py
+++ b/tests/integration/cli/projects/test/tests/test_fixture_isolation.py
@@ -1,8 +1,38 @@
 import pytest
 
+import ape
+
 INITIAL_BALANCE = 1_000_1 * 10**18
 
 
+@pytest.fixture(scope="function")
+def function_one(chain):
+    chain.mine(1)
+
+
+TOKEN_MAP = {
+    "WETH": "weth-token",
+    "DAI": "dai-token",
+    "BAT": "bat-token",
+}
+
+
+@ape.fixture(scope="module", chain_isolation=False, params=("WETH", "DAI", "BAT"))
+def token_key(request):
+    return TOKEN_MAP[request.param]
+
+
+def test_token_key(token_key):
+    # TODO: Improve this test - show token key doesn't trigger
+    #   resets in those unfortunate conditions.
+    assert True
+
+
+@pytest.fixture(scope="module", autouse=True)
+def module_one(chain):
+    chain.mine(3)
+
+
 @pytest.fixture(scope="session")
 def alice(accounts):
     yield accounts[0]
@@ -13,6 +43,12 @@ def bob(accounts):
     yield accounts[1]
 
 
+@pytest.fixture(params=(5, 6, 7))
+def parametrized_mining(chain, request):
+    chain.mine(request.param)
+    return request.param
+
+
 @pytest.fixture(scope="module", autouse=True)
 def setup(alice, bob, chain):
     start_number = chain.provider.get_block("latest").number
@@ -27,6 +63,27 @@ def start_block_number(chain):
     return chain.blocks.height
 
 
+def test_noop():
+    # This forces auto-use fixtures to fire off before
+    # any requested fixtures, for testing purposes.
+    assert True
+
+
+class TestClass:
+    @pytest.fixture(scope="class", autouse=True)
+    def classminer(self, chain):
+        chain.mine(9)
+
+    def test_chain(self, chain):
+        """
+        Sessions haven't run yet.
+        Module mined 3 + 1 = 4 total.
+        Class mined 9.
+        4 + 9 = 13
+        """
+        assert chain.blocks.height == 13
+
+
 def test_isolation_first(alice, bob, chain, start_block_number):
     assert chain.provider.get_block("latest").number == start_block_number
     assert bob.balance == INITIAL_BALANCE
@@ -36,3 +93,112 @@ def test_isolation_first(alice, bob, chain, start_block_number):
 def test_isolation_second(bob, chain, start_block_number):
     assert chain.provider.get_block("latest").number == start_block_number
     assert bob.balance == INITIAL_BALANCE
+
+
+def test_isolation_with_session_module_and_function(chain, session_one, session_two, function_one):
+    """
+    The sessions should be used, so that is 6.
+    Function is 1 and the module 3.
+    Also, setup does a transfer - that bumps up another 1.
+    Expected is 11.
+    """
+    # NOTE: Module is on autouse=True
+    assert chain.blocks.height == 11
+
+
+def test_isolation_module_ran_after(chain):
+    """
+    This test runs after the test above.
+    We should be back at the beginning of the state after
+    the session and module function but before the function.
+    Expected = sessions + module = 4 + 2 + 3 + 1 (from setup) = 10
+    """
+    assert chain.blocks.height == 10
+
+
+def test_parametrized_fixtures(start_block_number, chain, parametrized_mining):
+    assert chain.blocks.height == start_block_number + parametrized_mining
+
+
+@pytest.fixture(scope="session", params=(1, 2, 3))
+def parametrized_transaction(request, alice, bob):
+    """
+    3 more get added to the session here!
+    """
+    alice.transfer(bob, f"{request.param} wei")
+    return request.param
+
+
+@pytest.fixture(scope="session", params=(1, 2, 3))
+def second_parametrized_transaction(request, alice, bob):
+    """
+    2 more get added to the session here!
+    """
+    alice.transfer(bob, f"{request.param * 2} wei")
+    return request.param
+
+
+@pytest.fixture
+def functional_fixture_using_session(chain, session_one):
+    """
+    Showing the transactions in a functional-scoped
+    fixture that use a session-scoped fixture don't
+    persist on-chain.
+    """
+    _ = session_one
+    chain.mine()
+    return 11  # expected: 10 built up plus this 1.
+
+
+# Parametrized to show it works more than once.
+@pytest.mark.parametrize("it", (0, 1, 2))
+def test_functional_fixture_using_session(chain, functional_fixture_using_session, it):
+    assert chain.blocks.height == functional_fixture_using_session
+
+
+def test_use_parametrized_transaction(chain, parametrized_transaction):
+    starting = 10  # All session + module
+    assert chain.blocks.height == starting + parametrized_transaction
+
+
+def test_use_parametrized_transaction_again(chain, parametrized_transaction):
+    """
+    Should not have invalidated parametrized fixture.
+    """
+    starting = 10  # All session + module
+    assert chain.blocks.height == starting + parametrized_transaction
+
+
+@pytest.fixture
+def functional_fixture_using_parametrized_session(chain, parametrized_transaction):
+    chain.mine()
+    return 11 + parametrized_transaction
+
+
+def test_functional_fixture_using_parametrized_session(
+    chain, functional_fixture_using_parametrized_session
+):
+    assert chain.blocks.height == functional_fixture_using_parametrized_session
+
+
+@pytest.mark.parametrize("foo", (1, 2, 3))
+def test_parametrized_test(foo):
+    """
+    Ensuring parametrized tests don't mess up our isolation-fixture logic
+    (it was the case at one point!)
+    """
+    assert isinstance(foo, int)
+
+
+def test_use_isolate_in_test(chain, parametrized_transaction):
+    """
+    Show the isolation we control doesn't affect
+    the isolation fixtures.
+    """
+    _ = parametrized_transaction  # Using this for complexity.
+    start_block = chain.blocks.height
+    with chain.isolate():
+        chain.mine()
+        assert chain.blocks.height == start_block + 1
+
+    assert chain.blocks.height == start_block
diff --git a/tests/integration/cli/projects/test/tests/test_fixture_isolation_session.py b/tests/integration/cli/projects/test/tests/test_fixture_isolation_session.py
new file mode 100644
index 0000000000..37633d62af
--- /dev/null
+++ b/tests/integration/cli/projects/test/tests/test_fixture_isolation_session.py
@@ -0,0 +1,24 @@
+"""
+'test_fixture_isolation.py' runs before this module.
+We are testing that we go back to an expected session-level
+state without any of the module-level state from
+'test_fixture_isolation.py'.
+"""
+
+
+def test_session(chain):
+    """
+    `session_one` mines 4 and `session_two` mines 2,
+    so we expected 6. Then, at the end of the module,
+    3 more get added to the session.
+    """
+    assert chain.blocks.height == 9
+
+
+def test_session2(chain):
+    """
+    Session isolation doesn't revert other session fixtures,
+    so we are still at 6. Then, at the end of the module,
+    3 more get added to the session.
+    """
+    assert chain.blocks.height == 9
diff --git a/tests/integration/cli/test_test.py b/tests/integration/cli/test_test.py
index 90cf0b184c..23bd9ec58f 100644
--- a/tests/integration/cli/test_test.py
+++ b/tests/integration/cli/test_test.py
@@ -98,7 +98,7 @@ def setup(project):
                     [
                         x
                         for x in content.splitlines()
-                        if x.startswith("def test_") and not x.startswith("def test_fail_")
+                        if x.lstrip().startswith("def test_") and not x.startswith("def test_fail_")
                     ]
                 )
                 num_failed += len(
@@ -177,14 +177,16 @@ def run_gas_test(
 def test_test(setup_pytester, integ_project, pytester, eth_tester_provider):
     _ = eth_tester_provider  # Ensure using EthTester for this test.
     passed, failed = setup_pytester(integ_project)
+
     from ape.logging import logger
 
     logger.set_level("DEBUG")
     result = pytester.runpytest_subprocess(timeout=120)
-    try:
-        result.assert_outcomes(passed=passed, failed=failed), "\n".join(result.outlines)
-    except ValueError:
-        pytest.fail(str(result.stderr))
+    outcomes = result.parseoutcomes()
+    assert "failed" not in outcomes if failed == 0 else outcomes["failed"] == failed
+    if integ_project.name != "test":
+        assert outcomes["passed"] == passed
+    # else: too many parametrized tests to calculate. No fails is good enough.
 
 
 @skip_projects_except("with-contracts")
@@ -212,7 +214,7 @@ def test_show_internal(setup_pytester, integ_project, pytester, eth_tester_provi
 
 
 @skip_projects_except("test", "with-contracts")
-def test_test_isolation_disabled(setup_pytester, integ_project, pytester, eth_tester_provider):
+def test_isolation_disabled(setup_pytester, integ_project, pytester, eth_tester_provider):
     # check the disable isolation option actually disables built-in isolation
     _ = eth_tester_provider  # Ensure using EthTester for this test.
     setup_pytester(integ_project)

From e3657678c75b7eb4654ac70debdb9ade6e033ec8 Mon Sep 17 00:00:00 2001
From: Juliya Smith <jules@apeworx.io>
Date: Tue, 5 Nov 2024 10:52:38 -0600
Subject: [PATCH 2/3] chore: scopey fix

---
 src/ape/pytest/fixtures.py | 96 ++++++++++++--------------------------
 src/ape/pytest/runners.py  | 10 ++--
 2 files changed, 35 insertions(+), 71 deletions(-)

diff --git a/src/ape/pytest/fixtures.py b/src/ape/pytest/fixtures.py
index 563278aad1..469a3e4a05 100644
--- a/src/ape/pytest/fixtures.py
+++ b/src/ape/pytest/fixtures.py
@@ -4,7 +4,7 @@
 from collections.abc import Iterable, Iterator, Mapping
 from dataclasses import dataclass, field
 from fnmatch import fnmatch
-from functools import cached_property
+from functools import cached_property, singledispatchmethod
 from typing import TYPE_CHECKING, ClassVar, Optional
 
 import pytest
@@ -13,6 +13,7 @@
 
 from ape.exceptions import BlockNotFoundError, ChainError, ProviderNotConnectedError
 from ape.logging import logger
+from ape.pytest.utils import Scope
 from ape.utils.basemodel import ManagerAccessMixin
 from ape.utils.rpc import allow_disconnected
 
@@ -23,14 +24,13 @@
     from ape.managers.networks import NetworkManager
     from ape.managers.project import ProjectManager
     from ape.pytest.config import ConfigWrapper
-    from ape.pytest.utils import Scope
     from ape.types.vm import SnapshotID
 
 
 @dataclass()
 class FixtureRebase:
-    return_scope: "Scope"
-    invalid_fixtures: dict["Scope", list[str]]
+    return_scope: Scope
+    invalid_fixtures: dict[Scope, list[str]]
 
 
 class FixtureManager(ManagerAccessMixin):
@@ -106,7 +106,7 @@ def cache_fixtures(self, item) -> "FixtureMap":
 
         return fixture_map
 
-    def get_fixture_scope(self, fixture_name: str) -> Optional["Scope"]:
+    def get_fixture_scope(self, fixture_name: str) -> Optional[Scope]:
         return self._fixture_name_to_info.get(fixture_name, {}).get("scope")
 
     def is_stateful(self, name: str) -> Optional[bool]:
@@ -160,7 +160,7 @@ def add_fixture_info(self, name: str, **info):
     def _get_cached_fixtures(self, nodeid: str) -> Optional["FixtureMap"]:
         return self._nodeid_to_fixture_map.get(nodeid)
 
-    def rebase(self, scope: "Scope", fixtures: "FixtureMap"):
+    def rebase(self, scope: Scope, fixtures: "FixtureMap"):
         if not (rebase := self._get_rebase(scope)):
             # Rebase avoided: nothing would change.
             return
@@ -199,7 +199,7 @@ def rebase(self, scope: "Scope", fixtures: "FixtureMap"):
                 log = f"{log} invalidated-fixtures='{', '.join(invalidated)}'"
                 self.isolation_manager._records.append(log)
 
-    def _get_rebase(self, scope: "Scope") -> Optional[FixtureRebase]:
+    def _get_rebase(self, scope: Scope) -> Optional[FixtureRebase]:
         # Check for fixtures that are now invalid. For example, imagine a session
         # fixture comes into play after the module snapshot has been set.
         # Once we restore the module's state and move to the next module,
@@ -232,10 +232,8 @@ def _get_rebase(self, scope: "Scope") -> Optional[FixtureRebase]:
         )
 
 
-class FixtureMap(dict["Scope", list[str]]):
+class FixtureMap(dict[Scope, list[str]]):
     def __init__(self, item):
-        from ape.pytest.utils import Scope
-
         self._item = item
         self._parametrized_names: Optional[list[str]] = None
         super().__init__(
@@ -269,8 +267,6 @@ def names(self) -> list[str]:
         Outputs in correct order for item.fixturenames.
         Also, injects isolation fixtures if needed.
         """
-        from ape.pytest.utils import Scope
-
         result = []
         for scope, ls in self.items():
             # NOTE: For function scoped, we always add the isolation fixture.
@@ -317,21 +313,15 @@ def parametrized(self) -> dict[str, list]:
     def _arg2fixturedefs(self) -> Mapping:
         return self._item.session._fixturemanager._arg2fixturedefs
 
+    @singledispatchmethod
     def __setitem__(self, key, value):
-        # NOTE: Not using singledispatchmethod because it requires
-        #  types at runtime.
-        if isinstance(key, Scope):
-            self.__setitem_scope(key, value)
-        elif isinstance(key, str):
-            self.__setitem_str(key, value)
-        elif isinstance(key, int):
-            self.__setitem_int(key, value)
-
         raise NotImplementedError(type(key))
 
+    @__setitem__.register
     def __setitem_int(self, key: int, value: list[str]):
         super().__setitem__(Scope(key), value)
 
+    @__setitem__.register
     def __setitem_str(self, key: str, value: list[str]):
         for scope in Scope:
             if f"{scope}" == key:
@@ -340,38 +330,28 @@ def __setitem_str(self, key: str, value: list[str]):
 
         raise KeyError(key)
 
-    def __setitem_scope(self, key: "Scope", value: list[str]):
+    @__setitem__.register
+    def __setitem_scope(self, key: Scope, value: list[str]):
         super().__setitem__(key, value)
 
+    @singledispatchmethod
     def __getitem__(self, key):
-        # NOTE: Not using singledispatchmethod because it requires
-        #  types at runtime.
-        from ape.pytest.utils import Scope
-
-        if isinstance(key, Scope):
-            return self.__getitem_scope(key)
-        elif isinstance(key, str):
-            return self.__getitem_str(key)
-        elif isinstance(key, int):
-            return self.__getitem_int(key)
-
         raise NotImplementedError(type(key))
 
+    @__getitem__.register
     def __getitem_int(self, key: int) -> list[str]:
-        from ape.pytest.utils import Scope
-
         return super().__getitem__(Scope(key))
 
+    @__getitem__.register
     def __getitem_str(self, key: str) -> list[str]:
-        from ape.pytest.utils import Scope
-
         for scope in Scope:
             if f"{scope}" == key:
                 return super().__getitem__(scope)
 
         raise KeyError(key)
 
-    def __getitem_scope(self, key: "Scope") -> list[str]:
+    @__getitem__.register
+    def __getitem_scope(self, key: Scope) -> list[str]:
         return super().__getitem__(key)
 
     def get_info(self, name: str) -> list:
@@ -485,32 +465,22 @@ def Contract(self):
 
     @pytest.fixture(scope="session")
     def _session_isolation(self) -> Iterator[None]:
-        from ape.pytest.utils import Scope
-
         yield from self.isolation_manager.isolation(Scope.SESSION)
 
     @pytest.fixture(scope="package")
     def _package_isolation(self) -> Iterator[None]:
-        from ape.pytest.utils import Scope
-
         yield from self.isolation_manager.isolation(Scope.PACKAGE)
 
     @pytest.fixture(scope="module")
     def _module_isolation(self) -> Iterator[None]:
-        from ape.pytest.utils import Scope
-
         yield from self.isolation_manager.isolation(Scope.MODULE)
 
     @pytest.fixture(scope="class")
     def _class_isolation(self) -> Iterator[None]:
-        from ape.pytest.utils import Scope
-
         yield from self.isolation_manager.isolation(Scope.CLASS)
 
     @pytest.fixture(scope="function")
     def _function_isolation(self) -> Iterator[None]:
-        from ape.pytest.utils import Scope
-
         yield from self.isolation_manager.isolation(Scope.FUNCTION)
 
 
@@ -520,7 +490,7 @@ class Snapshot:
     All the data necessary for accurately supporting isolation.
     """
 
-    scope: "Scope"
+    scope: Scope
     """Corresponds to fixture scope."""
 
     identifier: Optional["SnapshotID"] = None
@@ -537,10 +507,8 @@ def append_fixtures(self, fixtures: Iterable[str]):
             self.fixtures.append(fixture)
 
 
-class SnapshotRegistry(dict["Scope", Snapshot]):
+class SnapshotRegistry(dict[Scope, Snapshot]):
     def __init__(self):
-        from ape.pytest.utils import Scope
-
         super().__init__(
             {
                 Scope.SESSION: Snapshot(Scope.SESSION),
@@ -551,22 +519,20 @@ def __init__(self):
             }
         )
 
-    def get_snapshot_id(self, scope: "Scope") -> Optional["SnapshotID"]:
+    def get_snapshot_id(self, scope: Scope) -> Optional["SnapshotID"]:
         return self[scope].identifier
 
-    def set_snapshot_id(self, scope: "Scope", snapshot_id: "SnapshotID"):
+    def set_snapshot_id(self, scope: Scope, snapshot_id: "SnapshotID"):
         self[scope].identifier = snapshot_id
 
-    def clear_snapshot_id(self, scope: "Scope"):
+    def clear_snapshot_id(self, scope: Scope):
         self[scope].identifier = None
 
-    def next_snapshots(self, scope: "Scope") -> Iterator[Snapshot]:
-        from ape.pytest.utils import Scope
-
+    def next_snapshots(self, scope: Scope) -> Iterator[Snapshot]:
         for scope_value in range(scope + 1, Scope.FUNCTION + 1):
             yield self[scope_value]  # type: ignore
 
-    def extend_fixtures(self, scope: "Scope", fixtures: Iterable[str]):
+    def extend_fixtures(self, scope: Scope, fixtures: Iterable[str]):
         self[scope].fixtures.extend(fixtures)
 
 
@@ -587,16 +553,16 @@ def _track_transactions(self) -> bool:
             and (self.config_wrapper.track_gas or self.config_wrapper.track_coverage)
         )
 
-    def get_snapshot(self, scope: "Scope") -> Snapshot:
+    def get_snapshot(self, scope: Scope) -> Snapshot:
         return self.snapshots[scope]
 
-    def extend_fixtures(self, scope: "Scope", fixtures: Iterable[str]):
+    def extend_fixtures(self, scope: Scope, fixtures: Iterable[str]):
         self.snapshots.extend_fixtures(scope, fixtures)
 
-    def next_snapshots(self, scope: "Scope") -> Iterator[Snapshot]:
+    def next_snapshots(self, scope: Scope) -> Iterator[Snapshot]:
         yield from self.snapshots.next_snapshots(scope)
 
-    def isolation(self, scope: "Scope") -> Iterator[None]:
+    def isolation(self, scope: Scope) -> Iterator[None]:
         """
         Isolation logic used to implement isolation fixtures for each pytest scope.
         When tracing support is available, will also assist in capturing receipts.
@@ -623,7 +589,7 @@ def isolation(self, scope: "Scope") -> Iterator[None]:
 
         self.restore(scope)
 
-    def set_snapshot(self, scope: "Scope"):
+    def set_snapshot(self, scope: Scope):
         # Also can be used to re-set snapshot.
         if not self.supported:
             return
@@ -654,7 +620,7 @@ def take_snapshot(self) -> Optional["SnapshotID"]:
         return None
 
     @allow_disconnected
-    def restore(self, scope: "Scope"):
+    def restore(self, scope: Scope):
         snapshot_id = self.snapshots.get_snapshot_id(scope)
         if snapshot_id is None:
             return
diff --git a/src/ape/pytest/runners.py b/src/ape/pytest/runners.py
index d6dfea143f..b77cee7871 100644
--- a/src/ape/pytest/runners.py
+++ b/src/ape/pytest/runners.py
@@ -4,14 +4,16 @@
 import click
 import pytest
 from _pytest._code.code import Traceback as PytestTraceback
-from _pytest.reports import TestReport
 from rich import print as rich_print
 
 from ape.exceptions import ConfigError, ProviderNotConnectedError
 from ape.logging import LogLevel
+from ape.pytest.utils import Scope
 from ape.utils.basemodel import ManagerAccessMixin
 
 if TYPE_CHECKING:
+    from _pytest.reports import TestReport
+
     from ape.api.networks import ProviderContextManager
     from ape.pytest.config import ConfigWrapper
     from ape.pytest.coverage import CoverageTracker
@@ -171,8 +173,6 @@ def pytest_runtest_setup(self, item):
             self._setup_isolation(item)
 
     def _setup_isolation(self, item):
-        from ape.pytest.utils import Scope
-
         fixtures = self.fixture_manager.get_fixtures(item)
         for scope in (Scope.SESSION, Scope.PACKAGE, Scope.MODULE, Scope.CLASS):
             if not (
@@ -273,8 +273,6 @@ def pytest_fixture_post_finalizer(self, fixturedef, request):
                 self.fixture_manager.add_fixture_info(fixture_name, teardown_block=block_number)
 
     def _track_fixture_blocks(self, fixture_name: str) -> bool:
-        from ape.pytest.utils import Scope
-
         if not self.fixture_manager.is_custom(fixture_name):
             return False
 
@@ -312,7 +310,7 @@ def _connect(self):
         self._provider_context.push_provider()
         self._provider_is_connected = True
 
-    def pytest_runtest_logreport(self, report: TestReport):
+    def pytest_runtest_logreport(self, report: "TestReport"):
         if self.config_wrapper.verbosity >= 3:
             self.isolation_manager.show_records()
 

From a2e9cb693cd777e47713b72e14f7b96145cce411 Mon Sep 17 00:00:00 2001
From: Juliya Smith <jules@apeworx.io>
Date: Tue, 5 Nov 2024 11:01:09 -0600
Subject: [PATCH 3/3] test: ify

---
 tests/functional/test_provider.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/functional/test_provider.py b/tests/functional/test_provider.py
index 60f3f8b698..9afcbe34fb 100644
--- a/tests/functional/test_provider.py
+++ b/tests/functional/test_provider.py
@@ -322,8 +322,9 @@ def test_gas_price(eth_tester_provider):
 
 def test_get_code(eth_tester_provider, vyper_contract_instance):
     address = vyper_contract_instance.address
+    block_number = vyper_contract_instance.creation_metadata.block
     assert eth_tester_provider.get_code(address) == eth_tester_provider.get_code(
-        address, block_id=1
+        address, block_id=block_number
     )