flaky_tests.py

#!/usr/bin/env python3
"""
Script to analyze test flakiness from WebKit results database.
Outputs CSV with test name, unexpected count, and total count.
"""

import csv
import json
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlencode

import requests


# Thread-local storage for requests sessions
thread_local = threading.local()


def get_session():
    """Get a requests session for the current thread."""
    if not hasattr(thread_local, 'session'):
        thread_local.session = requests.Session()
    return thread_local.session


def fetch_json(url):
    """Fetch JSON data from URL using thread-local session."""
    session = get_session()
    response = session.get(url)
    response.raise_for_status()
    return response.json()


def get_recent_cutoff(days=14):
    """Get timestamp cutoff for recent results (last N days)."""
    return int(time.time()) - (days * 24 * 60 * 60)


def create_base_params():
    """Create base parameters for API requests."""
    params = {
        'after_timestamp': get_recent_cutoff(),
        'platform': 'mac',
        'flavor': 'wk2'
    }
    return params


def fetch_test_list():
    """Fetch list of tests that have failures in recent period."""
    base_params = create_base_params()
    
    # Get failures data to identify which tests have failures
    failures_params = base_params.copy()
    failures_params.update({
        'collapsed': 'true',
        'unexpected': 'false'
    })
    
    failures_url = f"https://results.webkit.org/api/failures/layout-tests?{urlencode(failures_params)}"
    print(f"Fetching failures data from: {failures_url}", file=sys.stderr)
    
    failures_data = fetch_json(failures_url)
    
    # Extract unique test names from failures
    test_names = set()
    for config_result in failures_data:
        for result in config_result['results']:
            for key in result.keys():
                if key not in ('uuid', 'start_time'):
                    test_names.add(key)
    
    print(f"Found {len(test_names)} tests with failures", file=sys.stderr)
    return sorted(test_names)


def fetch_test_results(test_name):
    """Fetch detailed results for a specific test."""
    base_params = create_base_params()
    test_url = f"https://results.webkit.org/api/results/layout-tests/{requests.utils.quote(test_name, safe='')}?{urlencode(base_params)}"
    
    try:
        test_results = fetch_json(test_url)
        
        # Process results to count unexpected vs total
        unexpected_count = 0
        total_count = 0
        
        for config_result in test_results:
            for result in config_result['results']:
                total_count += 1
                # Count as unexpected if actual != expected (expected defaults to PASS)
                assert '.' not in result['actual'], repr(result['actual'])
                assert ' ' not in result['actual'], repr(result['actual'])
                actual = set(result['actual'].split())
                expected = set(result.get('expected', 'PASS').split())
                if 'FAIL' in expected:
                    expected |= {'TEXT', 'IMAGE', 'AUDIO'}
                if actual - expected:
                    unexpected_count += 1
        
        return {
            'test': test_name,
            'unexpected': unexpected_count,
            'total': total_count
        }
        
    except Exception as e:
        print(f"Error fetching results for {test_name}: {e}", file=sys.stderr)
        return {
            'test': test_name,
            'unexpected': 0,
            'total': 0
        }


def main():
    """Main function to orchestrate the flakiness analysis."""
    print("Fetching test list...", file=sys.stderr)
    
    # Get list of tests that have failures
    test_names = fetch_test_list()
    
    if not test_names:
        print("No tests with failures found", file=sys.stderr)
        return
    
    print(f"Fetching detailed results for {len(test_names)} tests...", file=sys.stderr)
    
    # Use ThreadPoolExecutor to fetch all test results concurrently
    results = []
    with ThreadPoolExecutor(max_workers=8) as executor:
        # Submit all tasks first without blocking
        future_to_test = {
            executor.submit(fetch_test_results, test_name): test_name 
            for test_name in test_names
        }
        
        # Collect results as they complete
        for future in as_completed(future_to_test):
            test_name = future_to_test[future]
            try:
                result = future.result()
                results.append(result)
                
                # Progress indicator
                if len(results) % 50 == 0:
                    print(f"Completed {len(results)}/{len(test_names)} tests", file=sys.stderr)
                    
            except Exception as e:
                print(f"Error processing {test_name}: {e}", file=sys.stderr)
                # Add a default result for failed requests
                results.append({
                    'test': test_name,
                    'unexpected': 0,
                    'total': 0
                })
    
    # Sort results by flakiness rate (unexpected/total) descending
    results.sort(key=lambda x: x['unexpected'] / max(x['total'], 1), reverse=True)
    
    # Output CSV to stdout
    writer = csv.DictWriter(sys.stdout, fieldnames=['test', 'unexpected', 'total'])
    writer.writeheader()
    writer.writerows(results)
    
    print(f"Analysis complete. Processed {len(results)} tests.", file=sys.stderr)


if __name__ == '__main__':
    main()