Regression Testing: Definition, Types, Tools, and Best Practices

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"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@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
    tags: List[str] = field(default_factory=list)


@dataclass
class RegressionSuite:
    """
    Manages a regression test suite with impact-based selection
    and execution tracking.
    """
    
    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.
        This is the core of risk-based regression selection.
        """
        selected = []
        for test in self.test_cases:
            if test.module in changed_modules:
                selected.append(test)
            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.
        """
        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 failed more than threshold times.
        These should be quarantined and fixed.
        """
        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.
        """
        return sum(t.avg_duration_ms for t in tests) / 1000.0
    
    def get_stats(self) -> Dict:
        """
        Return suite statistics.
        """
        total = len(self.test_cases)
        by_priority = {}
        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
suite = RegressionSuite(suite_name="main-regression")

suite.add_test(RegressionTestCase(
    test_id="TC-001",
    name="test_payment_processing",
    module="payments",
    priority=RegressionPriority.CRITICAL,
    tags=["payments", "locale", "currency"],
    avg_duration_ms=250.0
))

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

changed = {"notifications", "locale"}
selected = suite.select_by_impact(changed)
print(f"Selected {len(selected)} tests for changes in {changed}")
for test in selected:
    print(f"  {test.test_id}: {test.name} ({test.priority.value})")

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


class RegressionType(Enum):
    CORRECTIVE = "corrective"
    PROGRESSIVE = "progressive"
    SELECTIVE = "selective"
    COMPLETE = "complete"
    SMOKE = "smoke"
    UNIT = "unit"


class RegressionStrategy:
    """
    Implements different regression testing strategies
    based on change scope and risk level.
    """
    
    @staticmethod
    def corrective(
        suite: RegressionSuite,
        fixed_module: str
    ) -> List[RegressionTestCase]:
        """
        Corrective regression: re-test the module where the bug was fixed
        plus any directly dependent modules.
        """
        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 new module plus all modules
        it integrates with to verify no existing functionality broke.
        """
        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 pipelines.
        """
        return suite.select_by_impact(changed_modules)
    
    @staticmethod
    def complete(
        suite: RegressionSuite
    ) -> List[RegressionTestCase]:
        """
        Complete regression: run every test in the suite.
        Use before major releases or after infrastructure changes.
        """
        return suite.test_cases
    
    @staticmethod
    def smoke(
        suite: RegressionSuite
    ) -> List[RegressionTestCase]:
        """
        Smoke regression: run only critical-priority tests.
        Use for fast feedback in CI pipelines.
        """
        return suite.select_by_priority(RegressionPriority.CRITICAL)
    
    @staticmethod
    def recommend_strategy(
        change_scope: str,
        time_available_minutes: int,
        is_major_release: bool
    ) -> RegressionType:
        """
        Recommend the appropriate regression strategy.
        """
        if is_major_release:
            return RegressionType.COMPLETE
        
        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
strategy = RegressionStrategy.recommend_strategy(
    change_scope="new_feature",
    time_available_minutes=15,
    is_major_release=False
)
print(f"Recommended strategy: {strategy.value}")

from dataclasses import dataclass
from typing import List, Set, Dict, Optional
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 to select relevant regression tests.
    """
    
    def __init__(self):
        self.dependencies: Dict[str, List[str]] = {}
        self.reverse_dependencies: Dict[str, List[str]] = defaultdict(list)
        self.module_tests: Dict[str, List[str]] = defaultdict(list)
    
    def add_dependency(self, module: str, depends_on: List[str]) -> None:
        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:
        self.module_tests[module].append(test_id)
    
    def find_impacted_modules(self, changed_modules: Set[str]) -> Set[str]:
        """
        Find all modules impacted by changes using transitive
        reverse dependency traversal.
        """
        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)
        
        return impacted
    
    def find_impacted_tests(
        self,
        changed_modules: Set[str]
    ) -> Set[str]:
        """
        Find all test IDs that should run based on change impact.
        """
        impacted_modules = self.find_impacted_modules(changed_modules)
        test_ids = 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.
        """
        impacted = self.find_impacted_modules(changed_modules)
n        
        return {
            "changed_modules": list(changed_modules),
            "impacted_modules": list(impacted),
            "impacted_test_count": len(tests),
            "impact_radius": len(impacted) - len(changed_modules),
            "risk_level": "high" if len(impacted) > 5 else "medium" if len(impacted) > 2 else "low"
        }


class TestPrioritizer:
    """
    Prioritizes regression tests by failure probability
    and business impact for optimal defect detection.
    """
    
    @staticmethod
    def prioritize(
        tests: List[RegressionTestCase],
        changed_modules: Set[str]
    ) -> List[RegressionTestCase]:
        """
        Sort tests by priority score combining:
        - Direct impact (module was changed)
        - Historical failure rate
        - Business priority
        """
        def score(test: RegressionTestCase) -> float:
            s = 0.0
            
            if test.module in changed_modules:
                s += 100.0
            
            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)
            
            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.
        Use when time is constrained.
        """
        prioritized = TestPrioritizer.prioritize(tests, changed_modules)
        return prioritized[:n]


# Example
analyzer = ImpactAnalyzer()
analyzer.add_dependency("payments", ["locale", "currency"])
analyzer.add_dependency("notifications", ["locale", "email"])
analyzer.add_dependency("orders", ["payments", "inventory"])

analyzer.register_test("payments", "TC-001")
analyzer.register_test("notifications", "TC-002")
analyzer.register_test("orders", "TC-003")
analyzer.register_test("locale", "TC-004")

report = analyzer.get_impact_report({"locale"})
print(f"Changed: {report['changed_modules']}")
print(f"Impacted: {report['impacted_modules']}")
print(f"Tests to run: {report['impacted_test_count']}")
print(f"Risk level: {report['risk_level']}")

# Tiered regression testing pipeline
# Tier 1: Smoke tests on every push (< 2 minutes)
# Tier 2: Selective regression on PR (< 15 minutes)
# Tier 3: Complete regression on merge to main (< 60 minutes)
# Tier 4: Full suite including E2E before production deploy (< 120 minutes)

name: Regression Testing Pipeline

on:
  push:
    branches: [develop]
  pull_request:
    branches: [main]
  merge_group:
    branches: [main]

class RegressionPipeline:
    """
    Defines the regression testing pipeline tiers
    for CI/CD integration.
    """
    
    TIERS = {
        "tier_1_smoke": {
            "trigger": "every_push",
            "max_duration_minutes": 2,
            "test_count": "< 50",
            "strategy": "critical_priority_only",
            "purpose": "Fast feedback for obvious breakages"
        },
        "tier_2_selective": {
            "trigger": "pull_request",
            "max_duration_minutes": 15,
            "test_count": "< 500",
            "strategy": "impact_based_selection",
            "purpose": "Verify change does not break impacted modules"
        },
        "tier_3_complete": {
            "trigger": "merge_to_main",
            "max_duration_minutes": 60,
            "test_count": "all",
            "strategy": "complete_regression",
            "purpose": "Full verification before release candidate"
        },
        "tier_4_production": {
            "trigger": "before_deploy",
            "max_duration_minutes": 120,
            "test_count": "all_including_e2e",
            "strategy": "complete_plus_e2e",
            "purpose": "Final gate before production traffic"
        }
    }
    
    @staticmethod
    def should_block_deploy(tier_results: Dict[str, bool]) -> bool:
        """
        Determine if deployment should be blocked based on tier results.
        Any tier failure blocks deployment.
        """
        return not all(tier_results.values())
    
    @staticmethod
    def get_tier_for_event(event: str) -> str:
        """
        Map pipeline event to 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")


# Example pipeline execution
pipeline = RegressionPipeline()
print("Pipeline Tiers:")
for tier, config in pipeline.TIERS.items():
    print(f"  {tier}: {config['trigger']} - {config['max_duration_minutes']}min max")

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from datetime import datetime
import hashlib


@dataclass
class FlakyTestRecord:
    test_id: str
    name: str
    total_runs: int
    failures: int
    last_failure: Optional[datetime] = None
    failure_pattern: str = ""
    
    @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:
        return 0.0 < self.flakiness_rate < 0.5


class RegressionAutomationManager:
    """
    Manages automated regression test execution,
    flaky test detection, and suite health monitoring.
    """
    
    def __init__(self):
        self.test_history: Dict[str, List[bool]] = {}
        self.flaky_tests: List[FlakyTestRecord] = []
        self.quarantined: Set[str] = set()
    
    def record_result(self, test_id: str, passed: bool) -> None:
        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, window: int = 20) -> List[FlakyTestRecord]:
        """
        Detect flaky tests based on recent run history.
        A test is flaky if it has both passes and failures in the window.
        """
        flaky = []
        
        for test_id, history in self.test_history.items():
            recent = history[-window:]
            if len(recent) < 5:
                continue
            
            failures = sum(1 for r in recent if not r)
            passes = sum(1 for r in recent if r)
            
            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 so it does not block the pipeline.
        The test still runs but does not gate deployments.
        """
        self.quarantined.add(test_id)
        print(f"Quarantined {test_id}: {reason}")
    
    def get_executable_tests(
        self,
        all_tests: List[str]
    ) -> List[str]:
        """
        Return tests that are not quarantined.
        """
        return [t for t in all_tests if t not in self.quarantined]
    
    def get_suite_health(self) -> Dict:
        """
        Calculate overall suite health metrics.
        """
        total = len(self.test_history)
        if total == 0:
            return {"status": "no_data"}
        
        stable = sum(
            1 for history in self.test_history.values()
            if all(history[-10:]) if len(history) >= 10 else all(history)
        )
        
        return {
            "total_tests": total,
            "stable_tests": stable,
            "flaky_tests": len(self.flaky_tests),
            "quarantined_tests": len(self.quarantined),
            "stability_rate": stable / total if total > 0 else 0.0,
            "health_status": (
                "healthy" if stable / total > 0.95
                else "degraded" if stable / total > 0.85
                else "unhealthy"
            )
        }


class TestDataIsolator:
    """
    Ensures test data isolation to prevent test order dependencies.
    """
    
    @staticmethod
    def generate_unique_suffix() -> str:
        import uuid
        return str(uuid.uuid4())[:8]
    
    @staticmethod
    def create_isolated_database(test_name: str) -> str:
        """
        Create a unique database for each test run.
        """
        suffix = TestDataIsolator.generate_unique_suffix()
        return f"test_{test_name}_{suffix}"
    
    @staticmethod
    def cleanup_test_data(db_name: str) -> None:
        """
        Drop test database after test completion.
        """
        print(f"Dropping test database: {db_name}")


# Example
manager = RegressionAutomationManager()

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

flaky = manager.detect_flaky_tests()
print(f"Flaky tests detected: {len(flaky)}")
for test in flaky:
    print(f"  {test.test_id}: {test.flakiness_rate:.1%} failure rate")

health = manager.get_suite_health()
print(f"Suite health: {health['health_status']}")