Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
jerrrrry
infinicore
Commits
fe1ffc0b
Commit
fe1ffc0b
authored
Nov 07, 2025
by
wooway777
Browse files
issue/564 - updated test info, env settings, and readme
parent
39ec8f0e
Changes
7
Show whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
527 additions
and
145 deletions
+527
-145
README.md
README.md
+7
-4
scripts/set_env_linux.sh
scripts/set_env_linux.sh
+28
-0
test/infinicore/framework/__init__.py
test/infinicore/framework/__init__.py
+26
-10
test/infinicore/framework/config.py
test/infinicore/framework/config.py
+129
-33
test/infinicore/framework/devices.py
test/infinicore/framework/devices.py
+0
-1
test/infinicore/framework/utils.py
test/infinicore/framework/utils.py
+80
-0
test/infinicore/run.py
test/infinicore/run.py
+257
-97
No files found.
README.md
View file @
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
...
...
scripts/set_env_linux.sh
0 → 100644
View file @
fe1ffc0b
#!/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"
test/infinicore/framework/__init__.py
View file @
fe1ffc0b
...
...
@@ -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"
,
]
test/infinicore/framework/config.py
View file @
fe1ffc0b
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 performanc
e"
,
help
=
"
Enable performance benchmarking mod
e"
,
)
parser
.
add_argument
(
"--num_prerun"
,
type
=
lambda
x
:
max
(
0
,
int
(
x
)),
default
=
10
,
help
=
"
Set the n
umber of
pre-
runs before benchmarking
. D
efault
is
10
.
"
,
help
=
"
N
umber of
warm-up
runs before benchmarking
(d
efault
:
10
)
"
,
)
parser
.
add_argument
(
"--num_iterations"
,
type
=
lambda
x
:
max
(
0
,
int
(
x
)),
default
=
1000
,
help
=
"
Set the n
umber of iterations for benchmarking
. D
efault
is
1000
.
"
,
help
=
"
N
umber of iterations for benchmarking
(d
efault
:
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
:
try
:
import
torch.cuda
devices_to_test
.
append
(
InfiniDeviceEnum
.
NVIDIA
)
if
args
.
iluvatar
:
devices_to_test
.
append
(
InfiniDeviceEnum
.
ILUVATAR
)
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
:
try
:
import
torch
devices_to_test
.
append
(
InfiniDeviceEnum
.
METAX
)
except
ImportError
:
print
(
"Warning: Metax GPU support not available"
)
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
test/infinicore/framework/devices.py
View file @
fe1ffc0b
...
...
@@ -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"
,
...
...
test/infinicore/framework/utils.py
View file @
fe1ffc0b
...
...
@@ -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"
:
...
...
test/infinicore/run.py
View file @
fe1ffc0b
...
...
@@ -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
():
# 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
run_all_op_tests
(
ops_dir
=
None
,
verbose
=
False
,
specific_ops
=
None
,
extra_args
=
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
,
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
(
"
\n
STDERR:"
)
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
:.
1
f
}
%"
)
success_rate
=
passed
/
total
*
100
print
(
f
"Success rate:
{
success_rate
:.
1
f
}
%"
)
if
passed
==
total
:
if
not
failed_tests
:
print
(
"
\n
🎉 All tests passed!"
)
return
True
else
:
print
(
"
\n
Failed tests
:"
)
for
test_name
,
(
success
,
returncode
,
stdout
,
stderr
)
in
results
.
items
()
:
if
not
success
:
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
:
print
(
f
" Error:
{
error_lines
[
0
]
}
"
)
# 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
:
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
"
\n
Total:
{
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 argument
s
#
Core option
s
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
(
"
\n
Extra 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
not
ops_dir
.
exists
():
print
(
f
"Error: Ops directory '
{
ops_dir
}
' does not exist."
)
sys
.
exit
(
1
)
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
]
# Show what extra arguments will be passed
if
unknown_args
:
print
(
f
"Passing extra arguments to test scripts:
{
unknown_args
}
"
)
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
# Get available operators for display
available_operators
=
get_available_operators
(
ops_dir
)
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
}
"
)
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
:
print
(
f
"No operator test files found in
{
ops_dir
}
"
)
print
(
f
"Available Python files:
{
[
f
.
name
for
f
in
test_files
]
}
"
)
return
invalid_ops
.
append
(
op
)
# Show what extra arguments will be passed
if
unknown_args
:
print
(
f
"Passing extra arguments to test scripts:
{
unknown_args
}
"
)
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__"
:
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment