pluggy 1.6.0: PluginManager, HookspecMarker, HookimplMarker

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify pkg:pypi/pluggy@1.6.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/pluggy@1.6.0
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - packageManager: npm@10.9.8
  - runtime: node@22.23.2

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:1970384a87c71912c512a0ba9c1d88e2e771f99fcf427be59d2237641b848901","contract":["PluginManager discovers and executes hook implementations with tryfirst and trylast ordering","Hook specifications with firstresult=True return the first non-None result from registered implementations","Modern generator hook wrappers with wrapper=True and legacy hook wrappers with hookwrapper=True intercept and modify hook outcomes","Historic hooks record execution history and replay events to plugins registered after the call","PluginManager validates implementation argument specifications and raises PluginValidationError on unknown arguments","PluginManager controls plugin lifecycle including registration, unregistration, blocking, and subset caller creation"],"goal":"verify pkg:pypi/pluggy@1.6.0","kind":"HOW","packages":["pkg:pypi/pluggy@1.6.0"],"schemaVersion":1,"symbols":["PluginManager","HookspecMarker","HookimplMarker","PluginValidationError"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/pluggy@1.6.0"],"schemaVersion":1,"subject":"pkg:pypi/pluggy@1.6.0","symbols":["PluginManager","HookspecMarker","HookimplMarker","PluginValidationError"],"verifierAdapter":"python@1"}

[project]
name = "verify-pluggy"
version = "0.1.0"
dependencies = [
    "pluggy==1.6.0",
]

pluggy==1.6.0

{
  "schemaVersion": 1,
  "goal": "verify pkg:pypi/pluggy@1.6.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/pluggy@1.6.0"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "packageManager": "npm@10.9.8",
    "runtime": "node@22.23.2"
  }
}

"""Plugin management and hook specification module using pluggy."""

from typing import List, Optional
import pluggy

hookspec = pluggy.HookspecMarker("pluggy_demo")
hookimpl = pluggy.HookimplMarker("pluggy_demo")


class CalculationSpec:
    """Hook specifications for calculation operations."""

    @hookspec
    def compute(self, a: int, b: int) -> int:
        """Compute an integer result given a and b."""

    @hookspec(firstresult=True)
    def resolve_status(self, code: int) -> Optional[str]:
        """Resolve a status string for a given code, returning first non-None."""

    @hookspec(historic=True)
    def on_event(self, event_name: str) -> None:
        """Broadcast an event historically to current and future plugins."""


class AddPlugin:
    """Plugin implementing standard addition."""

    @hookimpl
    def compute(self, a: int, b: int) -> int:
        return a + b


class MultiplyPriorityPlugin:
    """Plugin implementing multiplication with tryfirst priority."""

    @hookimpl(tryfirst=True)
    def compute(self, a: int, b: int) -> int:
        return a * b


class SubtractFallbackPlugin:
    """Plugin implementing subtraction with trylast priority."""

    @hookimpl(trylast=True)
    def compute(self, a: int, b: int) -> int:
        return a - b


class StatusResolverPluginA:
    """Plugin resolving HTTP 200."""

    @hookimpl
    def resolve_status(self, code: int) -> Optional[str]:
        if code == 200:
            return "OK"
        return None


class StatusResolverPluginB:
    """Plugin resolving HTTP 404 with high priority."""

    @hookimpl(tryfirst=True)
    def resolve_status(self, code: int) -> Optional[str]:
        if code == 404:
            return "NOT_FOUND"
        return None


class ModernWrapperPlugin:
    """Modern wrapper plugin transforming execution outcomes."""

    @hookimpl(wrapper=True)
    def compute(self, a: int, b: int):
        results = yield
        return [item * 10 for item in results]


class LegacyWrapperPlugin:
    """Legacy hookwrapper plugin altering result via Result object."""

    @hookimpl(hookwrapper=True)
    def compute(self, a: int, b: int):
        outcome = yield
        raw_result = outcome.get_result()
        outcome.force_result([item + 1 for item in raw_result])


def create_plugin_manager() -> pluggy.PluginManager:
    """Create and initialize a PluginManager with CalculationSpec."""
    pm = pluggy.PluginManager("pluggy_demo")
    pm.add_hookspecs(CalculationSpec)
    return pm

"""Contract verification test for pkg:pypi/pluggy@1.6.0."""

import os
import sys

# Ensure repository root is on sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import pluggy
from src.plugin_system import (
    AddPlugin,
    CalculationSpec,
    LegacyWrapperPlugin,
    ModernWrapperPlugin,
    MultiplyPriorityPlugin,
    StatusResolverPluginA,
    StatusResolverPluginB,
    SubtractFallbackPlugin,
    create_plugin_manager,
    hookimpl,
    hookspec,
)


def test_hook_execution_ordering():
    """Verify PluginManager discovers and executes hook implementations with tryfirst and trylast ordering."""
    pm = create_plugin_manager()
    pm.register(AddPlugin(), name="add")
    pm.register(MultiplyPriorityPlugin(), name="multiply")
    pm.register(SubtractFallbackPlugin(), name="subtract")

    results = pm.hook.compute(a=4, b=2)
    # tryfirst (multiply: 8), normal (add: 6), trylast (subtract: 2)
    assert results == [8, 6, 2], f"Expected [8, 6, 2], got {results}"


def test_firstresult_hook():
    """Verify hook specifications with firstresult=True return first non-None result."""
    pm = create_plugin_manager()
    pm.register(StatusResolverPluginA(), name="status_a")
    pm.register(StatusResolverPluginB(), name="status_b")

    res_200 = pm.hook.resolve_status(code=200)
    assert res_200 == "OK", f"Expected 'OK', got {res_200}"

    res_404 = pm.hook.resolve_status(code=404)
    assert res_404 == "NOT_FOUND", f"Expected 'NOT_FOUND', got {res_404}"

    res_500 = pm.hook.resolve_status(code=500)
    assert res_500 is None, f"Expected None, got {res_500}"


def test_hook_wrappers():
    """Verify modern and legacy hook wrappers intercept and modify hook outcomes."""
    # Test modern wrapper (wrapper=True)
    pm_modern = create_plugin_manager()
    pm_modern.register(AddPlugin())
    pm_modern.register(ModernWrapperPlugin())

    res_modern = pm_modern.hook.compute(a=3, b=5)
    # Add returns [8], modern wrapper multiplies by 10 -> [80]
    assert res_modern == [80], f"Expected [80], got {res_modern}"

    # Test legacy wrapper (hookwrapper=True)
    pm_legacy = create_plugin_manager()
    pm_legacy.register(AddPlugin())
    pm_legacy.register(LegacyWrapperPlugin())

    res_legacy = pm_legacy.hook.compute(a=3, b=5)
    # Add returns [8], legacy wrapper adds 1 -> [9]
    assert res_legacy == [9], f"Expected [9], got {res_legacy}"


def test_historic_hooks():
    """Verify historic hooks record calls and replay them to newly registered plugins."""
    pm = create_plugin_manager()
    early_events = []
    late_events = []

    class EarlyListener:
        @hookimpl
        def on_event(self, event_name: str):
            early_events.append(event_name)

    class LateListener:
        @hookimpl
        def on_event(self, event_name: str):
            late_events.append(event_name)

    early = EarlyListener()
    pm.register(early)

    pm.hook.on_event.call_historic(kwargs={"event_name": "app_started"})
    pm.hook.on_event.call_historic(kwargs={"event_name": "ready"})

    assert early_events == ["app_started", "ready"], f"Expected early events, got {early_events}"
    assert late_events == [], "Late listener should not have received events yet"

    late = LateListener()
    pm.register(late)

    assert late_events == ["app_started", "ready"], f"Expected late events replayed, got {late_events}"


def test_argument_validation():
    """Verify PluginManager validates implementation arguments and raises PluginValidationError on mismatch."""
    pm = create_plugin_manager()

    class MismatchedPlugin:
        @hookimpl
        def compute(self, a: int, unknown_arg: int) -> int:
            return a + unknown_arg

    try:
        pm.register(MismatchedPlugin())
        assert False, "Expected PluginValidationError when registering mismatched arguments"
    except pluggy.PluginValidationError:
        pass


def test_plugin_lifecycle_management():
    """Verify PluginManager controls registration, lookup, blocking, unregistration, and subset caller."""
    pm = create_plugin_manager()
    plugin = AddPlugin()

    pm.register(plugin, name="my_adder")
    assert pm.is_registered(plugin) is True
    assert pm.get_plugin("my_adder") is plugin
    assert pm.get_name(plugin) == "my_adder"
    assert plugin in pm.get_plugins()

    subset_caller = pm.subset_hook_caller("compute", remove_plugins=[plugin])
    assert subset_caller(a=1, b=2) == []

    pm.set_blocked("blocked_plugin")
    assert pm.is_blocked("blocked_plugin") is True

    unregistered = pm.unregister(name="my_adder")
    assert unregistered is plugin
    assert pm.is_registered(plugin) is False
    assert pm.get_plugin("my_adder") is None

    pm.check_pending()


def main():
    test_hook_execution_ordering()
    test_firstresult_hook()
    test_hook_wrappers()
    test_historic_hooks()
    test_argument_validation()
    test_plugin_lifecycle_management()
    print("All pluggy contract tests passed successfully.")
    return 0


if __name__ == "__main__":
    sys.exit(main())