analyze_ci_cases.py

import yaml
import os
from collections import defaultdict

def parse_yaml_file(file_path):
    try:
        with open(file_path, 'r') as file:
            data = yaml.safe_load(file)

        file_path = file_path.split('/')[-1].split(r".")[0]
        # Use defaultdict to store results
        result_dict = defaultdict(dict)
        # Dictionary to store test entries and GPU configuration relationships
        test_config_dict = defaultdict(list)
        
        # Iterate through all condition blocks
        for condition_block in data.get(f'{file_path}', []):
            condition = condition_block.get('condition', {})
            ranges = condition.get('ranges', {})
            wildcards = condition.get('wildcards', {})
            tests = condition_block.get('tests', [])
            terms = condition.get("terms", {})
            # backend = terms.get('backend')

            # Get GPU count range
            gpu_count = ranges.get('system_gpu_count', {})
            gte_value = gpu_count.get('gte')
            lte_value = gpu_count.get('lte')
            
            # Get GPU values
            gpu_values = wildcards.get('gpu', [])
            
            # Create key
            key = f"gte{gte_value}_lte{lte_value}_{file_path}"
            
            # Store data in first dictionary
            result_dict[key] = {
                'gpu_values': gpu_values,
                'tests': tests
            }
            
            # Store each test entry as key in second dictionary
            for test in tests:
                test_config_dict[test].append({
                    'gte_value': gte_value,
                    'lte_value': lte_value,
                    'gpu_values': gpu_values,
                    'file_path': file_path,  # Add file path information
		    # 'backend': backend
		    'terms': terms
                })
            
        return result_dict, test_config_dict
            
    except FileNotFoundError:
        print(f"Error: File not found {file_path}")
        return None, None
    except yaml.YAMLError as e:
        print(f"YAML parsing error: {e}")
        return None, None

def process_directory(directory_path):
    # Dictionary to store results from all files
    all_results = defaultdict(dict)
    all_test_configs = defaultdict(list)

    all_files = []
    # Check if directory exists
    if not os.path.isdir(directory_path):
        if os.path.isfile(directory_path):
            all_files.append(directory_path)
        else:
            print(f"Error: {directory_path} is not a valid directory")
            return all_results, all_test_configs
    else:
        all_files = os.listdir(directory_path)
        all_files = [os.path.join(directory_path, f) for f in all_files]

    # Process only files in the specified directory
    for file in all_files:
        if file.endswith('.yml'):
            #file_path = os.path.join(directory_path, file)
            file_path = file
            # print(f"\nProcessing file: {file_path}")
            
            # Process single file
            result_dict, test_config_dict = parse_yaml_file(file_path)

            if result_dict and test_config_dict:
                # Merge results
                all_results.update(result_dict)
                for key, value in test_config_dict.items():
                    all_test_configs[key].extend(value)
    
    return all_results, all_test_configs

def analyze_test_configs(test_configs):
    # Dictionary to store test configurations
    test_config_list = defaultdict(list)

    # Group configurations by test
    for test, config in test_configs.items():
        test_config_list[test].extend(config)

    # Print tests with multiple configurations
    print("\nTests with multiple configurations:")
    for test, configs in test_config_list.items():
        has_multi_gpu = False
        for config in configs:
            if config.get('gte_value', 0) > 1 and config.get('lte_value', 0) > 1:
                has_multi_gpu = True

        if len(configs) > 1 and has_multi_gpu:
            print(f"\nTest: {test}")
            print(f"Number of configurations: {len(configs)}")
            #for i, config in enumerate(configs, 1):
            #    print(f"\nConfiguration {i}:")
	    #    recorder = f"GPU count, {config['gte_value']}-{config['lte_value']}, {}"
            #    print(f"GPU count range: gte={config['gte_value']}, lte={config['lte_value']}")
            #    print(f"GPU values: {config['gpu_values']}")
            #    #print(f"backend: {config['backend']}")
            #    print(f"terms: {config['terms']}")
            #    print(f"File path: {config['file_path']}")
            case_info = []
            for i, config in enumerate(configs, 1):
                info = f"case {i}; GPU {config['gte_value']}; {config['terms'].get('backend', None)}; {config['file_path']}"
                case_info.append(info)

            print(', '.join(case_info))

    return test_config_list

# Usage example
if __name__ == "__main__":
    directory_path = "./tests/integration/test_lists/test-db/" 
    #directory_path = "./tests/integration/test_lists/test-db/l0_dgx_h200.yml"
    all_results, all_test_configs = process_directory(directory_path)
    
    if all_results and all_test_configs:
        #print("\nResults from all files:")
        #for key, value in all_results.items():
        #    print(f"\nKey: {key}")
        #    print(f"GPU values: {value['gpu_values']}")
        #    print(f"Tests: {value['tests']}")
        #    
        #print("\nTest entry configuration relationships:")
        #for test, config in all_test_configs.items():
        #    print(f"\nTest: {test}")
        #    print(f"GPU count range: gte={config['gte_value']}, lte={config['lte_value']}")
        #    print(f"GPU values: {config['gpu_values']}")
        #    print(f"File path: {config['file_path']}")

	# Analyze test configurations
        test_config_list = analyze_test_configs(all_test_configs)