Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions pyrit/registry/discovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,9 @@ def discover_in_directory(
Discover all subclasses of base_class in a directory by loading Python files.

This function walks a directory, loads Python files dynamically, and yields
any classes that are subclasses of the specified base_class.
any classes that are subclasses of the specified base_class. Members are
inspected statically, with dynamic lookup limited to exports declared in
the module's ``__all__``.

Args:
directory: The directory to search for Python files.
Expand Down Expand Up @@ -70,8 +72,21 @@ def _process_file(*, file_path: Path, base_class: type[T]) -> Iterator[tuple[str
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)

for attr_name in dir(module):
attr = getattr(module, attr_name)
static_members = dict(inspect.getmembers_static(module))
member_names = set(static_members)
declared_exports = vars(module).get("__all__", ())
if isinstance(declared_exports, (list, tuple)):
member_names.update(name for name in declared_exports if isinstance(name, str))

for attr_name in sorted(member_names):
if attr_name in static_members:
attr = static_members[attr_name]
else:
try:
attr = getattr(module, attr_name)
except Exception as error:
logger.warning(f"Failed to load declared export '{attr_name}' from {file_path}: {error}")
continue
if (
inspect.isclass(attr)
and issubclass(attr, base_class)
Expand Down
97 changes: 97 additions & 0 deletions tests/unit/registry/test_initializer_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from pyrit.models.parameter import Parameter
from pyrit.registry.components.initializer_registry import PYRIT_PATH, InitializerRegistry
from pyrit.registry.custom_initializer_storage import CustomInitializerStorage
from pyrit.registry.discovery import discover_in_directory
from pyrit.setup.configuration_loader import ConfigurationLoader
from pyrit.setup.initialization import _execute_initializers_async
from pyrit.setup.pyrit_initializer import PyRITInitializer
Expand Down Expand Up @@ -525,6 +526,102 @@ def test_discover_directory_registers_and_lists_metadata():
assert any(m.registry_name == "solo" for m in metadata)


def test_discover_directory_does_not_evaluate_dynamic_module_attributes(caplog):
"""Test that discovery inspects ordinary module members without invoking dynamic attributes."""
script = (
"from pyrit.setup.pyrit_initializer import PyRITInitializer\n\n"
"class SafeInitializer(PyRITInitializer):\n"
" async def initialize_async(self) -> None:\n"
" pass\n\n"
"def __dir__():\n"
" return ['SafeInitializer', 'dynamic_attribute']\n\n"
"def __getattr__(name):\n"
" raise RuntimeError(f'dynamic attribute evaluated: {name}')\n"
)

with tempfile.TemporaryDirectory() as temp_dir:
script_path = Path(temp_dir) / "dynamic.py"
script_path.write_text(script)

discovered = list(discover_in_directory(directory=Path(temp_dir), base_class=PyRITInitializer))

assert [item[2] for item in discovered][0].__name__ == "SafeInitializer"
assert "Failed to load module" not in caplog.text


@pytest.mark.parametrize("include_lazy_export_in_dir", [False, True])
def test_discover_directory_registers_declared_lazy_exports(
tmp_path: Path, include_lazy_export_in_dir: bool, caplog: pytest.LogCaptureFixture
) -> None:
script = _SOLO_SCRIPT + (
"\n__all__ = ['SoloInitializer', 'TargetInitializer', 'non_class']\n"
"def __dir__():\n"
f" return ['SoloInitializer', 'unrelated'] + "
f"{['TargetInitializer'] if include_lazy_export_in_dir else []!r}\n"
"def __getattr__(name):\n"
" if name == 'TargetInitializer':\n"
" from pyrit.setup.initializers.targets import TargetInitializer\n"
" return TargetInitializer\n"
" if name == 'non_class':\n"
" return 42\n"
" raise RuntimeError(f'unexpected dynamic lookup: {name}')\n"
)
(tmp_path / "lazy.py").write_text(script, encoding="utf-8")

registry = InitializerRegistry(discovery_path=tmp_path)

assert set(registry.get_class_names()) == {"solo", "target"}
assert registry.get_class("target").__name__ == "TargetInitializer"
assert registry.is_builtin("target")
assert not caplog.records


@pytest.mark.parametrize("error_type", ["AttributeError", "RuntimeError"])
def test_discover_directory_logs_failed_lazy_export_and_keeps_other_classes(
tmp_path: Path, caplog: pytest.LogCaptureFixture, error_type: str
) -> None:
script = _SOLO_SCRIPT + (
"\n__all__ = ['BrokenInitializer', 'SoloInitializer', 'TargetInitializer']\n"
"def __getattr__(name):\n"
" if name == 'TargetInitializer':\n"
" from pyrit.setup.initializers.targets import TargetInitializer\n"
" return TargetInitializer\n"
f" raise {error_type}('export unavailable')\n"
)
(tmp_path / "lazy.py").write_text(script, encoding="utf-8")

registry = InitializerRegistry(discovery_path=tmp_path)

assert set(registry.get_class_names()) == {"solo", "target"}
assert "BrokenInitializer" in caplog.text
assert "export unavailable" in caplog.text
assert str(tmp_path / "lazy.py") in caplog.text


def test_discover_directory_resolves_declared_lazy_initializer_export(caplog):
"""Test that discovery resolves an initializer exposed through a module's __all__."""
script = (
"from pyrit.setup.pyrit_initializer import PyRITInitializer\n\n"
"__all__ = ['LazyInitializer']\n\n"
"def __getattr__(name):\n"
" if name != 'LazyInitializer':\n"
" raise AttributeError(name)\n\n"
" class LazyInitializer(PyRITInitializer):\n"
" async def initialize_async(self) -> None:\n"
" pass\n\n"
" return LazyInitializer\n"
)

with tempfile.TemporaryDirectory() as temp_dir:
script_path = Path(temp_dir) / "lazy.py"
script_path.write_text(script)

discovered = list(discover_in_directory(directory=Path(temp_dir), base_class=PyRITInitializer))

assert [item[2] for item in discovered][0].__name__ == "LazyInitializer"
assert "Failed to load module" not in caplog.text


def test_discover_single_file_registers_builtin():
"""Test that a discovery path pointing at a single file registers it as built-in."""
with tempfile.TemporaryDirectory() as temp_dir:
Expand Down
Loading