perf: make ape test --help faster

src/ape_test/__init__.py

@@ -5,20 +5,23 @@
 
 @plugins.register(plugins.Config)
 def config_class():
-    module = import_module("ape_test.config")
-    return module.ApeTestConfig
+    from ape_test.config import ApeTestConfig
+
+    return ApeTestConfig
 
 
 @plugins.register(plugins.AccountPlugin)
 def account_types():
-    module = import_module("ape_test.accounts")
-    return module.TestAccountContainer, module.TestAccount
+    from ape_test.accounts import TestAccount, TestAccountContainer
+
+    return TestAccountContainer, TestAccount
 
 
 @plugins.register(plugins.ProviderPlugin)
 def providers():
-    module = import_module("ape_test.provider")
-    yield "ethereum", "local", module.LocalProvider
+    from ape_test.provider import LocalProvider
+
+    yield "ethereum", "local", LocalProvider
 
 
 def __getattr__(name: str):

src/ape_test/_cli.py

@@ -1,80 +1,14 @@
 import sys
-import threading
-import time
-from datetime import datetime, timedelta
-from functools import cached_property
+from collections.abc import Iterable
 from pathlib import Path
-from subprocess import run as run_subprocess
 from typing import Any
 
 import click
 import pytest
 from click import Command
-from watchdog import events
-from watchdog.observers import Observer
 
 from ape.cli.options import ape_cli_context
 from ape.logging import LogLevel, _get_level
-from ape.utils.basemodel import ManagerAccessMixin as access
-
-# Copied from https://github.com/olzhasar/pytest-watcher/blob/master/pytest_watcher/watcher.py
-trigger_lock = threading.Lock()
-trigger = None
-
-
-def emit_trigger():
-    """
-    Emits trigger to run pytest
-    """
-
-    global trigger
-
-    with trigger_lock:
-        trigger = datetime.now()
-
-
-class EventHandler(events.FileSystemEventHandler):
-    EVENTS_WATCHED = (
-        events.EVENT_TYPE_CREATED,
-        events.EVENT_TYPE_DELETED,
-        events.EVENT_TYPE_MODIFIED,
-        events.EVENT_TYPE_MOVED,
-    )
-
-    def dispatch(self, event: events.FileSystemEvent) -> None:
-        if event.event_type in self.EVENTS_WATCHED:
-            self.process_event(event)
-
-    @cached_property
-    def _extensions_to_watch(self) -> list[str]:
-        return [".py", *access.compiler_manager.registered_compilers.keys()]
-
-    def _is_path_watched(self, filepath: str) -> bool:
-        """
-        Check if file should trigger pytest run
-        """
-        return any(map(filepath.endswith, self._extensions_to_watch))
-
-    def process_event(self, event: events.FileSystemEvent) -> None:
-        if self._is_path_watched(event.src_path):
-            emit_trigger()
-
-
-def _run_ape_test(*pytest_args):
-    return run_subprocess(["ape", "test", *[f"{a}" for a in pytest_args]])
-
-
-def _run_main_loop(delay: float, *pytest_args: str) -> None:
-    global trigger
-
-    now = datetime.now()
-    if trigger and now - trigger > timedelta(seconds=delay):
-        _run_ape_test(*pytest_args)
-
-        with trigger_lock:
-            trigger = None
-
-    time.sleep(delay)
 
 
 def _validate_pytest_args(*pytest_args) -> list[str]:
@@ -176,25 +110,7 @@ def cli(cli_ctx, watch, watch_folders, watch_delay, pytest_args):
 
     pytest_arg_ls = _validate_pytest_args(*pytest_arg_ls)
     if watch:
-        event_handler = _create_event_handler()
-        observer = _create_observer()
-
-        for folder in watch_folders:
-            if folder.is_dir():
-                observer.schedule(event_handler, folder, recursive=True)
-            else:
-                cli_ctx.logger.warning(f"Folder '{folder}' doesn't exist or isn't a folder.")
-
-        observer.start()
-
-        try:
-            _run_ape_test(*pytest_arg_ls)
-            while True:
-                _run_main_loop(watch_delay, *pytest_arg_ls)
-
-        finally:
-            observer.stop()
-            observer.join()
+        _run_with_observer(watch_folders, watch_delay, *pytest_arg_ls)
 
     else:
         return_code = pytest.main([*pytest_arg_ls], ["ape_test"])
@@ -203,11 +119,8 @@ def cli(cli_ctx, watch, watch_folders, watch_delay, pytest_args):
             sys.exit(return_code)
 
 
-def _create_event_handler():
+def _run_with_observer(watch_folders: Iterable[Path], watch_delay: float, *pytest_arg_ls: str):
     # Abstracted for testing purposes.
-    return EventHandler()
-
+    from ape_test._watch import run_with_observer as run
 
-def _create_observer():
-    # Abstracted for testing purposes.
-    return Observer()
+    run(watch_folders, watch_delay, *pytest_arg_ls)

src/ape_test/_watch.py

@@ -0,0 +1,105 @@
+import threading
+import time
+from collections.abc import Iterable
+from datetime import datetime, timedelta
+from functools import cached_property
+from pathlib import Path
+from subprocess import run as run_subprocess
+
+from watchdog import events
+from watchdog.observers import Observer
+
+from ape.logging import logger
+
+# Copied from https://github.com/olzhasar/pytest-watcher/blob/master/pytest_watcher/watcher.py
+trigger_lock = threading.Lock()
+trigger = None
+
+
+def run_with_observer(watch_folders: Iterable[Path], watch_delay: float, *pytest_arg_ls: str):
+    event_handler = _create_event_handler()
+    observer = _create_observer()
+
+    for folder in watch_folders:
+        if folder.is_dir():
+            observer.schedule(event_handler, folder, recursive=True)
+        else:
+            logger.warning(f"Folder '{folder}' doesn't exist or isn't a folder.")
+
+    observer.start()
+
+    try:
+        _run_ape_test(*pytest_arg_ls)
+        while True:
+            _run_main_loop(watch_delay, *pytest_arg_ls)
+
+    finally:
+        observer.stop()
+        observer.join()
+
+
+def emit_trigger():
+    """
+    Emits trigger to run pytest
+    """
+
+    global trigger
+
+    with trigger_lock:
+        trigger = datetime.now()
+
+
+class EventHandler(events.FileSystemEventHandler):
+    EVENTS_WATCHED = (
+        events.EVENT_TYPE_CREATED,
+        events.EVENT_TYPE_DELETED,
+        events.EVENT_TYPE_MODIFIED,
+        events.EVENT_TYPE_MOVED,
+    )
+
+    def dispatch(self, event: events.FileSystemEvent) -> None:
+        if event.event_type in self.EVENTS_WATCHED:
+            self.process_event(event)
+
+    @cached_property
+    def _extensions_to_watch(self) -> list[str]:
+        from ape.utils.basemodel import ManagerAccessMixin as access
+
+        return [".py", *access.compiler_manager.registered_compilers.keys()]
+
+    def _is_path_watched(self, filepath: str) -> bool:
+        """
+        Check if file should trigger pytest run
+        """
+        return any(map(filepath.endswith, self._extensions_to_watch))
+
+    def process_event(self, event: events.FileSystemEvent) -> None:
+        if self._is_path_watched(event.src_path):
+            emit_trigger()
+
+
+def _run_ape_test(*pytest_args):
+    return run_subprocess(["ape", "test", *[f"{a}" for a in pytest_args]])
+
+
+def _run_main_loop(delay: float, *pytest_args: str) -> None:
+    global trigger
+
+    now = datetime.now()
+    if trigger and now - trigger > timedelta(seconds=delay):
+        _run_ape_test(*pytest_args)
+
+        with trigger_lock:
+            trigger = None
+
+    time.sleep(delay)
+
+
+def _create_event_handler():
+    # Abstracted for testing purposes.
+    return EventHandler()
+
+
+def _create_observer():
+    # Abstracted for testing purposes.
+    return Observer()

tests/functional/test_test.py

@@ -1,8 +1,11 @@
+from pathlib import Path
+
 import pytest
 
 from ape.exceptions import ConfigError
 from ape.pytest.runners import PytestApeRunner
 from ape_test import ApeTestConfig
+from ape_test._watch import run_with_observer
 
 
 class TestApeTestConfig:
@@ -33,3 +36,30 @@ def test_connect_to_mainnet_by_default(mocker):
     )
     with pytest.raises(ConfigError, match=expected):
         runner._connect()
+
+
+def test_watch(mocker):
+    mock_event_handler = mocker.MagicMock()
+    event_handler_patch = mocker.patch("ape_test._watch._create_event_handler")
+    event_handler_patch.return_value = mock_event_handler
+
+    mock_observer = mocker.MagicMock()
+    observer_patch = mocker.patch("ape_test._watch._create_observer")
+    observer_patch.return_value = mock_observer
+
+    run_subprocess_patch = mocker.patch("ape_test._watch.run_subprocess")
+    run_main_loop_patch = mocker.patch("ape_test._watch._run_main_loop")
+    run_main_loop_patch.side_effect = SystemExit  # Avoid infinite loop.
+
+    # Only passing `-s` so we have an extra arg to test.
+    with pytest.raises(SystemExit):
+        run_with_observer((Path("contracts"),), 0.1, "-s")
+
+    # The observer started, then the main runner exits, and the observer stops + joins.
+    assert mock_observer.start.call_count == 1
+    assert mock_observer.stop.call_count == 1
+    assert mock_observer.join.call_count == 1
+
+    # NOTE: We had a bug once where the args it received were not strings.
+    #   (wasn't deconstructing), so this check is important.
+    run_subprocess_patch.assert_called_once_with(["ape", "test", "-s"])

tests/integration/cli/test_test.py

@@ -435,27 +435,10 @@ def test_fails():
 
 @skip_projects_except("with-contracts")
 def test_watch(mocker, integ_project, runner, ape_cli):
-    mock_event_handler = mocker.MagicMock()
-    event_handler_patch = mocker.patch("ape_test._cli._create_event_handler")
-    event_handler_patch.return_value = mock_event_handler
-
-    mock_observer = mocker.MagicMock()
-    observer_patch = mocker.patch("ape_test._cli._create_observer")
-    observer_patch.return_value = mock_observer
-
-    run_subprocess_patch = mocker.patch("ape_test._cli.run_subprocess")
-    run_main_loop_patch = mocker.patch("ape_test._cli._run_main_loop")
-    run_main_loop_patch.side_effect = SystemExit  # Avoid infinite loop.
+    runner_patch = mocker.patch("ape_test._cli._run_with_observer")
 
     # Only passing `-s` so we have an extra arg to test.
     result = runner.invoke(ape_cli, ("test", "--watch", "-s"))
     assert result.exit_code == 0
 
-    # The observer started, then the main runner exits, and the observer stops + joins.
-    assert mock_observer.start.call_count == 1
-    assert mock_observer.stop.call_count == 1
-    assert mock_observer.join.call_count == 1
-
-    # NOTE: We had a bug once where the args it received were not strings.
-    #   (wasn't deconstructing), so this check is important.
-    run_subprocess_patch.assert_called_once_with(["ape", "test", "-s"])
+    runner_patch.assert_called_once_with((Path("contracts"), Path("tests")), 0.5, "-s")