Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🌐 Web query:

pytest plugin architecture hooks design patterns

💡 Result:

Summary — pytest plugin architecture & common hook design patterns

Key components

Pluggy (hook framework): hook specs, hook implementations, PluginManager, HookCaller / _multicall. [1][2]
Pytest exposes hook specifications in _pytest.hookspec and uses pluggy to call implementations from builtin plugins, conftest.py and external plugins. [3][4]

Design patterns / best practices

Define clear hook specs as extension points (single-responsibility signatures). Use descriptive names and minimal args. [4][1]
Use hook wrappers for “around” behavior (pre/post) when callers expect to control flow; otherwise implement normal hookimpls. Mark wrappers with hookimpl(hookwrapper=True). [1][4]
Ordering & priority: use hookimpl(tryfirst=True/trylast=True) or explicit ordering when implementations must run in sequence. Favor idempotent implementations. [1]
Bootstrap / registration: add new specs at registration time via pytest_addhooks / pluginmanager.add_hookspecs; register plugins with entry points or pluginmanager.register for dynamic plugins. [4][3]
Keep plugin state minimal and thread-safe; prefer storing state on Config or Session objects, not module-level globals. Use fixtures for per-test state. [3][7]
Use conftest for local project hooks, distribution packages + entry_points for reusable plugins; avoid importing from conftest. [7]
Fail-fast & compatibility: validate input/output in hooks, handle missing hook implementations, and use deprecation warnings when evolving hook specs to preserve backward compatibility. [4][1]
Testing plugins: test hook behavior by registering fake plugins with PluginManager and exercising hook calls (use pytest --trace-config to inspect active plugins). [3][1]

Implementation tips

Prefer small focused hooks rather than one large generic hook.
Document hook contract (expected return values/exceptions and whether multicall aggregates results).
For long-running hooks, consider async support patterns or offloading work to fixtures/tasks.
Use pluggy’s markers (HookspecMarker/HookimplMarker) to keep namespace isolation between projects. [1][2]

Primary references

Pluggy project (architecture, API, examples). [1]
Pytest — Writing plugins / how hooks are used and loaded. [3]
Pytest hookspec source (detailed hook signatures & notes like pytest_addhooks). [4]
Pytest plugin usage / installing and disabling plugins. [2]

If you want, I can: (a) sketch a sample plugin + hookspec + hookimpl showing these patterns, or (b) review a specific hook design you’re considering.

@Agent-Hellboy, absolutely! pytest's architecture offers several powerful patterns that would dramatically improve your fuzzing reporter. Here's what you can adopt:

🔌 1. Hook-Based Plugin Architecture (Most Valuable)

Current Problem

# FuzzerReporter is monolithic with hard-coded formatters
class FuzzerReporter:
    def __init__(self, ...):
        self.json_formatter = deps.formatters.json
        self.text_formatter = deps.formatters.text
        self.csv_formatter = deps.formatters.csv
        # ... 7 formatters hard-coded

pytest Pattern: Hook Specifications

from typing import Protocol, Any
from pluggy import HookspecMarker, HookimplMarker

hookspec = HookspecMarker("mcp_fuzzer")
hookimpl = HookimplMarker("mcp_fuzzer")

class FuzzingHooks:
    """Hook specifications for fuzzing extensions."""
    
    @hookspec
    def fuzzer_runtest_setup(self, tool_name: str, test_case: dict) -> None:
        """Called before each fuzzing test runs."""
    
    @hookspec
    def fuzzer_runtest_call(self, tool_name: str, test_case: dict) -> dict | None:
        """Called to execute the fuzzing test. Return result or None."""
    
    @hookspec
    def fuzzer_runtest_teardown(self, tool_name: str, test_case: dict) -> None:
        """Called after each fuzzing test completes."""
    
    @hookspec(firstresult=True)
    def fuzzer_exception_handler(self, tool_name: str, exc: Exception) -> bool:
        """Handle exceptions during fuzzing. Return True if handled."""
    
    @hookspec
    def fuzzer_report_teststatus(self, tool_name: str, test_case: dict, 
                                 result: dict) -> tuple[str, str, str] | None:
        """Return (category, short_letter, verbose_word) for test status."""
    
    @hookspec
    def fuzzer_terminal_summary(self, collector: "ReportCollector", 
                                console: Console) -> None:
        """Add section to terminal summary report."""
    
    @hookspec
    def fuzzer_collection_finish(self, collector: "ReportCollector") -> None:
        """Called when result collection finishes."""
    
    @hookspec
    def fuzzer_report_header(self, collector: "ReportCollector") -> list[str]:
        """Return list of strings for report header."""
    
    @hookspec
    def fuzzer_sessionstart(self, session_id: str) -> None:
        """Called at the start of fuzzing session."""
    
    @hookspec
    def fuzzer_sessionfinish(self, session_id: str, 
                            collector: "ReportCollector") -> None:
        """Called at the end of fuzzing session."""

Implementation with PluginManager

from pluggy import PluginManager

class FuzzerPluginManager:
    """Manages fuzzing plugins using pluggy."""
    
    def __init__(self):
        self.manager = PluginManager("mcp_fuzzer")
        self.manager.add_hookspecs(FuzzingHooks)
        
        # Register built-in plugins
        self._register_builtin_plugins()
    
    def _register_builtin_plugins(self):
        """Register core plugins."""
        self.manager.register(TimingPlugin(), name="timing")
        self.manager.register(SafetyPlugin(), name="safety")
        self.manager.register(ConsoleReporterPlugin(), name="console")
    
    def register(self, plugin, name: str | None = None):
        """Register a custom plugin."""
        self.manager.register(plugin, name=name)
    
    def unregister(self, plugin=None, name: str | None = None):
        """Unregister a plugin."""
        return self.manager.unregister(plugin=plugin, name=name)
    
    def get_plugin(self, name: str):
        """Get a registered plugin by name."""
        return self.manager.get_plugin(name)
    
    def list_plugins(self) -> list[tuple[str, Any]]:
        """List all registered plugins."""
        return self.manager.list_name_plugin()

Example Plugins

Timing Plugin (Hook Wrapper Pattern)

class TimingPlugin:
    """Plugin that times each fuzzing test."""
    
    def __init__(self):
        self.timings: dict[str, list[float]] = {}
    
    @hookimpl(hookwrapper=True)  # Wraps around other implementations
    def fuzzer_runtest_call(self, tool_name: str, test_case: dict):
        """Time the test execution."""
        import time
        start = time.perf_counter()
        
        # Let other plugins run
        outcome = yield
        
        duration = time.perf_counter() - start
        self.timings.setdefault(tool_name, []).append(duration)
        
        # Modify result to add timing
        result = outcome.get_result()
        if result:
            result['duration'] = duration
    
    @hookimpl
    def fuzzer_terminal_summary(self, collector, console):
        """Print timing summary."""
        console.print("\n[bold cyan]⏱ Timing Summary[/bold cyan]")
        for tool, times in self.timings.items():
            avg = sum(times) / len(times)
            console.print(f"  {tool}: avg {avg:.3f}s ({len(times)} tests)")

JSON Reporter Plugin

class JSONReporterPlugin:
    """Plugin that exports results to JSON."""
    
    def __init__(self, output_dir: Path):
        self.output_dir = output_dir
    
    @hookimpl(trylast=True)  # Run after other plugins
    def fuzzer_sessionfinish(self, session_id: str, collector):
        """Export JSON report at session end."""
        from .formatters import JSONFormatter
        
        formatter = JSONFormatter()
        report = collector.get_report()
        json_content = formatter.format(report)
        
        output_file = self.output_dir / f"report_{session_id}.json"
        output_file.write_text(json_content, encoding='utf-8')

Safety Validator Plugin

class SafetyPlugin:
    """Plugin that validates safety constraints."""
    
    def __init__(self, safety_system):
        self.safety_system = safety_system
    
    @hookimpl(tryfirst=True)  # Run before test execution
    def fuzzer_runtest_setup(self, tool_name: str, test_case: dict):
        """Validate test case against safety rules before running."""
        if not self.safety_system.validate_input(test_case):
            raise SafetyViolation(f"Test case violates safety rules: {test_case}")
    
    @hookimpl(hookwrapper=True)
    def fuzzer_runtest_call(self, tool_name: str, test_case: dict):
        """Monitor test execution for safety violations."""
        outcome = yield
        
        result = outcome.get_result()
        if result:
            violations = self.safety_system.check_result(result)
            if violations:
                result['safety_violations'] = violations
    
    @hookimpl
    def fuzzer_terminal_summary(self, collector, console):
        """Print safety summary."""
        violations = self.safety_system.get_all_violations()
        if violations:
            console.print(f"\n[bold red]🚨 Safety Violations: {len(violations)}[/bold red]")

Usage Example

# Setup plugin manager
pm = FuzzerPluginManager()

# Register custom plugins
pm.register(JSONReporterPlugin(output_dir=Path("reports")), name="json_reporter")
pm.register(HTMLReporterPlugin(output_dir=Path("reports")), name="html_reporter")

# Run fuzzing with hooks
collector = ReportCollector()
session_id = generate_session_id()

# Session start hook
pm.manager.hook.fuzzer_sessionstart(session_id=session_id)

for tool_name, test_cases in fuzzing_plan.items():
    for test_case in test_cases:
        try:
            # Setup hook
            pm.manager.hook.fuzzer_runtest_setup(
                tool_name=tool_name,
                test_case=test_case
            )
            
            # Call hook (all plugins can modify/observe)
            result = pm.manager.hook.fuzzer_runtest_call(
                tool_name=tool_name,
                test_case=test_case
            )
            
            # Collect result
            collector.add_success(tool_name, test_case, result)
            
        except Exception as exc:
            # Exception handler hook (firstresult=True means first True stops)
            handled = pm.manager.hook.fuzzer_exception_handler(
                tool_name=tool_name,
                exc=exc
            )
            if not handled:
                collector.add_error(tool_name, test_case, exc)
        
        finally:
            # Teardown hook
            pm.manager.hook.fuzzer_runtest_teardown(
                tool_name=tool_name,
                test_case=test_case
            )

# Collection finish hook
pm.manager.hook.fuzzer_collection_finish(collector=collector)

# Terminal summary hook (all plugins contribute)
pm.manager.hook.fuzzer_terminal_summary(
    collector=collector,
    console=console
)

# Session finish hook
pm.manager.hook.fuzzer_sessionfinish(
    session_id=session_id,
    collector=collector
)

🔧 2. Fixture-Like Dependency Injection

pytest Pattern: Fixtures

from typing import Generator, Any
from functools import wraps

class FixtureRegistry:
    """Registry for fuzzing fixtures (test dependencies)."""
    
    def __init__(self):
        self._fixtures: dict[str, callable] = {}
        self._cache: dict[str, Any] = {}
        self._scope_cache: dict[str, dict[str, Any]] = {
            "session": {},
            "tool": {},
            "test": {}
        }
    
    def fixture(self, scope: str = "test", name: str | None = None):
        """Decorator to register a fixture."""
        def decorator(func):
            fixture_name = name or func.__name__
            
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Check cache based on scope
                if scope == "session" and fixture_name in self._scope_cache["session"]:
                    return self._scope_cache["session"][fixture_name]
                
                # Execute fixture
                result = func(*args, **kwargs)
                
                # Cache based on scope
                if scope == "session":
                    self._scope_cache["session"][fixture_name] = result
                
                return result
            
            self._fixtures[fixture_name] = wrapper
            return wrapper
        
        return decorator
    
    def get_fixture(self, name: str) -> Any:
        """Get fixture by name."""
        if name not in self._fixtures:
            raise ValueError(f"Fixture '{name}' not found")
        return self._fixtures[name]()
    
    def clear_scope(self, scope: str):
        """Clear cached fixtures for a scope."""
        self._scope_cache[scope].clear()

# Usage
registry = FixtureRegistry()

@registry.fixture(scope="session")
def fuzzer_config():
    """Session-scoped config fixture."""
    return FuzzerConfig.load_from_file("config.yaml")

@registry.fixture(scope="session")
def mcp_client(fuzzer_config):
    """Session-scoped MCP client."""
    config = fuzzer_config  # Can depend on other fixtures
    client = MCPClient(config.server_url)
    yield client  # Setup
    client.close()  # Teardown

@registry.fixture(scope="tool")
def tool_fuzzer(mcp_client):
    """Tool-scoped fuzzer instance."""
    return ToolFuzzer(mcp_client)

@registry.fixture(scope="test")
def test_data_generator():
    """Test-scoped data generator."""
    return FuzzDataGenerator(seed=random.randint(0, 1000000))

# In fuzzing loop
config = registry.get_fixture("fuzzer_config")
client = registry.get_fixture("mcp_client")

for tool_name in tools:
    fuzzer = registry.get_fixture("tool_fuzzer")
    
    for i in range(num_tests):
        generator = registry.get_fixture("test_data_generator")
        test_case = generator.generate()
        # Run test...
        
        registry.clear_scope("test")  # Clean up test-scoped fixtures
    
    registry.clear_scope("tool")  # Clean up tool-scoped fixtures

📊 3. Parametrization Pattern

from dataclasses import dataclass
from typing import Any

@dataclass
class FuzzParameter:
    """Represents a fuzz parameter configuration."""
    name: str
    values: list[Any]
    ids: list[str] | None = None

class ParametrizedFuzzer:
    """Fuzzer with parametrization support."""
    
    def __init__(self):
        self.parameters: dict[str, FuzzParameter] = {}
    
    def parametrize(self, param_name: str, values: list[Any], 
                   ids: list[str] | None = None):
        """Register a parameter similar to pytest.mark.parametrize."""
        self.parameters[param_name] = FuzzParameter(param_name, values, ids)
        return self
    
    def generate_test_cases(self) -> list[dict[str, Any]]:
        """Generate cartesian product of all parameters."""
        if not self.parameters:
            return [{}]
        
        import itertools
        
        param_names = list(self.parameters.keys())
        param_values = [self.parameters[name].values for name in param_names]
        
        test_cases = []
        for combination in itertools.product(*param_values):
            test_case = dict(zip(param_names, combination))
            test_cases.append(test_case)
        
        return test_cases

# Usage
fuzzer = ParametrizedFuzzer()
fuzzer.parametrize("method", ["GET", "POST", "PUT", "DELETE"])
fuzzer.parametrize("auth", [None, "Bearer token", "Basic auth"])
fuzzer.parametrize("payload_size", [0, 100, 1000, 10000])

for test_case in fuzzer.generate_test_cases():
    # Each test_case is: {"method": "GET", "auth": None, "payload_size": 0}
    result = run_fuzzing_test(test_case)
    collector.record_result(test_case, result)

🎯 4. Marks/Tags System

from enum import Enum
from typing import Callable

class FuzzMark(Enum):
    """Fuzzing test marks."""
    SLOW = "slow"
    SAFETY_CRITICAL = "safety_critical"
    FLAKY = "flaky"
    SKIP = "skip"
    XFAIL = "xfail"  # Expected to fail

class MarkedTest:
    """Test case with marks attached."""
    
    def __init__(self, tool_name: str, test_case: dict):
        self.tool_name = tool_name
        self.test_case = test_case
        self.marks: set[FuzzMark] = set()
        self.metadata: dict[str, Any] = {}
    
    def add_mark(self, mark: FuzzMark, reason: str = ""):
        """Add a mark to this test."""
        self.marks.add(mark)
        if reason:
            self.metadata[f"{mark.value}_reason"] = reason
        return self
    
    def has_mark(self, mark: FuzzMark) -> bool:
        """Check if test has a specific mark."""
        return mark in self.marks

class TestSelector:
    """Select tests based on marks (like pytest -m)."""
    
    def __init__(self, run_slow: bool = False, skip_flaky: bool = True):
        self.run_slow = run_slow
        self.skip_flaky = skip_flaky
    
    def should_run(self, test: MarkedTest) -> bool:
        """Determine if test should run."""
        if test.has_mark(FuzzMark.SKIP):
            return False
        
        if test.has_mark(FuzzMark.FLAKY) and self.skip_flaky:
            return False
        
        if test.has_mark(FuzzMark.SLOW) and not self.run_slow:
            return False
        
        return True

# Usage
test = MarkedTest("call_tool", {"arg": "invalid"})
test.add_mark(FuzzMark.SAFETY_CRITICAL, reason="Tests input validation")
test.add_mark(FuzzMark.SLOW, reason="Requires 10s timeout")

selector = TestSelector(run_slow=False, skip_flaky=True)
if selector.should_run(test):
    run_fuzzing_test(test)

📋 5. Improved Result Collection (Items Pattern)

from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal

@dataclass
class FuzzItem:
    """Represents a single fuzz test item (like pytest.Item)."""
    tool_name: str
    test_case: dict
    node_id: str  # Unique identifier like "list_tools[test_0]"
    
    # Execution state
    when: Literal["setup", "call", "teardown"] | None = None
    outcome: Literal["passed", "failed", "skipped", "error"] | None = None
    
    # Timing
    start_time: datetime | None = None
    stop_time: datetime | None = None
    duration: float = 0.0
    
    # Results
    result: Any = None
    exception: Exception | None = None
    traceback: str | None = None
    
    # Metadata
    marks: set[FuzzMark] = field(default_factory=set)
    user_properties: dict[str, Any] = field(default_factory=dict)
    
    def set_outcome(self, when: str, outcome: str, exception: Exception | None = None):
        """Set the outcome of this item."""
        self.when = when
        self.outcome = outcome
        self.exception = exception
        if exception:
            import traceback as tb
            self.traceback = "".join(tb.format_exception(type(exception), exception, exception.__traceback__))

@dataclass
class FuzzReport:
    """Test report (like pytest.TestReport)."""
    item: FuzzItem
    outcome: str
    longrepr: str | None = None  # Long representation of failure
    sections: list[tuple[str, str]] = field(default_factory=list)  # Extra info sections
    
    def add_section(self, name: str, content: str):
        """Add an extra section to the report."""
        self.sections.append((name, content))

class ImprovedReportCollector:
    """Collector using Item pattern."""
    
    def __init__(self):
        self.items: list[FuzzItem] = []
        self.reports: list[FuzzReport] = []
    
    def create_item(self, tool_name: str, test_case: dict, index: int) -> FuzzItem:
        """Create a new test item."""
        node_id = f"{tool_name}[test_{index}]"
        item = FuzzItem(tool_name=tool_name, test_case=test_case, node_id=node_id)
        self.items.append(item)
        return item
    
    def add_report(self, item: FuzzItem, outcome: str, longrepr: str | None = None):
        """Add a test report."""
        report = FuzzReport(item=item, outcome=outcome, longrepr=longrepr)
        
        # Plugins can add sections
        if item.exception:
            report.add_section("Exception", str(item.exception))
        if item.traceback:
            report.add_section("Traceback", item.traceback)
        
        self.reports.append(report)
    
    def get_stats(self) -> dict[str, int]:
        """Get test statistics."""
        stats = {"total": len(self.items), "passed": 0, "failed": 0, "skipped": 0, "error": 0}
        for item in self.items:
            if item.outcome:
                stats[item.outcome] += 1
        return stats

🚀 Implementation Strategy

Phase 1: Add Plugin System (High Impact)

# Install pluggy
# pip install pluggy

# 1. Define hooks
# 2. Create PluginManager
# 3. Convert existing formatters to plugins
# 4. Register built-in plugins

Phase 2: Refactor with Hooks

# Replace direct method calls with hooks:
# OLD: self.json_formatter.format(report)
# NEW: self.pm.hook.fuzzer_format_json(report=report)

Phase 3: Add Fixture System (Optional)

# For managing test dependencies and setup/teardown

Phase 4: Add Parametrization (Optional)

# For systematic input variation

📊 Comparison

Pattern	pytest	Your System	Benefit
Plugins	✅ Hook-based	❌ Monolithic	Extensibility without modification
Fixtures	✅ DI + scoping	⚠️ Manual	Automated setup/teardown
Parametrization	✅ Built-in	❌ Manual loops	Systematic test generation
Marks	✅ Decorators	❌ None	Test filtering and categorization
Items	✅ Rich metadata	⚠️ Basic	Better tracking and reporting
Hooks	✅ Lifecycle	⚠️ Partial	Complete control over execution

🎯 Recommended Priority

Hook System 🔥🔥🔥 (Highest ROI)
- Solves extensibility problem
- Enables custom reporters without core changes
- Cleaner separation of concerns
Item Pattern 🔥🔥 (High value)
- Better result tracking
- Richer metadata
- Easier debugging
Marks System 🔥 (Medium value)
- Test filtering
- Categorization
- Selective execution
Fixture System 💡 (Nice to have)
- Cleaner setup/teardown
- Dependency injection
- Resource management
Parametrization 💡 (Nice to have)
- Systematic input generation
- Better coverage

Would you like me to:

Create a complete plugin-based refactoring of FuzzerReporter?
Generate a working example with pluggy integration?
Show migration path from current to plugin-based architecture?

  /)_/)
 (='.'=)  Plugins make 
 (")_(")  everything better!

Uh oh!

Review all code flow of reporter components #141

Description

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions