Regression Testing — Locale Utility Payment Failures

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Set, Dict, Optional
from datetime import datetime


class TestStatus(Enum):
    PASSED = "passed"
    FAILED = "failed"
    SKIPPED = "skipped"
    FLAKY = "flaky"


class RegressionPriority(Enum):
    CRITICAL = "critical"   # Payment, auth, data integrity — always run
    HIGH = "high"           # Core user flows — run on every PR
    MEDIUM = "medium"       # Supporting features — run on merge to main
    LOW = "low"             # Edge cases — run on full nightly suite


@dataclass
class RegressionTestCase:
    test_id: str
    name: str
    module: str
    priority: RegressionPriority
    last_run: Optional[datetime] = None
    last_status: TestStatus = TestStatus.SKIPPED
    avg_duration_ms: float = 0.0
    failure_count: int = 0    # Cumulative failures — high count signals flakiness
    tags: List[str] = field(default_factory=list)  # Used for cross-module impact matching


@dataclass
class RegressionSuite:
    """
    Manages a regression test suite with impact-based selection
    and execution tracking.

    Key design decisions:
    - Tests are tagged with module names they exercise, not just the module they live in.
      A payment test may tag 'locale' and 'currency' because it exercises those utilities.
    - select_by_impact uses tags for cross-module matching, catching invisible coupling.
    - get_flaky_tests uses a configurable threshold — tune this per team tolerance.
    """

    suite_name: str
    test_cases: List[RegressionTestCase] = field(default_factory=list)

    def add_test(self, test: RegressionTestCase) -> None:
        self.test_cases.append(test)

    def select_by_impact(self, changed_modules: Set[str]) -> List[RegressionTestCase]:
        """
        Select tests that cover modules affected by code changes.
        Matches both primary module and tags — critical for catching
        cross-module regressions from shared utilities.
        """
        selected = []
        for test in self.test_cases:
            # Direct module match: the test lives in a changed module
            if test.module in changed_modules:
                selected.append(test)
            # Tag match: the test exercises a changed module as a dependency
            # This is what catches the locale-utility-breaks-payments class of bugs
            elif any(tag in changed_modules for tag in test.tags):
                selected.append(test)
        return selected

    def select_by_priority(
        self, min_priority: RegressionPriority
    ) -> List[RegressionTestCase]:
        """
        Select tests at or above a minimum priority level.
        Used for smoke runs where impact analysis is not available
        (e.g., infrastructure changes with unknown blast radius).
        """
        priority_order = {
            RegressionPriority.CRITICAL: 4,
            RegressionPriority.HIGH: 3,
            RegressionPriority.MEDIUM: 2,
            RegressionPriority.LOW: 1
        }
        min_level = priority_order[min_priority]
        return [
            t for t in self.test_cases
            if priority_order[t.priority] >= min_level
        ]

    def get_flaky_tests(
        self, threshold: int = 3
    ) -> List[RegressionTestCase]:
        """
        Identify tests that have accumulated failures above the threshold.
        These candidates should be quarantined and fixed — not retried.
        Threshold of 3 is conservative; teams with high deployment frequency
        may need to lower this to 2 to catch instability faster.
        """
        return [t for t in self.test_cases if t.failure_count >= threshold]

    def estimate_execution_time(
        self, tests: List[RegressionTestCase]
    ) -> float:
        """Estimate total execution time in seconds for a given test list."""
        return sum(t.avg_duration_ms for t in tests) / 1000.0

    def get_stats(self) -> Dict:
        """Return suite statistics useful for health dashboards."""
        total = len(self.test_cases)
        by_priority: Dict[str, int] = {}
        for test in self.test_cases:
            key = test.priority.value
            by_priority[key] = by_priority.get(key, 0) + 1

        return {
            "total_tests": total,
            "by_priority": by_priority,
            "flaky_count": len(self.get_flaky_tests()),
            "estimated_full_runtime_sec": self.estimate_execution_time(self.test_cases)
        }


# Example usage — illustrating the locale-utility coupling scenario
suite = RegressionSuite(suite_name="main-regression")

suite.add_test(RegressionTestCase(
    test_id="TC-001",
    name="test_payment_processing_eu_locale",
    module="payments",
    priority=RegressionPriority.CRITICAL,
    # Tags include 'locale' — so changes to the locale utility trigger this test
    tags=["payments", "locale", "currency"],
    avg_duration_ms=250.0
))

suite.add_test(RegressionTestCase(
    test_id="TC-002",
    name="test_email_notification_timestamp",
    module="notifications",
    priority=RegressionPriority.HIGH,
    tags=["notifications", "locale"],
    avg_duration_ms=180.0
))

# A change to the locale utility selects BOTH tests — not just the notification test
changed = {"locale"}
selected = suite.select_by_impact(changed)
print(f"Selected {len(selected)} tests for changes in: {changed}")
for test in selected:
    print(f"  [{test.priority.value.upper()}] {test.test_id}: {test.name}")

stats = suite.get_stats()
print(f"\nSuite stats: {stats}")

from enum import Enum
from typing import List, Set
from io.thecodeforge.testing.regression import (
    RegressionSuite, RegressionTestCase, RegressionPriority
)


class RegressionType(Enum):
    CORRECTIVE = "corrective"   # Bug fix verification
    PROGRESSIVE = "progressive" # New feature integration verification
    SELECTIVE = "selective"     # Impact-based subset — default for CI/CD
    COMPLETE = "complete"       # Full suite — pre-release gate
    SMOKE = "smoke"             # Critical path only — fastest feedback
    UNIT = "unit"               # Module-level only — fastest possible


class RegressionStrategy:
    """
    Implements different regression testing strategies based on
    change scope and available time budget.

    The recommend_strategy method encodes the decision logic that
    most teams apply informally and inconsistently. Making it explicit
    forces the conversation about what 'low risk' actually means.
    """

    @staticmethod
    def corrective(
        suite: RegressionSuite,
        fixed_module: str
    ) -> List[RegressionTestCase]:
        """
        Corrective regression: re-test the fixed module plus any test
        that tags the fixed module as a dependency.
        This catches cases where the bug fix introduced a side effect
        in a module that imports the fixed one.
        """
        return [
            t for t in suite.test_cases
            if t.module == fixed_module
            or fixed_module in t.tags
        ]

    @staticmethod
    def progressive(
        suite: RegressionSuite,
        new_module: str,
        integration_modules: Set[str]
    ) -> List[RegressionTestCase]:
        """
        Progressive regression: test the new module plus every module
        it integrates with. integration_modules should include all
        modules the new feature calls, imports, or shares state with.
        """
        affected = {new_module} | integration_modules
        return suite.select_by_impact(affected)

    @staticmethod
    def selective(
        suite: RegressionSuite,
        changed_modules: Set[str]
    ) -> List[RegressionTestCase]:
        """
        Selective regression: run only tests impacted by the change.
        Most efficient for CI/CD pull request gates.
        Requires accurate module-to-test mapping and transitive
        dependency traversal to be effective.
        """
        return suite.select_by_impact(changed_modules)

    @staticmethod
    def complete(
        suite: RegressionSuite
    ) -> List[RegressionTestCase]:
        """
        Complete regression: run every test in the suite.
        The only strategy that guarantees full coverage.
        Run before major releases, after dependency upgrades,
        and after any infrastructure change.
        """
        return suite.test_cases

    @staticmethod
    def smoke(
        suite: RegressionSuite
    ) -> List[RegressionTestCase]:
        """
        Smoke regression: run only CRITICAL-priority tests.
        Designed for fast feedback — must complete in under 2 minutes.
        Catches obvious breakages; does not catch subtle regressions.
        """
        return suite.select_by_priority(RegressionPriority.CRITICAL)

    @staticmethod
    def recommend_strategy(
        change_scope: str,
        time_available_minutes: int,
        is_major_release: bool,
        touches_shared_utility: bool = False
    ) -> RegressionType:
        """
        Recommend the appropriate regression strategy.

        touches_shared_utility overrides time constraints because
        shared utility changes have unpredictable blast radius.
        Selective regression is not safe for them without thorough
        transitive dependency analysis.
        """
        if is_major_release:
            return RegressionType.COMPLETE

        # Shared utilities require at minimum selective with full transitive analysis
        # Time pressure does not reduce this requirement
        if touches_shared_utility and time_available_minutes < 30:
            return RegressionType.SELECTIVE  # with full transitive deps — not smoke

        if time_available_minutes < 5:
            return RegressionType.SMOKE

        if time_available_minutes < 30:
            return RegressionType.SELECTIVE

        if change_scope == "bug_fix":
            return RegressionType.CORRECTIVE

        if change_scope == "new_feature":
            return RegressionType.PROGRESSIVE

        return RegressionType.SELECTIVE


# Example — demonstrating strategy recommendation with edge cases
scenarios = [
    {"change_scope": "bug_fix", "time_available_minutes": 45,
     "is_major_release": False, "touches_shared_utility": False},
    {"change_scope": "config_change", "time_available_minutes": 3,
     "is_major_release": False, "touches_shared_utility": True},
    {"change_scope": "new_feature", "time_available_minutes": 20,
     "is_major_release": True, "touches_shared_utility": False},
]

for scenario in scenarios:
    strategy = RegressionStrategy.recommend_strategy(**scenario)
    print(f"Scope: {scenario['change_scope']}, "
          f"Time: {scenario['time_available_minutes']}min, "
          f"Shared utility: {scenario['touches_shared_utility']} "
          f"→ {strategy.value}")

from dataclasses import dataclass
from typing import List, Set, Dict
from collections import defaultdict
from io.thecodeforge.testing.regression import (
    RegressionTestCase, RegressionPriority
)


@dataclass
class ModuleDependency:
    module: str
    depends_on: List[str]


class ImpactAnalyzer:
    """
    Analyzes the impact of code changes across the module dependency
    graph using transitive reverse dependency traversal.

    Why transitive traversal matters:
    Module A imports B. Module B imports C (the locale utility).
    Changing C does not show A in C's direct reverse deps.
    But A is affected because its behavior changes when B's behavior changes.
    Only full transitive traversal catches this.
    """

    def __init__(self):
        self.dependencies: Dict[str, List[str]] = {}
        # reverse_dependencies[C] = [B, D] means B and D import C
        self.reverse_dependencies: Dict[str, List[str]] = defaultdict(list)
        # module_tests[module] = [test_id_1, test_id_2]
        self.module_tests: Dict[str, List[str]] = defaultdict(list)

    def add_dependency(self, module: str, depends_on: List[str]) -> None:
        """Register that 'module' imports everything in 'depends_on'."""
        self.dependencies[module] = depends_on
        for dep in depends_on:
            self.reverse_dependencies[dep].append(module)

    def register_test(self, module: str, test_id: str) -> None:
        """Map a test to the module it primarily exercises."""
        self.module_tests[module].append(test_id)

    def find_impacted_modules(self, changed_modules: Set[str]) -> Set[str]:
        """
        BFS traversal of the reverse dependency graph.
        Finds every module that transitively depends on any changed module.
        Starting with the changed modules and expanding outward until
        no new modules are found.
        """
        impacted = set(changed_modules)
        to_visit = list(changed_modules)

        while to_visit:
            current = to_visit.pop()
            for dependent in self.reverse_dependencies.get(current, []):
                if dependent not in impacted:
                    impacted.add(dependent)
                    to_visit.append(dependent)  # Continue traversing outward

        return impacted

    def find_impacted_tests(self, changed_modules: Set[str]) -> Set[str]:
        """
        Find all test IDs that should run based on transitive change impact.
        Returns the union of tests registered for all impacted modules.
        """
        impacted_modules = self.find_impacted_modules(changed_modules)
        test_ids: Set[str] = set()
        for module in impacted_modules:
            test_ids.update(self.module_tests.get(module, []))
        return test_ids

    def get_impact_report(self, changed_modules: Set[str]) -> Dict:
        """
        Generate a detailed impact report for a set of changes.
        impact_radius = how many additional modules beyond the changed ones
        are affected — a high radius signals a high-risk change.
        """
        impacted = self.find_impacted_modules(changed_modules)
        tests = self.find_impacted_tests(changed_modules)
        impact_radius = len(impacted) - len(changed_modules)

        return {
            "changed_modules": sorted(changed_modules),
            "impacted_modules": sorted(impacted),
            "impacted_test_count": len(tests),
            "impact_radius": impact_radius,
            # Thresholds are heuristics — tune for your codebase size
            "risk_level": (
                "high" if impact_radius > 5
                else "medium" if impact_radius > 2
                else "low"
            )
        }


class TestPrioritizer:
    """
    Ranks regression tests by a composite score combining:
    - Direct impact (the test's module was directly changed)
    - Business priority (CRITICAL > HIGH > MEDIUM > LOW)
    - Historical failure rate (tests that have failed before are more likely to fail again)

    Higher scores run first, giving faster failure feedback on the
    most important and most failure-prone tests.
    """

    @staticmethod
    def prioritize(
        tests: List[RegressionTestCase],
        changed_modules: Set[str]
    ) -> List[RegressionTestCase]:
        def score(test: RegressionTestCase) -> float:
            s = 0.0

            # Direct impact: this test's module was directly changed
            # Gets highest weight — the change directly affects this test
            if test.module in changed_modules:
                s += 100.0

            # Business priority weight
            priority_weights = {
                RegressionPriority.CRITICAL: 50.0,
                RegressionPriority.HIGH: 30.0,
                RegressionPriority.MEDIUM: 15.0,
                RegressionPriority.LOW: 5.0
            }
            s += priority_weights.get(test.priority, 0.0)

            # Historical failure correlation: cap at 40 to prevent
            # a very flaky test from dominating the ordering
            s += min(test.failure_count * 10.0, 40.0)

            return s

        return sorted(tests, key=score, reverse=True)

    @staticmethod
    def select_top_n(
        tests: List[RegressionTestCase],
        n: int,
        changed_modules: Set[str]
    ) -> List[RegressionTestCase]:
        """
        Select the top N highest-priority tests for time-constrained runs.
        Use this only when you have documented the risk of not running the rest.
        """
        prioritized = TestPrioritizer.prioritize(tests, changed_modules)
        return prioritized[:n]


# Example — demonstrating the locale utility cascading impact
analyzer = ImpactAnalyzer()

# Dependency declarations — who imports whom
analyzer.add_dependency("payments", ["locale", "currency"])
analyzer.add_dependency("notifications", ["locale", "email"])
analyzer.add_dependency("orders", ["payments", "inventory"])
analyzer.add_dependency("reports", ["payments", "locale"])

# Test-to-module registration
analyzer.register_test("payments", "TC-001")
analyzer.register_test("notifications", "TC-002")
analyzer.register_test("orders", "TC-003")
analyzer.register_test("locale", "TC-004")
analyzer.register_test("reports", "TC-005")

# Changing only the locale utility — how far does it reach?
report = analyzer.get_impact_report({"locale"})
print(f"Changed modules: {report['changed_modules']}")
print(f"Impacted modules: {report['impacted_modules']}")
print(f"Tests to run: {report['impacted_test_count']}")
print(f"Impact radius: {report['impact_radius']} additional modules")
print(f"Risk level: {report['risk_level']}")
# Output: locale change impacts payments, notifications, orders, and reports
# — four modules beyond the one that was touched

from typing import Dict, List, Optional
from dataclasses import dataclass


@dataclass
class TierConfig:
    trigger: str
    max_duration_minutes: int
    test_count_limit: str
    strategy: str
    purpose: str
    is_blocking: bool  # Whether failure blocks the pipeline event
    override_allowed: bool  # Should almost always be False in production


class RegressionPipeline:
    """
    Defines the regression testing pipeline tiers for CI/CD integration.

    Design principles:
    - Every tier is blocking by default — no override path for routine merges
    - Time budgets are hard constraints, not targets
    - If a tier exceeds its time budget, fix the suite — do not raise the budget
    - Tier 4 (production gate) never has an override path, period
    """

    TIERS: Dict[str, TierConfig] = {
        "tier_1_smoke": TierConfig(
            trigger="every_push",
            max_duration_minutes=2,
            test_count_limit="< 50",
            strategy="critical_priority_only",
            purpose="Fast feedback for obvious breakages — catches complete failures",
            is_blocking=True,
            override_allowed=False
        ),
        "tier_2_selective": TierConfig(
            trigger="pull_request",
            max_duration_minutes=15,
            test_count_limit="< 500",
            strategy="impact_based_selection_with_transitive_deps",
            purpose="Verify change does not break impacted modules",
            is_blocking=True,
            override_allowed=False  # Removing the override is the critical decision
        ),
        "tier_3_complete": TierConfig(
            trigger="merge_to_main",
            max_duration_minutes=60,
            test_count_limit="all",
            strategy="complete_regression",
            purpose="Full verification before release candidate creation",
            is_blocking=True,
            override_allowed=False
        ),
        "tier_4_production": TierConfig(
            trigger="before_production_deploy",
            max_duration_minutes=120,
            test_count_limit="all_including_e2e",
            strategy="complete_plus_end_to_end",
            purpose="Final gate before production traffic receives the change",
            is_blocking=True,
            override_allowed=False  # Never. Not for hotfixes. Not for time pressure.
        )
    }

    @staticmethod
    def should_block_deploy(tier_results: Dict[str, bool]) -> bool:
        """
        Any tier failure blocks deployment.
        Partial success is not success.
        """
        return not all(tier_results.values())

    @staticmethod
    def get_tier_for_event(event: str) -> str:
        """Map a pipeline event to its corresponding regression tier."""
        event_map = {
            "push": "tier_1_smoke",
            "pull_request": "tier_2_selective",
            "merge": "tier_3_complete",
            "deploy": "tier_4_production"
        }
        return event_map.get(event, "tier_1_smoke")

    @staticmethod
    def validate_tier_health(
        tier_name: str,
        actual_duration_minutes: float,
        config: TierConfig
    ) -> Dict:
        """
        Validate that a tier completed within its time budget.
        A tier consistently exceeding its budget needs suite optimization,
        not a looser budget.
        """
        within_budget = actual_duration_minutes <= config.max_duration_minutes
        overage_pct = (
            (actual_duration_minutes - config.max_duration_minutes)
            / config.max_duration_minutes * 100
            if not within_budget else 0.0
        )
        return {
            "tier": tier_name,
            "within_budget": within_budget,
            "actual_minutes": actual_duration_minutes,
            "budget_minutes": config.max_duration_minutes,
            "overage_percent": round(overage_pct, 1),
            "action_required": (
                "optimize_suite" if overage_pct > 20
                else "monitor" if not within_budget
                else "none"
            )
        }


# Example pipeline configuration output
pipeline = RegressionPipeline()
print("Pipeline Tiers (all blocking, no overrides):")
for tier_name, config in pipeline.TIERS.items():
    status = "HARD GATE" if not config.override_allowed else "SOFT GATE"
    print(
        f"  [{status}] {tier_name}: "
        f"{config.trigger} → {config.max_duration_minutes}min max "
        f"({config.strategy})"
    )

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from datetime import datetime
import uuid


@dataclass
class FlakyTestRecord:
    test_id: str
    name: str
    total_runs: int
    failures: int
    last_failure: Optional[datetime] = None
    failure_pattern: str = ""  # "intermittent", "time_sensitive", "order_dependent"

    @property
    def flakiness_rate(self) -> float:
        if self.total_runs == 0:
            return 0.0
        return self.failures / self.total_runs

    @property
    def is_flaky(self) -> bool:
        # A test with 0% or 100% failure rate is not flaky — it is broken or reliable
        # Flakiness is the unpredictable middle ground
        return 0.0 < self.flakiness_rate < 0.9


class RegressionAutomationManager:
    """
    Manages automated regression execution, flaky test detection,
    and suite health monitoring.

    Key behaviors:
    - Flaky test detection uses a sliding window, not cumulative counts
    - Quarantine removes tests from blocking gates but keeps them running
    - Suite health tracks stability rate — target > 95% stable tests
    """

    def __init__(self, flakiness_window: int = 20):
        self.test_history: Dict[str, List[bool]] = {}
        self.flaky_tests: List[FlakyTestRecord] = []
        self.quarantined: Set[str] = set()
        self.quarantine_reasons: Dict[str, str] = {}
        self.flakiness_window = flakiness_window

    def record_result(self, test_id: str, passed: bool) -> None:
        """Record a single test run result."""
        if test_id not in self.test_history:
            self.test_history[test_id] = []
        self.test_history[test_id].append(passed)

    def detect_flaky_tests(
        self, min_runs: int = 5
    ) -> List[FlakyTestRecord]:
        """
        Detect flaky tests using a sliding window of recent results.
        A test is flaky if it has BOTH passes and failures in the window.
        Requires at least min_runs results before flagging as flaky —
        avoids false positives on tests with only 1-2 runs.
        """
        flaky = []
        for test_id, history in self.test_history.items():
            recent = history[-self.flakiness_window:]
            if len(recent) < min_runs:
                continue

            failures = sum(1 for r in recent if not r)
            passes = sum(1 for r in recent if r)

            # Both passes AND failures in the window = flaky
            # All failures = broken (fix immediately, different process)
            if failures > 0 and passes > 0:
                flaky.append(FlakyTestRecord(
                    test_id=test_id,
                    name=test_id,
                    total_runs=len(recent),
                    failures=failures,
                    failure_pattern="intermittent"
                ))

        self.flaky_tests = flaky
        return flaky

    def quarantine_test(self, test_id: str, reason: str) -> None:
        """
        Quarantine a flaky test.
        Quarantined tests still run and report results but do not
        gate pipeline progression. This prevents flakes from blocking
        deployments while keeping the signal visible.

        A quarantined test with an unfixed root cause after one sprint
        should be deleted, not carried indefinitely.
        """
        self.quarantined.add(test_id)
        self.quarantine_reasons[test_id] = reason
        print(f"[QUARANTINE] {test_id}: {reason}")
        print(f"  Action required: fix root cause within one sprint or delete test")

    def get_executable_tests(
        self, all_tests: List[str], include_quarantined: bool = False
    ) -> List[str]:
        """
        Return tests eligible to gate the pipeline.
        include_quarantined=True runs all tests but marks quarantined ones
        as non-blocking — useful for visibility without impact.
        """
        if include_quarantined:
            return all_tests
        return [t for t in all_tests if t not in self.quarantined]

    def get_suite_health(self) -> Dict:
        """
        Calculate overall suite health metrics.
        Health status thresholds:
        - healthy: > 95% stable
        - degraded: 85-95% stable (flaky tests need attention)
        - unhealthy: < 85% stable (suite is unreliable, trust is eroded)
        """
        total = len(self.test_history)
        if total == 0:
            return {"health_status": "no_data"}

        stable = sum(
            1 for history in self.test_history.values()
            if all(history[-10:]) if len(history) >= 10 else all(history)
        )
        stability_rate = stable / total

        return {
            "total_tests": total,
            "stable_tests": stable,
            "flaky_tests": len(self.flaky_tests),
            "quarantined_tests": len(self.quarantined),
            "stability_rate": round(stability_rate, 3),
            "health_status": (
                "healthy" if stability_rate > 0.95
                else "degraded" if stability_rate > 0.85
                else "unhealthy"
            ),
            # Action guidance based on health status
            "recommended_action": (
                "none" if stability_rate > 0.95
                else "quarantine_and_fix_flaky_tests" if stability_rate > 0.85
                else "halt_feature_work_and_stabilize_suite"
            )
        }


class TestDataIsolator:
    """
    Provides utilities for test data isolation.
    Isolation prevents test order dependencies — the most common
    source of flaky behavior in automated regression suites.
    """

    @staticmethod
    def generate_unique_suffix() -> str:
        """Generate a short unique suffix for test resource naming."""
        return str(uuid.uuid4())[:8]

    @staticmethod
    def create_isolated_schema(test_name: str) -> str:
        """
        Create a unique database schema for a test.
        Schema isolation is lighter weight than full database isolation
        and works well for PostgreSQL environments.
        """
        suffix = TestDataIsolator.generate_unique_suffix()
        return f"test_{test_name[:20]}_{suffix}"

    @staticmethod
    def cleanup_test_schema(schema_name: str) -> None:
        """Drop test schema after test completion."""
        print(f"[CLEANUP] Dropping schema: {schema_name}")


# Example — simulating flaky test detection
manager = RegressionAutomationManager(flakiness_window=20)

for i in range(20):
    manager.record_result("TC-001", i % 5 != 0)   # Fails every 5th run (20% flaky)
    manager.record_result("TC-002", True)           # Always passes — stable
    manager.record_result("TC-003", i % 3 != 0)   # Fails every 3rd run (33% flaky)

flaky = manager.detect_flaky_tests()
print(f"Flaky tests detected: {len(flaky)}")
for test in flaky:
    print(f"  {test.test_id}: {test.flakiness_rate:.0%} failure rate — quarantine immediately")
    manager.quarantine_test(test.test_id, f"Intermittent failure at {test.flakiness_rate:.0%} rate")

health = manager.get_suite_health()
print(f"\nSuite health: {health['health_status']}")
print(f"Stability rate: {health['stability_rate']:.1%}")
print(f"Recommended action: {health['recommended_action']}")

from dataclasses import dataclass, field
from typing import Dict, Any, Optional, Callable, List
from datetime import datetime, timedelta
import random
import uuid


class TestDataManager:
    """
    Manages test data lifecycle for regression tests.
    Enforces isolation (each test owns its data) and cleanup
    (each test removes its data after completion).

    Using a registry pattern so factory functions are defined once
    and reused consistently — prevents fixture drift where different
    tests create slightly different versions of 'a user'.
    """

    def __init__(self):
        self._fixtures: Dict[str, Callable] = {}
        self._active_data: Dict[str, Any] = {}

    def register_fixture(
        self, name: str, factory: Callable
    ) -> None:
        """Register a named factory function. Factories are called fresh for each create()."""
        self._fixtures[name] = factory

    def create(
        self, fixture_name: str, test_id: str, **overrides
    ) -> Any:
        """
        Create test data from a registered fixture.
        test_id scopes the data — each test's data is namespaced separately.
        overrides allow per-test customization without duplicating factory logic.
        """
        if fixture_name not in self._fixtures:
            raise ValueError(
                f"Unknown fixture: '{fixture_name}'. "
                f"Register it with register_fixture() before use."
            )
        data = self._fixtures[fixture_name](**overrides)
        key = f"{test_id}:{fixture_name}:{uuid.uuid4().hex[:6]}"
        self._active_data[key] = data
        return data

    def cleanup(self, test_id: str) -> None:
        """Remove all data created by a specific test. Call this in teardown."""
        keys_to_remove = [
            k for k in self._active_data if k.startswith(f"{test_id}:")
        ]
        for key in keys_to_remove:
            del self._active_data[key]

    def cleanup_all(self) -> None:
        """Remove all test data — use after full suite completion."""
        self._active_data.clear()


def create_test_user(
    locale: str = "en_US",
    seed: Optional[int] = None,
    **overrides
) -> Dict[str, Any]:
    """
    Deterministic test user factory.
    Uses seed for reproducibility — the same seed produces the same data
    across different machines and CI environments.

    Locale parameter is explicit rather than defaulting to en_US everywhere —
    callers must consciously choose a locale, which prevents the realism gap
    where all fixtures accidentally use the same locale.
    """
    rng = random.Random(seed)  # Seeded RNG — not the global random state
    base = {
        "user_id": str(uuid.UUID(int=rng.getrandbits(128))),
        "email": f"test_{rng.randint(10000, 99999)}@example.thecodeforge.io",
        "locale": locale,
        "created_at": datetime.now().isoformat(),
        "plan": rng.choice(["basic", "premium", "enterprise"]),
        # Edge cases included by default, not just the happy path
        "profile_complete": rng.choice([True, True, True, False]),  # 25% incomplete
        "payment_method_valid": rng.choice([True, True, False]),    # 33% invalid
    }
    base.update(overrides)  # Per-test overrides take precedence
    return base


def create_test_transaction(
    user_id: str,
    currency: str = "USD",
    seed: Optional[int] = None,
    **overrides
) -> Dict[str, Any]:
    """
    Deterministic test transaction factory.
    Currency is explicit — forces callers to test non-USD paths.
    """
    rng = random.Random(seed)
    base = {
        "transaction_id": str(uuid.UUID(int=rng.getrandbits(128))),
        "user_id": user_id,
        "amount": round(rng.uniform(1.0, 9999.99), 2),
        "currency": currency,
        "timestamp": datetime.now().isoformat(),
        # Edge case: some transactions have null metadata
        "metadata": None if rng.random() < 0.1 else {"source": "web"},
    }
    base.update(overrides)
    return base


# Locales that production actually serves — not just en_US
PRODUCTION_LOCALES = ["en_US", "de_DE", "fr_FR", "ja_JP", "ar_SA", "pt_BR"]
PRODUCTION_CURRENCIES = ["USD", "EUR", "GBP", "JPY", "BRL", "SAR"]


# Example — creating realistic test data with locale coverage
manager = TestDataManager()
manager.register_fixture("user", create_test_user)
manager.register_fixture("transaction", create_test_transaction)

# Test that exercises a European locale — the one the production incident missed
eu_user = manager.create("user", "TC-001", locale="de_DE", seed=42)
eu_transaction = manager.create(
    "transaction", "TC-001",
    user_id=eu_user["user_id"],
    currency="EUR",
    seed=42
)

print("Test user (de_DE locale):")
for k, v in eu_user.items():
    print(f"  {k}: {v}")

print("\nTest transaction (EUR):")
for k, v in eu_transaction.items():
    print(f"  {k}: {v}")

# Cleanup scoped to TC-001 only
manager.cleanup("TC-001")
print("\n[CLEANUP] TC-001 data removed")

from dataclasses import dataclass
from typing import List, Dict, Tuple


@dataclass
class TestExecution:
    test_id: str
    estimated_duration_sec: float
    module: str
    # Historical p95 duration — used when estimated_duration is stale
    p95_duration_sec: float = 0.0


class ParallelSharder:
    """
    Distributes tests across parallel workers using duration-aware bin packing.
    Minimizes wall-clock time by balancing worker loads, not test counts.

    Algorithm: Longest Processing Time First (LPT)
    - Sort tests by duration descending
    - Assign each test to the worker with the least current load
    - This greedy approach produces near-optimal load balancing

    Why not round-robin:
    A 5-minute test and a 10-second test in the same pool means
    round-robin creates a worker imbalance that wastes wall-clock time.
    LPT minimizes the maximum worker runtime.
    """

    @staticmethod
    def shard_by_duration(
        tests: List[TestExecution],
        num_workers: int
    ) -> Dict[int, List[TestExecution]]:
        # Sort longest-first — this is critical for good load balancing
        sorted_tests = sorted(
            tests, key=lambda t: t.estimated_duration_sec, reverse=True
        )

        worker_loads = [0.0] * num_workers
        worker_assignments: Dict[int, List[TestExecution]] = {
            i: [] for i in range(num_workers)
        }

        for test in sorted_tests:
            # Assign to worker with the least current load
            lightest = min(range(num_workers), key=lambda w: worker_loads[w])
            worker_assignments[lightest].append(test)
            worker_loads[lightest] += test.estimated_duration_sec

        return worker_assignments

    @staticmethod
    def estimate_speedup(
        tests: List[TestExecution],
        num_workers: int
    ) -> Dict:
        serial_time = sum(t.estimated_duration_sec for t in tests)
        shards = ParallelSharder.shard_by_duration(tests, num_workers)

        worker_times = {
            w: sum(t.estimated_duration_sec for t in shard)
            for w, shard in shards.items()
        }
        parallel_time = max(worker_times.values()) if worker_times else 0.0

        utilization = {
            w: round(load / parallel_time, 3) if parallel_time > 0 else 0.0
            for w, load in worker_times.items()
        }

        return {
            "serial_time_sec": round(serial_time, 1),
            "parallel_time_sec": round(parallel_time, 1),
            "speedup": round(serial_time / parallel_time, 1) if parallel_time > 0 else 0,
            "num_workers": num_workers,
            "worker_utilization": utilization,
            # Low min utilization means uneven sharding — some workers idle
            "min_worker_utilization": min(utilization.values()) if utilization else 0.0,
            "sharding_efficiency": "good" if min(utilization.values()) > 0.7 else "poor"
        }


class SuiteOptimizer:
    """
    Identifies optimization opportunities before parallelization.
    Optimize first, parallelize second.
    """

    @staticmethod
    def find_slow_tests(
        tests: List[TestExecution],
        threshold_sec: float = 30.0
    ) -> List[TestExecution]:
        """
        Tests exceeding the threshold are candidates for:
        - Mocking external service calls (most common root cause)
        - Splitting into multiple focused tests
        - Moving to a nightly suite if they cannot be optimized
        """
        return sorted(
            [t for t in tests if t.estimated_duration_sec > threshold_sec],
            key=lambda t: t.estimated_duration_sec,
            reverse=True
        )

    @staticmethod
    def find_redundant_tests(
        tests: List[TestExecution],
        module_coverage: Dict[str, List[str]]
    ) -> List[str]:
        """
        Tests whose module coverage is a strict subset of another test
        may be redundant. This is a signal for review — not automatic deletion.
        Always verify before removing — the subset test may be faster or
        have a different assertion focus.
        """
        redundant = []
        for i, test_a in enumerate(tests):
            for j, test_b in enumerate(tests):
                if i == j:
                    continue
                modules_a = set(module_coverage.get(test_a.test_id, []))
                modules_b = set(module_coverage.get(test_b.test_id, []))
                if modules_b and modules_b.issubset(modules_a):
                    redundant.append(test_b.test_id)
        return list(set(redundant))

    @staticmethod
    def optimization_report(
        tests: List[TestExecution],
        slow_threshold_sec: float = 30.0,
        num_workers: int = 8
    ) -> Dict:
        """Generate a prioritized optimization report."""
        slow = SuiteOptimizer.find_slow_tests(tests, slow_threshold_sec)
        speedup = ParallelSharder.estimate_speedup(tests, num_workers)

        return {
            "total_tests": len(tests),
            "slow_test_count": len(slow),
            "slow_test_ids": [t.test_id for t in slow[:5]],  # Top 5 slowest
            "time_saved_if_slow_fixed_sec": sum(
                t.estimated_duration_sec - slow_threshold_sec for t in slow
            ),
            "parallel_speedup": speedup,
            "recommended_action": (
                "fix_slow_tests_first" if len(slow) > 5
                else "parallelize_now"
            )
        }


# Example
import random
random.seed(42)

tests = [
    TestExecution(
        test_id=f"TC-{i:03d}",
        estimated_duration_sec=random.uniform(0.5, 120.0),
        module=f"module_{i % 10}"
    )
    for i in range(200)
]

report = SuiteOptimizer.optimization_report(tests, slow_threshold_sec=60.0, num_workers=8)
print(f"Total tests: {report['total_tests']}")
print(f"Slow tests (>60s): {report['slow_test_count']}")
print(f"Time saved if slow tests fixed: {report['time_saved_if_slow_fixed_sec']:.0f}s")
print(f"\nParallel execution (8 workers):")
print(f"  Serial: {report['parallel_speedup']['serial_time_sec']}s")
print(f"  Parallel: {report['parallel_speedup']['parallel_time_sec']}s")
print(f"  Speedup: {report['parallel_speedup']['speedup']}x")
print(f"  Sharding efficiency: {report['parallel_speedup']['sharding_efficiency']}")
print(f"\nRecommendation: {report['recommended_action']}")

// io.thecodeforge.regression
import org.junit.jupiter.api.Test;

// Simulates triggering regression on a payment hotfix
public class RegressionTriggerTest {
    @Test
    void verifyPaymentAfterBugFix() {
        PaymentService svc = new PaymentService();
        Invoice inv = svc.processPayment(new CreditCard("4111-1111-1111-1111", 2999));
        
        // New bug fix: ensure refund idempotency
        assert inv.isCompleted() : "Payment did not finalize";
        assert inv.getTotal() == 2999 : "Total mismatch after fix";
        
        // Regression check: old invoice path still works
        Invoice legacy = svc.processPaymentFromLegacySystem("order-42");
        assert legacy.getStatus() != InvoiceStatus.FAILED : "Legacy path regressed";
    }
}

// io.thecodeforge.selection
import java.util.*;

// Stub for change-impact test selection strategy
public class ImpactSelector {
    private final Map<String, Set<String>> testCoverage = new HashMap<>();
    
    public Set<String> selectTestsFor(Set<String> changedFiles) {
        Set<String> impacted = new HashSet<>();
        for (String file : changedFiles) {
            if (testCoverage.containsKey(file)) {
                impacted.addAll(testCoverage.get(file));
            }
        }
        // Always include mandatory core tests
        impacted.addAll(CORE_TESTS);
        return impacted;
    }
    
    public static final Set<String> CORE_TESTS = Set.of(
        "testLoginFlow", "testPaymentIdempotency", "testDataIntegrity"
    );
}