pluggy 1.6.0

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

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:e9841afd533e6f3720ccd77b27517e29b0d621de1c61e43544dd19907e893551","contract":["PluginManager discovers and executes hook implementations registered across multiple plugins with deterministic ordering","Hook specifications with firstresult=True return the first non-None result from registered implementations","Hook wrappers intercept execution and allow inspecting or transforming hook return values","Historic hooks record calls and automatically replay them to plugins registered later","PluginManager manages plugin lifecycle including registration, lookup, blocking, and unregistration"],"goal":"verify pkg:pypi/pluggy@1.6.0","kind":"HOW","packages":["pkg:pypi/pluggy@1.6.0"],"schemaVersion":1},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/pluggy@1.6.0"],"schemaVersion":1,"subject":"pkg:pypi/pluggy@1.6.0","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"
  ]
}

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

import sys
import pluggy

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


class SampleSpecification:
    """Hook specifications for sample project."""

    @hookspec
    def calculate(self, x: int, y: int) -> int:
        """Calculate a value given x and y."""

    @hookspec(firstresult=True)
    def resolve_status(self, code: int) -> str:
        """Resolve a status string for a given code."""

    @hookspec(historic=True)
    def notify_event(self, event_name: str) -> None:
        """Historic hook to record and broadcast events."""


class AddPlugin:
    @hookimpl
    def calculate(self, x: int, y: int) -> int:
        return x + y


class MultiplyPlugin:
    @hookimpl(tryfirst=True)
    def calculate(self, x: int, y: int) -> int:
        return x * y


class StatusPluginA:
    @hookimpl
    def resolve_status(self, code: int):
        if code == 200:
            return "OK"
        return None


class StatusPluginB:
    @hookimpl(tryfirst=True)
    def resolve_status(self, code: int):
        if code == 404:
            return "NOT_FOUND"
        return None


class TransformWrapperPlugin:
    @hookimpl(wrapper=True)
    def calculate(self, x: int, y: int):
        res = yield
        return [item * 10 for item in res]


class LegacyHookWrapperPlugin:
    @hookimpl(hookwrapper=True)
    def calculate(self, x: int, y: int):
        outcome = yield
        raw_res = outcome.get_result()
        outcome.force_result([item + 1 for item in raw_res])


def test_hook_registration_and_execution():
    """Verify PluginManager registers hooks and executes them with expected ordering."""
    pm = pluggy.PluginManager("sample_project")
    pm.add_hookspecs(SampleSpecification)

    p1 = AddPlugin()
    p2 = MultiplyPlugin()
    pm.register(p1, name="add_plugin")
    pm.register(p2, name="multiply_plugin")

    results = pm.hook.calculate(x=3, y=4)
    assert results == [12, 7]


def test_firstresult_hook():
    """Verify firstresult=True stops at the first non-None result."""
    pm = pluggy.PluginManager("sample_project")
    pm.add_hookspecs(SampleSpecification)

    pm.register(StatusPluginA())
    pm.register(StatusPluginB())

    res_200 = pm.hook.resolve_status(code=200)
    assert res_200 == "OK"

    res_404 = pm.hook.resolve_status(code=404)
    assert res_404 == "NOT_FOUND"

    res_500 = pm.hook.resolve_status(code=500)
    assert res_500 is None


def test_hook_wrappers():
    """Verify wrapper=True and hookwrapper=True can inspect and transform results."""
    pm = pluggy.PluginManager("sample_project")
    pm.add_hookspecs(SampleSpecification)
    pm.register(AddPlugin())
    pm.register(TransformWrapperPlugin())

    res_wrapper = pm.hook.calculate(x=3, y=4)
    assert res_wrapper == [70]

    pm_legacy = pluggy.PluginManager("sample_project")
    pm_legacy.add_hookspecs(SampleSpecification)
    pm_legacy.register(AddPlugin())
    pm_legacy.register(LegacyHookWrapperPlugin())

    res_legacy = pm_legacy.hook.calculate(x=3, y=4)
    assert res_legacy == [8]


def test_historic_hooks():
    """Verify historic hooks replay recorded calls to newly registered plugins."""
    pm = pluggy.PluginManager("sample_project")
    pm.add_hookspecs(SampleSpecification)

    events_early = []
    events_late = []

    class EarlySubscriber:
        @hookimpl
        def notify_event(self, event_name: str):
            events_early.append(event_name)

    class LateSubscriber:
        @hookimpl
        def notify_event(self, event_name: str):
            events_late.append(event_name)

    early = EarlySubscriber()
    pm.register(early)

    pm.hook.notify_event.call_historic(kwargs={"event_name": "initialized"})
    pm.hook.notify_event.call_historic(kwargs={"event_name": "started"})

    assert events_early == ["initialized", "started"]
    assert events_late == []

    late = LateSubscriber()
    pm.register(late)

    assert events_late == ["initialized", "started"]


def test_plugin_lifecycle():
    """Verify plugin registration, lookup, blocking, and unregistration."""
    pm = pluggy.PluginManager("sample_project")
    pm.add_hookspecs(SampleSpecification)

    plugin = AddPlugin()
    pm.register(plugin, name="my_plugin")

    assert pm.is_registered(plugin) is True
    assert pm.get_plugin("my_plugin") is plugin
    assert pm.get_name(plugin) == "my_plugin"
    assert plugin in pm.get_plugins()

    subset_caller = pm.subset_hook_caller("calculate", remove_plugins=[plugin])
    assert subset_caller(x=2, y=3) == []

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

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

    pm.check_pending()


def main():
    test_hook_registration_and_execution()
    test_firstresult_hook()
    test_hook_wrappers()
    test_historic_hooks()
    test_plugin_lifecycle()
    print("All pluggy contract tests passed successfully.")
    return 0


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