From c867eaa9cdcf575a7a93821cc38cda114a84db1f Mon Sep 17 00:00:00 2001 From: Fei Zheng <44449748+feizheng10@users.noreply.github.com> Date: Tue, 8 Jul 2025 06:51:20 -0600 Subject: [PATCH] format with isort (#793) [ROCm/rocprofiler-compute commit: 239e6550f4d6abe5b8272a4c317ba6b55b63dc5e] --- .../tests/test_analyze_commands.py | 306 ++- .../tests/test_db_connector.py | 294 +- .../tests/test_gpu_specs.py | 126 +- .../rocprofiler-compute/tests/test_utils.py | 2395 +++++++++-------- 4 files changed, 1749 insertions(+), 1372 deletions(-) diff --git a/projects/rocprofiler-compute/tests/test_analyze_commands.py b/projects/rocprofiler-compute/tests/test_analyze_commands.py index 0f157badc8..68b105ffc4 100644 --- a/projects/rocprofiler-compute/tests/test_analyze_commands.py +++ b/projects/rocprofiler-compute/tests/test_analyze_commands.py @@ -683,10 +683,12 @@ def test_baseline(binary_handler_analyze_rocprof_compute): ) assert code == 1 + # ============================================================================= # Test cases for Parser.py # ============================================================================= + @pytest.mark.misc def test_dependency_MI100(binary_handler_analyze_rocprof_compute): for dir in indirs: @@ -697,229 +699,256 @@ def test_dependency_MI100(binary_handler_analyze_rocprof_compute): 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 from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import to_min, to_max, to_avg, to_median, to_std, to_int, to_quantile, to_round, to_mod, to_concat - import pandas as pd + import numpy as np + import pandas as pd + + from utils.parser import ( + to_avg, + to_concat, + to_int, + to_max, + to_median, + to_min, + to_mod, + to_quantile, + to_round, + to_std, + ) try: - result = to_min(None, None) + 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 result is None, "to_median should return None 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 result is None, "to_int should return None 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 result is None, "to_quantile should return None 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 from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import build_eval_string, update_denom_string, calc_builtin_var - + + from utils.parser import build_eval_string, calc_builtin_var, update_denom_string + try: build_eval_string("AVG(SQ_WAVES)", None) 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") == "" assert update_denom_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_parser_error_handling(): """Test parser error handling paths""" import sys from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import build_eval_string, update_denom_string, calc_builtin_var - + + from utils.parser import build_eval_string, calc_builtin_var, update_denom_string + try: build_eval_string("AVG(SQ_WAVES)", None) 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") == "" assert update_denom_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 import os - + import tempfile + with tempfile.TemporaryDirectory() as temp_dir: - code = binary_handler_analyze_rocprof_compute( - ["analyze", "--path", temp_dir] - ) + code = binary_handler_analyze_rocprof_compute(["analyze", "--path", temp_dir]) assert code != 0 - -@pytest.mark.misc + + +@pytest.mark.misc def test_ast_transformer_edge_cases(): """Simplified test focusing on the actual code paths""" import sys from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import CodeTransformer + 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=[] + 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': + 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)}" - + 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=[] + 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}" + 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 from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import eval_metric - import pandas as pd + import numpy as np - + 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') + 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'} - + + mock_dfs_type = {1: "metric_table"} + class MockSysInfo: ip_blocks = "standard" se_per_gpu = 4 @@ -937,41 +966,43 @@ def test_analyze_with_debug_mode(binary_handler_analyze_rocprof_compute): 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] - }) + "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) except Exception as e: pass -@pytest.mark.misc +@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 @@ -981,61 +1012,70 @@ def test_apply_filters_direct(): """Test apply_filters function directly to cover filter branches""" import sys from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import apply_filters + 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 = 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_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 - + 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 + 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 tempfile 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) + csv_path = os.path.join(workload_dir, csv_file) if os.path.exists(csv_path): os.remove(csv_path) - + code = binary_handler_analyze_rocprof_compute( ["analyze", "--path", workload_dir] ) @@ -1047,27 +1087,29 @@ def test_pc_sampling_basic_coverage(): """Test PC sampling functions with minimal data""" import sys from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - - from utils.parser import load_pc_sampling_data, search_pc_sampling_record + 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 @@ -1077,26 +1119,27 @@ def test_build_dfs_edge_cases(): """Test build_dfs and gen_counter_list with various configurations""" import sys from pathlib import Path + 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 - + assert visited + visited, counters = gen_counter_list("INVALID SYNTAX !!!") assert not visited @@ -1106,23 +1149,24 @@ def test_update_functions_coverage(): """Test update_denom_string and update_normUnit_string branches""" import sys from pathlib import Path + sys.path.insert(0, str(Path(__file__).parent.parent / "src")) - + from utils.parser import update_denom_string, update_normUnit_string - + result = update_denom_string("AVG(SQ_WAVES / $denom)", "per_wave") assert "$denom" not in result assert "SQ_WAVES" in result - - result = update_denom_string("AVG(DATA / $denom)", "per_cycle") + + result = update_denom_string("AVG(DATA / $denom)", "per_cycle") assert "$GRBM_GUI_ACTIVE_PER_XCD" in result - + result = update_denom_string("AVG(DATA / $denom)", "per_second") assert "End_Timestamp - Start_Timestamp" in result - + result = update_denom_string("AVG(DATA / $denom)", "unsupported_unit") assert "$denom" in result - + result = update_normUnit_string("(Prefix + $normUnit)", "per_wave") assert "per wave" in result.lower() - assert result[0].isupper() \ No newline at end of file + assert result[0].isupper() diff --git a/projects/rocprofiler-compute/tests/test_db_connector.py b/projects/rocprofiler-compute/tests/test_db_connector.py index e896f449af..6d7908df16 100644 --- a/projects/rocprofiler-compute/tests/test_db_connector.py +++ b/projects/rocprofiler-compute/tests/test_db_connector.py @@ -23,31 +23,37 @@ ##############################################################################el -import pytest -import tempfile +import logging import shutil import sys -import logging -from unittest.mock import Mock, patch, MagicMock, call +import tempfile from pathlib import Path +from unittest.mock import MagicMock, Mock, call, patch + import pandas as pd +import pytest logging.TRACE = logging.DEBUG - 5 logging.addLevelName(logging.TRACE, "TRACE") + + def trace_logger(message, *args, **kwargs): logging.log(logging.TRACE, message, *args, **kwargs) + + setattr(logging, "trace", trace_logger) from db_connector import DatabaseConnector """ -Tests for the DatabaseConnector class that tests almost methods with initialization, -CSV import, database removal, and error handling. +Tests for the DatabaseConnector class that tests almost methods with initialization, +CSV import, database removal, and error handling. The tests use mocks instead of a real MongoDB server for speed and reliability. """ + class TestDatabaseConnector: - + @pytest.fixture def mock_args_import(self): """Mock arguments for import operation""" @@ -62,7 +68,7 @@ class TestDatabaseConnector: args.remove = False args.kernel_verbose = False return args - + @pytest.fixture def mock_args_remove(self): """Mock arguments for remove operation""" @@ -77,15 +83,15 @@ class TestDatabaseConnector: args.remove = True args.kernel_verbose = False return args - + def test_init(self, mock_args_import): """Test DatabaseConnector initialization""" connector = DatabaseConnector(mock_args_import) - + assert connector.args == mock_args_import assert isinstance(connector.cache, dict) assert len(connector.cache) == 0 - + expected_connection_info = { "username": "test_user", "password": "test_pass", @@ -99,130 +105,140 @@ class TestDatabaseConnector: assert connector.interaction_type is None assert connector.client is None - @patch('db_connector.pd.read_csv') - @patch('db_connector.Path') + @patch("db_connector.pd.read_csv") + @patch("db_connector.Path") def test_prep_import_success(self, mock_path, mock_read_csv, mock_args_import): """Test successful prep_import""" # Setup mocks mock_path.return_value.joinpath.return_value = "/fake/path/sysinfo.csv" mock_path.return_value.is_file.return_value = True - - mock_sysinfo = pd.DataFrame({ - 'gpu_model': ['MI100 '], - 'workload_name': [' test_workload'] - }) + + mock_sysinfo = pd.DataFrame( + {"gpu_model": ["MI100 "], "workload_name": [" test_workload"]} + ) mock_read_csv.return_value = mock_sysinfo - + connector = DatabaseConnector(mock_args_import) connector.prep_import() - + expected_db = "rocprofiler-compute_test_team_test_workload_MI100" assert connector.connection_info["db"] == expected_db - @patch('db_connector.pd.read_csv') - @patch('db_connector.Path') + @patch("db_connector.pd.read_csv") + @patch("db_connector.Path") def test_prep_import_missing_file(self, mock_path, mock_read_csv, mock_args_import): """Test prep_import when sysinfo.csv is missing""" mock_path.return_value.joinpath.return_value = "/fake/path/sysinfo.csv" mock_path.return_value.is_file.return_value = False - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)) as mock_console_error: + + with patch( + "db_connector.console_error", side_effect=SystemExit(1) + ) as mock_console_error: with pytest.raises(SystemExit): connector.prep_import() - + mock_console_error.assert_called_with( "database", "Unable to parse SoC and/or workload name from sysinfo.csv" ) - @patch('db_connector.pd.read_csv') - @patch('db_connector.Path') + @patch("db_connector.pd.read_csv") + @patch("db_connector.Path") def test_prep_import_key_error(self, mock_path, mock_read_csv, mock_args_import): """Test prep_import when required fields are missing""" mock_path.return_value.joinpath.return_value = "/fake/path/sysinfo.csv" mock_path.return_value.is_file.return_value = True - - mock_sysinfo = pd.DataFrame({'other_column': ['value']}) + + mock_sysinfo = pd.DataFrame({"other_column": ["value"]}) mock_read_csv.return_value = mock_sysinfo - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)) as mock_console_error: + + with patch( + "db_connector.console_error", side_effect=SystemExit(1) + ) as mock_console_error: with pytest.raises(SystemExit): connector.prep_import() - + assert mock_console_error.called error_call = mock_console_error.call_args[0][0] assert "Outdated workload" in error_call - @patch('db_connector.tqdm') - @patch('db_connector.os.listdir') - @patch('db_connector.console_log') - @patch('db_connector.console_warning') - @patch('db_connector.kernel_name_shortener') - @patch('db_connector.MongoClient') - @patch('db_connector.pd.read_csv') - def test_db_import_success(self, mock_read_csv, mock_mongo_client, mock_kernel_shortener, - mock_console_warning, mock_console_log, mock_listdir, - mock_tqdm, mock_args_import): + @patch("db_connector.tqdm") + @patch("db_connector.os.listdir") + @patch("db_connector.console_log") + @patch("db_connector.console_warning") + @patch("db_connector.kernel_name_shortener") + @patch("db_connector.MongoClient") + @patch("db_connector.pd.read_csv") + def test_db_import_success( + self, + mock_read_csv, + mock_mongo_client, + mock_kernel_shortener, + mock_console_warning, + mock_console_log, + mock_listdir, + mock_tqdm, + mock_args_import, + ): """Test successful database import""" - mock_listdir.return_value = ['test_data.csv', 'empty_file.csv', 'non_csv.txt'] + mock_listdir.return_value = ["test_data.csv", "empty_file.csv", "non_csv.txt"] mock_tqdm.return_value = mock_listdir.return_value - - test_df = pd.DataFrame({'col1': [1, 2], 'col2': [3, 4]}) - mock_read_csv.side_effect = [ - test_df, - pd.errors.EmptyDataError() - ] - + + test_df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]}) + mock_read_csv.side_effect = [test_df, pd.errors.EmptyDataError()] + mock_client_instance = MagicMock() mock_db = MagicMock() mock_collection = MagicMock() mock_workload_db = MagicMock() mock_workload_col = MagicMock() - + mock_mongo_client.return_value = mock_client_instance mock_client_instance.__getitem__.side_effect = lambda x: { - 'rocprofiler-compute_test_team_test_workload_MI100': mock_db, - 'workload_names': mock_workload_db + "rocprofiler-compute_test_team_test_workload_MI100": mock_db, + "workload_names": mock_workload_db, }.get(x, mock_db) mock_db.__getitem__.return_value = mock_collection mock_workload_db.__getitem__.return_value = mock_workload_col - + connector = DatabaseConnector(mock_args_import) connector.connection_info["workload"] = "/fake/workload/path" connector.client = mock_client_instance - - with patch.object(connector, 'prep_import') as mock_prep: + + with patch.object(connector, "prep_import") as mock_prep: mock_prep.return_value = None - connector.connection_info["db"] = "rocprofiler-compute_test_team_test_workload_MI100" - + connector.connection_info["db"] = ( + "rocprofiler-compute_test_team_test_workload_MI100" + ) + connector.db_import() - + mock_collection.insert_many.assert_called_once() mock_workload_col.replace_one.assert_called_once() - @patch('db_connector.console_log') + @patch("db_connector.console_log") def test_db_remove_success(self, mock_console_log, mock_args_remove): """Test successful database removal""" mock_client = MagicMock() mock_db_to_remove = MagicMock() mock_workload_names_db = MagicMock() mock_names_col = MagicMock() - + mock_client.__getitem__.side_effect = lambda x: { - 'rocprofiler-compute_test_team_workload_mi100': mock_db_to_remove, - 'workload_names': mock_workload_names_db + "rocprofiler-compute_test_team_workload_mi100": mock_db_to_remove, + "workload_names": mock_workload_names_db, }[x] mock_workload_names_db.__getitem__.return_value = mock_names_col - mock_db_to_remove.list_collection_names.return_value = ['col1', 'col2'] - + mock_db_to_remove.list_collection_names.return_value = ["col1", "col2"] + connector = DatabaseConnector(mock_args_remove) connector.client = mock_client - + connector.db_remove() - + mock_client.drop_database.assert_called_once_with(mock_db_to_remove) mock_names_col.delete_many.assert_called_once_with( {"name": "rocprofiler-compute_test_team_workload_mi100"} @@ -232,20 +248,20 @@ class TestDatabaseConnector: """Test pre_processing when neither upload nor remove is specified""" mock_args_import.upload = False mock_args_import.remove = False - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() def test_pre_processing_remove_invalid_workload_name(self, mock_args_remove): """Test pre_processing remove with invalid workload name""" mock_args_remove.workload = "invalid_name" - + connector = DatabaseConnector(mock_args_remove) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() @@ -253,106 +269,121 @@ class TestDatabaseConnector: """Test pre_processing remove with missing host/username""" mock_args_remove.host = None mock_args_remove.username = None - + connector = DatabaseConnector(mock_args_remove) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() def test_pre_processing_remove_protected_database(self, mock_args_remove): """Test pre_processing remove with protected database names""" mock_args_remove.workload = "admin" - + connector = DatabaseConnector(mock_args_remove) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() - @patch('db_connector.Path') - @patch('db_connector.is_workload_empty') - @patch('db_connector.getpass.getpass') - @patch('db_connector.console_log') - @patch('db_connector.MongoClient') - def test_pre_processing_import_password_prompt_success(self, mock_mongo_client, mock_console_log, - mock_getpass, mock_is_workload_empty, - mock_path, mock_args_import): + @patch("db_connector.Path") + @patch("db_connector.is_workload_empty") + @patch("db_connector.getpass.getpass") + @patch("db_connector.console_log") + @patch("db_connector.MongoClient") + def test_pre_processing_import_password_prompt_success( + self, + mock_mongo_client, + mock_console_log, + mock_getpass, + mock_is_workload_empty, + mock_path, + mock_args_import, + ): """Test pre_processing import with password prompt success""" mock_args_import.password = "" mock_getpass.return_value = "prompted_password" - + mock_path.return_value.absolute.return_value.is_dir.return_value = True - mock_path.return_value.absolute.return_value.resolve.return_value = "/resolved/path" - + mock_path.return_value.absolute.return_value.resolve.return_value = ( + "/resolved/path" + ) + mock_client_instance = MagicMock() mock_mongo_client.return_value = mock_client_instance mock_client_instance.server_info.return_value = {} - + connector = DatabaseConnector(mock_args_import) connector.pre_processing() - + mock_getpass.assert_called_once() mock_console_log.assert_called_with("database", "Password received") - @patch('db_connector.Path') - @patch('db_connector.is_workload_empty') - @patch('db_connector.MongoClient') - def test_pre_processing_import_connection_failure(self, mock_mongo_client, mock_is_workload_empty, - mock_path, mock_args_import): + @patch("db_connector.Path") + @patch("db_connector.is_workload_empty") + @patch("db_connector.MongoClient") + def test_pre_processing_import_connection_failure( + self, mock_mongo_client, mock_is_workload_empty, mock_path, mock_args_import + ): """Test pre_processing import with MongoDB connection failure""" mock_path.return_value.absolute.return_value.is_dir.return_value = True - mock_path.return_value.absolute.return_value.resolve.return_value = "/resolved/path" - + mock_path.return_value.absolute.return_value.resolve.return_value = ( + "/resolved/path" + ) + mock_client_instance = MagicMock() mock_mongo_client.return_value = mock_client_instance mock_client_instance.server_info.side_effect = Exception("Connection failed") - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() - @patch('db_connector.Path') - @patch('db_connector.is_workload_empty') - def test_pre_processing_import_missing_required_fields(self, mock_is_workload_empty, mock_path, mock_args_import): + @patch("db_connector.Path") + @patch("db_connector.is_workload_empty") + def test_pre_processing_import_missing_required_fields( + self, mock_is_workload_empty, mock_path, mock_args_import + ): """Test pre_processing import with missing required fields""" mock_args_import.host = None - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() - @patch('db_connector.Path') - def test_pre_processing_import_invalid_workload_path(self, mock_path, mock_args_import): + @patch("db_connector.Path") + def test_pre_processing_import_invalid_workload_path( + self, mock_path, mock_args_import + ): """Test pre_processing import with invalid workload path""" mock_path.return_value.absolute.return_value.is_dir.return_value = False - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() def test_pre_processing_import_team_name_too_long(self, mock_args_import): """Test pre_processing import with team name exceeding limit""" mock_args_import.team = "this_team_name_is_way_too_long" - + connector = DatabaseConnector(mock_args_import) - - with patch('db_connector.console_error', side_effect=SystemExit(1)): + + with patch("db_connector.console_error", side_effect=SystemExit(1)): with pytest.raises(SystemExit): connector.pre_processing() class TestDatabaseConnectorIntegration: """Simple integration test""" - - @patch('db_connector.Path') - @patch('db_connector.pd.read_csv') + + @patch("db_connector.Path") + @patch("db_connector.pd.read_csv") def test_prep_import_with_real_workload_path(self, mock_read_csv, mock_path): """Test prep_import with actual workload path structure""" args = Mock() @@ -365,22 +396,23 @@ class TestDatabaseConnectorIntegration: args.upload = True args.remove = False args.kernel_verbose = False - - mock_path.return_value.joinpath.return_value = "/app/tests/workloads/device_filter/MI100/sysinfo.csv" + + mock_path.return_value.joinpath.return_value = ( + "/app/tests/workloads/device_filter/MI100/sysinfo.csv" + ) mock_path.return_value.is_file.return_value = True - - mock_sysinfo = pd.DataFrame({ - 'gpu_model': ['MI100'], - 'workload_name': ['device_filter'] - }) + + mock_sysinfo = pd.DataFrame( + {"gpu_model": ["MI100"], "workload_name": ["device_filter"]} + ) mock_read_csv.return_value = mock_sysinfo - + connector = DatabaseConnector(args) connector.prep_import() - + expected_db = "rocprofiler-compute_test_team_device_filter_MI100" assert connector.connection_info["db"] == expected_db if __name__ == "__main__": - pytest.main([__file__, "-v"]) \ No newline at end of file + pytest.main([__file__, "-v"]) diff --git a/projects/rocprofiler-compute/tests/test_gpu_specs.py b/projects/rocprofiler-compute/tests/test_gpu_specs.py index 570eb599fc..87238294ab 100644 --- a/projects/rocprofiler-compute/tests/test_gpu_specs.py +++ b/projects/rocprofiler-compute/tests/test_gpu_specs.py @@ -22,19 +22,18 @@ # SOFTWARE. ##############################################################################el +import os import re import subprocess import sys -import pytest -import yaml import tempfile -import os from importlib.machinery import SourceFileLoader -from unittest.mock import patch, mock_open, MagicMock from pathlib import Path +from unittest.mock import MagicMock, mock_open, patch import pandas as pd import pytest +import yaml from src.utils.specs import generate_machine_specs @@ -201,185 +200,214 @@ def test_num_xcds_cli_output(): assert compute_partition_actual is not None assert int(num_xcd_actual) == num_xcds.get(compute_partition_actual.lower(), -1) + @pytest.mark.misc def test_load_yaml_file_not_found(): """Test _load_yaml with non-existent file - covers lines 104-105""" from src.utils.mi_gpu_spec import MIGPUSpecs - + non_existent_path = "/path/that/does/not/exist/file.yaml" - + with pytest.raises(SystemExit): MIGPUSpecs._load_yaml(non_existent_path) + @pytest.mark.misc def test_load_yaml_invalid_yaml(): """Test _load_yaml with corrupted YAML - covers lines 106-107""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: - f.write('invalid: yaml: content: [\nunclosed bracket') + + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + f.write("invalid: yaml: content: [\nunclosed bracket") temp_path = f.name - + try: with pytest.raises(SystemExit): MIGPUSpecs._load_yaml(temp_path) finally: os.unlink(temp_path) + @pytest.mark.misc def test_load_yaml_generic_exception(): """Test _load_yaml generic exception handling - covers lines 108-111""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch('builtins.open', side_effect=PermissionError("Access denied")): + + with patch("builtins.open", side_effect=PermissionError("Access denied")): with pytest.raises(SystemExit): MIGPUSpecs._load_yaml("some_file.yaml") + @pytest.mark.misc def test_get_gpu_series_dict_uninitialized(): """Test get_gpu_series_dict when dict not populated - covers lines 182-185""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_gpu_series_dict', {}): + + with patch.object(MIGPUSpecs, "_gpu_series_dict", {}): with pytest.raises(SystemExit): MIGPUSpecs.get_gpu_series_dict() + @pytest.mark.misc def test_get_gpu_series_uninitialized(): """Test get_gpu_series when dict not populated - covers lines 191-194""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_gpu_series_dict', {}): + + with patch.object(MIGPUSpecs, "_gpu_series_dict", {}): with pytest.raises(SystemExit): result = MIGPUSpecs.get_gpu_series("gfx942") + @pytest.mark.misc def test_get_perfmon_config_uninitialized(): """Test get_perfmon_config when dict not populated - covers lines 210-213""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_perfmon_config', {}): + + with patch.object(MIGPUSpecs, "_perfmon_config", {}): with pytest.raises(SystemExit): MIGPUSpecs.get_perfmon_config("gfx942") + @pytest.mark.misc def test_get_gpu_model_uninitialized(): """Test get_gpu_model when dict not populated - covers lines 223-226""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_gpu_model_dict', {}): + + with patch.object(MIGPUSpecs, "_gpu_model_dict", {}): with pytest.raises(SystemExit): MIGPUSpecs.get_gpu_model("gfx942", "29857") + @pytest.mark.misc def test_get_gpu_model_invalid_chip_id(): """Test get_gpu_model with invalid chip_id - covers lines 235-236""" from src.utils.mi_gpu_spec import MIGPUSpecs - + result = MIGPUSpecs.get_gpu_model("gfx942", "99999") assert result is None + @pytest.mark.misc def test_get_gpu_model_invalid_arch(): """Test get_gpu_model with invalid architecture - covers lines 243-244""" from src.utils.mi_gpu_spec import MIGPUSpecs - + result = MIGPUSpecs.get_gpu_model("gfx999", "12345") assert result is None + @pytest.mark.misc def test_get_gpu_model_none_result(): """Test get_gpu_model when result is None - covers lines 246-248""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_chip_id_dict', {999: None}): + + with patch.object(MIGPUSpecs, "_chip_id_dict", {999: None}): result = MIGPUSpecs.get_gpu_model("gfx942", "999") assert result is None -@pytest.mark.misc + +@pytest.mark.misc def test_get_num_xcds_no_compute_partition_data(): """Test get_num_xcds when no compute partition data found - covers lines 307-309""" from src.utils.mi_gpu_spec import MIGPUSpecs - + mock_dict = {"gfx942": None} - with patch.object(MIGPUSpecs, '_gpu_arch_to_compute_partition_dict', mock_dict): + with patch.object(MIGPUSpecs, "_gpu_arch_to_compute_partition_dict", mock_dict): result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx942") + @pytest.mark.misc def test_get_num_xcds_uninitialized_dict(): """Test get_num_xcds when XCD dict not populated - covers lines 315-317""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_num_xcds_dict', {}): + + with patch.object(MIGPUSpecs, "_num_xcds_dict", {}): with pytest.raises(SystemExit): MIGPUSpecs.get_num_xcds(gpu_arch="gfx950", gpu_model="MI350") + @pytest.mark.misc def test_get_num_xcds_unknown_gpu_model(): """Test get_num_xcds with unknown gpu model - covers lines 319-321""" from src.utils.mi_gpu_spec import MIGPUSpecs - + result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx950", gpu_model="UNKNOWN_MODEL") + @pytest.mark.misc def test_get_num_xcds_no_compute_partition(): """Test get_num_xcds with no compute partition - covers lines 325-327""" from src.utils.mi_gpu_spec import MIGPUSpecs - - result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx950", gpu_model="MI350", compute_partition="") + + result = MIGPUSpecs.get_num_xcds( + gpu_arch="gfx950", gpu_model="MI350", compute_partition="" + ) + @pytest.mark.misc def test_get_num_xcds_unknown_compute_partition(): """Test get_num_xcds with unknown compute partition - covers lines 329-332""" from src.utils.mi_gpu_spec import MIGPUSpecs - - result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx950", gpu_model="MI350", compute_partition="UNKNOWN") + + result = MIGPUSpecs.get_num_xcds( + gpu_arch="gfx950", gpu_model="MI350", compute_partition="UNKNOWN" + ) + @pytest.mark.misc def test_get_num_xcds_none_partition_value(): """Test get_num_xcds when partition value is None - covers lines 338-340""" from src.utils.mi_gpu_spec import MIGPUSpecs - + mock_dict = {"mi350": {"spx": None}} - with patch.object(MIGPUSpecs, '_num_xcds_dict', mock_dict): - result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx950", gpu_model="MI350", compute_partition="spx") + with patch.object(MIGPUSpecs, "_num_xcds_dict", mock_dict): + result = MIGPUSpecs.get_num_xcds( + gpu_arch="gfx950", gpu_model="MI350", compute_partition="spx" + ) + @pytest.mark.misc def test_get_num_xcds_no_gpu_model(): """Test get_num_xcds with no gpu model - covers line 342""" from src.utils.mi_gpu_spec import MIGPUSpecs - - result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx950", gpu_model="", compute_partition="spx") + + result = MIGPUSpecs.get_num_xcds( + gpu_arch="gfx950", gpu_model="", compute_partition="spx" + ) + @pytest.mark.misc def test_get_chip_id_dict_empty(): """Test get_chip_id_dict when dict is empty - covers line 352""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_chip_id_dict', {}): - with patch('src.utils.mi_gpu_spec.console_error') as mock_error: + + with patch.object(MIGPUSpecs, "_chip_id_dict", {}): + with patch("src.utils.mi_gpu_spec.console_error") as mock_error: result = MIGPUSpecs.get_chip_id_dict() mock_error.assert_called_once() + @pytest.mark.misc def test_get_num_xcds_dict_empty(): """Test get_num_xcds_dict when dict is empty - covers line 359""" from src.utils.mi_gpu_spec import MIGPUSpecs - - with patch.object(MIGPUSpecs, '_num_xcds_dict', {}): - with patch('src.utils.mi_gpu_spec.console_error') as mock_error: + + with patch.object(MIGPUSpecs, "_num_xcds_dict", {}): + with patch("src.utils.mi_gpu_spec.console_error") as mock_error: result = MIGPUSpecs.get_num_xcds_dict() mock_error.assert_called_once() + + @pytest.mark.misc def test_normal_functionality_still_works(): """Ensure that normal paths still work after adding error handling tests""" from src.utils.mi_gpu_spec import MIGPUSpecs - + result = MIGPUSpecs.get_gpu_model("gfx906", None) assert result is not None - + result = MIGPUSpecs.get_gpu_series("gfx906") assert result is not None - + result = MIGPUSpecs.get_num_xcds(gpu_arch="gfx906") - assert result == 1 \ No newline at end of file + assert result == 1 diff --git a/projects/rocprofiler-compute/tests/test_utils.py b/projects/rocprofiler-compute/tests/test_utils.py index ded8209080..504944ccaa 100644 --- a/projects/rocprofiler-compute/tests/test_utils.py +++ b/projects/rocprofiler-compute/tests/test_utils.py @@ -27,12 +27,14 @@ import logging logging.trace = lambda *args, **kwargs: None import builtins +import glob import inspect import io import json import locale import logging import os +import pathlib import re import selectors import shutil @@ -40,9 +42,6 @@ import subprocess import tempfile from pathlib import Path from unittest import mock -import pathlib -import glob -import shutil import pandas as pd import pytest @@ -3091,183 +3090,241 @@ def test_run_prof_tcc_flattening_mi300(tmp_path, monkeypatch): assert flatten_called + import utils.utils as utils_mod - -class MockMSpec: - def __init__(self, gpu_model="mi300a", gpu_arch="gfx942", compute_partition=None, l2_banks=32): - self.gpu_model = gpu_model - self.gpu_arch = gpu_arch - self.compute_partition = compute_partition - self._l2_banks = l2_banks - -def test_run_prof_sdk_creates_new_env_copy(tmp_path, monkeypatch): - """ - Covers: new_env = os.environ.copy() - when rocprof_cmd == "rocprofiler-sdk" and new_env was not previously set - by the mspec.gpu_model check. - """ - fname_str = str(tmp_path / "counters.txt") - pathlib.Path(fname_str).touch() - workload_dir_str = str(tmp_path) - - monkeypatch.setattr("utils.utils.rocprof_cmd", "rocprofiler-sdk") - monkeypatch.setattr("utils.utils.using_v3", lambda: False) - monkeypatch.setattr("utils.utils.process_rocprofv3_output", lambda *a, **k: []) - - - capture_subprocess_called_with_env = None - def mock_capture_subprocess(app_cmd, new_env=None, profileMode=False): - nonlocal capture_subprocess_called_with_env - capture_subprocess_called_with_env = new_env - return (True, "Success") - monkeypatch.setattr("utils.utils.capture_subprocess_output", mock_capture_subprocess) - - def mock_console_error_no_exit(msg, exit=True): - print(f"Mocked console_error: {msg}, exit={exit}") - monkeypatch.setattr("utils.utils.console_error", mock_console_error_no_exit) - monkeypatch.setattr("utils.utils.console_debug", lambda *a, **k: None) - monkeypatch.setattr("utils.utils.parse_text", lambda *a, **k: ["COUNTER1", "COUNTER2"]) - - mock_fname_path_obj = mock.Mock(spec=pathlib.Path) - mock_fname_path_obj.stem = "counters" - mock_fname_path_obj.name = "counters.txt" - mock_fname_path_obj.with_suffix.return_value.exists.return_value = False - mock_out_path_obj = mock.Mock(spec=pathlib.Path) - mock_out_path_obj.exists.return_value = False - - def path_side_effect(p_arg, *args): - if isinstance(p_arg, pathlib.Path): - if p_arg.name == "counters.txt": return mock_fname_path_obj - return p_arg - if isinstance(p_arg, str): - if p_arg.endswith("/out"): return mock_out_path_obj - if p_arg.endswith("counters.txt"): return mock_fname_path_obj - if p_arg == mock_fname_path_obj and args == () and hasattr(p_arg, 'with_suffix'): - return mock_fname_path_obj + + +class MockMSpec: + def __init__( + self, gpu_model="mi300a", gpu_arch="gfx942", compute_partition=None, l2_banks=32 + ): + self.gpu_model = gpu_model + self.gpu_arch = gpu_arch + self.compute_partition = compute_partition + self._l2_banks = l2_banks + + +def test_run_prof_sdk_creates_new_env_copy(tmp_path, monkeypatch): + """ + Covers: new_env = os.environ.copy() + when rocprof_cmd == "rocprofiler-sdk" and new_env was not previously set + by the mspec.gpu_model check. + """ + fname_str = str(tmp_path / "counters.txt") + pathlib.Path(fname_str).touch() + workload_dir_str = str(tmp_path) + + monkeypatch.setattr("utils.utils.rocprof_cmd", "rocprofiler-sdk") + monkeypatch.setattr("utils.utils.using_v3", lambda: False) + monkeypatch.setattr("utils.utils.process_rocprofv3_output", lambda *a, **k: []) + + capture_subprocess_called_with_env = None + + def mock_capture_subprocess(app_cmd, new_env=None, profileMode=False): + nonlocal capture_subprocess_called_with_env + capture_subprocess_called_with_env = new_env + return (True, "Success") + + monkeypatch.setattr("utils.utils.capture_subprocess_output", mock_capture_subprocess) + + def mock_console_error_no_exit(msg, exit=True): + print(f"Mocked console_error: {msg}, exit={exit}") + + monkeypatch.setattr("utils.utils.console_error", mock_console_error_no_exit) + monkeypatch.setattr("utils.utils.console_debug", lambda *a, **k: None) + monkeypatch.setattr( + "utils.utils.parse_text", lambda *a, **k: ["COUNTER1", "COUNTER2"] + ) + + mock_fname_path_obj = mock.Mock(spec=pathlib.Path) + mock_fname_path_obj.stem = "counters" + mock_fname_path_obj.name = "counters.txt" + mock_fname_path_obj.with_suffix.return_value.exists.return_value = False + mock_out_path_obj = mock.Mock(spec=pathlib.Path) + mock_out_path_obj.exists.return_value = False + + def path_side_effect(p_arg, *args): + if isinstance(p_arg, pathlib.Path): + if p_arg.name == "counters.txt": + return mock_fname_path_obj + return p_arg + if isinstance(p_arg, str): + if p_arg.endswith("/out"): + return mock_out_path_obj + if p_arg.endswith("counters.txt"): + return mock_fname_path_obj + if p_arg == mock_fname_path_obj and args == () and hasattr(p_arg, "with_suffix"): + return mock_fname_path_obj return mock_fname_path_obj - monkeypatch.setattr("utils.utils.path", path_side_effect) - - - original_env_var = "original_value" - monkeypatch.setenv("EXISTING_VAR", original_env_var) - monkeypatch.delenv("ROCPROFILER_INDIVIDUAL_XCC_MODE", raising=False) - - profiler_options = {"APP_CMD": "my_app --arg"} - mspec = MockMSpec(gpu_model="mi250") - loglevel = logging.DEBUG - format_rocprof_output = True - - dummy_df = pd.DataFrame({'Dispatch_ID': [0], 'A': [1]}) - monkeypatch.setattr("pandas.read_csv", lambda *a, **k: dummy_df.copy()) - monkeypatch.setattr("pandas.DataFrame.to_csv", lambda self, *a, **k: None) - monkeypatch.setattr("shutil.copyfile", lambda *a, **k: None) - monkeypatch.setattr("shutil.rmtree", lambda *a, **k: None) - monkeypatch.setattr("utils.utils.console_warning", lambda *a, **k: None) - - utils_mod.run_prof(fname_str, profiler_options.copy(), workload_dir_str, mspec, loglevel, format_rocprof_output) - - assert capture_subprocess_called_with_env is not None, "new_env should have been created" - assert "EXISTING_VAR" in capture_subprocess_called_with_env, "new_env should be a copy of os.environ" - assert capture_subprocess_called_with_env["EXISTING_VAR"] == original_env_var - assert "ROCPROF_COUNTERS" in capture_subprocess_called_with_env - assert "APP_CMD" not in capture_subprocess_called_with_env - -def test_run_prof_v3_sdk_and_cli_calls_trace_processing(tmp_path, monkeypatch): - """ - Covers: - Line 3 (SDK): if "ROCPROF_HIP_RUNTIME_API_TRACE" in options: process_hip_trace_output(...) - Line 4 (CLI): if "--kokkos-trace" in options: process_kokkos_trace_output(...) - Line 5 (CLI): elif "--hip-trace" in options: process_hip_trace_output(...) - """ - fname_str = str(tmp_path / "counters.txt") - pathlib.Path(fname_str).touch() - fbase_str = "counters" - workload_dir_str = str(tmp_path) - (tmp_path / "out" / "pmc_1").mkdir(parents=True, exist_ok=True) - - monkeypatch.setattr("utils.utils.capture_subprocess_output", lambda *a, **k: (True, "Success")) - monkeypatch.setattr("utils.utils.process_rocprofv3_output", lambda *a, **k: [str(tmp_path / "results1.csv")]) - - hip_trace_called_with = None - def mock_hip_trace(wd, fb): - nonlocal hip_trace_called_with - hip_trace_called_with = (wd, fb) - monkeypatch.setattr("utils.utils.process_hip_trace_output", mock_hip_trace) - - kokkos_trace_called_with = None - def mock_kokkos_trace(wd, fb): - nonlocal kokkos_trace_called_with - kokkos_trace_called_with = (wd, fb) - monkeypatch.setattr("utils.utils.process_kokkos_trace_output", mock_kokkos_trace) - - monkeypatch.setattr("utils.utils.console_debug", lambda *a, **k: None) - monkeypatch.setattr("utils.utils.console_warning", lambda *a, **k: None) - monkeypatch.setattr("utils.utils.parse_text", lambda *a, **k: ["C1"]) - - mock_fname_path_obj = mock.Mock(spec=pathlib.Path) - mock_fname_path_obj.stem = fbase_str - mock_fname_path_obj.name = "counters.txt" - mock_fname_path_obj.with_suffix.return_value.exists.return_value = False - - mock_out_path_obj = mock.Mock(spec=pathlib.Path) - mock_out_path_obj.exists.return_value = True - - def path_side_effect(p_arg, *args): - if isinstance(p_arg, pathlib.Path) and p_arg.name == "counters.txt": return mock_fname_path_obj - if isinstance(p_arg, str) and p_arg.endswith("/out"): return mock_out_path_obj - if isinstance(p_arg, str) and p_arg.endswith("counters.txt"): return mock_fname_path_obj - if p_arg == mock_fname_path_obj and args == () and hasattr(p_arg, 'with_suffix'): return mock_fname_path_obj - return mock_fname_path_obj - monkeypatch.setattr("utils.utils.path", path_side_effect) - - dummy_df = pd.DataFrame({'Dispatch_ID': [0], 'A': [1]}) - monkeypatch.setattr("pandas.read_csv", lambda *a, **k: dummy_df.copy()) - monkeypatch.setattr("pandas.DataFrame.to_csv", lambda self, *a, **k: None) - monkeypatch.setattr("shutil.copyfile", lambda *a, **k: None) - monkeypatch.setattr("shutil.rmtree", lambda *a, **k: None) - monkeypatch.setattr("utils.utils.flatten_tcc_info_across_xcds", lambda df, *a: df) - monkeypatch.setattr("utils.utils.mi_gpu_specs.get_num_xcds", lambda *a: 1) - - mspec = MockMSpec() - loglevel = logging.INFO - format_rocprof_output = True - - monkeypatch.setattr("utils.utils.rocprof_cmd", "rocprofiler-sdk") - monkeypatch.setattr("utils.utils.using_v3", lambda: True) - - profiler_options_sdk_hip = { - "APP_CMD": "my_app", - "ROCPROF_HIP_RUNTIME_API_TRACE": "1" - } - hip_trace_called_with = None - kokkos_trace_called_with = None - - utils_mod.run_prof(fname_str, profiler_options_sdk_hip.copy(), workload_dir_str, mspec, loglevel, format_rocprof_output) - assert hip_trace_called_with == (workload_dir_str, fbase_str) - assert kokkos_trace_called_with is None - - monkeypatch.setattr("utils.utils.rocprof_cmd", "rocprof_cli_v3") - - profiler_options_cli_kokkos = ["--kokkos-trace", "--other-opt"] - hip_trace_called_with = None - kokkos_trace_called_with = None - - utils_mod.run_prof(fname_str, profiler_options_cli_kokkos, workload_dir_str, mspec, loglevel, format_rocprof_output) - assert kokkos_trace_called_with == (workload_dir_str, fbase_str) - assert hip_trace_called_with is None - - profiler_options_cli_hip = ["--hip-trace", "--other-opt"] - hip_trace_called_with = None - kokkos_trace_called_with = None - - utils_mod.run_prof(fname_str, profiler_options_cli_hip, workload_dir_str, mspec, loglevel, format_rocprof_output) - assert hip_trace_called_with == (workload_dir_str, fbase_str) - assert kokkos_trace_called_with is None + + monkeypatch.setattr("utils.utils.path", path_side_effect) + + original_env_var = "original_value" + monkeypatch.setenv("EXISTING_VAR", original_env_var) + monkeypatch.delenv("ROCPROFILER_INDIVIDUAL_XCC_MODE", raising=False) + + profiler_options = {"APP_CMD": "my_app --arg"} + mspec = MockMSpec(gpu_model="mi250") + loglevel = logging.DEBUG + format_rocprof_output = True + + dummy_df = pd.DataFrame({"Dispatch_ID": [0], "A": [1]}) + monkeypatch.setattr("pandas.read_csv", lambda *a, **k: dummy_df.copy()) + monkeypatch.setattr("pandas.DataFrame.to_csv", lambda self, *a, **k: None) + monkeypatch.setattr("shutil.copyfile", lambda *a, **k: None) + monkeypatch.setattr("shutil.rmtree", lambda *a, **k: None) + monkeypatch.setattr("utils.utils.console_warning", lambda *a, **k: None) + + utils_mod.run_prof( + fname_str, + profiler_options.copy(), + workload_dir_str, + mspec, + loglevel, + format_rocprof_output, + ) + + assert ( + capture_subprocess_called_with_env is not None + ), "new_env should have been created" + assert ( + "EXISTING_VAR" in capture_subprocess_called_with_env + ), "new_env should be a copy of os.environ" + assert capture_subprocess_called_with_env["EXISTING_VAR"] == original_env_var + assert "ROCPROF_COUNTERS" in capture_subprocess_called_with_env + assert "APP_CMD" not in capture_subprocess_called_with_env + + +def test_run_prof_v3_sdk_and_cli_calls_trace_processing(tmp_path, monkeypatch): + """ + Covers: + Line 3 (SDK): if "ROCPROF_HIP_RUNTIME_API_TRACE" in options: process_hip_trace_output(...) + Line 4 (CLI): if "--kokkos-trace" in options: process_kokkos_trace_output(...) + Line 5 (CLI): elif "--hip-trace" in options: process_hip_trace_output(...) + """ + fname_str = str(tmp_path / "counters.txt") + pathlib.Path(fname_str).touch() + fbase_str = "counters" + workload_dir_str = str(tmp_path) + (tmp_path / "out" / "pmc_1").mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr( + "utils.utils.capture_subprocess_output", lambda *a, **k: (True, "Success") + ) + monkeypatch.setattr( + "utils.utils.process_rocprofv3_output", + lambda *a, **k: [str(tmp_path / "results1.csv")], + ) + + hip_trace_called_with = None + + def mock_hip_trace(wd, fb): + nonlocal hip_trace_called_with + hip_trace_called_with = (wd, fb) + + monkeypatch.setattr("utils.utils.process_hip_trace_output", mock_hip_trace) + + kokkos_trace_called_with = None + + def mock_kokkos_trace(wd, fb): + nonlocal kokkos_trace_called_with + kokkos_trace_called_with = (wd, fb) + + monkeypatch.setattr("utils.utils.process_kokkos_trace_output", mock_kokkos_trace) + + monkeypatch.setattr("utils.utils.console_debug", lambda *a, **k: None) + monkeypatch.setattr("utils.utils.console_warning", lambda *a, **k: None) + monkeypatch.setattr("utils.utils.parse_text", lambda *a, **k: ["C1"]) + + mock_fname_path_obj = mock.Mock(spec=pathlib.Path) + mock_fname_path_obj.stem = fbase_str + mock_fname_path_obj.name = "counters.txt" + mock_fname_path_obj.with_suffix.return_value.exists.return_value = False + + mock_out_path_obj = mock.Mock(spec=pathlib.Path) + mock_out_path_obj.exists.return_value = True + + def path_side_effect(p_arg, *args): + if isinstance(p_arg, pathlib.Path) and p_arg.name == "counters.txt": + return mock_fname_path_obj + if isinstance(p_arg, str) and p_arg.endswith("/out"): + return mock_out_path_obj + if isinstance(p_arg, str) and p_arg.endswith("counters.txt"): + return mock_fname_path_obj + if p_arg == mock_fname_path_obj and args == () and hasattr(p_arg, "with_suffix"): + return mock_fname_path_obj + return mock_fname_path_obj + + monkeypatch.setattr("utils.utils.path", path_side_effect) + + dummy_df = pd.DataFrame({"Dispatch_ID": [0], "A": [1]}) + monkeypatch.setattr("pandas.read_csv", lambda *a, **k: dummy_df.copy()) + monkeypatch.setattr("pandas.DataFrame.to_csv", lambda self, *a, **k: None) + monkeypatch.setattr("shutil.copyfile", lambda *a, **k: None) + monkeypatch.setattr("shutil.rmtree", lambda *a, **k: None) + monkeypatch.setattr("utils.utils.flatten_tcc_info_across_xcds", lambda df, *a: df) + monkeypatch.setattr("utils.utils.mi_gpu_specs.get_num_xcds", lambda *a: 1) + + mspec = MockMSpec() + loglevel = logging.INFO + format_rocprof_output = True + + monkeypatch.setattr("utils.utils.rocprof_cmd", "rocprofiler-sdk") + monkeypatch.setattr("utils.utils.using_v3", lambda: True) + + profiler_options_sdk_hip = {"APP_CMD": "my_app", "ROCPROF_HIP_RUNTIME_API_TRACE": "1"} + hip_trace_called_with = None + kokkos_trace_called_with = None + + utils_mod.run_prof( + fname_str, + profiler_options_sdk_hip.copy(), + workload_dir_str, + mspec, + loglevel, + format_rocprof_output, + ) + assert hip_trace_called_with == (workload_dir_str, fbase_str) + assert kokkos_trace_called_with is None + + monkeypatch.setattr("utils.utils.rocprof_cmd", "rocprof_cli_v3") + + profiler_options_cli_kokkos = ["--kokkos-trace", "--other-opt"] + hip_trace_called_with = None + kokkos_trace_called_with = None + + utils_mod.run_prof( + fname_str, + profiler_options_cli_kokkos, + workload_dir_str, + mspec, + loglevel, + format_rocprof_output, + ) + assert kokkos_trace_called_with == (workload_dir_str, fbase_str) + assert hip_trace_called_with is None + + profiler_options_cli_hip = ["--hip-trace", "--other-opt"] + hip_trace_called_with = None + kokkos_trace_called_with = None + + utils_mod.run_prof( + fname_str, + profiler_options_cli_hip, + workload_dir_str, + mspec, + loglevel, + format_rocprof_output, + ) + assert hip_trace_called_with == (workload_dir_str, fbase_str) + assert kokkos_trace_called_with is None + # ============================================================================= # ROCPROFV3 OUTPUT PROCESSING TESTS # ============================================================================= + def test_process_rocprofv3_output_json_format(tmp_path, monkeypatch): """ Test process_rocprofv3_output with json format converts JSON files to CSV. @@ -3743,11 +3800,14 @@ def test_process_kokkos_trace_output_multiple_files(tmp_path, monkeypatch): assert output_file.exists(), "The primary output file was not created." df = pd.read_csv(output_file) - assert len(df) == 4, "The final DataFrame does not contain the correct number of rows." + assert ( + len(df) == 4 + ), "The final DataFrame does not contain the correct number of rows." assert set(df["timestamp"]) == {1000, 2000, 3000, 4000} assert "kokkos_malloc" in df["marker_name"].values assert "kokkos_parallel_reduce" in df["marker_name"].values + def test_process_kokkos_trace_output_no_files_found(tmp_path, monkeypatch): """ Test process_kokkos_trace_output when no marker API trace files are found. @@ -4164,7 +4224,6 @@ File I/O errors """ - def test_process_hip_trace_output_multiple_files(tmp_path, monkeypatch): """ Test process_hip_trace_output with multiple valid CSV files. @@ -4201,7 +4260,9 @@ def test_process_hip_trace_output_multiple_files(tmp_path, monkeypatch): assert output_file.exists(), "The primary output file was not created." df = pd.read_csv(output_file) - assert len(df) == 4, "The final DataFrame does not contain the correct number of rows." + assert ( + len(df) == 4 + ), "The final DataFrame does not contain the correct number of rows." assert set(df["timestamp"]) == {1000, 2000, 3000, 4000} assert "hipMalloc" in df["api_name"].values assert "hipLaunchKernel" in df["api_name"].values @@ -4209,8 +4270,11 @@ def test_process_hip_trace_output_multiple_files(tmp_path, monkeypatch): copied_file = tmp_path / f"{fbase}_hip_api_trace.csv" assert copied_file.exists(), "The copied output file was not created." df_copy = pd.read_csv(copied_file) - assert df.equals(df_copy), "The copied file content does not match the primary output." - + assert df.equals( + df_copy + ), "The copied file content does not match the primary output." + + def test_process_hip_trace_output_single_file(tmp_path, monkeypatch): """ Test process_hip_trace_output with a single CSV file. @@ -8518,951 +8582,1160 @@ def test_convert_metric_id_to_panel_idx_edge_case_dot_only(): with pytest.raises(ValueError): utils.convert_metric_id_to_panel_idx(".02") - + + # ============================================================================= -# --- New test functions for add_counter_extra_config_input_yaml --- +# --- New test functions for add_counter_extra_config_input_yaml --- # ============================================================================= -def test_add_counter_invalid_architectures_type(): - """ - Test that add_counter_extra_config_input_yaml raises TypeError - if 'architectures' is not a list. - """ - data = {} - with pytest.raises(TypeError, match="'architectures' must be a list, got str"): - utils.add_counter_extra_config_input_yaml( - data=data, - counter_name="test_counter", - description="A test counter", - expression="expr1", - architectures="not_a_list", # Invalid type - properties=["prop1"] - ) - with pytest.raises(TypeError, match="'architectures' must be a list, got int"): - utils.add_counter_extra_config_input_yaml( - data=data, - counter_name="test_counter_2", - description="A test counter 2", - expression="expr2", - architectures=123, # Invalid type - properties=["prop1"] - ) - -def test_add_counter_invalid_properties_type(): - """ - Test that add_counter_extra_config_input_yaml raises TypeError - if 'properties' is not a list (and not None). - """ - data = {} - with pytest.raises(TypeError, match="'properties' must be a list, got str"): - utils.add_counter_extra_config_input_yaml( - data=data, - counter_name="test_counter", - description="A test counter", - expression="expr1", - architectures=["arch1"], - properties="not_a_list" # Invalid type - ) - with pytest.raises(TypeError, match="'properties' must be a list, got dict"): - utils.add_counter_extra_config_input_yaml( - data=data, - counter_name="test_counter_2", - description="A test counter 2", - expression="expr2", - architectures=["arch1"], - properties={"key": "value"} # Invalid type - ) - -def test_add_counter_overwrite_existing(): - """ - Test that add_counter_extra_config_input_yaml overwrites an existing counter - with the same name. - """ - data = {} - counter_name = "MY_COUNTER" - initial_description = "Initial version" - initial_expression = "initial_expr" - initial_architectures = ["gfx900"] - initial_properties = ["P_INIT"] - - # Add the counter for the first time - data = utils.add_counter_extra_config_input_yaml( - data=data, - counter_name=counter_name, - description=initial_description, - expression=initial_expression, - architectures=initial_architectures, - properties=initial_properties - ) - - assert len(data["rocprofiler-sdk"]["counters"]) == 1 - assert data["rocprofiler-sdk"]["counters"][0]["name"] == counter_name - assert data["rocprofiler-sdk"]["counters"][0]["description"] == initial_description - assert data["rocprofiler-sdk"]["counters"][0]["definitions"][0]["expression"] == initial_expression - - updated_description = "Updated version" - updated_expression = "updated_expr" - updated_architectures = ["gfx906", "gfx908"] - updated_properties = ["P_UPDATED", "P_NEW"] + +def test_add_counter_invalid_architectures_type(): + """ + Test that add_counter_extra_config_input_yaml raises TypeError + if 'architectures' is not a list. + """ + data = {} + with pytest.raises(TypeError, match="'architectures' must be a list, got str"): + utils.add_counter_extra_config_input_yaml( + data=data, + counter_name="test_counter", + description="A test counter", + expression="expr1", + architectures="not_a_list", # Invalid type + properties=["prop1"], + ) + with pytest.raises(TypeError, match="'architectures' must be a list, got int"): + utils.add_counter_extra_config_input_yaml( + data=data, + counter_name="test_counter_2", + description="A test counter 2", + expression="expr2", + architectures=123, # Invalid type + properties=["prop1"], + ) + + +def test_add_counter_invalid_properties_type(): + """ + Test that add_counter_extra_config_input_yaml raises TypeError + if 'properties' is not a list (and not None). + """ + data = {} + with pytest.raises(TypeError, match="'properties' must be a list, got str"): + utils.add_counter_extra_config_input_yaml( + data=data, + counter_name="test_counter", + description="A test counter", + expression="expr1", + architectures=["arch1"], + properties="not_a_list", # Invalid type + ) + with pytest.raises(TypeError, match="'properties' must be a list, got dict"): + utils.add_counter_extra_config_input_yaml( + data=data, + counter_name="test_counter_2", + description="A test counter 2", + expression="expr2", + architectures=["arch1"], + properties={"key": "value"}, # Invalid type + ) + + +def test_add_counter_overwrite_existing(): + """ + Test that add_counter_extra_config_input_yaml overwrites an existing counter + with the same name. + """ + data = {} + counter_name = "MY_COUNTER" + initial_description = "Initial version" + initial_expression = "initial_expr" + initial_architectures = ["gfx900"] + initial_properties = ["P_INIT"] + + # Add the counter for the first time + data = utils.add_counter_extra_config_input_yaml( + data=data, + counter_name=counter_name, + description=initial_description, + expression=initial_expression, + architectures=initial_architectures, + properties=initial_properties, + ) + + assert len(data["rocprofiler-sdk"]["counters"]) == 1 + assert data["rocprofiler-sdk"]["counters"][0]["name"] == counter_name + assert data["rocprofiler-sdk"]["counters"][0]["description"] == initial_description + assert ( + data["rocprofiler-sdk"]["counters"][0]["definitions"][0]["expression"] + == initial_expression + ) + + updated_description = "Updated version" + updated_expression = "updated_expr" + updated_architectures = ["gfx906", "gfx908"] + updated_properties = ["P_UPDATED", "P_NEW"] + # ================================================================================= # Test extract counter info extra config input yaml # ================================================================================= -def test_extract_counter_info_returns_none_when_not_found(): - """ - Test that extract_counter_info_extra_config_input_yaml returns None - when the counter is not found or data structure is incomplete. - """ - data_empty = {} - assert utils.extract_counter_info_extra_config_input_yaml(data_empty, "ANY_COUNTER") is None - - data_no_counters_key = {"rocprofiler-sdk": {}} - assert utils.extract_counter_info_extra_config_input_yaml(data_no_counters_key, "ANY_COUNTER") is None - - data_empty_counters_list = {"rocprofiler-sdk": {"counters": []}} - assert utils.extract_counter_info_extra_config_input_yaml(data_empty_counters_list, "ANY_COUNTER") is None - - data_with_other_counters = { - "rocprofiler-sdk": { - "counters": [ - {"name": "EXISTING_COUNTER_1", "value": "val1"}, - {"name": "EXISTING_COUNTER_2", "value": "val2"}, - ] - } - } - assert utils.extract_counter_info_extra_config_input_yaml(data_with_other_counters, "NON_EXISTENT_COUNTER") is None - - data_with_malformed_counter = { - "rocprofiler-sdk": { - "counters": [ - {"value": "val1"}, # No 'name' key - {"name": "EXISTING_COUNTER_2", "value": "val2"}, - ] - } - } - assert utils.extract_counter_info_extra_config_input_yaml(data_with_malformed_counter, "EXISTING_COUNTER_1") is None - assert utils.extract_counter_info_extra_config_input_yaml(data_with_malformed_counter, "EXISTING_COUNTER_2") is not None - - -def test_extract_counter_info_returns_counter_when_found(): - """ - Test that extract_counter_info_extra_config_input_yaml returns the correct - counter dictionary when the counter is found. - """ - counter1_details = {"name": "MY_COUNTER_1", "description": "Desc 1", "expression": "expr1"} - counter2_details = {"name": "MY_COUNTER_2", "description": "Desc 2", "expression": "expr2"} - data = { - "rocprofiler-sdk": { - "counters-schema-version": 1, - "counters": [ - counter1_details, - counter2_details, - ] - } - } - - extracted_counter1 = utils.extract_counter_info_extra_config_input_yaml(data, "MY_COUNTER_1") - assert extracted_counter1 is not None - assert extracted_counter1 == counter1_details - - extracted_counter2 = utils.extract_counter_info_extra_config_input_yaml(data, "MY_COUNTER_2") - assert extracted_counter2 is not None - assert extracted_counter2 == counter2_details + +def test_extract_counter_info_returns_none_when_not_found(): + """ + Test that extract_counter_info_extra_config_input_yaml returns None + when the counter is not found or data structure is incomplete. + """ + data_empty = {} + assert ( + utils.extract_counter_info_extra_config_input_yaml(data_empty, "ANY_COUNTER") + is None + ) + + data_no_counters_key = {"rocprofiler-sdk": {}} + assert ( + utils.extract_counter_info_extra_config_input_yaml( + data_no_counters_key, "ANY_COUNTER" + ) + is None + ) + + data_empty_counters_list = {"rocprofiler-sdk": {"counters": []}} + assert ( + utils.extract_counter_info_extra_config_input_yaml( + data_empty_counters_list, "ANY_COUNTER" + ) + is None + ) + + data_with_other_counters = { + "rocprofiler-sdk": { + "counters": [ + {"name": "EXISTING_COUNTER_1", "value": "val1"}, + {"name": "EXISTING_COUNTER_2", "value": "val2"}, + ] + } + } + assert ( + utils.extract_counter_info_extra_config_input_yaml( + data_with_other_counters, "NON_EXISTENT_COUNTER" + ) + is None + ) + + data_with_malformed_counter = { + "rocprofiler-sdk": { + "counters": [ + {"value": "val1"}, # No 'name' key + {"name": "EXISTING_COUNTER_2", "value": "val2"}, + ] + } + } + assert ( + utils.extract_counter_info_extra_config_input_yaml( + data_with_malformed_counter, "EXISTING_COUNTER_1" + ) + is None + ) + assert ( + utils.extract_counter_info_extra_config_input_yaml( + data_with_malformed_counter, "EXISTING_COUNTER_2" + ) + is not None + ) + + +def test_extract_counter_info_returns_counter_when_found(): + """ + Test that extract_counter_info_extra_config_input_yaml returns the correct + counter dictionary when the counter is found. + """ + counter1_details = { + "name": "MY_COUNTER_1", + "description": "Desc 1", + "expression": "expr1", + } + counter2_details = { + "name": "MY_COUNTER_2", + "description": "Desc 2", + "expression": "expr2", + } + data = { + "rocprofiler-sdk": { + "counters-schema-version": 1, + "counters": [ + counter1_details, + counter2_details, + ], + } + } + + extracted_counter1 = utils.extract_counter_info_extra_config_input_yaml( + data, "MY_COUNTER_1" + ) + assert extracted_counter1 is not None + assert extracted_counter1 == counter1_details + + extracted_counter2 = utils.extract_counter_info_extra_config_input_yaml( + data, "MY_COUNTER_2" + ) + assert extracted_counter2 is not None + assert extracted_counter2 == counter2_details + # ============================================================================= # Test add_counter_from_source_to_target_extra_config_input_yaml valueError cases # ============================================================================= -def test_add_counter_from_source_value_error_counter_not_found(): - """ - Test that add_counter_from_source_to_target_extra_config_input_yaml - raises ValueError if the counter_name is not found in source_data. - """ - source_data_empty = {} - source_data_with_other_counters = { - "rocprofiler-sdk": { - "counters": [ - {"name": "OTHER_COUNTER", "description": "desc", "definitions": [{"architectures": ["gfx900"], "expression": "expr"}]} - ] - } - } - target_data = {} - counter_name_to_find = "MISSING_COUNTER" - - with pytest.raises(ValueError, match=f"Counter '{counter_name_to_find}' not found in source data"): - utils.add_counter_from_source_to_target_extra_config_input_yaml( - source_data_empty, target_data, counter_name_to_find - ) - - with pytest.raises(ValueError, match=f"Counter '{counter_name_to_find}' not found in source data"): - utils.add_counter_from_source_to_target_extra_config_input_yaml( - source_data_with_other_counters, target_data, counter_name_to_find - ) - -def test_add_counter_from_source_value_error_no_definitions(): - """ - Test that add_counter_from_source_to_target_extra_config_input_yaml - raises ValueError if the found counter has no 'definitions'. - """ - counter_name_no_defs = "COUNTER_NO_DEFS" - source_data_no_defs = { - "rocprofiler-sdk": { - "counters": [ - { - "name": counter_name_no_defs, - "description": "A counter without definitions", - "properties": ["prop1"] - } - ] - } - } - source_data_empty_defs_list = { - "rocprofiler-sdk": { - "counters": [ - { - "name": counter_name_no_defs, - "description": "A counter with empty definitions list", - "properties": ["prop1"], - "definitions": [] - } - ] - } - } - target_data = {} - - with pytest.raises(ValueError, match=f"Counter '{counter_name_no_defs}' has no definitions"): - utils.add_counter_from_source_to_target_extra_config_input_yaml( - source_data_no_defs, target_data, counter_name_no_defs - ) - - with pytest.raises(ValueError, match=f"Counter '{counter_name_no_defs}' has no definitions"): - utils.add_counter_from_source_to_target_extra_config_input_yaml( - source_data_empty_defs_list, target_data, counter_name_no_defs - ) - -def test_add_counter_from_source_success(): - """ - Test successful addition of a counter from source to target. - """ - counter_name = "MY_VALID_COUNTER" - source_data = { - "rocprofiler-sdk": { - "counters": [ - { - "name": counter_name, - "description": "Valid Counter Description", - "properties": ["propA", "propB"], - "definitions": [ - { - "architectures": ["gfx900", "gfx906"], - "expression": "SOME_EXPRESSION" - } - ] - } - ] - } - } - target_data_initial = {} - - updated_target_data = utils.add_counter_from_source_to_target_extra_config_input_yaml( - source_data, target_data_initial, counter_name - ) - - assert "rocprofiler-sdk" in updated_target_data - assert "counters" in updated_target_data["rocprofiler-sdk"] - assert len(updated_target_data["rocprofiler-sdk"]["counters"]) == 1 - - added_counter = updated_target_data["rocprofiler-sdk"]["counters"][0] - assert added_counter["name"] == counter_name - assert added_counter["description"] == "Valid Counter Description" - assert added_counter["properties"] == ["propA", "propB"] - assert len(added_counter["definitions"]) == 1 - assert added_counter["definitions"][0]["architectures"] == ["gfx900", "gfx906"] - assert added_counter["definitions"][0]["expression"] == "SOME_EXPRESSION" - - target_data_existing = { - "rocprofiler-sdk": { - "counters-schema-version": 1, - "counters": [ - {"name": "EXISTING_ONE", "description": "desc", "properties": [], "definitions": [{"architectures": [], "expression": ""}]} - ] - } - } - updated_target_data_existing = utils.add_counter_from_source_to_target_extra_config_input_yaml( - source_data, target_data_existing, counter_name - ) - assert len(updated_target_data_existing["rocprofiler-sdk"]["counters"]) == 2 - found_newly_added = False - for c in updated_target_data_existing["rocprofiler-sdk"]["counters"]: - if c["name"] == counter_name: - found_newly_added = True - assert c["description"] == "Valid Counter Description" - break - assert found_newly_added + +def test_add_counter_from_source_value_error_counter_not_found(): + """ + Test that add_counter_from_source_to_target_extra_config_input_yaml + raises ValueError if the counter_name is not found in source_data. + """ + source_data_empty = {} + source_data_with_other_counters = { + "rocprofiler-sdk": { + "counters": [ + { + "name": "OTHER_COUNTER", + "description": "desc", + "definitions": [{"architectures": ["gfx900"], "expression": "expr"}], + } + ] + } + } + target_data = {} + counter_name_to_find = "MISSING_COUNTER" + + with pytest.raises( + ValueError, match=f"Counter '{counter_name_to_find}' not found in source data" + ): + utils.add_counter_from_source_to_target_extra_config_input_yaml( + source_data_empty, target_data, counter_name_to_find + ) + + with pytest.raises( + ValueError, match=f"Counter '{counter_name_to_find}' not found in source data" + ): + utils.add_counter_from_source_to_target_extra_config_input_yaml( + source_data_with_other_counters, target_data, counter_name_to_find + ) -def test_is_spi_pipe_counter_returns_true_when_a_pattern_matches(monkeypatch): - """ - Tests that is_spi_pipe_counter returns True if the counter name - matches at least one regex in spi_pipe_counter_regexs. - """ - sample_regexs = [ - r"SQ_WAVE_CYCLES", - r"TA_DATA_STALL_([A-Z_]+)", - r"TCP_BUSY" - ] - - monkeypatch.setattr(utils, 'spi_pipe_counter_regexs', sample_regexs) - - counter_matches_first = "SQ_WAVE_CYCLES" - assert utils.is_spi_pipe_counter(counter_matches_first) is True, f"Expected True for '{counter_matches_first}'" - - counter_matches_second = "TA_DATA_STALL_SPI_BUSY" - assert utils.is_spi_pipe_counter(counter_matches_second) is True, f"Expected True for '{counter_matches_second}'" - - counter_matches_third = "TCP_BUSY_STATE" # "TCP_BUSY" is a prefix - assert utils.is_spi_pipe_counter(counter_matches_third) is True, f"Expected True for '{counter_matches_third}'" - - non_matching_counter = "SOME_OTHER_COUNTER" - assert utils.is_spi_pipe_counter(non_matching_counter) is False, f"Expected False for '{non_matching_counter}' with the test regexes" +def test_add_counter_from_source_value_error_no_definitions(): + """ + Test that add_counter_from_source_to_target_extra_config_input_yaml + raises ValueError if the found counter has no 'definitions'. + """ + counter_name_no_defs = "COUNTER_NO_DEFS" + source_data_no_defs = { + "rocprofiler-sdk": { + "counters": [ + { + "name": counter_name_no_defs, + "description": "A counter without definitions", + "properties": ["prop1"], + } + ] + } + } + source_data_empty_defs_list = { + "rocprofiler-sdk": { + "counters": [ + { + "name": counter_name_no_defs, + "description": "A counter with empty definitions list", + "properties": ["prop1"], + "definitions": [], + } + ] + } + } + target_data = {} + + with pytest.raises( + ValueError, match=f"Counter '{counter_name_no_defs}' has no definitions" + ): + utils.add_counter_from_source_to_target_extra_config_input_yaml( + source_data_no_defs, target_data, counter_name_no_defs + ) + + with pytest.raises( + ValueError, match=f"Counter '{counter_name_no_defs}' has no definitions" + ): + utils.add_counter_from_source_to_target_extra_config_input_yaml( + source_data_empty_defs_list, target_data, counter_name_no_defs + ) + + +def test_add_counter_from_source_success(): + """ + Test successful addition of a counter from source to target. + """ + counter_name = "MY_VALID_COUNTER" + source_data = { + "rocprofiler-sdk": { + "counters": [ + { + "name": counter_name, + "description": "Valid Counter Description", + "properties": ["propA", "propB"], + "definitions": [ + { + "architectures": ["gfx900", "gfx906"], + "expression": "SOME_EXPRESSION", + } + ], + } + ] + } + } + target_data_initial = {} + + updated_target_data = utils.add_counter_from_source_to_target_extra_config_input_yaml( + source_data, target_data_initial, counter_name + ) + + assert "rocprofiler-sdk" in updated_target_data + assert "counters" in updated_target_data["rocprofiler-sdk"] + assert len(updated_target_data["rocprofiler-sdk"]["counters"]) == 1 + + added_counter = updated_target_data["rocprofiler-sdk"]["counters"][0] + assert added_counter["name"] == counter_name + assert added_counter["description"] == "Valid Counter Description" + assert added_counter["properties"] == ["propA", "propB"] + assert len(added_counter["definitions"]) == 1 + assert added_counter["definitions"][0]["architectures"] == ["gfx900", "gfx906"] + assert added_counter["definitions"][0]["expression"] == "SOME_EXPRESSION" + + target_data_existing = { + "rocprofiler-sdk": { + "counters-schema-version": 1, + "counters": [ + { + "name": "EXISTING_ONE", + "description": "desc", + "properties": [], + "definitions": [{"architectures": [], "expression": ""}], + } + ], + } + } + updated_target_data_existing = ( + utils.add_counter_from_source_to_target_extra_config_input_yaml( + source_data, target_data_existing, counter_name + ) + ) + assert len(updated_target_data_existing["rocprofiler-sdk"]["counters"]) == 2 + found_newly_added = False + for c in updated_target_data_existing["rocprofiler-sdk"]["counters"]: + if c["name"] == counter_name: + found_newly_added = True + assert c["description"] == "Valid Counter Description" + break + assert found_newly_added + + +def test_is_spi_pipe_counter_returns_true_when_a_pattern_matches(monkeypatch): + """ + Tests that is_spi_pipe_counter returns True if the counter name + matches at least one regex in spi_pipe_counter_regexs. + """ + sample_regexs = [r"SQ_WAVE_CYCLES", r"TA_DATA_STALL_([A-Z_]+)", r"TCP_BUSY"] + + monkeypatch.setattr(utils, "spi_pipe_counter_regexs", sample_regexs) + + counter_matches_first = "SQ_WAVE_CYCLES" + assert ( + utils.is_spi_pipe_counter(counter_matches_first) is True + ), f"Expected True for '{counter_matches_first}'" + + counter_matches_second = "TA_DATA_STALL_SPI_BUSY" + assert ( + utils.is_spi_pipe_counter(counter_matches_second) is True + ), f"Expected True for '{counter_matches_second}'" + + counter_matches_third = "TCP_BUSY_STATE" # "TCP_BUSY" is a prefix + assert ( + utils.is_spi_pipe_counter(counter_matches_third) is True + ), f"Expected True for '{counter_matches_third}'" + + non_matching_counter = "SOME_OTHER_COUNTER" + assert ( + utils.is_spi_pipe_counter(non_matching_counter) is False + ), f"Expected False for '{non_matching_counter}' with the test regexes" + # ============================================================================= # test get_base_spi_pipe_counter # ============================================================================= -def test_get_base_spi_counter_match_found_returns_group1(monkeypatch): - """ - Covers: - - for pattern in spi_pipe_counter_regexs: (iterates) - - match = re.match(pattern, counter) (gets a match object) - - if match: (condition is True) - - return match.group(1) (executes and returns) - """ - sample_regexs = [ + +def test_get_base_spi_counter_match_found_returns_group1(monkeypatch): + """ + Covers: + - for pattern in spi_pipe_counter_regexs: (iterates) + - match = re.match(pattern, counter) (gets a match object) + - if match: (condition is True) + - return match.group(1) (executes and returns) + """ + sample_regexs = [ r"UNRELATED_PATTERN_([A-Z]+)", - r"PREFIX_([A-Z0-9_]+)_SUFFIX", - r"ANOTHER_PATTERN_(.*)" - ] - monkeypatch.setattr(utils, 'spi_pipe_counter_regexs', sample_regexs) - - counter_name = "PREFIX_MY_BASE_COUNTER_SUFFIX" - expected_base = "MY_BASE_COUNTER" - - result = utils.get_base_spi_pipe_counter(counter_name) - assert result == expected_base, f"Expected '{expected_base}', got '{result}'" - -def test_get_base_spi_counter_no_match_returns_empty_string(monkeypatch): - """ - Covers: - - for pattern in spi_pipe_counter_regexs: (iterates through all) - - match = re.match(pattern, counter) (match is None for all patterns) - - if match: (condition is always False) - - return "" (executes after loop finishes) - """ - sample_regexs = [ - r"PATTERN_A_([A-Z]+)", - r"PATTERN_B_([0-9]+)" - ] - monkeypatch.setattr(utils, 'spi_pipe_counter_regexs', sample_regexs) - - counter_name = "UNRELATED_COUNTER_NAME" - expected_base = "" - - result = utils.get_base_spi_pipe_counter(counter_name) - assert result == expected_base, f"Expected empty string, got '{result}'" - -def test_get_base_spi_counter_empty_regex_list_returns_empty_string(monkeypatch): - """ - Covers: - - for pattern in spi_pipe_counter_regexs: (loop does not run) - - return "" (executes immediately after non-loop) - """ - monkeypatch.setattr(utils, 'spi_pipe_counter_regexs', []) - - counter_name = "ANY_COUNTER_NAME" - expected_base = "" - - result = utils.get_base_spi_pipe_counter(counter_name) - assert result == expected_base, f"Expected empty string for empty regex list, got '{result}'" - -def test_get_base_spi_counter_match_but_no_group1_raises_indexerror(monkeypatch): - """ - Covers: - - for pattern in spi_pipe_counter_regexs: (iterates) - - match = re.match(pattern, counter) (gets a match object) - - if match: (condition is True) - - return match.group(1) (this line will be attempted and raise IndexError) - This test verifies the behavior of the code as written when a pattern matches - but doesn't have a capturing group 1. - """ - sample_regexs = [ - r"SIMPLE_MATCH_PATTERN" - ] - monkeypatch.setattr(utils, 'spi_pipe_counter_regexs', sample_regexs) - + r"PREFIX_([A-Z0-9_]+)_SUFFIX", + r"ANOTHER_PATTERN_(.*)", + ] + monkeypatch.setattr(utils, "spi_pipe_counter_regexs", sample_regexs) + + counter_name = "PREFIX_MY_BASE_COUNTER_SUFFIX" + expected_base = "MY_BASE_COUNTER" + + result = utils.get_base_spi_pipe_counter(counter_name) + assert result == expected_base, f"Expected '{expected_base}', got '{result}'" + + +def test_get_base_spi_counter_no_match_returns_empty_string(monkeypatch): + """ + Covers: + - for pattern in spi_pipe_counter_regexs: (iterates through all) + - match = re.match(pattern, counter) (match is None for all patterns) + - if match: (condition is always False) + - return "" (executes after loop finishes) + """ + sample_regexs = [r"PATTERN_A_([A-Z]+)", r"PATTERN_B_([0-9]+)"] + monkeypatch.setattr(utils, "spi_pipe_counter_regexs", sample_regexs) + + counter_name = "UNRELATED_COUNTER_NAME" + expected_base = "" + + result = utils.get_base_spi_pipe_counter(counter_name) + assert result == expected_base, f"Expected empty string, got '{result}'" + + +def test_get_base_spi_counter_empty_regex_list_returns_empty_string(monkeypatch): + """ + Covers: + - for pattern in spi_pipe_counter_regexs: (loop does not run) + - return "" (executes immediately after non-loop) + """ + monkeypatch.setattr(utils, "spi_pipe_counter_regexs", []) + + counter_name = "ANY_COUNTER_NAME" + expected_base = "" + + result = utils.get_base_spi_pipe_counter(counter_name) + assert ( + result == expected_base + ), f"Expected empty string for empty regex list, got '{result}'" + + +def test_get_base_spi_counter_match_but_no_group1_raises_indexerror(monkeypatch): + """ + Covers: + - for pattern in spi_pipe_counter_regexs: (iterates) + - match = re.match(pattern, counter) (gets a match object) + - if match: (condition is True) + - return match.group(1) (this line will be attempted and raise IndexError) + This test verifies the behavior of the code as written when a pattern matches + but doesn't have a capturing group 1. + """ + sample_regexs = [r"SIMPLE_MATCH_PATTERN"] + monkeypatch.setattr(utils, "spi_pipe_counter_regexs", sample_regexs) + counter_name = "SIMPLE_MATCH_PATTERN_EXTRA" - - with pytest.raises(IndexError, match="no such group"): - utils.get_base_spi_pipe_counter(counter_name) - -def test_get_base_spi_counter_match_with_group0_only_raises_indexerror(monkeypatch): - """ - Similar to the above, but explicitly tests a regex that produces a match object - where group(0) exists but group(1) does not. - Covers the same lines as test_get_base_spi_counter_match_but_no_group1_raises_indexerror. - """ - sample_regexs = [ - r"MY_WHOLE_MATCH_STRING" - ] - monkeypatch.setattr(utils, 'spi_pipe_counter_regexs', sample_regexs) - - counter_name = "MY_WHOLE_MATCH_STRING" - - with pytest.raises(IndexError, match="no such group"): - utils.get_base_spi_pipe_counter(counter_name) - + + with pytest.raises(IndexError, match="no such group"): + utils.get_base_spi_pipe_counter(counter_name) + + +def test_get_base_spi_counter_match_with_group0_only_raises_indexerror(monkeypatch): + """ + Similar to the above, but explicitly tests a regex that produces a match object + where group(0) exists but group(1) does not. + Covers the same lines as test_get_base_spi_counter_match_but_no_group1_raises_indexerror. + """ + sample_regexs = [r"MY_WHOLE_MATCH_STRING"] + monkeypatch.setattr(utils, "spi_pipe_counter_regexs", sample_regexs) + + counter_name = "MY_WHOLE_MATCH_STRING" + + with pytest.raises(IndexError, match="no such group"): + utils.get_base_spi_pipe_counter(counter_name) + + # ============================================================================= # test using_v1 function # ============================================================================= -def test_using_v1_rocprof_set_and_ends_with_rocprof_returns_true(): - """ - Covers the case where "ROCPROF" is in os.environ and its value ends with "rocprof". - This makes the entire expression True, so the function returns True. - """ - with mock.patch.dict(os.environ, {"ROCPROF": "/opt/rocm/bin/rocprof", "OTHER_VAR": "value"}): + +def test_using_v1_rocprof_set_and_ends_with_rocprof_returns_true(): + """ + Covers the case where "ROCPROF" is in os.environ and its value ends with "rocprof". + This makes the entire expression True, so the function returns True. + """ + with mock.patch.dict( + os.environ, {"ROCPROF": "/opt/rocm/bin/rocprof", "OTHER_VAR": "value"} + ): assert utils.using_v1() is True - -def test_using_v1_rocprof_set_but_not_ends_with_rocprof_returns_false(): - """ - Covers the case where "ROCPROF" is in os.environ, but its value does NOT end with "rocprof". - The second part of the 'and' (os.environ["ROCPROF"].endswith("rocprof")) is False. - So the function returns False. - """ - with mock.patch.dict(os.environ, {"ROCPROF": "/opt/rocm/bin/rocprofv2", "OTHER_VAR": "value"}): + + +def test_using_v1_rocprof_set_but_not_ends_with_rocprof_returns_false(): + """ + Covers the case where "ROCPROF" is in os.environ, but its value does NOT end with "rocprof". + The second part of the 'and' (os.environ["ROCPROF"].endswith("rocprof")) is False. + So the function returns False. + """ + with mock.patch.dict( + os.environ, {"ROCPROF": "/opt/rocm/bin/rocprofv2", "OTHER_VAR": "value"} + ): assert utils.using_v1() is False - - with mock.patch.dict(os.environ, {"ROCPROF": "some/path/to/rocprof_tool", "OTHER_VAR": "value"}): + + with mock.patch.dict( + os.environ, {"ROCPROF": "some/path/to/rocprof_tool", "OTHER_VAR": "value"} + ): assert utils.using_v1() is False - -def test_using_v1_rocprof_not_in_environ_returns_false(): - """ - Covers the case where "ROCPROF" is NOT in os.environ. - The first part of the 'and' ("ROCPROF" in os.environ.keys()) is False. - Due to short-circuiting, the second part is not evaluated. - So the function returns False. - """ - current_env = os.environ.copy() - if "ROCPROF" in current_env: - del current_env["ROCPROF"] - - with mock.patch.dict(os.environ, current_env, clear=True): - assert utils.using_v1() is False - -def test_using_v1_rocprof_is_empty_string_returns_false(): - """ - Covers the case where "ROCPROF" is in os.environ but is an empty string. - The second part (os.environ["ROCPROF"].endswith("rocprof")) will be False. - So the function returns False. - """ - with mock.patch.dict(os.environ, {"ROCPROF": "", "OTHER_VAR": "value"}): - assert utils.using_v1() is False - + + +def test_using_v1_rocprof_not_in_environ_returns_false(): + """ + Covers the case where "ROCPROF" is NOT in os.environ. + The first part of the 'and' ("ROCPROF" in os.environ.keys()) is False. + Due to short-circuiting, the second part is not evaluated. + So the function returns False. + """ + current_env = os.environ.copy() + if "ROCPROF" in current_env: + del current_env["ROCPROF"] + + with mock.patch.dict(os.environ, current_env, clear=True): + assert utils.using_v1() is False + + +def test_using_v1_rocprof_is_empty_string_returns_false(): + """ + Covers the case where "ROCPROF" is in os.environ but is an empty string. + The second part (os.environ["ROCPROF"].endswith("rocprof")) will be False. + So the function returns False. + """ + with mock.patch.dict(os.environ, {"ROCPROF": "", "OTHER_VAR": "value"}): + assert utils.using_v1() is False + + # ============================================================================= # additional test detect_rocprof console error # ============================================================================= -class MockArgs: - def __init__(self, rocprofiler_sdk_library_path): - self.rocprofiler_sdk_library_path = rocprofiler_sdk_library_path - -@mock.patch.dict(os.environ, {"ROCPROF": "rocprofiler-sdk"}, clear=True) -@mock.patch('utils.utils.console_error') -@mock.patch('utils.utils.path') -def test_detect_rocprof_calls_console_error_if_sdk_path_invalid( - mock_path_constructor, mock_console_error_func -): - """ - Tests that detect_rocprof calls console_error when ROCPROF is 'rocprofiler-sdk' - and the rocprofiler_sdk_library_path does not exist. - Focuses on the console_error call. - """ - mock_path_instance = mock.Mock() - mock_path_instance.exists.return_value = False - mock_path_constructor.return_value = mock_path_instance - - fake_library_path = "/some/invalid/path/to/librocprofiler_sdk.so" - args = MockArgs(rocprofiler_sdk_library_path=fake_library_path) - - with mock.patch('utils.utils.console_debug') as mock_console_debug: - utils.detect_rocprof(args) - - expected_error_message = ( - "Could not find rocprofiler-sdk library at " + fake_library_path - ) - mock_console_error_func.assert_called_once_with(expected_error_message) - - mock_path_constructor.assert_called_once_with(fake_library_path) - mock_path_instance.exists.assert_called_once() +class MockArgs: + def __init__(self, rocprofiler_sdk_library_path): + self.rocprofiler_sdk_library_path = rocprofiler_sdk_library_path -class MockArgs: - def __init__(self, **kwargs): - self.__dict__.update(kwargs) - - def __eq__(self, other): - if not isinstance(other, MockArgs): - return NotImplemented - return self.__dict__ == other.__dict__ - -def test_store_app_cmd_sets_global_rocprof_args(): - """ - Tests that store_app_cmd correctly assigns the passed 'args' - object to the global 'rocprof_args'. - """ - sample_args_object = MockArgs( - rocprofiler_sdk_library_path="/path/to/sdk", - input_file="input.txt", - some_other_option=True - ) - - if hasattr(utils, 'rocprof_args'): + +@mock.patch.dict(os.environ, {"ROCPROF": "rocprofiler-sdk"}, clear=True) +@mock.patch("utils.utils.console_error") +@mock.patch("utils.utils.path") +def test_detect_rocprof_calls_console_error_if_sdk_path_invalid( + mock_path_constructor, mock_console_error_func +): + """ + Tests that detect_rocprof calls console_error when ROCPROF is 'rocprofiler-sdk' + and the rocprofiler_sdk_library_path does not exist. + Focuses on the console_error call. + """ + mock_path_instance = mock.Mock() + mock_path_instance.exists.return_value = False + mock_path_constructor.return_value = mock_path_instance + + fake_library_path = "/some/invalid/path/to/librocprofiler_sdk.so" + args = MockArgs(rocprofiler_sdk_library_path=fake_library_path) + + with mock.patch("utils.utils.console_debug") as mock_console_debug: + utils.detect_rocprof(args) + + expected_error_message = ( + "Could not find rocprofiler-sdk library at " + fake_library_path + ) + mock_console_error_func.assert_called_once_with(expected_error_message) + + mock_path_constructor.assert_called_once_with(fake_library_path) + mock_path_instance.exists.assert_called_once() + + +class MockArgs: + def __init__(self, **kwargs): + self.__dict__.update(kwargs) + + def __eq__(self, other): + if not isinstance(other, MockArgs): + return NotImplemented + return self.__dict__ == other.__dict__ + + +def test_store_app_cmd_sets_global_rocprof_args(): + """ + Tests that store_app_cmd correctly assigns the passed 'args' + object to the global 'rocprof_args'. + """ + sample_args_object = MockArgs( + rocprofiler_sdk_library_path="/path/to/sdk", + input_file="input.txt", + some_other_option=True, + ) + + if hasattr(utils, "rocprof_args"): utils.rocprof_args = None - else: - pass - utils.store_app_cmd(sample_args_object) - assert utils.rocprof_args is sample_args_object, "Global rocprof_args should be the same object as the passed args" - - + else: + pass + utils.store_app_cmd(sample_args_object) + assert ( + utils.rocprof_args is sample_args_object + ), "Global rocprof_args should be the same object as the passed args" + + # ============================================================================= # additional tests for v3_counter_csv_to_v2_csv function # ============================================================================= -def create_csv_string(data_dict): - return pd.DataFrame(data_dict).to_csv(index=False) - -@mock.patch('utils.utils.console_error') -@mock.patch('utils.utils.console_debug') -def test_v3_to_v2_agent_id_parsing_success_and_error(mock_console_debug, mock_console_error, tmp_path): - """ - Tests Line 1: Successful parsing of 'Agent Id' string. - Tests Line 2: Error during parsing of 'Agent Id' string, triggering console_error. - """ - agent_info_content = create_csv_string({ - "Node_Id": [0, 1], - "Agent_Type": ["CPU", "GPU"], - "Wave_Front_Size": [0, 64] - }) - agent_info_filepath = tmp_path / "agent_info.csv" - agent_info_filepath.write_text(agent_info_content) - converted_csv_filepath = tmp_path / "converted.csv" - counter_content_success = create_csv_string({ - "Correlation_Id": [1], "Dispatch_Id": [10], "Agent_Id": ["Agent 1"], "Queue_Id": [100], - "Process_Id": [1000], "Thread_Id": [10000], "Grid_Size": [256], "Kernel_Id": [1], - "Kernel_Name": ["kernelA"], "Workgroup_Size": [64], "LDS_Block_Size": [32], - "Scratch_Size": [0], "VGPR_Count": [16], "Accum_VGPR_Count": [0], "SGPR_Count": [32], - "Start_Timestamp": [100000], "End_Timestamp": [100100], - "Counter_Name": ["Cycles"], "Counter_Value": [5000] - }) - counter_filepath_success = tmp_path / "counter_success.csv" - counter_filepath_success.write_text(counter_content_success) - - utils.v3_counter_csv_to_v2_csv(str(counter_filepath_success), str(agent_info_filepath), str(converted_csv_filepath)) - - mock_console_error.assert_not_called() - result_df_success = pd.read_csv(converted_csv_filepath) - assert "GPU_ID" in result_df_success.columns + +def create_csv_string(data_dict): + return pd.DataFrame(data_dict).to_csv(index=False) + + +@mock.patch("utils.utils.console_error") +@mock.patch("utils.utils.console_debug") +def test_v3_to_v2_agent_id_parsing_success_and_error( + mock_console_debug, mock_console_error, tmp_path +): + """ + Tests Line 1: Successful parsing of 'Agent Id' string. + Tests Line 2: Error during parsing of 'Agent Id' string, triggering console_error. + """ + agent_info_content = create_csv_string( + {"Node_Id": [0, 1], "Agent_Type": ["CPU", "GPU"], "Wave_Front_Size": [0, 64]} + ) + agent_info_filepath = tmp_path / "agent_info.csv" + agent_info_filepath.write_text(agent_info_content) + converted_csv_filepath = tmp_path / "converted.csv" + counter_content_success = create_csv_string( + { + "Correlation_Id": [1], + "Dispatch_Id": [10], + "Agent_Id": ["Agent 1"], + "Queue_Id": [100], + "Process_Id": [1000], + "Thread_Id": [10000], + "Grid_Size": [256], + "Kernel_Id": [1], + "Kernel_Name": ["kernelA"], + "Workgroup_Size": [64], + "LDS_Block_Size": [32], + "Scratch_Size": [0], + "VGPR_Count": [16], + "Accum_VGPR_Count": [0], + "SGPR_Count": [32], + "Start_Timestamp": [100000], + "End_Timestamp": [100100], + "Counter_Name": ["Cycles"], + "Counter_Value": [5000], + } + ) + counter_filepath_success = tmp_path / "counter_success.csv" + counter_filepath_success.write_text(counter_content_success) + + utils.v3_counter_csv_to_v2_csv( + str(counter_filepath_success), + str(agent_info_filepath), + str(converted_csv_filepath), + ) + + mock_console_error.assert_not_called() + result_df_success = pd.read_csv(converted_csv_filepath) + assert "GPU_ID" in result_df_success.columns assert result_df_success["GPU_ID"].iloc[0] == 0 - assert result_df_success["GPU_ID"].dtype == "int64" - - mock_console_error.reset_mock() - - counter_content_error = create_csv_string({ - "Correlation_Id": [2], "Dispatch_Id": [20], "Agent_Id": ["Malformed Agent X"], "Queue_Id": [200], - "Process_Id": [2000], "Thread_Id": [20000], "Grid_Size": [512], "Kernel_Id": [2], - "Kernel_Name": ["kernelB"], "Workgroup_Size": [128], "LDS_Block_Size": [64], - "Scratch_Size": [0], "VGPR_Count": [32], "Accum_VGPR_Count": [0], "SGPR_Count": [64], - "Start_Timestamp": [200000], "End_Timestamp": [200200], - "Counter_Name": ["Instructions"], "Counter_Value": [10000] - }) - counter_filepath_error = tmp_path / "counter_error.csv" - counter_filepath_error.write_text(counter_content_error) - - try: - utils.v3_counter_csv_to_v2_csv(str(counter_filepath_error), str(agent_info_filepath), str(converted_csv_filepath)) - except Exception: - pass - - mock_console_error.assert_called_once() - call_args = mock_console_error.call_args[0][0] - assert 'Parsing rocprofv3 csv output: Error of getting "Agent_Id"' in call_args - assert "AttributeError" in call_args or "'NoneType' object has no attribute 'group'" in call_args - -@mock.patch('utils.utils.console_debug') # To suppress debug output -def test_v3_to_v2_accum_column_rename(mock_console_debug, tmp_path): - """ - Tests Line 3: Renaming of a column ending with '_ACCUM' to 'SQ_ACCUM_PREV_HIRES'. - """ - # --- Setup --- - agent_info_content = create_csv_string({ - "Node_Id": [0], "Agent_Type": ["GPU"], "Wave_Front_Size": [64] - }) - agent_info_filepath = tmp_path / "agent_info.csv" - agent_info_filepath.write_text(agent_info_content) - converted_csv_filepath = tmp_path / "converted_accum.csv" - - counter_data = { - "Correlation_Id": [1, 1], - "Dispatch_Id": [10, 10], - "Agent_Id": [0, 0], - "Queue_Id": [100, 100], - "Process_Id": [1000, 1000], - "Thread_Id": [10000, 10000], - "Grid_Size": [256, 256], - "Kernel_Id": [1, 1], - "Kernel_Name": ["kernelA", "kernelA"], - "Workgroup_Size": [64, 64], - "LDS_Block_Size": [32, 32], - "Scratch_Size": [0, 0], - "VGPR_Count": [16, 16], - "Accum_VGPR_Count": [0, 0], - "SGPR_Count": [32, 32], - "Start_Timestamp": [100000, 100000], - "End_Timestamp": [100100, 100100], - "Counter_Name": ["FETCH_SIZE_ACCUM", "CYCLES"], - "Counter_Value": [12345, 5000] - } - counter_content = create_csv_string(counter_data) - counter_filepath = tmp_path / "counter_accum.csv" - counter_filepath.write_text(counter_content) - - utils.v3_counter_csv_to_v2_csv(str(counter_filepath), str(agent_info_filepath), str(converted_csv_filepath)) - - result_df = pd.read_csv(converted_csv_filepath) - assert "SQ_ACCUM_PREV_HIRES" in result_df.columns - assert "FETCH_SIZE_ACCUM" not in result_df.columns - assert "CYCLES" in result_df.columns - assert result_df["SQ_ACCUM_PREV_HIRES"].iloc[0] == 12345 + assert result_df_success["GPU_ID"].dtype == "int64" + + mock_console_error.reset_mock() + + counter_content_error = create_csv_string( + { + "Correlation_Id": [2], + "Dispatch_Id": [20], + "Agent_Id": ["Malformed Agent X"], + "Queue_Id": [200], + "Process_Id": [2000], + "Thread_Id": [20000], + "Grid_Size": [512], + "Kernel_Id": [2], + "Kernel_Name": ["kernelB"], + "Workgroup_Size": [128], + "LDS_Block_Size": [64], + "Scratch_Size": [0], + "VGPR_Count": [32], + "Accum_VGPR_Count": [0], + "SGPR_Count": [64], + "Start_Timestamp": [200000], + "End_Timestamp": [200200], + "Counter_Name": ["Instructions"], + "Counter_Value": [10000], + } + ) + counter_filepath_error = tmp_path / "counter_error.csv" + counter_filepath_error.write_text(counter_content_error) + + try: + utils.v3_counter_csv_to_v2_csv( + str(counter_filepath_error), + str(agent_info_filepath), + str(converted_csv_filepath), + ) + except Exception: + pass + + mock_console_error.assert_called_once() + call_args = mock_console_error.call_args[0][0] + assert 'Parsing rocprofv3 csv output: Error of getting "Agent_Id"' in call_args + assert ( + "AttributeError" in call_args + or "'NoneType' object has no attribute 'group'" in call_args + ) + + +@mock.patch("utils.utils.console_debug") # To suppress debug output +def test_v3_to_v2_accum_column_rename(mock_console_debug, tmp_path): + """ + Tests Line 3: Renaming of a column ending with '_ACCUM' to 'SQ_ACCUM_PREV_HIRES'. + """ + # --- Setup --- + agent_info_content = create_csv_string( + {"Node_Id": [0], "Agent_Type": ["GPU"], "Wave_Front_Size": [64]} + ) + agent_info_filepath = tmp_path / "agent_info.csv" + agent_info_filepath.write_text(agent_info_content) + converted_csv_filepath = tmp_path / "converted_accum.csv" + + counter_data = { + "Correlation_Id": [1, 1], + "Dispatch_Id": [10, 10], + "Agent_Id": [0, 0], + "Queue_Id": [100, 100], + "Process_Id": [1000, 1000], + "Thread_Id": [10000, 10000], + "Grid_Size": [256, 256], + "Kernel_Id": [1, 1], + "Kernel_Name": ["kernelA", "kernelA"], + "Workgroup_Size": [64, 64], + "LDS_Block_Size": [32, 32], + "Scratch_Size": [0, 0], + "VGPR_Count": [16, 16], + "Accum_VGPR_Count": [0, 0], + "SGPR_Count": [32, 32], + "Start_Timestamp": [100000, 100000], + "End_Timestamp": [100100, 100100], + "Counter_Name": ["FETCH_SIZE_ACCUM", "CYCLES"], + "Counter_Value": [12345, 5000], + } + counter_content = create_csv_string(counter_data) + counter_filepath = tmp_path / "counter_accum.csv" + counter_filepath.write_text(counter_content) + + utils.v3_counter_csv_to_v2_csv( + str(counter_filepath), str(agent_info_filepath), str(converted_csv_filepath) + ) + + result_df = pd.read_csv(converted_csv_filepath) + assert "SQ_ACCUM_PREV_HIRES" in result_df.columns + assert "FETCH_SIZE_ACCUM" not in result_df.columns + assert "CYCLES" in result_df.columns + assert result_df["SQ_ACCUM_PREV_HIRES"].iloc[0] == 12345 assert result_df["CYCLES"].iloc[0] == 5000 - -@mock.patch('utils.utils.console_debug') -def test_v3_to_v2_default_accum_vgpr_count(mock_console_debug, tmp_path): - """ - Tests Line 4: 'Accum_VGPR_Count' is added and set to 0 if not present in input. - """ - agent_info_content = create_csv_string({ - "Node_Id": [0], "Agent_Type": ["GPU"], "Wave_Front_Size": [64] - }) - agent_info_filepath = tmp_path / "agent_info.csv" - agent_info_filepath.write_text(agent_info_content) - converted_csv_filepath = tmp_path / "converted_no_accum_vgpr.csv" - - counter_content = create_csv_string({ - "Correlation_Id": [1], "Dispatch_Id": [10], "Agent_Id": [0], "Queue_Id": [100], - "Process_Id": [1000], "Thread_Id": [10000], "Grid_Size": [256], "Kernel_Id": [1], - "Kernel_Name": ["kernelA"], "Workgroup_Size": [64], "LDS_Block_Size": [32], - "Scratch_Size": [0], "VGPR_Count": [16], - "SGPR_Count": [32], "Start_Timestamp": [100000], "End_Timestamp": [100100], - "Counter_Name": ["Cycles"], "Counter_Value": [5000] - }) - counter_filepath = tmp_path / "counter_no_accum_vgpr.csv" - counter_filepath.write_text(counter_content) - - utils.v3_counter_csv_to_v2_csv(str(counter_filepath), str(agent_info_filepath), str(converted_csv_filepath)) - - result_df = pd.read_csv(converted_csv_filepath) - assert "Accum_VGPR" in result_df.columns - assert result_df["Accum_VGPR"].iloc[0] == 0 - assert result_df["Accum_VGPR"].dtype == "int64" + + +@mock.patch("utils.utils.console_debug") +def test_v3_to_v2_default_accum_vgpr_count(mock_console_debug, tmp_path): + """ + Tests Line 4: 'Accum_VGPR_Count' is added and set to 0 if not present in input. + """ + agent_info_content = create_csv_string( + {"Node_Id": [0], "Agent_Type": ["GPU"], "Wave_Front_Size": [64]} + ) + agent_info_filepath = tmp_path / "agent_info.csv" + agent_info_filepath.write_text(agent_info_content) + converted_csv_filepath = tmp_path / "converted_no_accum_vgpr.csv" + + counter_content = create_csv_string( + { + "Correlation_Id": [1], + "Dispatch_Id": [10], + "Agent_Id": [0], + "Queue_Id": [100], + "Process_Id": [1000], + "Thread_Id": [10000], + "Grid_Size": [256], + "Kernel_Id": [1], + "Kernel_Name": ["kernelA"], + "Workgroup_Size": [64], + "LDS_Block_Size": [32], + "Scratch_Size": [0], + "VGPR_Count": [16], + "SGPR_Count": [32], + "Start_Timestamp": [100000], + "End_Timestamp": [100100], + "Counter_Name": ["Cycles"], + "Counter_Value": [5000], + } + ) + counter_filepath = tmp_path / "counter_no_accum_vgpr.csv" + counter_filepath.write_text(counter_content) + + utils.v3_counter_csv_to_v2_csv( + str(counter_filepath), str(agent_info_filepath), str(converted_csv_filepath) + ) + + result_df = pd.read_csv(converted_csv_filepath) + assert "Accum_VGPR" in result_df.columns + assert result_df["Accum_VGPR"].iloc[0] == 0 + assert result_df["Accum_VGPR"].dtype == "int64" # =================================================================== # Test PC_sampling function # =================================================================== -@mock.patch('utils.utils.capture_subprocess_output') -@mock.patch('utils.utils.console_error') -@mock.patch('utils.utils.console_debug') -def test_pc_sampling_prof_sdk_path_nonexistent_librocprofiler_sdk_tool( - mock_console_debug, mock_console_error, mock_capture_subprocess, tmp_path -): - """ - Edge Case: rocprofiler_sdk_library_path is valid, but librocprofiler-sdk-tool.so - is NOT found next to it (or in rocprofiler-sdk subdir). - This test primarily checks if the paths are constructed. The actual check for - file existence before `capture_subprocess_output` is not in the provided snippet, - but we test the path construction. + +@mock.patch("utils.utils.capture_subprocess_output") +@mock.patch("utils.utils.console_error") +@mock.patch("utils.utils.console_debug") +def test_pc_sampling_prof_sdk_path_nonexistent_librocprofiler_sdk_tool( + mock_console_debug, mock_console_error, mock_capture_subprocess, tmp_path +): """ - with mock.patch('utils.utils.rocprof_cmd', "rocprofiler-sdk"): - method = "host_trap" - interval = 1000 - workload_dir = str(tmp_path) - appcmd = "my_app --arg" - - sdk_lib_dir = tmp_path / "rocm_sdk" / "lib" - sdk_lib_dir.mkdir(parents=True, exist_ok=True) - rocprofiler_sdk_library_path = str(sdk_lib_dir / "librocprofiler_sdk.so") + Edge Case: rocprofiler_sdk_library_path is valid, but librocprofiler-sdk-tool.so + is NOT found next to it (or in rocprofiler-sdk subdir). + This test primarily checks if the paths are constructed. The actual check for + file existence before `capture_subprocess_output` is not in the provided snippet, + but we test the path construction. + """ + with mock.patch("utils.utils.rocprof_cmd", "rocprofiler-sdk"): + method = "host_trap" + interval = 1000 + workload_dir = str(tmp_path) + appcmd = "my_app --arg" + + sdk_lib_dir = tmp_path / "rocm_sdk" / "lib" + sdk_lib_dir.mkdir(parents=True, exist_ok=True) + rocprofiler_sdk_library_path = str(sdk_lib_dir / "librocprofiler_sdk.so") pathlib.Path(rocprofiler_sdk_library_path).touch() - - expected_tool_path = str(sdk_lib_dir / "rocprofiler-sdk" / "librocprofiler-sdk-tool.so") - + + expected_tool_path = str( + sdk_lib_dir / "rocprofiler-sdk" / "librocprofiler-sdk-tool.so" + ) + mock_capture_subprocess.return_value = (True, "Success output") - - utils.pc_sampling_prof(method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path) - - assert mock_capture_subprocess.called - call_args = mock_capture_subprocess.call_args - called_env = call_args.kwargs.get('new_env', {}) - - assert "LD_PRELOAD" in called_env - ld_preload_paths = called_env["LD_PRELOAD"].split(':') - assert expected_tool_path in ld_preload_paths - assert rocprofiler_sdk_library_path in ld_preload_paths - - mock_console_error.assert_not_called() - - -@mock.patch('utils.utils.capture_subprocess_output') -@mock.patch('utils.utils.console_error') -@mock.patch('utils.utils.console_debug') -def test_pc_sampling_prof_subprocess_fails( - mock_console_debug, mock_console_error, mock_capture_subprocess, tmp_path -): - """ - Edge Case: The capture_subprocess_output returns success=False. - This should trigger the console_error("PC sampling failed."). - """ - with mock.patch('utils.utils.rocprof_cmd', "rocprof_cli_tool"): - method = "stochastic" - interval = 5000 - workload_dir = str(tmp_path) - appcmd = "another_app" - rocprofiler_sdk_library_path = "/some/path/librocprofiler_sdk.so" - - mock_capture_subprocess.return_value = (False, "Error output from subprocess") - - utils.pc_sampling_prof(method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path) - - mock_capture_subprocess.assert_called_once() - mock_console_error.assert_called_once_with("PC sampling failed.") - - mock_capture_subprocess.reset_mock() - mock_console_error.reset_mock() - with mock.patch('utils.utils.rocprof_cmd', "rocprofiler-sdk"): - sdk_lib_dir = tmp_path / "rocm_sdk_fail" / "lib" - sdk_lib_dir.mkdir(parents=True, exist_ok=True) - rocprofiler_sdk_library_path_sdk = str(sdk_lib_dir / "librocprofiler_sdk.so") - pathlib.Path(rocprofiler_sdk_library_path_sdk).touch() - - tool_dir = sdk_lib_dir / "rocprofiler-sdk" - tool_dir.mkdir(parents=True, exist_ok=True) - (tool_dir / "librocprofiler-sdk-tool.so").touch() - - mock_capture_subprocess.return_value = (False, "Error output from SDK subprocess") - - utils.pc_sampling_prof(method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path_sdk) - - mock_capture_subprocess.assert_called_once() - mock_console_error.assert_called_once_with("PC sampling failed.") - - -@mock.patch('utils.utils.capture_subprocess_output') -@mock.patch('utils.utils.console_error') -@mock.patch('utils.utils.console_debug') -def test_pc_sampling_prof_empty_appcmd( - mock_console_debug, mock_console_error, mock_capture_subprocess, tmp_path -): - """ - Edge Case: The appcmd is an empty string. - The function should still attempt to run it. The behavior of - capture_subprocess_output with an empty command is external to this function. - """ - with mock.patch('utils.utils.rocprof_cmd', "rocprof_cli_tool"): - method = "host_trap" - interval = 100 - workload_dir = str(tmp_path) + + utils.pc_sampling_prof( + method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path + ) + + assert mock_capture_subprocess.called + call_args = mock_capture_subprocess.call_args + called_env = call_args.kwargs.get("new_env", {}) + + assert "LD_PRELOAD" in called_env + ld_preload_paths = called_env["LD_PRELOAD"].split(":") + assert expected_tool_path in ld_preload_paths + assert rocprofiler_sdk_library_path in ld_preload_paths + + mock_console_error.assert_not_called() + + +@mock.patch("utils.utils.capture_subprocess_output") +@mock.patch("utils.utils.console_error") +@mock.patch("utils.utils.console_debug") +def test_pc_sampling_prof_subprocess_fails( + mock_console_debug, mock_console_error, mock_capture_subprocess, tmp_path +): + """ + Edge Case: The capture_subprocess_output returns success=False. + This should trigger the console_error("PC sampling failed."). + """ + with mock.patch("utils.utils.rocprof_cmd", "rocprof_cli_tool"): + method = "stochastic" + interval = 5000 + workload_dir = str(tmp_path) + appcmd = "another_app" + rocprofiler_sdk_library_path = "/some/path/librocprofiler_sdk.so" + + mock_capture_subprocess.return_value = (False, "Error output from subprocess") + + utils.pc_sampling_prof( + method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path + ) + + mock_capture_subprocess.assert_called_once() + mock_console_error.assert_called_once_with("PC sampling failed.") + + mock_capture_subprocess.reset_mock() + mock_console_error.reset_mock() + with mock.patch("utils.utils.rocprof_cmd", "rocprofiler-sdk"): + sdk_lib_dir = tmp_path / "rocm_sdk_fail" / "lib" + sdk_lib_dir.mkdir(parents=True, exist_ok=True) + rocprofiler_sdk_library_path_sdk = str(sdk_lib_dir / "librocprofiler_sdk.so") + pathlib.Path(rocprofiler_sdk_library_path_sdk).touch() + + tool_dir = sdk_lib_dir / "rocprofiler-sdk" + tool_dir.mkdir(parents=True, exist_ok=True) + (tool_dir / "librocprofiler-sdk-tool.so").touch() + + mock_capture_subprocess.return_value = (False, "Error output from SDK subprocess") + + utils.pc_sampling_prof( + method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path_sdk + ) + + mock_capture_subprocess.assert_called_once() + mock_console_error.assert_called_once_with("PC sampling failed.") + + +@mock.patch("utils.utils.capture_subprocess_output") +@mock.patch("utils.utils.console_error") +@mock.patch("utils.utils.console_debug") +def test_pc_sampling_prof_empty_appcmd( + mock_console_debug, mock_console_error, mock_capture_subprocess, tmp_path +): + """ + Edge Case: The appcmd is an empty string. + The function should still attempt to run it. The behavior of + capture_subprocess_output with an empty command is external to this function. + """ + with mock.patch("utils.utils.rocprof_cmd", "rocprof_cli_tool"): + method = "host_trap" + interval = 100 + workload_dir = str(tmp_path) appcmd = "" - rocprofiler_sdk_library_path = "/some/path/librocprofiler_sdk.so" - - mock_capture_subprocess.return_value = (True, "Output with empty appcmd") - - utils.pc_sampling_prof(method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path) - - assert mock_capture_subprocess.called - options_list = mock_capture_subprocess.call_args[0][0] + rocprofiler_sdk_library_path = "/some/path/librocprofiler_sdk.so" + + mock_capture_subprocess.return_value = (True, "Output with empty appcmd") + + utils.pc_sampling_prof( + method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path + ) + + assert mock_capture_subprocess.called + options_list = mock_capture_subprocess.call_args[0][0] assert options_list[-1] == "" - mock_console_error.assert_not_called() - - mock_capture_subprocess.reset_mock() - mock_console_error.reset_mock() - with mock.patch('utils.utils.rocprof_cmd', "rocprofiler-sdk"): - sdk_lib_dir = tmp_path / "rocm_sdk_empty" / "lib" - sdk_lib_dir.mkdir(parents=True, exist_ok=True) - rocprofiler_sdk_library_path_sdk = str(sdk_lib_dir / "librocprofiler_sdk.so") - pathlib.Path(rocprofiler_sdk_library_path_sdk).touch() - tool_dir = sdk_lib_dir / "rocprofiler-sdk" - tool_dir.mkdir(parents=True, exist_ok=True) - (tool_dir / "librocprofiler-sdk-tool.so").touch() - - mock_capture_subprocess.return_value = (True, "Output with empty appcmd SDK") - - utils.pc_sampling_prof(method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path_sdk) - - assert mock_capture_subprocess.called - assert mock_capture_subprocess.call_args[0][0] == "" - mock_console_error.assert_not_called() - + mock_console_error.assert_not_called() + + mock_capture_subprocess.reset_mock() + mock_console_error.reset_mock() + with mock.patch("utils.utils.rocprof_cmd", "rocprofiler-sdk"): + sdk_lib_dir = tmp_path / "rocm_sdk_empty" / "lib" + sdk_lib_dir.mkdir(parents=True, exist_ok=True) + rocprofiler_sdk_library_path_sdk = str(sdk_lib_dir / "librocprofiler_sdk.so") + pathlib.Path(rocprofiler_sdk_library_path_sdk).touch() + tool_dir = sdk_lib_dir / "rocprofiler-sdk" + tool_dir.mkdir(parents=True, exist_ok=True) + (tool_dir / "librocprofiler-sdk-tool.so").touch() + + mock_capture_subprocess.return_value = (True, "Output with empty appcmd SDK") + + utils.pc_sampling_prof( + method, interval, workload_dir, appcmd, rocprofiler_sdk_library_path_sdk + ) + + assert mock_capture_subprocess.called + assert mock_capture_subprocess.call_args[0][0] == "" + mock_console_error.assert_not_called() + # ============================================================================= # test replace_timestamps function # ============================================================================= -def create_dummy_csv(filepath, data_dict): - df = pd.DataFrame(data_dict) - df.to_csv(filepath, index=False) - -@mock.patch('utils.utils.console_warning') -@mock.patch('utils.utils.path') -def test_replace_timestamps_no_timestamps_csv_returns_early(mock_path_util, mock_console_warning, tmp_path): - """ - Edge Case: timestamps.csv does not exist in workload_dir. - The function should return early. - Covers: if not path(workload_dir, "timestamps.csv").is_file(): return - """ - workload_dir = str(tmp_path) - - mock_timestamps_path_obj = mock.Mock() + +def create_dummy_csv(filepath, data_dict): + df = pd.DataFrame(data_dict) + df.to_csv(filepath, index=False) + + +@mock.patch("utils.utils.console_warning") +@mock.patch("utils.utils.path") +def test_replace_timestamps_no_timestamps_csv_returns_early( + mock_path_util, mock_console_warning, tmp_path +): + """ + Edge Case: timestamps.csv does not exist in workload_dir. + The function should return early. + Covers: if not path(workload_dir, "timestamps.csv").is_file(): return + """ + workload_dir = str(tmp_path) + + mock_timestamps_path_obj = mock.Mock() mock_timestamps_path_obj.is_file.return_value = False - - mock_path_util.side_effect = lambda *args: mock_timestamps_path_obj if args[1] == "timestamps.csv" else mock.DEFAULT - - utils.replace_timestamps(workload_dir) - - mock_path_util.assert_any_call(workload_dir, "timestamps.csv") - mock_timestamps_path_obj.is_file.assert_called_once() - mock_console_warning.assert_not_called() - -@mock.patch('utils.utils.console_warning') -@mock.patch('glob.glob') # Mock glob.glob -@mock.patch('utils.utils.path') -def test_replace_timestamps_timestamps_csv_missing_columns_warns(mock_path_util, mock_glob, mock_console_warning, tmp_path): - """ - Edge Case: timestamps.csv exists but is missing 'Start_Timestamp' or 'End_Timestamp'. - The function should call console_warning. - Covers: else: console_warning(...) - """ - workload_dir = str(tmp_path) - timestamps_csv_path_str = os.path.join(workload_dir, "timestamps.csv") - - create_dummy_csv(timestamps_csv_path_str, {"Some_Other_Column": [123]}) - - mock_timestamps_path_obj = mock.Mock() - mock_timestamps_path_obj.is_file.return_value = True + + mock_path_util.side_effect = lambda *args: ( + mock_timestamps_path_obj if args[1] == "timestamps.csv" else mock.DEFAULT + ) + + utils.replace_timestamps(workload_dir) + + mock_path_util.assert_any_call(workload_dir, "timestamps.csv") + mock_timestamps_path_obj.is_file.assert_called_once() + mock_console_warning.assert_not_called() + + +@mock.patch("utils.utils.console_warning") +@mock.patch("glob.glob") # Mock glob.glob +@mock.patch("utils.utils.path") +def test_replace_timestamps_timestamps_csv_missing_columns_warns( + mock_path_util, mock_glob, mock_console_warning, tmp_path +): + """ + Edge Case: timestamps.csv exists but is missing 'Start_Timestamp' or 'End_Timestamp'. + The function should call console_warning. + Covers: else: console_warning(...) + """ + workload_dir = str(tmp_path) + timestamps_csv_path_str = os.path.join(workload_dir, "timestamps.csv") + + create_dummy_csv(timestamps_csv_path_str, {"Some_Other_Column": [123]}) + + mock_timestamps_path_obj = mock.Mock() + mock_timestamps_path_obj.is_file.return_value = True mock_timestamps_path_obj.name = "timestamps.csv" - - mock_path_util.side_effect = lambda *args, **kwargs: mock_timestamps_path_obj if args[-1] == "timestamps.csv" else mock.DEFAULT - - utils.replace_timestamps(workload_dir) - - mock_path_util.assert_any_call(workload_dir, "timestamps.csv") - mock_timestamps_path_obj.is_file.assert_called_once() - mock_console_warning.assert_called_once_with( - "Incomplete profiling data detected. Unable to update timestamps.\n" - ) + + mock_path_util.side_effect = lambda *args, **kwargs: ( + mock_timestamps_path_obj if args[-1] == "timestamps.csv" else mock.DEFAULT + ) + + utils.replace_timestamps(workload_dir) + + mock_path_util.assert_any_call(workload_dir, "timestamps.csv") + mock_timestamps_path_obj.is_file.assert_called_once() + mock_console_warning.assert_called_once_with( + "Incomplete profiling data detected. Unable to update timestamps.\n" + ) mock_glob.assert_not_called() - -@mock.patch('utils.utils.console_warning') -@mock.patch('glob.glob') -@mock.patch('utils.utils.path') -def test_replace_timestamps_updates_other_csvs_skips_sysinfo(mock_path_util, mock_glob, mock_console_warning, tmp_path): - """ - Edge Case: timestamps.csv is valid. Other CSVs exist, including sysinfo.csv. - Only non-sysinfo.csv files should be updated. - Covers: for fname in glob.glob(...): if path(fname).name != "sysinfo.csv": ... - """ - workload_dir = str(tmp_path) - timestamps_csv_path_str = os.path.join(workload_dir, "timestamps.csv") - data_csv_path_str = os.path.join(workload_dir, "data.csv") - sysinfo_csv_path_str = os.path.join(workload_dir, "sysinfo.csv") - - new_start_ts = [1000, 2000] - new_end_ts = [1500, 2500] - create_dummy_csv(timestamps_csv_path_str, {"Start_Timestamp": new_start_ts, "End_Timestamp": new_end_ts}) - - create_dummy_csv(data_csv_path_str, {"Kernel_Name": ["A", "B"], "Start_Timestamp": [1, 2], "End_Timestamp": [3, 4]}) - create_dummy_csv(sysinfo_csv_path_str, {"Info": ["CPU", "MEM"], "Start_Timestamp": [5, 6], "End_Timestamp": [7, 8]}) - - def path_side_effect(*args, **kwargs): - p_obj = mock.Mock() - full_path = args[0] if len(args) == 1 else os.path.join(args[0], args[1]) - - if full_path == timestamps_csv_path_str: - p_obj.is_file.return_value = True - p_obj.name = "timestamps.csv" - elif full_path == data_csv_path_str: + + +@mock.patch("utils.utils.console_warning") +@mock.patch("glob.glob") +@mock.patch("utils.utils.path") +def test_replace_timestamps_updates_other_csvs_skips_sysinfo( + mock_path_util, mock_glob, mock_console_warning, tmp_path +): + """ + Edge Case: timestamps.csv is valid. Other CSVs exist, including sysinfo.csv. + Only non-sysinfo.csv files should be updated. + Covers: for fname in glob.glob(...): if path(fname).name != "sysinfo.csv": ... + """ + workload_dir = str(tmp_path) + timestamps_csv_path_str = os.path.join(workload_dir, "timestamps.csv") + data_csv_path_str = os.path.join(workload_dir, "data.csv") + sysinfo_csv_path_str = os.path.join(workload_dir, "sysinfo.csv") + + new_start_ts = [1000, 2000] + new_end_ts = [1500, 2500] + create_dummy_csv( + timestamps_csv_path_str, + {"Start_Timestamp": new_start_ts, "End_Timestamp": new_end_ts}, + ) + + create_dummy_csv( + data_csv_path_str, + {"Kernel_Name": ["A", "B"], "Start_Timestamp": [1, 2], "End_Timestamp": [3, 4]}, + ) + create_dummy_csv( + sysinfo_csv_path_str, + {"Info": ["CPU", "MEM"], "Start_Timestamp": [5, 6], "End_Timestamp": [7, 8]}, + ) + + def path_side_effect(*args, **kwargs): + p_obj = mock.Mock() + full_path = args[0] if len(args) == 1 else os.path.join(args[0], args[1]) + + if full_path == timestamps_csv_path_str: p_obj.is_file.return_value = True - p_obj.name = "data.csv" - elif full_path == sysinfo_csv_path_str: + p_obj.name = "timestamps.csv" + elif full_path == data_csv_path_str: p_obj.is_file.return_value = True - p_obj.name = "sysinfo.csv" + p_obj.name = "data.csv" + elif full_path == sysinfo_csv_path_str: + p_obj.is_file.return_value = True + p_obj.name = "sysinfo.csv" else: - p_obj.is_file.return_value = False - p_obj.name = os.path.basename(full_path) - return p_obj - - mock_path_util.side_effect = path_side_effect - - mock_glob.return_value = [data_csv_path_str, sysinfo_csv_path_str, timestamps_csv_path_str] - - utils.replace_timestamps(workload_dir) - - mock_console_warning.assert_not_called() - - df_data_updated = pd.read_csv(data_csv_path_str) - pd.testing.assert_series_equal(df_data_updated["Start_Timestamp"], pd.Series(new_start_ts, name="Start_Timestamp")) - pd.testing.assert_series_equal(df_data_updated["End_Timestamp"], pd.Series(new_end_ts, name="End_Timestamp")) - - df_sysinfo_original = pd.read_csv(sysinfo_csv_path_str) + p_obj.is_file.return_value = False + p_obj.name = os.path.basename(full_path) + return p_obj + + mock_path_util.side_effect = path_side_effect + + mock_glob.return_value = [ + data_csv_path_str, + sysinfo_csv_path_str, + timestamps_csv_path_str, + ] + + utils.replace_timestamps(workload_dir) + + mock_console_warning.assert_not_called() + + df_data_updated = pd.read_csv(data_csv_path_str) + pd.testing.assert_series_equal( + df_data_updated["Start_Timestamp"], + pd.Series(new_start_ts, name="Start_Timestamp"), + ) + pd.testing.assert_series_equal( + df_data_updated["End_Timestamp"], pd.Series(new_end_ts, name="End_Timestamp") + ) + + df_sysinfo_original = pd.read_csv(sysinfo_csv_path_str) assert list(df_sysinfo_original["Start_Timestamp"]) == [5, 6] assert list(df_sysinfo_original["End_Timestamp"]) == [7, 8] - -@mock.patch('utils.utils.console_warning') -@mock.patch('glob.glob') -@mock.patch('utils.utils.path') -def test_replace_timestamps_no_other_csvs_to_update(mock_path_util, mock_glob, mock_console_warning, tmp_path): - """ - Edge Case: timestamps.csv is valid, but no other *.csv files (or only sysinfo.csv) exist. - The loop for updating files should not do anything or not run. - Covers: The for loop not iterating if glob returns empty or only sysinfo. - """ - workload_dir = str(tmp_path) - timestamps_csv_path_str = os.path.join(workload_dir, "timestamps.csv") - sysinfo_csv_path_str = os.path.join(workload_dir, "sysinfo.csv") - - create_dummy_csv(timestamps_csv_path_str, {"Start_Timestamp": [100], "End_Timestamp": [200]}) - create_dummy_csv(sysinfo_csv_path_str, {"Info": ["CPU"], "Start_Timestamp": [5], "End_Timestamp": [7]}) - - - def path_side_effect(*args, **kwargs): - p_obj = mock.Mock() - full_path = args[0] if len(args) == 1 else os.path.join(args[0], args[1]) - if full_path == timestamps_csv_path_str: - p_obj.is_file.return_value = True - p_obj.name = "timestamps.csv" - elif full_path == sysinfo_csv_path_str: - p_obj.is_file.return_value = True - p_obj.name = "sysinfo.csv" - else: - p_obj.is_file.return_value = False - p_obj.name = os.path.basename(full_path) - return p_obj - mock_path_util.side_effect = path_side_effect - - mock_glob.return_value = [timestamps_csv_path_str, sysinfo_csv_path_str] - - utils.replace_timestamps(workload_dir) - - mock_console_warning.assert_not_called() - df_sysinfo_original = pd.read_csv(sysinfo_csv_path_str) - assert list(df_sysinfo_original["Start_Timestamp"]) == [5] - assert list(df_sysinfo_original["End_Timestamp"]) == [7] + +@mock.patch("utils.utils.console_warning") +@mock.patch("glob.glob") +@mock.patch("utils.utils.path") +def test_replace_timestamps_no_other_csvs_to_update( + mock_path_util, mock_glob, mock_console_warning, tmp_path +): + """ + Edge Case: timestamps.csv is valid, but no other *.csv files (or only sysinfo.csv) exist. + The loop for updating files should not do anything or not run. + Covers: The for loop not iterating if glob returns empty or only sysinfo. + """ + workload_dir = str(tmp_path) + timestamps_csv_path_str = os.path.join(workload_dir, "timestamps.csv") + sysinfo_csv_path_str = os.path.join(workload_dir, "sysinfo.csv") + + create_dummy_csv( + timestamps_csv_path_str, {"Start_Timestamp": [100], "End_Timestamp": [200]} + ) + create_dummy_csv( + sysinfo_csv_path_str, + {"Info": ["CPU"], "Start_Timestamp": [5], "End_Timestamp": [7]}, + ) + + def path_side_effect(*args, **kwargs): + p_obj = mock.Mock() + full_path = args[0] if len(args) == 1 else os.path.join(args[0], args[1]) + if full_path == timestamps_csv_path_str: + p_obj.is_file.return_value = True + p_obj.name = "timestamps.csv" + elif full_path == sysinfo_csv_path_str: + p_obj.is_file.return_value = True + p_obj.name = "sysinfo.csv" + else: + p_obj.is_file.return_value = False + p_obj.name = os.path.basename(full_path) + return p_obj + + mock_path_util.side_effect = path_side_effect + + mock_glob.return_value = [timestamps_csv_path_str, sysinfo_csv_path_str] + + utils.replace_timestamps(workload_dir) + + mock_console_warning.assert_not_called() + df_sysinfo_original = pd.read_csv(sysinfo_csv_path_str) + assert list(df_sysinfo_original["Start_Timestamp"]) == [5] + assert list(df_sysinfo_original["End_Timestamp"]) == [7]