Unverified Commit 581eddb1 authored by PanZezhong1725's avatar PanZezhong1725 Committed by GitHub
Browse files

Merge pull request #570 from InfiniTensor/issue/564

issue/564 - updated test info, env settings, and readme
parents 39ec8f0e fe1ffc0b
......@@ -50,6 +50,13 @@ git clone --recursive https://github.com/InfiniTensor/InfiniCore.git
git submodule update --init --recursive
```
配置`INFINI_ROOT``LD_LIBRARY_PATH` 环境变量。
默认`INFINI_ROOT``$HOME/.infini`,可以使用以下命令自动配置:
```shell
source scripts/set_env_linux.sh
```
如果你需要在本地开发九齿算子(即需要对九齿算子库进行修改),推荐单独克隆[九齿算子库](https://github.com/InfiniTensor/ntops),并从本地安装:
```shell
......@@ -148,10 +155,6 @@ python scripts/install.py [XMAKE_CONFIG_FLAGS]
xmake build && xmake install
```
3. 设置环境变量
按输出提示设置 `INFINI_ROOT``LD_LIBRARY_PATH` 环境变量。
#### 2. 安装 C++ 库
```shell
......
#!/bin/bash
# Define the content to be added
content='export INFINI_ROOT="$HOME/.infini"
export LD_LIBRARY_PATH="$INFINI_ROOT/lib:$LD_LIBRARY_PATH"'
# Check if bashrc file exists
bashrc_file="$HOME/.bashrc"
if [ ! -f "$bashrc_file" ]; then
echo "Creating $bashrc_file file"
touch "$bashrc_file"
fi
# Check if the content already exists
if grep -q "export INFINI_ROOT=" "$bashrc_file"; then
echo "INFINI_ROOT configuration already exists in $bashrc_file"
else
echo "Adding configuration to $bashrc_file"
echo "$content" >> "$bashrc_file"
echo "Configuration added successfully"
fi
# Reload bashrc
echo "Reloading $bashrc_file"
source "$bashrc_file"
echo "Done! INFINI_ROOT and LD_LIBRARY_PATH have been set"
......@@ -9,33 +9,49 @@ from .utils import (
profile_operation,
rearrange_tensor,
convert_infinicore_to_torch,
get_operator_help_info,
print_operator_testing_tips,
)
from .config import (
get_args,
get_hardware_args_group,
get_hardware_help_text,
get_supported_hardware_platforms,
get_test_devices,
)
from .config import get_test_devices, get_args
from .devices import InfiniDeviceEnum, InfiniDeviceNames, torch_device_map
from .datatypes import to_torch_dtype, to_infinicore_dtype
from .runner import GenericTestRunner
__all__ = [
"TensorSpec",
# Core types and classes
"BaseOperatorTest",
"GenericTestRunner",
"InfiniDeviceEnum",
"InfiniDeviceNames",
"TensorInitializer",
"TensorSpec",
"TestCase",
"TestConfig",
"TestRunner",
"TestCase",
"BaseOperatorTest",
# Core functions
"compare_results",
"create_test_comparator",
"convert_infinicore_to_torch",
"create_test_comparator",
"debug",
"get_args",
"get_hardware_args_group",
"get_hardware_help_text",
"get_operator_help_info",
"get_supported_hardware_platforms",
"get_test_devices",
"get_tolerance",
"infinicore_tensor_from_torch",
"print_operator_testing_tips",
"profile_operation",
"rearrange_tensor",
"InfiniDeviceEnum",
"InfiniDeviceNames",
"torch_device_map",
"to_torch_dtype",
# Utility functions
"to_infinicore_dtype",
"GenericTestRunner",
"to_torch_dtype",
"torch_device_map",
]
import argparse
from .devices import InfiniDeviceEnum
# hardware_info.py
"""
Shared hardware platform information for the InfiniCore testing framework
"""
def get_supported_hardware_platforms():
"""
Get list of supported hardware platforms with descriptions.
Returns:
List of tuples (flag, description)
"""
return [
("--cpu", "Standard CPU execution"),
("--nvidia", "NVIDIA GPUs with CUDA support"),
("--cambricon", "Cambricon MLU accelerators (requires torch_mlu)"),
("--ascend", "Huawei Ascend NPUs (requires torch_npu)"),
("--iluvatar", "Iluvatar GPUs"),
("--metax", "Metax GPUs"),
("--moore", "Moore Threads GPUs (requires torch_musa)"),
("--kunlun", "Kunlun XPUs (requires torch_xmlir)"),
("--hygon", "Hygon DCUs"),
]
def get_hardware_help_text():
"""
Get formatted help text for hardware platforms.
Returns:
str: Formatted help text for argument parsers
"""
platforms = get_supported_hardware_platforms()
help_lines = ["Supported Hardware Platforms:"]
for flag, description in platforms:
# Remove leading dashes for cleaner display
name = flag.lstrip("-")
help_lines.append(f" - {name.upper():<10} {description}")
return "\n".join(help_lines)
def get_hardware_args_group(parser):
"""
Add hardware platform arguments to an argument parser.
Args:
parser: argparse.ArgumentParser instance
Returns:
The argument group for hardware platforms
"""
hardware_group = parser.add_argument_group("Hardware Platform Options")
for flag, description in get_supported_hardware_platforms():
hardware_group.add_argument(flag, action="store_true", help=description)
return hardware_group
def get_args():
"""Parse command line arguments"""
parser = argparse.ArgumentParser(description="Test Operator")
"""Parse command line arguments for operator testing"""
parser = argparse.ArgumentParser(
description="Test InfiniCore operators across multiple hardware platforms",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=f"""
Examples:
# Run all tests on CPU only
python test_operator.py --cpu
# Run with benchmarking on NVIDIA GPU
python test_operator.py --nvidia --bench
# Run with debug mode on multiple devices
python test_operator.py --cpu --nvidia --debug
# Run performance profiling with custom iterations
python test_operator.py --nvidia --bench --num_prerun 50 --num_iterations 5000
{get_hardware_help_text()}
""",
)
# Core testing options
parser.add_argument(
"--bench",
action="store_true",
help="Whether to benchmark performance",
help="Enable performance benchmarking mode",
)
parser.add_argument(
"--num_prerun",
type=lambda x: max(0, int(x)),
default=10,
help="Set the number of pre-runs before benchmarking. Default is 10.",
help="Number of warm-up runs before benchmarking (default: 10)",
)
parser.add_argument(
"--num_iterations",
type=lambda x: max(0, int(x)),
default=1000,
help="Set the number of iterations for benchmarking. Default is 1000.",
help="Number of iterations for benchmarking (default: 1000)",
)
parser.add_argument(
"--debug",
action="store_true",
help="Whether to turn on debug mode.",
help="Enable debug mode for detailed tensor comparison",
)
# Device options
device_group = parser.add_argument_group("Device options")
device_group.add_argument("--cpu", action="store_true", help="Run CPU test")
device_group.add_argument(
"--nvidia", action="store_true", help="Run NVIDIA GPU test"
)
device_group.add_argument(
"--cambricon", action="store_true", help="Run Cambricon MLU test"
)
device_group.add_argument(
"--ascend", action="store_true", help="Run ASCEND NPU test"
)
device_group.add_argument(
"--iluvatar", action="store_true", help="Run Iluvatar GPU test"
)
device_group.add_argument("--metax", action="store_true", help="Run METAX GPU test")
device_group.add_argument(
"--moore", action="store_true", help="Run MTHREADS GPU test"
)
device_group.add_argument(
"--kunlun", action="store_true", help="Run KUNLUN XPU test"
)
# Device options using shared hardware info
hardware_group = get_hardware_args_group(parser)
return parser.parse_args()
......@@ -57,15 +119,24 @@ def get_args():
def get_test_devices(args):
"""
Determine which devices to test based on command line arguments
Returns:
List[InfiniDeviceEnum]: List of devices to test
"""
devices_to_test = []
# Check each hardware platform with proper dependency validation
if args.cpu:
devices_to_test.append(InfiniDeviceEnum.CPU)
if args.nvidia:
devices_to_test.append(InfiniDeviceEnum.NVIDIA)
if args.iluvatar:
devices_to_test.append(InfiniDeviceEnum.ILUVATAR)
try:
import torch.cuda
devices_to_test.append(InfiniDeviceEnum.NVIDIA)
except ImportError:
print("Warning: CUDA not available, skipping NVIDIA tests")
if args.cambricon:
try:
import torch_mlu
......@@ -73,6 +144,7 @@ def get_test_devices(args):
devices_to_test.append(InfiniDeviceEnum.CAMBRICON)
except ImportError:
print("Warning: torch_mlu not available, skipping Cambricon tests")
if args.ascend:
try:
import torch
......@@ -82,10 +154,15 @@ def get_test_devices(args):
devices_to_test.append(InfiniDeviceEnum.ASCEND)
except ImportError:
print("Warning: torch_npu not available, skipping Ascend tests")
if args.metax:
import torch
try:
import torch
devices_to_test.append(InfiniDeviceEnum.METAX)
except ImportError:
print("Warning: Metax GPU support not available")
devices_to_test.append(InfiniDeviceEnum.METAX)
if args.moore:
try:
import torch
......@@ -94,6 +171,16 @@ def get_test_devices(args):
devices_to_test.append(InfiniDeviceEnum.MOORE)
except ImportError:
print("Warning: torch_musa not available, skipping Moore tests")
if args.iluvatar:
try:
# Iluvatar GPU detection
import torch
devices_to_test.append(InfiniDeviceEnum.ILUVATAR)
except ImportError:
print("Warning: Iluvatar GPU support not available")
if args.kunlun:
try:
import torch_xmlir
......@@ -102,8 +189,17 @@ def get_test_devices(args):
except ImportError:
print("Warning: torch_xmlir not available, skipping Kunlun tests")
if args.hygon:
try:
import torch
devices_to_test.append(InfiniDeviceEnum.HYGON)
except ImportError:
print("Warning: Hygon DCU support not available")
# Default to CPU if no devices specified
if not devices_to_test:
devices_to_test = [InfiniDeviceEnum.CPU]
print("No devices specified, defaulting to CPU")
return devices_to_test
......@@ -22,7 +22,6 @@ InfiniDeviceNames = {
InfiniDeviceEnum.HYGON: "Hygon",
}
# Mapping that maps InfiniDeviceEnum to torch device string
torch_device_map = {
InfiniDeviceEnum.CPU: "cpu",
InfiniDeviceEnum.NVIDIA: "cuda",
......
......@@ -4,6 +4,86 @@ import infinicore
from .datatypes import to_infinicore_dtype, to_torch_dtype
def get_operator_help_info():
"""
Get help information for operator testing framework
Returns:
str: Comprehensive help information about the testing framework
"""
return """
InfiniCore Operator Testing Framework
This framework provides comprehensive testing for InfiniCore operators across
multiple hardware platforms with the following features:
Key Features:
-------------
1. Multi-platform Support: CPU, NVIDIA, Cambricon, Ascend, Iluvatar, Metax,
Moore, Kunlun, and Hygon devices
2. Flexible Testing: Out-of-place and in-place operations
3. Performance Benchmarking: Accurate timing with warm-up runs
4. Debug Capabilities: Detailed tensor comparison and discrepancy analysis
5. Tolerance Control: Configurable absolute and relative tolerances per data type
Usage Patterns:
--------------
Basic testing:
python test_operator.py --cpu --nvidia
With benchmarking:
python test_operator.py --nvidia --bench --num_iterations 1000
Debug mode:
python test_operator.py --cpu --debug
Multiple devices:
python test_operator.py --cpu --nvidia
Data Type Support:
-----------------
- Floating point: float16, bfloat16, float32
- Integer: int8, int16, int32, int64, uint8
- Boolean: bool
Tensor Initialization Modes:
---------------------------
- RANDOM: Random values using torch.rand
- ZEROS: All zeros using torch.zeros
- ONES: All ones using torch.ones
- RANDINT: Random integers using torch.randint
- MANUAL: Use pre-existing tensor with shape/strides validation
- BINARY: Use pre-existing tensor with shape validation only
- FROM_FILE: Load tensor data from file
For detailed examples and advanced usage, refer to the individual operator
test files and the framework documentation.
"""
def print_operator_testing_tips():
"""Print useful tips for operator testing"""
tips = """
Operator Testing Tips:
---------------------
1. Start with CPU tests for basic functionality validation
2. Use --debug flag to identify precision issues in early development
3. Benchmark with sufficient iterations (--num_iterations) for stable results
4. Set appropriate tolerances for different data types (float16 needs higher tolerance)
5. Test both contiguous and non-contiguous tensor layouts
6. Validate in-place operations separately from out-of-place operations
7. Check edge cases: empty tensors, broadcasting, different tensor shapes
Common Tolerance Settings:
-------------------------
- float32: atol=1e-5, rtol=1e-3
- float16: atol=1e-3, rtol=1e-2
- bfloat16: atol=1e-2, rtol=1e-1
- Integer types: exact equality (atol=0, rtol=0)
"""
print(tips)
def synchronize_device(torch_device):
"""Device synchronization"""
if torch_device == "cuda":
......
......@@ -3,82 +3,126 @@ import sys
import subprocess
import argparse
from pathlib import Path
from typing import Dict, Tuple, List
def find_ops_directory(start_dir=None):
"""
Find the ops directory by searching from start_dir upwards.
Args:
start_dir: Starting directory for search (default: current file's parent)
Returns:
Path: Path to ops directory or None if not found
"""
if start_dir is None:
start_dir = Path(__file__).parent
ops_dir = start_dir / "ops"
if ops_dir.exists() and (ops_dir / "rms_norm.py").exists():
return ops_dir
# Look for ops directory in common locations
possible_locations = [
start_dir / "ops",
start_dir / ".." / "ops",
start_dir / ".." / "test" / "ops",
start_dir / "test" / "ops",
]
for location in possible_locations:
ops_dir = location.resolve()
if ops_dir.exists() and any(ops_dir.glob("*.py")):
return ops_dir
return None
def get_available_operators(ops_dir):
"""
Get list of available operators from ops directory.
Args:
ops_dir: Path to ops directory
Returns:
List of operator names
"""
if not ops_dir or not ops_dir.exists():
return []
test_files = list(ops_dir.glob("*.py"))
current_script = Path(__file__).name
test_files = [f for f in test_files if f.name != current_script]
operators = []
for test_file in test_files:
try:
with open(test_file, "r", encoding="utf-8") as f:
content = f.read()
if "infinicore" in content and (
"BaseOperatorTest" in content or "GenericTestRunner" in content
):
operators.append(test_file.stem)
except:
continue
return sorted(operators)
def run_all_op_tests(ops_dir=None, verbose=False, specific_ops=None, extra_args=None):
def run_all_op_tests(ops_dir=None, specific_ops=None, extra_args=None):
"""
Run all operator test scripts in the ops directory.
Args:
ops_dir (str, optional): Path to the ops directory. If None, uses the current directory.
verbose (bool): Whether to print detailed output.
specific_ops (list, optional): List of specific operator names to test (e.g., ['add', 'matmul']).
ops_dir (str, optional): Path to the ops directory. If None, uses auto-detection.
specific_ops (list, optional): List of specific operator names to test.
extra_args (list, optional): Extra command line arguments to pass to test scripts.
Returns:
dict: Results dictionary with test names as keys and (success, return_code, output) as values.
dict: Results dictionary with test names as keys and (success, return_code, stdout, stderr) as values.
"""
if ops_dir is None:
ops_dir = find_ops_directory()
else:
ops_dir = Path(ops_dir)
if not ops_dir.exists():
if not ops_dir or not ops_dir.exists():
print(f"Error: Ops directory '{ops_dir}' does not exist.")
return {}
print(f"Looking for test files in: {ops_dir}")
# Find all Python test files (looking for actual operator test files)
# Find all Python test files
test_files = list(ops_dir.glob("*.py"))
# Filter out this script itself and non-operator test files
current_script = Path(__file__).name
test_files = [f for f in test_files if f.name != current_script]
# Further filter to include only files that look like operator tests
# (they typically import infinicore and BaseOperatorTest)
# Filter to include only files that look like operator tests
operator_test_files = []
for test_file in test_files:
try:
with open(test_file, "r", encoding="utf-8") as f:
content = f.read()
if "infinicore" in content and "BaseOperatorTest" in content:
# Look for characteristic patterns of operator tests
if "infinicore" in content and (
"BaseOperatorTest" in content or "GenericTestRunner" in content
):
operator_test_files.append(test_file)
elif verbose:
print(f" Skipping {test_file.name}: not an operator test file")
except Exception as e:
if verbose:
print(f" Could not read {test_file.name}: {e}")
continue
# Filter for specific operators if requested
if specific_ops:
# Filter for specific operators (case insensitive)
filtered_files = []
for test_file in operator_test_files:
test_name = test_file.stem.lower()
if any(op.lower() in test_name for op in specific_ops):
filtered_files.append(test_file)
elif verbose:
print(f" Filtered out {test_file.name}: not in specific_ops list")
operator_test_files = filtered_files
if not operator_test_files:
print(f"No operator test files found in {ops_dir}")
print(f"Available Python files: {[f.name for f in test_files]}")
print(f"Current directory: {Path.cwd()}")
return {}
print(f"Found {len(operator_test_files)} operator test files:")
......@@ -98,12 +142,14 @@ def run_all_op_tests(ops_dir=None, verbose=False, specific_ops=None, extra_args=
if extra_args:
cmd.extend(extra_args)
if verbose:
print(f"Command: {' '.join(cmd)}")
print(f"Working directory: {ops_dir}")
# Always capture output to display it
result = subprocess.run(cmd, cwd=ops_dir, capture_output=True, text=True)
# Run with captured output
result = subprocess.run(
cmd,
cwd=ops_dir,
capture_output=True,
text=True,
timeout=300, # 5 minute timeout per test
)
success = result.returncode == 0
results[test_name] = (
......@@ -114,75 +160,180 @@ def run_all_op_tests(ops_dir=None, verbose=False, specific_ops=None, extra_args=
)
# Print the output from the test script
print(f"\n{'='*60}")
print(f"TEST: {test_name}")
print(f"{'='*60}")
if result.stdout:
print(result.stdout)
print(result.stdout.rstrip())
if result.stderr:
print("STDERR:")
print(result.stderr)
print("\nSTDERR:")
print(result.stderr.rstrip())
if success:
print(f"✅ {test_name}: PASSED (return code: {result.returncode})")
else:
print(f"❌ {test_name}: FAILED (return code: {result.returncode})")
status_icon = "✅" if success else "❌"
print(
f"\n{status_icon} {test_name}: {'PASSED' if success else 'FAILED'} (return code: {result.returncode})"
)
except subprocess.TimeoutExpired:
print(f"⏰ {test_name}: TIMEOUT (exceeded 5 minutes)")
results[test_name] = (False, -2, "", "Test execution timed out")
except Exception as e:
print(f" {test_name}: ERROR - {str(e)}")
print(f"💥 {test_name}: ERROR - {str(e)}")
results[test_name] = (False, -1, "", str(e))
return results
def print_summary(results):
"""Print a summary of test results."""
"""Print a comprehensive summary of test results."""
print(f"\n{'='*80}")
print("TEST SUMMARY")
print(f"{'='*80}")
if not results:
print("No tests were run.")
return
return False
passed = sum(1 for success, _, _, _ in results.values() if success)
total = len(results)
failed_tests = [name for name, (success, _, _, _) in results.items() if not success]
print(f"Total tests: {total}")
print(f"Passed: {passed}")
print(f"Failed: {total - passed}")
if total > 0:
print(f"Success rate: {passed/total*100:.1f}%")
success_rate = passed / total * 100
print(f"Success rate: {success_rate:.1f}%")
if passed == total:
if not failed_tests:
print("\n🎉 All tests passed!")
return True
else:
print(f"\n{len(failed_tests)} tests failed:")
for test_name in failed_tests:
success, returncode, stdout, stderr = results[test_name]
print(f" - {test_name} (return code: {returncode})")
# Print brief error info for failed tests
if stderr:
error_lines = stderr.strip().split("\n")
if error_lines:
# Take first meaningful error line
for line in error_lines:
if line.strip() and not line.startswith("Warning:"):
print(f" Error: {line.strip()}")
break
return False
def list_available_tests(ops_dir=None):
"""List all available operator test files."""
if ops_dir is None:
ops_dir = find_ops_directory()
else:
print("\nFailed tests:")
for test_name, (success, returncode, stdout, stderr) in results.items():
if not success:
print(f" - {test_name} (return code: {returncode})")
# Print brief error info for failed tests
if stderr:
error_lines = stderr.strip().split("\n")
if error_lines:
print(f" Error: {error_lines[0]}")
ops_dir = Path(ops_dir)
if not ops_dir or not ops_dir.exists():
print(f"Error: Ops directory '{ops_dir}' does not exist.")
return
operators = get_available_operators(ops_dir)
if operators:
print(f"Available operator test files in {ops_dir}:")
for operator in operators:
print(f" - {operator}")
print(f"\nTotal: {len(operators)} operators")
else:
print(f"No operator test files found in {ops_dir}")
# Show available Python files for debugging
test_files = list(ops_dir.glob("*.py"))
current_script = Path(__file__).name
test_files = [f for f in test_files if f.name != current_script]
if test_files:
print(f"Available Python files: {[f.name for f in test_files]}")
def generate_help_epilog(ops_dir):
"""
Generate dynamic help epilog with available operators and hardware platforms.
Args:
ops_dir: Path to ops directory
Returns:
str: Formatted help text
"""
# Get available operators
operators = get_available_operators(ops_dir)
# Build epilog text
epilog_parts = []
# Examples section
epilog_parts.append("Examples:")
epilog_parts.append(" # Run all operator tests on CPU")
epilog_parts.append(" python run.py --cpu")
epilog_parts.append("")
epilog_parts.append(" # Run specific operators with benchmarking")
epilog_parts.append(" python run.py --ops add matmul --nvidia --bench")
epilog_parts.append("")
epilog_parts.append(" # Run with debug mode on multiple devices")
epilog_parts.append(" python run.py --cpu --nvidia --debug")
epilog_parts.append("")
epilog_parts.append(" # List available tests without running")
epilog_parts.append(" python run.py --list")
epilog_parts.append("")
epilog_parts.append(" # Run with custom performance settings")
epilog_parts.append(
" python run.py --nvidia --bench --num_prerun 50 --num_iterations 5000"
)
epilog_parts.append("")
# Available operators section
if operators:
epilog_parts.append("Available Operators:")
# Group operators for better display
operators_per_line = 4
for i in range(0, len(operators), operators_per_line):
line_ops = operators[i : i + operators_per_line]
epilog_parts.append(f" {', '.join(line_ops)}")
epilog_parts.append("")
else:
epilog_parts.append("Available Operators: (none detected)")
epilog_parts.append("")
# Additional notes
epilog_parts.append("Note:")
epilog_parts.append(
" - Use '--' to pass additional arguments to individual test scripts"
)
epilog_parts.append(
" - Operators are automatically discovered from the ops directory"
)
return "\n".join(epilog_parts)
def main():
"""Main entry point with command line argument parsing."""
"""Main entry point with comprehensive command line argument parsing."""
# First, find ops directory for dynamic help generation
ops_dir = find_ops_directory()
parser = argparse.ArgumentParser(
description="Run all operator tests in the ops directory", add_help=False
description="Run InfiniCore operator tests across multiple hardware platforms",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=generate_help_epilog(ops_dir),
)
# Our script's specific arguments
# Core options
parser.add_argument(
"--ops-dir", type=str, help="Path to the ops directory (default: auto-detect)"
)
parser.add_argument(
"-v",
"--verbose",
action="store_true",
help="Print detailed command information for each test",
)
parser.add_argument(
"--ops", nargs="+", help="Run specific operators only (e.g., --ops add matmul)"
)
......@@ -191,69 +342,78 @@ def main():
action="store_true",
help="List all available test files without running them",
)
parser.add_argument(
"-h", "--help", action="store_true", help="Show this help message and exit"
)
# Hardware platform options using shared function
from framework import get_hardware_args_group
hardware_group = get_hardware_args_group(parser)
# Parse known args first, leave the rest for the test scripts
args, unknown_args = parser.parse_known_args()
if args.help:
parser.print_help()
print("\nExtra arguments that will be passed to test scripts:")
print(" --nvidia, --cpu, --bench, --debug, etc.")
# Handle list command
if args.list:
list_available_tests(args.ops_dir)
return
# Auto-detect ops directory if not provided
if args.ops_dir is None:
ops_dir = find_ops_directory()
if not ops_dir:
print(
"Error: Could not auto-detect ops directory. Please specify with --ops-dir"
)
sys.exit(1)
else:
ops_dir = Path(args.ops_dir)
if args.list:
# Just list available test files
test_files = list(ops_dir.glob("*.py"))
current_script = Path(__file__).name
test_files = [f for f in test_files if f.name != current_script]
operator_test_files = []
for test_file in test_files:
try:
with open(test_file, "r", encoding="utf-8") as f:
content = f.read()
if "infinicore" in content and "BaseOperatorTest" in content:
operator_test_files.append(test_file)
except:
continue
if operator_test_files:
print(f"Available operator test files in {ops_dir}:")
for test_file in operator_test_files:
print(f" - {test_file.name}")
else:
print(f"No operator test files found in {ops_dir}")
print(f"Available Python files: {[f.name for f in test_files]}")
return
if not ops_dir.exists():
print(f"Error: Ops directory '{ops_dir}' does not exist.")
sys.exit(1)
# Show what extra arguments will be passed
if unknown_args:
print(f"Passing extra arguments to test scripts: {unknown_args}")
# Get available operators for display
available_operators = get_available_operators(ops_dir)
print(f"InfiniCore Operator Test Runner")
print(f"Operating directory: {ops_dir}")
print(f"Available operators: {len(available_operators)}")
if args.ops:
# Validate requested operators
valid_ops = []
invalid_ops = []
for op in args.ops:
if op in available_operators:
valid_ops.append(op)
else:
invalid_ops.append(op)
if invalid_ops:
print(f"Warning: Unknown operators: {', '.join(invalid_ops)}")
print(f"Available operators: {', '.join(available_operators)}")
if valid_ops:
print(f"Testing operators: {', '.join(valid_ops)}")
else:
print("No valid operators specified. Running all available tests.")
else:
print("Testing all available operators")
print()
# Run all tests
results = run_all_op_tests(
ops_dir=ops_dir,
verbose=args.verbose,
specific_ops=args.ops,
extra_args=unknown_args,
)
print_summary(results)
# Exit with appropriate code
if results and all(success for success, _, _, _ in results.values()):
sys.exit(0)
else:
sys.exit(1)
# Print summary and exit with appropriate code
all_passed = print_summary(results)
sys.exit(0 if all_passed else 1)
if __name__ == "__main__":
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment