vedithal-amd
769d3dd67a
* Data imputation strategy for iteration multiplexing * Implement data imputation methodology to handle missing counter values in case of iteration multiplexing * Enable dispatch filtering with iteration multiplexing since we are no longer merging dispatches * Bugfix to prevent check for missing counter values when using csv format when profiling with iteration multiplexing * Move warning and info message in case of iteration multiplexing to sanitize function which comes earlier in analyze mode * Address review comments * Fix typo in documentation * Move profiling config init. after path check in sanitize() * Graceful handling of dispatches with all counters empty within data imputation logic * Improve info message for iteration multiplexing based analysis * Ensure proper error message when trying to run iteration multiplexing with attach/detach * fix test case
1691 righe
49 KiB
Python
1691 righe
49 KiB
Python
##############################################################################
|
|
# MIT License
|
|
#
|
|
# Copyright (c) 2025 Advanced Micro Devices, Inc. All Rights Reserved.
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
# of this software and associated documentation files (the "Software"), to deal
|
|
# in the Software without restriction, including without limitation the rights
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
# furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
# THE SOFTWARE.
|
|
|
|
##############################################################################
|
|
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
import test_utils
|
|
|
|
config = {}
|
|
config["cleanup"] = True if "PYTEST_XDIST_WORKER_COUNT" in os.environ else False
|
|
|
|
indirs = [
|
|
"tests/workloads/vcopy/MI100",
|
|
"tests/workloads/vcopy/MI200",
|
|
"tests/workloads/vcopy/MI300A_A1",
|
|
"tests/workloads/vcopy/MI300X_A1",
|
|
"tests/workloads/vcopy/MI350",
|
|
]
|
|
|
|
time_units = {"s": 10**9, "ms": 10**6, "us": 10**3, "ns": 1}
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_valid_path(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_list_kernels(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--list-stats",
|
|
])
|
|
assert code == 0
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.list_metrics
|
|
def test_list_metrics_gfx90a(binary_handler_analyze_rocprof_compute):
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--list-metrics",
|
|
"gfx90a",
|
|
])
|
|
assert code == 1
|
|
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--list-metrics",
|
|
"gfx90a",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.list_metrics
|
|
def test_list_metrics_gfx908(binary_handler_analyze_rocprof_compute):
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--list-metrics",
|
|
"gfx908",
|
|
])
|
|
assert code == 1
|
|
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--list-metrics",
|
|
"gfx908",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.list_metrics
|
|
def test_list_metrics_gfx908_with_block(binary_handler_analyze_rocprof_compute):
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--list-metrics",
|
|
"gfx908",
|
|
"--block",
|
|
"1",
|
|
])
|
|
assert code == 1
|
|
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--list-metrics",
|
|
"gfx908",
|
|
"--block",
|
|
"1",
|
|
])
|
|
assert code == 1
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.list_metrics
|
|
def test_list_available_metrics(binary_handler_analyze_rocprof_compute, capsys):
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--list-available-metrics",
|
|
])
|
|
assert code == 1
|
|
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--list-available-metrics",
|
|
])
|
|
assert code == 0
|
|
|
|
# Test output
|
|
output = capsys.readouterr().out
|
|
assert "0. Top Stats" in output
|
|
assert "1. System Info" in output
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.list_metrics
|
|
def test_list_available_metrics_with_block(
|
|
binary_handler_analyze_rocprof_compute, capsys
|
|
):
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--list-available-metrics",
|
|
"--block",
|
|
"1",
|
|
])
|
|
assert code == 1
|
|
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--list-available-metrics",
|
|
"--block",
|
|
"1",
|
|
])
|
|
assert code == 1
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.filter_block
|
|
def test_filter_block_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"1",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.filter_block
|
|
def test_filter_block_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"5",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.filter_block
|
|
def test_filter_block_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"5.2.2",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.filter_block
|
|
def test_filter_block_4(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"6.1",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.filter_block
|
|
def test_filter_block_5(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"10",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.filter_block
|
|
def test_filter_block_6(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"100",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_filter_kernel_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel",
|
|
"0",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_filter_kernel_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel",
|
|
"1",
|
|
])
|
|
assert code == 1
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_filter_kernel_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel",
|
|
"0",
|
|
"1",
|
|
])
|
|
assert code == 1
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_dispatch_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--dispatch",
|
|
"0",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_dispatch_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--dispatch",
|
|
"1",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_dispatch_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--dispatch",
|
|
"2",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_dispatch_4(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--dispatch",
|
|
"1",
|
|
"4",
|
|
])
|
|
assert code == 1
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.serial
|
|
def test_dispatch_5(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--dispatch",
|
|
"5",
|
|
"6",
|
|
])
|
|
assert code == 1
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_gpu_ids(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
if dir == "tests/workloads/vcopy/MI350":
|
|
gpu_id = "0"
|
|
else:
|
|
gpu_id = "2"
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--gpu-id",
|
|
gpu_id,
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.normal_unit
|
|
def test_normal_unit_per_wave(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--normal-unit",
|
|
"per_wave",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.normal_unit
|
|
def test_normal_unit_per_cycle(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--normal-unit",
|
|
"per_cycle",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.normal_unit
|
|
def test_normal_unit_per_second(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--normal-unit",
|
|
"per_second",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.normal_unit
|
|
def test_normal_unit_per_kernel(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--normal-unit",
|
|
"per_kernel",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.max_stat
|
|
def test_max_stat_num_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--max-stat-num",
|
|
"0",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.max_stat
|
|
def test_max_stat_num_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--max-stat-num",
|
|
"5",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.max_stat
|
|
def test_max_stat_num_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--max-stat-num",
|
|
"10",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.max_stat
|
|
def test_max_stat_num_4(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--max-stat-num",
|
|
"15",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.time_unit
|
|
def test_time_unit_s(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--time-unit",
|
|
"s",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.time_unit
|
|
def test_time_unit_ms(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--time-unit",
|
|
"ms",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.time_unit
|
|
def test_time_unit_us(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--time-unit",
|
|
"us",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.time_unit
|
|
def test_time_unit_ns(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--time-unit",
|
|
"ns",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.decimal
|
|
def test_decimal_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--decimal",
|
|
"0",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.decimal
|
|
def test_decimal_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--decimal",
|
|
"1",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.decimal
|
|
def test_decimal_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--decimal",
|
|
"4",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_save_dfs(binary_handler_analyze_rocprof_compute):
|
|
output_path = test_utils.get_output_dir()
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--output-format",
|
|
"csv",
|
|
"--output-name",
|
|
output_path,
|
|
])
|
|
assert code == 0
|
|
|
|
files_in_workload = os.listdir(output_path)
|
|
for file_name in files_in_workload:
|
|
df = pd.read_csv(output_path + "/" + file_name)
|
|
assert len(df.index) >= 1
|
|
|
|
shutil.rmtree(output_path)
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
test_utils.clean_output_dir(config["cleanup"], output_path)
|
|
|
|
|
|
@pytest.mark.col
|
|
def test_col_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--cols",
|
|
"0",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.col
|
|
def test_col_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--cols",
|
|
"2",
|
|
"--include-cols",
|
|
"Description",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.col
|
|
def test_col_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--cols",
|
|
"0",
|
|
"2",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_g(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"-g",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_0(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"0",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_1(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"1",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_2(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"2",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_3(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"3",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_4(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"4",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_5(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"5",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.kernel_verbose
|
|
def test_kernel_verbose_6(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--kernel-verbose",
|
|
"6",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_baseline(binary_handler_analyze_rocprof_compute):
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
"tests/workloads/vcopy/MI200",
|
|
"--path",
|
|
"tests/workloads/vcopy/MI100",
|
|
])
|
|
assert code == 0
|
|
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
"tests/workloads/vcopy/MI200",
|
|
"--path",
|
|
"tests/workloads/vcopy/MI200",
|
|
])
|
|
assert code == 1
|
|
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
"tests/workloads/vcopy/MI100",
|
|
"--path",
|
|
"tests/workloads/vcopy/MI100",
|
|
])
|
|
assert code == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Test cases for Parser.py
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_dependency_MI100(binary_handler_analyze_rocprof_compute):
|
|
for dir in indirs:
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--dependency",
|
|
])
|
|
assert code == 0
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_parser_utility_functions():
|
|
"""Test parser utility functions edge cases"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from utils.parser import (
|
|
to_concat,
|
|
to_int,
|
|
to_max,
|
|
to_median,
|
|
to_min,
|
|
to_mod,
|
|
to_quantile,
|
|
to_round,
|
|
to_std,
|
|
)
|
|
|
|
try:
|
|
result = to_min(None, None)
|
|
assert np.isnan(result), "to_min with all None should return nan"
|
|
except TypeError:
|
|
pass
|
|
|
|
try:
|
|
result = to_min(None, 5)
|
|
assert False, "Should have crashed"
|
|
except TypeError:
|
|
pass
|
|
|
|
result = to_min(7, 3, 9, 1)
|
|
assert result == 1, "to_min should return minimum value"
|
|
|
|
try:
|
|
result = to_max(None, None)
|
|
assert np.isnan(result), "to_max with all None should return nan"
|
|
except TypeError:
|
|
pass
|
|
|
|
try:
|
|
result = to_max(None, 5)
|
|
assert False, "Should have crashed"
|
|
except TypeError:
|
|
pass
|
|
|
|
result = to_max(7, 3, 9, 1)
|
|
assert result == 9, "to_max should return maximum value"
|
|
|
|
result = to_median(None)
|
|
assert np.isnan(result), "to_median should return np.nan for None input"
|
|
|
|
try:
|
|
to_median("invalid_string")
|
|
assert False, "to_median should raise exception for invalid type"
|
|
except Exception as e:
|
|
assert "unsupported type" in str(e)
|
|
|
|
try:
|
|
to_std("invalid_string")
|
|
assert False, "to_std should raise exception for invalid type"
|
|
except Exception as e:
|
|
assert "unsupported type" in str(e)
|
|
|
|
result = to_int(None)
|
|
assert np.isnan(result), "to_int should return np.nan for None input"
|
|
|
|
try:
|
|
to_int(["list", "not", "supported"])
|
|
assert False, "to_int should raise exception for invalid type"
|
|
except Exception as e:
|
|
assert "unsupported type" in str(e)
|
|
|
|
result = to_quantile(None, 0.5)
|
|
assert np.isnan(result), "to_quantile should return np.nan for None input"
|
|
|
|
try:
|
|
to_quantile("invalid_string", 0.5)
|
|
assert False, "to_quantile should raise exception for invalid type"
|
|
except Exception as e:
|
|
assert "unsupported type" in str(e)
|
|
|
|
result = to_concat("hello", "world")
|
|
assert result == "helloworld", "to_concat should concatenate strings"
|
|
|
|
result = to_concat(123, 456)
|
|
assert result == "123456", "to_concat should convert to strings and concatenate"
|
|
|
|
series = pd.Series([1.234, 2.567, 3.890])
|
|
result = to_round(series, 2)
|
|
expected = pd.Series([1.23, 2.57, 3.89])
|
|
pd.testing.assert_series_equal(result, expected)
|
|
|
|
result = to_round(3.14159, 2)
|
|
assert result == 3.14, "to_round should round scalar values"
|
|
|
|
series = pd.Series([10, 15, 20])
|
|
result = to_mod(series, 3)
|
|
expected = pd.Series([1, 0, 2])
|
|
pd.testing.assert_series_equal(result, expected)
|
|
|
|
result = to_mod(10, 3)
|
|
assert result == 1, "to_mod should return modulo for scalars"
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_parser_error_handling():
|
|
"""Test parser error handling paths"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from utils.parser import (
|
|
build_eval_string,
|
|
calc_builtin_var,
|
|
update_denominator_string,
|
|
)
|
|
|
|
try:
|
|
build_eval_string("AVG(SQ_WAVES)", None, config={})
|
|
assert False, "Should have raised exception for None coll_level"
|
|
except Exception as e:
|
|
assert "coll_level can not be None" in str(e)
|
|
|
|
assert build_eval_string("", "pmc_perf", config={}) == ""
|
|
assert update_denominator_string("", "per_wave") == ""
|
|
|
|
class MockSysInfo:
|
|
total_l2_chan = 32
|
|
|
|
sys_info = MockSysInfo()
|
|
try:
|
|
calc_builtin_var("$unsupported_var", sys_info)
|
|
assert False, "Should have raised exception for unsupported var"
|
|
except SystemExit:
|
|
pass
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_missing_file_handling(binary_handler_analyze_rocprof_compute):
|
|
"""Test handling of missing files"""
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
code = binary_handler_analyze_rocprof_compute(["analyze", "--path", temp_dir])
|
|
assert code != 0
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_ast_transformer_edge_cases():
|
|
"""Simplified test focusing on the actual code paths"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import ast
|
|
|
|
from utils.parser import CodeTransformer
|
|
|
|
transformer = CodeTransformer()
|
|
|
|
unknown_call = ast.Call(
|
|
func=ast.Name(id="UNKNOWN_FUNCTION", ctx=ast.Load()),
|
|
args=[ast.Constant(value=5) if hasattr(ast, "Constant") else ast.Num(n=5)],
|
|
keywords=[],
|
|
)
|
|
|
|
try:
|
|
result = transformer.visit_Call(unknown_call)
|
|
if hasattr(result.func, "id") and result.func.id == "UNKNOWN_FUNCTION":
|
|
assert False, "Function name should have been changed or exception raised"
|
|
except Exception as e:
|
|
assert "Unknown call" in str(e), (
|
|
f"Expected 'Unknown call' in error, got: {str(e)}"
|
|
)
|
|
|
|
SUPPORTED_CALL = ast.Call(
|
|
func=ast.Name(id="MIN", ctx=ast.Load()),
|
|
args=[ast.Constant(value=5) if hasattr(ast, "Constant") else ast.Num(n=5)],
|
|
keywords=[],
|
|
)
|
|
|
|
try:
|
|
result = transformer.visit_Call(SUPPORTED_CALL)
|
|
assert result.func.id == "to_min", f"Expected 'to_min', got: {result.func.id}"
|
|
except Exception as e:
|
|
assert False, f"Supported function call should not raise exception: {e}"
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_analyze_with_debug_mode(binary_handler_analyze_rocprof_compute):
|
|
"""Test analyze to cover debug paths in eval_metric - using direct function call"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import pandas as pd
|
|
|
|
from utils.parser import eval_metric
|
|
|
|
mock_dfs = {
|
|
1: pd.DataFrame({
|
|
"Metric_ID": ["1.1.0"],
|
|
"Metric": ["Test Metric"],
|
|
"Expr": ["AVG(SQ_WAVES)"],
|
|
"coll_level": ["pmc_perf"],
|
|
}).set_index("Metric_ID")
|
|
}
|
|
|
|
mock_dfs_type = {1: "metric_table"}
|
|
|
|
class MockSysInfo:
|
|
ip_blocks = "standard"
|
|
se_per_gpu = 4
|
|
pipes_per_gpu = 4
|
|
cu_per_gpu = 64
|
|
simd_per_cu = 4
|
|
sqc_per_gpu = 16
|
|
lds_banks_per_cu = 32
|
|
cur_sclk = 1800.0
|
|
cur_mclk = 1200.0
|
|
max_sclk = 2100.0
|
|
max_mclk = 1600.0
|
|
max_waves_per_cu = 40
|
|
num_hbm_channels = 4
|
|
total_l2_chan = 32
|
|
num_xcd = 1
|
|
wave_size = 64
|
|
|
|
sys_info = MockSysInfo()
|
|
|
|
raw_pmc_df = {
|
|
"pmc_perf": pd.DataFrame({
|
|
"SQ_WAVES": [100, 200, 150],
|
|
"GRBM_GUI_ACTIVE": [1000, 2000, 1500],
|
|
"End_Timestamp": [1000000, 2000000, 1500000],
|
|
"Start_Timestamp": [0, 1000000, 500000],
|
|
})
|
|
}
|
|
|
|
try:
|
|
eval_metric(
|
|
mock_dfs, mock_dfs_type, sys_info, raw_pmc_df, debug=True, config={}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_filter_combinations_coverage(binary_handler_analyze_rocprof_compute):
|
|
"""Test basic filters that should work"""
|
|
for dir in ["tests/workloads/vcopy/MI100", "tests/workloads/vcopy/MI200"]:
|
|
if os.path.exists(dir):
|
|
workload_dir = test_utils.setup_workload_dir(dir)
|
|
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
])
|
|
assert code == 0
|
|
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
"--block",
|
|
"SQ",
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|
|
break
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_apply_filters_direct():
|
|
"""Test apply_filters function directly to cover filter branches"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import pandas as pd
|
|
|
|
from utils.parser import apply_filters
|
|
|
|
class MockWorkload:
|
|
def __init__(self):
|
|
self.raw_pmc = pd.DataFrame({
|
|
("pmc_perf", "GPU_ID"): [0, 0, 1, 1],
|
|
("pmc_perf", "Kernel_Name"): [
|
|
"vecCopy",
|
|
"vecAdd",
|
|
"vecCopy",
|
|
"vecMul",
|
|
],
|
|
("pmc_perf", "Dispatch_ID"): [0, 1, 2, 3],
|
|
("pmc_perf", "Node"): ["node0", "node0", "node1", "node1"],
|
|
})
|
|
self.raw_pmc.columns = pd.MultiIndex.from_tuples(self.raw_pmc.columns)
|
|
|
|
filter_nodes = None
|
|
filter_gpu_ids = None
|
|
filter_kernel_ids = None
|
|
filter_dispatch_ids = None
|
|
|
|
workload = MockWorkload()
|
|
|
|
workload.filter_gpu_ids = "0"
|
|
result = apply_filters(workload, "/tmp", False, False)
|
|
assert len(result) == 2
|
|
|
|
workload.filter_gpu_ids = None
|
|
workload.filter_kernel_ids = ["vecCopy"]
|
|
result = apply_filters(workload, "/tmp", False, False)
|
|
assert len(result) == 2
|
|
|
|
workload.filter_kernel_ids = None
|
|
workload.filter_dispatch_ids = ["0", "1"]
|
|
result = apply_filters(workload, "/tmp", False, False)
|
|
assert len(result) == 2
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_missing_files_scenarios(binary_handler_analyze_rocprof_compute):
|
|
"""Test scenarios with missing files to cover error paths"""
|
|
import shutil
|
|
import tempfile
|
|
|
|
for dir in ["tests/workloads/vcopy/MI100", "tests/workloads/vcopy/MI200"]:
|
|
if os.path.exists(dir):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
workload_dir = os.path.join(temp_dir, "incomplete_workload")
|
|
shutil.copytree(dir, workload_dir)
|
|
|
|
csv_files = ["pmc_perf_1.csv", "pmc_perf_2.csv", "timestamps.csv"]
|
|
for csv_file in csv_files:
|
|
csv_path = os.path.join(workload_dir, csv_file)
|
|
if os.path.exists(csv_path):
|
|
os.remove(csv_path)
|
|
|
|
binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
])
|
|
break
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_pc_sampling_basic_coverage():
|
|
"""Test PC sampling functions with minimal data"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import tempfile
|
|
|
|
from utils.parser import load_pc_sampling_data, search_pc_sampling_record
|
|
|
|
class MockWorkload:
|
|
filter_kernel_ids = []
|
|
|
|
workload = MockWorkload()
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
result = load_pc_sampling_data(workload, temp_dir, "none", "count")
|
|
assert result.empty
|
|
|
|
result = load_pc_sampling_data(workload, temp_dir, "missing", "count")
|
|
assert result.empty
|
|
|
|
workload.filter_kernel_ids = [0, 1, 2] # Multiple kernels
|
|
result = load_pc_sampling_data(workload, temp_dir, "test", "count")
|
|
assert result.empty
|
|
|
|
result = search_pc_sampling_record([])
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_build_dfs_edge_cases():
|
|
"""Test build_dfs and gen_counter_list with various configurations"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from utils.parser import gen_counter_list
|
|
|
|
visited, counters = gen_counter_list(None)
|
|
assert not visited
|
|
assert counters == []
|
|
|
|
visited, counters = gen_counter_list(123)
|
|
assert not visited
|
|
assert counters == []
|
|
|
|
visited, counters = gen_counter_list("AVG(SQ_WAVES + TCC_HIT)")
|
|
assert visited
|
|
assert "SQ_WAVES" in counters
|
|
assert "TCC_HIT" in counters
|
|
|
|
visited, counters = gen_counter_list("Start_Timestamp + End_Timestamp")
|
|
assert visited
|
|
|
|
visited, counters = gen_counter_list("INVALID SYNTAX !!!")
|
|
assert not visited
|
|
|
|
|
|
@pytest.mark.misc
|
|
def test_update_functions_coverage():
|
|
"""Test update_denominator_string and update_norm_unit_string branches"""
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from utils.parser import update_denominator_string, update_normal_unit_string
|
|
|
|
result = update_denominator_string("AVG(SQ_WAVES / $denom)", "per_wave")
|
|
assert "$denom" not in result
|
|
assert "SQ_WAVES" in result
|
|
|
|
result = update_denominator_string("AVG(DATA / $denom)", "per_cycle")
|
|
assert "$GRBM_GUI_ACTIVE_PER_XCD" in result
|
|
|
|
result = update_denominator_string("AVG(DATA / $denom)", "per_second")
|
|
assert "End_Timestamp - Start_Timestamp" in result
|
|
|
|
result = update_denominator_string("AVG(DATA / $denom)", "unsupported_unit")
|
|
assert "$denom" in result
|
|
|
|
result = update_normal_unit_string("(Prefix + $normUnit)", "per_wave")
|
|
assert "per wave" in result.lower()
|
|
assert result[0].isupper()
|
|
|
|
|
|
def test_metric_evaluation_no_valid_data():
|
|
"""Test emetric evaluation with no valid data"""
|
|
import numpy as np
|
|
|
|
from utils.parser import MetricEvaluator
|
|
|
|
metric_evaluator = MetricEvaluator({}, {}, {})
|
|
with patch("builtins.eval") as mock_eval, patch("builtins.compile"):
|
|
# Test when eval returns None
|
|
mock_eval.return_value = None
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
# Test when eval returns NaN
|
|
mock_eval.return_value = np.nan
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
# Test when eval raises an exception
|
|
mock_eval.side_effect = TypeError("Mock exception")
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
mock_eval.side_effect = NameError("empirical_peak")
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
mock_eval.side_effect = KeyError("Some KeyError")
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
with patch("sys.exit"):
|
|
mock_eval.side_effect = AttributeError("Some AttributeError")
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
mock_eval.side_effect = AttributeError(
|
|
"'NoneType' object has no attribute 'get'"
|
|
)
|
|
assert metric_evaluator.eval_expression("Mock Metric") == "N/A"
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_time_data():
|
|
return pd.DataFrame({
|
|
"Metric_ID": ["7.2.0", "7.2.1", "7.2.2"],
|
|
"Metric": [
|
|
"Kernel Time",
|
|
"Kernel Time (Cycles)",
|
|
"Non-Time Metric",
|
|
],
|
|
"Avg": [3446.64, 64499.39, 1000.0],
|
|
"Min": [1769.25, 17269.25, 500.0],
|
|
"Max": [12532.12, 337030.50, 2000.0],
|
|
"Unit": ["ns", "Cycle", "Count"],
|
|
})
|
|
|
|
|
|
@pytest.fixture
|
|
def original_ns_values():
|
|
return {"Avg": 3446.64, "Min": 1769.25, "Max": 12532.12}
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_has_time_data_detection(sample_time_data):
|
|
from utils.tty import has_time_data
|
|
|
|
assert has_time_data(sample_time_data)
|
|
|
|
no_time_data = pd.DataFrame({
|
|
"Metric": ["Non-Time Metric"],
|
|
"Avg": [1000.0],
|
|
"Unit": ["Count"],
|
|
})
|
|
assert not has_time_data(no_time_data)
|
|
|
|
no_unit_column = pd.DataFrame({"Metric": ["Some Metric"], "Avg": [1000.0]})
|
|
assert not has_time_data(no_unit_column)
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_default_unit_is_nanoseconds(sample_time_data):
|
|
time_rows = sample_time_data["Unit"].str.lower().str.contains("ns", na=False)
|
|
assert time_rows.any()
|
|
assert sample_time_data.loc[0, "Unit"] == "ns"
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_time_unit_conversion_to_seconds(sample_time_data, original_ns_values):
|
|
from utils.tty import convert_time_columns
|
|
|
|
converted_df = convert_time_columns(sample_time_data, "s")
|
|
|
|
assert converted_df.loc[0, "Unit"] == "s"
|
|
|
|
expected_avg = original_ns_values["Avg"] / time_units["s"]
|
|
expected_min = original_ns_values["Min"] / time_units["s"]
|
|
expected_max = original_ns_values["Max"] / time_units["s"]
|
|
|
|
assert abs(converted_df.loc[0, "Avg"] - expected_avg) < 1e-10
|
|
assert abs(converted_df.loc[0, "Min"] - expected_min) < 1e-10
|
|
assert abs(converted_df.loc[0, "Max"] - expected_max) < 1e-10
|
|
|
|
assert converted_df.loc[1, "Unit"] == "Cycle"
|
|
assert converted_df.loc[2, "Unit"] == "Count"
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_time_unit_conversion_to_milliseconds(sample_time_data, original_ns_values):
|
|
from utils.tty import convert_time_columns
|
|
|
|
converted_df = convert_time_columns(sample_time_data, "ms")
|
|
|
|
assert converted_df.loc[0, "Unit"] == "ms"
|
|
|
|
expected_avg = original_ns_values["Avg"] / time_units["ms"]
|
|
expected_min = original_ns_values["Min"] / time_units["ms"]
|
|
expected_max = original_ns_values["Max"] / time_units["ms"]
|
|
|
|
assert abs(converted_df.loc[0, "Avg"] - expected_avg) < 1e-6
|
|
assert abs(converted_df.loc[0, "Min"] - expected_min) < 1e-6
|
|
assert abs(converted_df.loc[0, "Max"] - expected_max) < 1e-6
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_time_unit_conversion_to_microseconds(sample_time_data, original_ns_values):
|
|
from utils.tty import convert_time_columns
|
|
|
|
converted_df = convert_time_columns(sample_time_data, "us")
|
|
|
|
assert converted_df.loc[0, "Unit"] == "us"
|
|
|
|
expected_avg = original_ns_values["Avg"] / time_units["us"]
|
|
expected_min = original_ns_values["Min"] / time_units["us"]
|
|
expected_max = original_ns_values["Max"] / time_units["us"]
|
|
|
|
assert abs(converted_df.loc[0, "Avg"] - expected_avg) < 1e-3
|
|
assert abs(converted_df.loc[0, "Min"] - expected_min) < 1e-3
|
|
assert abs(converted_df.loc[0, "Max"] - expected_max) < 1e-3
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_time_unit_conversion_to_nanoseconds(sample_time_data, original_ns_values):
|
|
from utils.tty import convert_time_columns
|
|
|
|
converted_df = convert_time_columns(sample_time_data, "ns")
|
|
|
|
assert converted_df.loc[0, "Unit"] == "ns"
|
|
|
|
assert abs(converted_df.loc[0, "Avg"] - original_ns_values["Avg"]) < 1e-10
|
|
assert abs(converted_df.loc[0, "Min"] - original_ns_values["Min"]) < 1e-10
|
|
assert abs(converted_df.loc[0, "Max"] - original_ns_values["Max"]) < 1e-10
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_non_time_rows_unchanged(sample_time_data):
|
|
from utils.tty import convert_time_columns
|
|
|
|
converted_df = convert_time_columns(sample_time_data, "ms")
|
|
|
|
assert converted_df.loc[1, "Unit"] == "Cycle"
|
|
assert converted_df.loc[2, "Unit"] == "Count"
|
|
assert converted_df.loc[1, "Avg"] == 64499.39
|
|
assert converted_df.loc[2, "Avg"] == 1000.0
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_invalid_time_unit_handling(sample_time_data):
|
|
from utils.tty import convert_time_columns
|
|
|
|
original_df = sample_time_data.copy()
|
|
converted_df = convert_time_columns(sample_time_data, "invalid_unit")
|
|
|
|
pd.testing.assert_frame_equal(converted_df, original_df)
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_missing_unit_column():
|
|
from utils.tty import convert_time_columns
|
|
|
|
df_no_unit = pd.DataFrame({"Metric": ["Test Metric"], "Avg": [1000.0]})
|
|
converted_df = convert_time_columns(df_no_unit, "ms")
|
|
|
|
pd.testing.assert_frame_equal(converted_df, df_no_unit)
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_conversion_with_missing_columns(sample_time_data, original_ns_values):
|
|
from utils.tty import convert_time_columns
|
|
|
|
df_partial = sample_time_data[["Metric_ID", "Metric", "Avg", "Unit"]].copy()
|
|
converted_df = convert_time_columns(df_partial, "ms")
|
|
|
|
assert converted_df.loc[0, "Unit"] == "ms"
|
|
expected_avg = original_ns_values["Avg"] / time_units["ms"]
|
|
assert abs(converted_df.loc[0, "Avg"] - expected_avg) < 1e-6
|
|
|
|
|
|
@pytest.mark.time_unit_conversion
|
|
def test_mathematical_correctness_all_units(sample_time_data, original_ns_values):
|
|
from utils.tty import convert_time_columns
|
|
|
|
test_cases = [
|
|
("s", 10**9), # 1 second = 10^9 nanoseconds
|
|
("ms", 10**6), # 1 millisecond = 10^6 nanoseconds
|
|
("us", 10**3), # 1 microsecond = 10^3 nanoseconds
|
|
("ns", 1), # 1 nanosecond = 1 nanosecond
|
|
]
|
|
|
|
for target_unit, divisor in test_cases:
|
|
converted_df = convert_time_columns(sample_time_data, target_unit)
|
|
|
|
expected_avg = original_ns_values["Avg"] / divisor
|
|
expected_min = original_ns_values["Min"] / divisor
|
|
expected_max = original_ns_values["Max"] / divisor
|
|
|
|
assert abs(converted_df.loc[0, "Avg"] - expected_avg) < 1e-10
|
|
assert abs(converted_df.loc[0, "Min"] - expected_min) < 1e-10
|
|
assert abs(converted_df.loc[0, "Max"] - expected_max) < 1e-10
|
|
assert converted_df.loc[0, "Unit"] == target_unit
|
|
|
|
|
|
# Integration tests with show_all functionality
|
|
@pytest.mark.time_unit_integration
|
|
def test_integration_conversion_flow():
|
|
from utils.tty import convert_time_columns, has_time_data
|
|
|
|
mock_args = Mock()
|
|
mock_args.time_unit = "ms"
|
|
mock_args.decimal = 2
|
|
|
|
sample_df = pd.DataFrame({
|
|
"Metric_ID": ["7.2.0"],
|
|
"Metric": ["Kernel Time"],
|
|
"Avg": [3446640.0], # 3.44664 ms in nanoseconds
|
|
"Min": [1769250.0], # 1.76925 ms in nanoseconds
|
|
"Max": [12532120.0], # 12.53212 ms in nanoseconds
|
|
"Unit": ["ns"],
|
|
})
|
|
|
|
if has_time_data(sample_df):
|
|
converted_df = convert_time_columns(sample_df, mock_args.time_unit)
|
|
else:
|
|
converted_df = sample_df
|
|
|
|
assert converted_df.loc[0, "Unit"] == "ms"
|
|
assert abs(converted_df.loc[0, "Avg"] - 3.44664) < 1e-5
|
|
assert abs(converted_df.loc[0, "Min"] - 1.76925) < 1e-5
|
|
assert abs(converted_df.loc[0, "Max"] - 12.53212) < 1e-5
|
|
|
|
|
|
@pytest.mark.time_unit_integration
|
|
def test_show_all_with_time_unit_conversion():
|
|
from utils.tty import convert_time_columns
|
|
|
|
test_data = pd.DataFrame({
|
|
"Metric_ID": ["7.2.0"],
|
|
"Metric": ["Kernel Time"],
|
|
"Avg": [3446.64],
|
|
"Min": [1769.25],
|
|
"Max": [12532.12],
|
|
"Unit": ["Ns"],
|
|
})
|
|
|
|
for time_unit in ["s", "ms", "us", "ns"]:
|
|
converted_df = convert_time_columns(test_data, time_unit)
|
|
|
|
assert converted_df.loc[0, "Unit"] == time_unit
|
|
|
|
expected_avg = 3446.64 / time_units[time_unit]
|
|
assert abs(converted_df.loc[0, "Avg"] - expected_avg) < 1e-10
|
|
|
|
|
|
@pytest.mark.time_unit_edge_cases
|
|
def test_edge_cases_and_error_handling():
|
|
from utils.tty import convert_time_columns
|
|
|
|
empty_df = pd.DataFrame()
|
|
result = convert_time_columns(empty_df, "ms")
|
|
assert result.empty
|
|
|
|
nan_df = pd.DataFrame({"Avg": [float("nan"), 1000.0], "Unit": ["ns", "Count"]})
|
|
result = convert_time_columns(nan_df, "ms")
|
|
assert result.loc[0, "Unit"] == "ms"
|
|
|
|
mixed_case_df = pd.DataFrame({"Avg": [1000.0, 2000.0], "Unit": ["ns", "NS"]})
|
|
result = convert_time_columns(mixed_case_df, "ms")
|
|
assert result.loc[0, "Unit"] == "ms"
|
|
assert result.loc[1, "Unit"] == "ms"
|
|
|
|
|
|
@pytest.mark.iteration_multiplexing
|
|
def test_iteration_multiplexing(binary_handler_analyze_rocprof_compute):
|
|
workload = "tests/workloads/vcopy_iteration_multiplexing/MI350"
|
|
workload_dir = test_utils.setup_workload_dir(workload)
|
|
|
|
# Test with dispatch filtering
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--dispatch",
|
|
"0",
|
|
"--path",
|
|
workload_dir,
|
|
])
|
|
assert code == 0
|
|
|
|
# Test without dispatch filtering
|
|
code = binary_handler_analyze_rocprof_compute([
|
|
"analyze",
|
|
"--path",
|
|
workload_dir,
|
|
])
|
|
assert code == 0
|
|
|
|
test_utils.clean_output_dir(config["cleanup"], workload_dir)
|