How to Implement AI-Powered Testing with PyTest 8.0 and Testim 2.0

import pytest
import os
import json
import logging
from typing import Optional, Dict, Any
from dataclasses import dataclass

# Configure logging for test fixture diagnostics
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("ai_adaptive_fixtures")

@dataclass
class TestEnvironmentConfig:
    """Stores environment-specific config with AI-suggested overrides"""
    base_url: str
    db_connection_string: Optional[str]
    ai_adaptive_enabled: bool
    flakiness_threshold: float = 0.15  # 15% flakiness threshold before AI intervention

def _load_env_config() -> Dict[str, Any]:
    """Load environment config from .env.test or AI-generated fallback"""
    config = {}
    # Try loading from local .env first
    if os.path.exists(".env.test"):
        with open(".env.test", "r") as f:
            for line in f:
                if line.strip() and not line.startswith("#"):
                    key, value = line.strip().split("=", 1)
                    config[key] = value
        logger.info("Loaded config from .env.test")
    else:
        # Fallback to AI-generated config if no local file exists
        logger.warning("No .env.test found, loading AI-generated default config")
        config = {
            "BASE_URL": "https://staging.example.com",
            "DB_CONNECTION_STRING": os.getenv("CI_DB_STRING", "postgresql://user:pass@localhost:5432/test"),
            "AI_ADAPTIVE_ENABLED": "true"
        }
    return config

@pytest.fixture(scope="ai_adaptive")  # PyTest 8.0’s new adaptive scope: auto-adjusts based on test run history
def test_env_config() -> TestEnvironmentConfig:
    """AI-adaptive fixture that adjusts scope based on test flakiness history"""
    try:
        raw_config = _load_env_config()
        config = TestEnvironmentConfig(
            base_url=raw_config.get("BASE_URL", "https://default.example.com"),
            db_connection_string=raw_config.get("DB_CONNECTION_STRING"),
            ai_adaptive_enabled=raw_config.get("AI_ADAPTIVE_ENABLED", "false").lower() == "true"
        )
        # Log AI fixture activation if enabled
        if config.ai_adaptive_enabled:
            logger.info(f"AI-adaptive fixture activated for {config.base_url}")
        return config
    except Exception as e:
        logger.error(f"Failed to load test environment config: {e}")
        raise pytest.FixtureLookupError("test_env_config", f"Config load failed: {e}")

@pytest.fixture(scope="ai_adaptive")
def db_session(test_env_config: TestEnvironmentConfig):
    """AI-adaptive DB fixture that automatically retries flaky connections"""
    if not test_env_config.db_connection_string:
        logger.warning("No DB connection string provided, skipping DB fixture")
        yield None
        return
    import psycopg2  # Lazy import to avoid requiring DB deps for non-DB tests
    max_retries = 3
    retry_delay = 1  # seconds
    conn = None
    for attempt in range(max_retries):
        try:
            conn = psycopg2.connect(test_env_config.db_connection_string)
            logger.info(f"DB connection established on attempt {attempt + 1}")
            yield conn
            break
        except psycopg2.OperationalError as e:
            logger.warning(f"DB connection attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                raise pytest.FixtureLookupError("db_session", f"Failed to connect to DB after {max_retries} attempts: {e}")
            import time
            time.sleep(retry_delay * (2 ** attempt))  # Exponential backoff
    if conn:
        conn.close()
        logger.info("DB connection closed")

import os
import json
import subprocess
import pytest
import logging
from typing import List, Dict, Any, Optional
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("testim_integration")

@dataclass
class TestimTestResult:
    """Structured result from a Testim 2.0 test run"""
    test_id: str
    name: str
    status: str  # "passed", "failed", "flaky", "skipped"
    duration_ms: int
    error_message: Optional[str]
    ai_suggested_fix: Optional[str]

def _run_testim_cli(test_ids: List[str], base_url: str, timeout: int = 300) -> List[Dict[str, Any]]:
    """Execute Testim 2.0 CLI with AI self-healing enabled"""
    testim_token = os.getenv("TESTIM_API_TOKEN")
    if not testim_token:
        raise ValueError("TESTIM_API_TOKEN environment variable is required")

    # Build Testim CLI command with AI self-healing flags
    cmd = [
        "testim",
        "run",
        "--token", testim_token,
        "--project", os.getenv("TESTIM_PROJECT_ID", "default-project"),
        "--test-ids", ",".join(test_ids),
        "--base-url", base_url,
        "--self-healing-enabled",  # Testim 2.0’s AI self-healing flag
        "--ai-root-cause-enabled",  # Enable AI failure analysis
        "--timeout", str(timeout * 1000),  # Testim expects timeout in ms
        "--format", "json",
        "--output", "testim_results.json"
    ]

    logger.info(f"Running Testim CLI: {' '.join(cmd)}")
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout + 30  # Buffer for CLI startup
        )
        if result.returncode != 0:
            logger.error(f"Testim CLI failed: {result.stderr}")
            raise RuntimeError(f"Testim run failed with exit code {result.returncode}: {result.stderr}")

        # Load results from JSON output
        with open("testim_results.json", "r") as f:
            raw_results = json.load(f)
        return raw_results.get("tests", [])
    except subprocess.TimeoutExpired:
        logger.error(f"Testim run timed out after {timeout} seconds")
        raise
    except FileNotFoundError:
        logger.error("Testim CLI not found. Install via: npm install -g testim-cli@2.0")
        raise
    finally:
        # Clean up temporary results file
        if os.path.exists("testim_results.json"):
            os.remove("testim_results.json")

@pytest.mark.testim  # Custom marker to identify Testim integration tests
def test_run_testim_e2e_suite(test_env_config):
    """Run Testim 2.0 E2E suite and validate results with PyTest"""
    # List of Testim test IDs to run (from Testim dashboard)
    testim_test_ids = [
        "test-1234-e2e-login",
        "test-5678-e2e-checkout",
        "test-9012-e2e-user-profile"
    ]

    try:
        raw_results = _run_testim_cli(
            test_ids=testim_test_ids,
            base_url=test_env_config.base_url,
            timeout=300
        )

        # Parse results into structured objects
        parsed_results: List[TestimTestResult] = []
        for raw in raw_results:
            parsed_results.append(TestimTestResult(
                test_id=raw.get("id"),
                name=raw.get("name"),
                status=raw.get("status"),
                duration_ms=raw.get("durationMs", 0),
                error_message=raw.get("errorMessage"),
                ai_suggested_fix=raw.get("aiSuggestedFix")
            ))

        # Assert no critical failures
        failed_tests = [r for r in parsed_results if r.status == "failed"]
        if failed_tests:
            error_details = "\n".join([
                f"Test {t.name} ({t.test_id}) failed: {t.error_message}"
                f"AI Suggested Fix: {t.ai_suggested_fix}" if t.ai_suggested_fix else ""
                for t in failed_tests
            ])
            pytest.fail(f"Testim E2E suite failed with {len(failed_tests)} failures:\n{error_details}")

        # Log AI fixes if available
        ai_fixes = [r for r in parsed_results if r.ai_suggested_fix]
        if ai_fixes:
            logger.info(f"Testim AI suggested {len(ai_fixes)} fixes:")
            for fix in ai_fixes:
                logger.info(f"  {fix.name}: {fix.ai_suggested_fix}")

        logger.info(f"Testim E2E suite passed: {len(parsed_results)} tests run")
    except Exception as e:
        logger.error(f"Testim integration test failed: {e}")
        raise

import json
import os
import argparse
import logging
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, asdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("unified_report")

@dataclass
class UnifiedTestResult:
    """Unified test result across PyTest and Testim"""
    source: str  # "pytest" or "testim"
    test_id: str
    name: str
    status: str
    duration_ms: int
    flakiness_score: Optional[float]  # 0.0 (stable) to 1.0 (flaky)
    ai_suggested_fix: Optional[str]
    error_message: Optional[str]

def _load_pytest_results(pytest_json_path: str) -> List[UnifiedTestResult]:
    """Load PyTest 8.0 JSON report (generated with --json-report flag)"""
    try:
        with open(pytest_json_path, "r") as f:
            raw = json.load(f)
        results = []
        for test in raw.get("tests", []):
            # Calculate flakiness score from PyTest 8.0’s run history
            flakiness = test.get("metadata", {}).get("ai_flakiness_score")
            results.append(UnifiedTestResult(
                source="pytest",
                test_id=test.get("nodeid"),
                name=test.get("nodeid").split("::")[-1],
                status=test.get("outcome"),
                duration_ms=int(test.get("duration", 0) * 1000),
                flakiness_score=flakiness,
                ai_suggested_fix=test.get("metadata", {}).get("ai_suggested_fix"),
                error_message=test.get("call", {}).get("longrepr") if test.get("outcome") == "failed" else None
            ))
        logger.info(f"Loaded {len(results)} PyTest results from {pytest_json_path}")
        return results
    except FileNotFoundError:
        logger.error(f"PyTest JSON report not found at {pytest_json_path}")
        return []
    except json.JSONDecodeError:
        logger.error(f"Invalid JSON in PyTest report {pytest_json_path}")
        return []

def _load_testim_results(testim_json_path: str) -> List[UnifiedTestResult]:
    """Load Testim 2.0 JSON results"""
    try:
        with open(testim_json_path, "r") as f:
            raw = json.load(f)
        results = []
        for test in raw.get("tests", []):
            results.append(UnifiedTestResult(
                source="testim",
                test_id=test.get("id"),
                name=test.get("name"),
                status=test.get("status"),
                duration_ms=test.get("durationMs", 0),
                flakiness_score=test.get("flakinessScore"),
                ai_suggested_fix=test.get("aiSuggestedFix"),
                error_message=test.get("errorMessage")
            ))
        logger.info(f"Loaded {len(results)} Testim results from {testim_json_path}")
        return results
    except FileNotFoundError:
        logger.error(f"Testim JSON report not found at {testim_json_path}")
        return []

def generate_unified_report(pytest_json: str, testim_json: str, output_path: str):
    """Generate unified HTML report with AI insights"""
    pytest_results = _load_pytest_results(pytest_json)
    testim_results = _load_testim_results(testim_json)
    all_results = pytest_results + testim_results

    # Calculate aggregate metrics
    total_tests = len(all_results)
    passed = len([r for r in all_results if r.status == "passed"])
    failed = len([r for r in all_results if r.status == "failed"])
    flaky = len([r for r in all_results if r.status == "flaky"])
    avg_flakiness = sum(r.flakiness_score for r in all_results if r.flakiness_score) / len(all_results) if all_results else 0.0

    # Generate HTML report
    html_content = f"""




        Unified AI-Powered Test Report
        Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
        Aggregate Metrics

            Total Tests: {total_tests}
            Passed: {passed}
            Failed: {failed}
            Flaky: {flaky}
            Average Flakiness Score: {avg_flakiness:.2f}

        Test Results

            {"".join([
                f""
                f""
                f""
                f""
                f""
                f""
                f""
                f""
                for r in all_results
            ])}


                Source
                Test Name
                Status
                Duration (ms)
                Flakiness Score
                AI Suggested Fix
            {r.source}{r.name}{r.status}{r.duration_ms}{r.flakiness_score:.2f if r.flakiness_score else 'N/A'}{r.ai_suggested_fix if r.ai_suggested_fix else 'None'}


    """

    with open(output_path, "w") as f:
        f.write(html_content)
    logger.info(f"Unified report generated at {output_path}: {total_tests} total tests, {failed} failures")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Generate unified test report from PyTest and Testim results")
    parser.add_argument("--pytest-json", required=True, help="Path to PyTest JSON report")
    parser.add_argument("--testim-json", required=True, help="Path to Testim JSON report")
    parser.add_argument("--output", default="unified_report.html", help="Output path for HTML report")
    args = parser.parse_args()
    generate_unified_report(args.pytest_json, args.testim_json, args.output)

[pytest]
ai_flakiness_threshold = 0.08
ai_adaptive_enabled = true
log_file = pytest_ai.log
log_file_level = INFO

# Install testim pre-commit hook
testim hooks install --pre-commit --project your-project-id --token $TESTIM_API_TOKEN
# Add to .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: testim-selector-validate
        name: Validate Testim Selectors
        entry: testim pre-commit validate
        language: system
        files: '\.(js|ts|jsx|tsx)$'

@pytest.hookimpl(tryfirst=True)
def pytest_runtest_makereport(item, call):
    if call.excinfo and item.get_closest_marker("testim"):
        # Pull AI root cause from Testim result
        testim_result = item.stash[testim_result_key]
        if testim_result.ai_suggested_fix and testim_result.flakiness_score < 0.1:
            item.stash[ai_triage_decision] = "auto_fixed"
            item.add_marker(pytest.mark.skip(reason=f"AI auto-fixed: {testim_result.ai_suggested_fix}"))

ai-powered-testing-pytest-testim/
├── conftest.py                  # PyTest 8.0 AI-adaptive fixtures (from Code Example 1)
├── tests/
│   ├── unit/                    # PyTest unit tests with AI fixtures
│   │   ├── test_auth.py
│   │   └── test_payments.py
│   ├── integration/             # PyTest integration tests with DB fixtures
│   │   ├── test_db.py
│   │   └── test_api.py
│   └── e2e/                     # Testim integration tests (from Code Example 2)
│       ├── test_testim_e2e.py
│       └── testim_test_ids.json
├── scripts/
│   ├── generate_unified_report.py  # Unified report generator (from Code Example 3)
│   └── ci_run_tests.sh             # CI pipeline script
├── .github/
│   └── workflows/
│       └── test-pipeline.yml       # GitHub Actions CI workflow
├── pytest.ini                      # PyTest 8.0 configuration
├── .testimrc                       # Testim 2.0 configuration
├── .env.test.example               # Example environment config
└── README.md                       # Setup and usage instructions