wrapt 2.2.2: Register post-import hooks and resolve target attribute paths for deferred or immediate module monkey-patching

# wrapt Post-Import Hooks and Path Resolution Contracts

## Prior Search Results
A search via `search_known_solution` for `wrapt` returned existing samples:
- `sha256:7a13d82fc8fc99b353d10c003fde6ab931c335ba8d42b8cf245f5a18f4524a1d`: focused on `@wrapt.decorator` 4-argument signature protocol and `wrap_function_wrapper` anti-patterns.
- `sha256:7ea6f89cb5cb7fdfeb8506e92a3a052d6c25e5a6ef58d5cde04cede70d7b5656`: focused on `WRAPT_DISABLE_EXTENSIONS` environment variable handling and decorator `enabled` flags.
- `sha256:8daf2a76a9fa4801ddad9b3f8d796f82d1757a4ab55890032b9c836fb6b694d8`: focused on `@wrapt.lru_cache` method recursion on inheritance.

To avoid duplicating existing coverage, this sample proves the behavior of `wrapt`'s post-import hook subsystem and path-resolution patching utilities (`wrapt.register_post_import_hook`, `wrapt.when_imported`, `wrapt.resolve_path`, `wrapt.patch_function_wrapper`, `wrapt.apply_patch`, and `wrapt.notify_module_loaded`).

## Target Release
- Pinned release: `wrapt==2.2.2` (PyPI)

## Observed Behaviour
1. **Deferred Hook Triggering**: When `@wrapt.when_imported(module_name)` or `wrapt.register_post_import_hook(hook, module_name)` is registered for a module not yet in `sys.modules`, the hook registration registers an import hook in Python's import finder chain (`sys.meta_path`). The hook is deferred until the module is first imported.
2. **Immediate Execution for Loaded Modules**: If the target module is already present in `sys.modules` at registration time, `wrapt.when_imported` executes the callback synchronously during the decorator/function call.
3. **FIFO Hook Sequence**: Multiple post-import hooks registered for the same target module execute sequentially in order of registration upon import.
4. **Target Path Resolution**: `wrapt.resolve_path(module_or_name, dotted_path)` resolves nested dot-separated attribute paths (such as nested classes or methods) and returns a 3-tuple: `(parent_namespace, attribute_name, target_object)`. Resolving a non-existent path segment raises `AttributeError`.
5. **In-Place Function Patching**: `@wrapt.patch_function_wrapper(parent, attribute_name)` decorates a 4-argument wrapper `(wrapped, instance, args, kwargs)` and replaces the target callable on `parent` in-place.
6. **Namespace Patch Swapping & Restorations**: `wrapt.apply_patch(parent, attribute_name, replacement)` mutates the attribute on `parent` with the replacement object and allows restoring the original attribute reference.
7. **Manual Module Notification**: `wrapt.notify_module_loaded(module_object)` directly fires registered hooks for manually instantiated module objects without requiring standard filesystem imports.

{"case":{"caseId":"case:sha256:8f99607c43c96ebfa3fe43a206c27fd313537747dcf8daa8228febae59091f23","contract":["wrapt.when_imported defers hook execution until the module is loaded via the import machinery or executes immediately if already in sys.modules","assert multiple registered post-import hooks execute sequentially in order of registration","assert wrapt.resolve_path returns a tuple of parent object, attribute name, and target reference for dotted target strings","assert wrapt.resolve_path raises AttributeError when resolving non-existent path segments","assert @wrapt.patch_function_wrapper installs a 4-argument wrapper on a resolved target function in-place","assert wrapt.apply_patch replaces the attribute on the parent namespace and allows unpatching to the original target","assert wrapt.notify_module_loaded triggers pending hooks for manually instantiated module objects"],"goal":"Register post-import hooks and resolve target attribute paths for deferred or immediate module monkey-patching","kind":"HOW","packages":["pkg:pypi/wrapt@2.2.2"],"schemaVersion":1,"symbols":["wrapt.register_post_import_hook","wrapt.when_imported","wrapt.resolve_path","wrapt.patch_function_wrapper","wrapt.apply_patch","wrapt.notify_module_loaded"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/wrapt@2.2.2"],"schemaVersion":1,"symbols":["wrapt.register_post_import_hook","wrapt.when_imported","wrapt.resolve_path","wrapt.patch_function_wrapper","wrapt.apply_patch","wrapt.notify_module_loaded"],"verifierAdapter":"python@1"}

wrapt==2.2.2

"""Sample package for testing wrapt post-import hooks and monkey-patching."""

"""Hook management and dynamic patching helpers utilizing wrapt."""

import wrapt


def register_audit_hook(module_name: str, target_attr: str, audit_log: list):
    """Register a post-import hook to audit invocations of target_attr."""
    @wrapt.when_imported(module_name)
    def _hook(module):
        parent, name, target = wrapt.resolve_path(module, target_attr)

        @wrapt.patch_function_wrapper(parent, name)
        def _wrapper(wrapped, instance, args, kwargs):
            result = wrapped(*args, **kwargs)
            audit_log.append({
                "target": f"{parent.__name__}.{name}" if hasattr(parent, "__name__") else name,
                "has_instance": instance is not None,
                "result": result,
            })
            return result

"""Target service classes and helper functions for post-import patching."""


class DataFormatter:
    """Service formatting records."""

    def __init__(self, tag: str = "default"):
        self.tag = tag

    def format_entry(self, key: str, value: str) -> str:
        return f"[{self.tag}] {key}={value}"


def aggregate_values(numbers: list) -> dict:
    """Compute basic sum and count for numeric values."""
    if not numbers:
        return {"count": 0, "total": 0}
    return {"count": len(numbers), "total": sum(numbers)}


class ProcessingPipeline:
    """Hierarchical pipeline with nested processing steps."""

    class Transformer:
        """Inner transformer component."""

        def transform(self, text: str) -> str:
            return text.strip().upper()

"""Test package for contract suite."""

"""Contract verification suite for wrapt post-import hooks, path resolution, and monkey-patching."""

import os
import sys
import types
import wrapt

# Ensure src package is discoverable
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))


def test_deferred_and_immediate_hooks():
    """Verify wrapt.when_imported defers hook execution until import or runs immediately if loaded."""
    execution_order = []

    # 1. Register hook on an unimported module (src.services is not yet in sys.modules)
    assert "src.services" not in sys.modules

    @wrapt.when_imported("src.services")
    def on_services_loaded(module):
        execution_order.append(("deferred_hook", module.__name__))

    assert len(execution_order) == 0, "Hook should not execute before module import"

    # Now import src.services - the deferred hook must fire during import
    import src.services
    assert "src.services" in sys.modules
    assert execution_order == [("deferred_hook", "src.services")], (
        "Deferred hook must execute when module is loaded"
    )

    # 2. Register hook on an already-imported module - must fire immediately
    @wrapt.when_imported("src.services")
    def on_services_already_loaded(module):
        execution_order.append(("immediate_hook", module.__name__))

    assert execution_order == [
        ("deferred_hook", "src.services"),
        ("immediate_hook", "src.services"),
    ], "Hook on already imported module must execute immediately upon registration"


def test_multiple_hooks_sequence():
    """Verify multiple registered post-import hooks execute sequentially in order of registration."""
    sequence = []

    wrapt.register_post_import_hook(lambda m: sequence.append("hook_1"), "synthetic_mod_a")
    wrapt.register_post_import_hook(lambda m: sequence.append("hook_2"), "synthetic_mod_a")
    wrapt.register_post_import_hook(lambda m: sequence.append("hook_3"), "synthetic_mod_a")

    mod_a = types.ModuleType("synthetic_mod_a")
    sys.modules["synthetic_mod_a"] = mod_a
    wrapt.notify_module_loaded(mod_a)

    assert sequence == ["hook_1", "hook_2", "hook_3"], (
        "Registered post-import hooks must execute in first-in first-out registration order"
    )


def test_resolve_path():
    """Verify wrapt.resolve_path resolves nested dotted target strings to (parent, name, target)."""
    import src.services

    # Resolve nested class method from module object
    parent, name, target = wrapt.resolve_path(
        src.services, "ProcessingPipeline.Transformer.transform"
    )
    assert parent is src.services.ProcessingPipeline.Transformer
    assert name == "transform"
    assert target is src.services.ProcessingPipeline.Transformer.transform
    assert target(src.services.ProcessingPipeline.Transformer(), "  hello  ") == "HELLO"

    # Resolve from module name string
    p2, n2, t2 = wrapt.resolve_path(
        "src.services", "DataFormatter.format_entry"
    )
    assert p2 is src.services.DataFormatter
    assert n2 == "format_entry"
    assert t2 is src.services.DataFormatter.format_entry


def test_resolve_path_error():
    """Verify wrapt.resolve_path raises AttributeError when resolving non-existent path segments."""
    import src.services

    try:
        wrapt.resolve_path(src.services, "NonExistentClass.method")
        raise AssertionError("resolve_path did not raise AttributeError on missing class")
    except AttributeError:
        pass

    try:
        wrapt.resolve_path(src.services, "ProcessingPipeline.NonExistentMethod")
        raise AssertionError("resolve_path did not raise AttributeError on missing method")
    except AttributeError:
        pass


def test_patch_function_wrapper():
    """Verify @wrapt.patch_function_wrapper installs a 4-argument wrapper on a resolved target in-place."""
    import src.services

    formatter = src.services.DataFormatter(tag="audit")

    # Before patching
    initial_res = formatter.format_entry("status", "ok")
    assert initial_res == "[audit] status=ok"

    # Apply patch_function_wrapper
    @wrapt.patch_function_wrapper(src.services.DataFormatter, "format_entry")
    def wrap_format_entry(wrapped, instance, args, kwargs):
        assert instance is not None, "instance should be passed for method invocations"
        res = wrapped(*args, **kwargs)
        return f"WRAPPED({res})"

    patched_res = formatter.format_entry("status", "ok")
    assert patched_res == "WRAPPED([audit] status=ok)", (
        "patch_function_wrapper must intercept method execution"
    )


def test_apply_patch():
    """Verify wrapt.apply_patch replaces the attribute on the parent namespace and allows unpatching."""
    import src.services

    original_fn = src.services.aggregate_values
    assert original_fn([10, 20]) == {"count": 2, "total": 30}

    def replacement_fn(numbers):
        return {"mocked": True, "count": len(numbers)}

    # Apply custom replacement
    wrapt.apply_patch(src.services, "aggregate_values", replacement_fn)
    assert src.services.aggregate_values([10, 20]) == {"mocked": True, "count": 2}

    # Restore original using apply_patch
    wrapt.apply_patch(src.services, "aggregate_values", original_fn)
    assert src.services.aggregate_values([10, 20]) == {"count": 2, "total": 30}
    assert src.services.aggregate_values is original_fn


def test_notify_module_loaded():
    """Verify wrapt.notify_module_loaded triggers pending hooks for manually instantiated module objects."""
    received_modules = []

    wrapt.register_post_import_hook(
        lambda m: received_modules.append(m.__name__),
        "synthetic.manual_target",
    )

    assert len(received_modules) == 0

    manual_mod = types.ModuleType("synthetic.manual_target")
    manual_mod.version = "1.0.0"

    wrapt.notify_module_loaded(manual_mod)

    assert received_modules == ["synthetic.manual_target"], (
        "notify_module_loaded must execute pending hooks for the loaded module instance"
    )


def main():
    test_deferred_and_immediate_hooks()
    test_multiple_hooks_sequence()
    test_resolve_path()
    test_resolve_path_error()
    test_patch_function_wrapper()
    test_apply_patch()
    test_notify_module_loaded()
    print("All wrapt post-import hook and monkey-patching contract assertions passed.")


if __name__ == "__main__":
    main()