cleveragents-core/robot/helper_predictive_error_prevention.py

"""Helper utilities for Predictive Error Prevention Robot integration tests.

Covers Layer 4: Error Pattern Database — recording, matching, and
preventive guidance generation.
"""

from __future__ import annotations

from cleveragents.application.services.error_pattern_service import (
    ErrorPatternService,
)
from cleveragents.domain.models.core.error_pattern import ErrorPattern


def _record_and_verify_pattern() -> None:
    """Integration test: record a failure and verify pattern creation."""
    service = ErrorPatternService()
    pattern = service.record_failure(
        pattern_text="Race condition in payment module",
        failure_description="Payment confirmation timeout",
        preventive_checks=("Add explicit locking",),
        keywords=("race", "payment"),
    )
    assert isinstance(pattern, ErrorPattern), (
        f"Expected ErrorPattern, got {type(pattern)}"
    )
    assert pattern.frequency == 1, f"Expected frequency 1, got {pattern.frequency}"
    assert len(pattern.historical_failures) == 1, (
        f"Expected 1 historical failure, got {len(pattern.historical_failures)}"
    )
    assert pattern.pattern == "Race condition in payment module"

    stats = service.get_statistics()
    assert stats["total_patterns"] == 1, (
        f"Expected 1 pattern, got {stats['total_patterns']}"
    )
    print("record-and-verify-ok")


def _match_patterns_returns_guidance() -> None:
    """Integration test: match context against patterns returns guidance."""
    service = ErrorPatternService()
    service.record_failure(
        pattern_text="Async conversion failure",
        failure_description="Timeout during async conversion",
        preventive_checks=("Add transaction boundaries", "Use retry logic"),
        keywords=("async", "conversion"),
    )

    guidance = service.match_patterns(
        "We need to perform async conversion in the payment module"
    )
    assert guidance.has_guidance, "Expected guidance to be non-empty"
    assert len(guidance.preventive_checks) == 2, (
        f"Expected 2 checks, got {len(guidance.preventive_checks)}"
    )
    assert "Add transaction boundaries" in guidance.preventive_checks
    assert "Use retry logic" in guidance.preventive_checks

    formatted = guidance.format_for_context()
    assert "PREVENTIVE GUIDANCE" in formatted, (
        f"Expected header in formatted output: {formatted}"
    )
    print("match-patterns-guidance-ok")


def _frequency_increments_on_duplicate() -> None:
    """Integration test: duplicate pattern recording increments frequency."""
    service = ErrorPatternService()
    service.record_failure(
        pattern_text="Memory leak in cache",
        failure_description="OOM after 24h",
        preventive_checks=("Check dispose patterns",),
        keywords=("memory", "leak", "cache"),
    )
    updated = service.record_failure(
        pattern_text="Memory leak in cache",
        failure_description="High memory usage after 12h",
    )
    assert updated.frequency == 2, f"Expected frequency 2, got {updated.frequency}"
    assert len(updated.historical_failures) == 2, (
        f"Expected 2 failures, got {len(updated.historical_failures)}"
    )

    stats = service.get_statistics()
    assert stats["total_patterns"] == 1, (
        f"Expected 1 pattern, got {stats['total_patterns']}"
    )
    assert stats["total_occurrences"] == 2, (
        f"Expected 2 occurrences, got {stats['total_occurrences']}"
    )
    print("frequency-increments-ok")


if __name__ == "__main__":
    _record_and_verify_pattern()
    _match_patterns_returns_guidance()
    _frequency_increments_on_duplicate()
    print("all-predictive-error-prevention-tests-passed")