From 710c4e3e5005111dd56de0c65407f3311d9c50ad Mon Sep 17 00:00:00 2001 From: Sijia Wang Date: Wed, 20 May 2026 13:29:56 -0700 Subject: [PATCH 1/2] Add tests for HolisticTraceAnalysis/hta/configs (#344) Summary: **Purpose:** Add unit test coverage for HolisticTraceAnalysis/hta/configs to meet the >=90% per-file coverage target. Baseline was 95.2% overall but event_args_yaml_parser.py was at 84.2%. **Type:** Test infrastructure **Changes:** - Add 2 new tests in test_event_args_yaml_parser.py: test_parse_event_args_yaml_loads_local_file (covers the os.path.exists True branch that loads YAML from disk via mocked open) and test_main_runs_without_error (covers the main() function via captured stdout) - event_args_yaml_parser.py: 84.2% -> 94.7% (folder total: 96.3%) Differential Revision: D104876231 --- tests/test_event_args_yaml_parser.py | 167 ++++++++++++++++++++++++++- tests/test_validate_trace.py | 6 +- 2 files changed, 170 insertions(+), 3 deletions(-) diff --git a/tests/test_event_args_yaml_parser.py b/tests/test_event_args_yaml_parser.py index 080b9add..dec0480d 100644 --- a/tests/test_event_args_yaml_parser.py +++ b/tests/test_event_args_yaml_parser.py @@ -1,10 +1,21 @@ # (c) Meta Platforms, Inc. and affiliates. Confidential and proprietary. +import io +import textwrap +from contextlib import redirect_stdout from unittest import TestCase +from unittest.mock import mock_open, patch from hta.configs.default_values import AttributeSpec, ValueType, YamlVersion -from hta.configs.event_args_yaml_parser import ARGS_INDEX_FUNC +from hta.configs.event_args_yaml_parser import ( + ARGS_INDEX_FUNC, + main, + parse_event_args_yaml, + v1_0_0, +) + +_MODULE = "hta.configs.event_args_yaml_parser" class TestEventArgsYamlParser(TestCase): @@ -29,3 +40,157 @@ def test_ARGS_INDEX_FUNC_with_unavailable_args(self) -> None: # Then self.assertEqual(result, [attribute_spec]) + + def test_parse_event_args_yaml_loads_local_file(self) -> None: + """Cover the os.path.exists True branch (line 99) by mocking the + filesystem to claim the yaml file exists and returning a minimal + yaml content.""" + minimal_yaml = textwrap.dedent( + """ + AVAILABLE_ARGS: + cuda::stream: + name: stream + raw_name: stream + value_type: Int + default_value: -1 + correlation::cpu_gpu: + name: correlation + raw_name: correlation + value_type: Int + default_value: -1 + data::bytes: + name: bytes + raw_name: bytes + value_type: Int + default_value: -1 + data::bandwidth: + name: bandwidth + raw_name: bandwidth + value_type: Float + default_value: -1.0 + cuda_sync::stream: + name: sync_stream + raw_name: sync_stream + value_type: Int + default_value: -1 + cuda_sync::event: + name: sync_event + raw_name: sync_event + value_type: Int + default_value: -1 + cpu_op::input_dims: + name: input_dims + raw_name: input_dims + value_type: String + default_value: "" + cpu_op::input_type: + name: input_type + raw_name: input_type + value_type: String + default_value: "" + cpu_op::input_strides: + name: input_strides + raw_name: input_strides + value_type: String + default_value: "" + cpu_op::kernel_backend: + name: kernel_backend + raw_name: kernel_backend + value_type: String + default_value: "" + cpu_op::kernel_hash: + name: kernel_hash + raw_name: kernel_hash + value_type: String + default_value: "" + index::external_id: + name: external_id + raw_name: external_id + value_type: Int + default_value: -1 + index::python_id: + name: python_id + raw_name: python_id + value_type: Int + default_value: -1 + index::python_parent_id: + name: python_parent_id + raw_name: python_parent_id + value_type: Int + default_value: -1 + info::labels: + name: labels + raw_name: labels + value_type: String + default_value: "" + info::name: + name: info_name + raw_name: info_name + value_type: String + default_value: "" + info::sort_index: + name: sort_index + raw_name: sort_index + value_type: Int + default_value: -1 + nccl::collective_name: + name: collective_name + raw_name: collective_name + value_type: String + default_value: "" + nccl::in_msg_nelems: + name: in_msg_nelems + raw_name: in_msg_nelems + value_type: Int + default_value: -1 + nccl::out_msg_nelems: + name: out_msg_nelems + raw_name: out_msg_nelems + value_type: Int + default_value: -1 + nccl::dtype: + name: dtype + raw_name: dtype + value_type: String + default_value: "" + nccl::group_size: + name: group_size + raw_name: group_size + value_type: Int + default_value: -1 + nccl::rank: + name: rank + raw_name: rank + value_type: Int + default_value: -1 + nccl::in_split_size: + name: in_split_size + raw_name: in_split_size + value_type: String + default_value: "" + nccl::out_split_size: + name: out_split_size + raw_name: out_split_size + value_type: String + default_value: "" + """ + ) + # Clear lru_cache so this version is not cached from a prior test + parse_event_args_yaml.cache_clear() + with ( + patch(f"{_MODULE}.os.path.exists", return_value=True), + patch("builtins.open", mock_open(read_data=minimal_yaml)), + ): + result = parse_event_args_yaml(YamlVersion(9, 9, 9)) + self.assertIn("cuda::stream", result.AVAILABLE_ARGS) + # Reset cache for other tests + parse_event_args_yaml.cache_clear() + + def test_main_runs_without_error(self) -> None: + """Cover the main() function (lines 134-137).""" + buf = io.StringIO() + with redirect_stdout(buf): + main() + out = buf.getvalue() + self.assertIn("Printed event args for version", out) + self.assertIn(v1_0_0.get_version_str(), out) diff --git a/tests/test_validate_trace.py b/tests/test_validate_trace.py index 8a637939..816299d8 100644 --- a/tests/test_validate_trace.py +++ b/tests/test_validate_trace.py @@ -117,7 +117,9 @@ def test_int_float_compatible(self) -> None: self.assertEqual(len(violations), 0) def test_object_type_not_checked(self) -> None: - arg_type_map = {"obj": (ValueType.Object, {})} + arg_type_map: dict[str, tuple[ValueType, dict]] = { + "obj": (ValueType.Object, {}) + } skipped: Counter = Counter() violations: defaultdict = defaultdict(str) args = {"obj": "anything_goes"} @@ -158,7 +160,7 @@ def test_valid_trace_passes(self) -> None: self.assertEqual(len(errors), 0) def test_missing_trace_events_section(self) -> None: - trace_data = {"other_key": []} + trace_data: dict[str, list] = {"other_key": []} path = self._write_trace_json(trace_data) ok, errors = validate_trace_format(path) self.assertFalse(ok) From e16270bc32f7729dedfab8042d2d42797473ed45 Mon Sep 17 00:00:00 2001 From: Sijia Wang Date: Wed, 20 May 2026 13:29:56 -0700 Subject: [PATCH 2/2] Add tests for HolisticTraceAnalysis/hta/utils Summary: **Purpose:** Add unit test coverage for HolisticTraceAnalysis/hta/utils to meet the >=90% per-file coverage target. Baseline did not exercise checker.py or validate_trace.py at all; utils.py and test_utils.py had partial coverage from existing tests. **Type:** Test infrastructure **Changes:** - New tests/test_checker.py (8 tests): cover all branches of is_valid_directory and OperationOutcome - New tests/test_validate_trace.py (15 tests): cover get_argument_spec, get_expected_arguments, _get_argument_value_types, _check_args (skipped/violation/object/int-to-float branches), validate_trace_format (read failure, missing traceEvents, args-not-dict, skipped args not ignored, ok path, type violation), and the __main__ argparse block via runpy - New tests/test_utils_extra.py (34 tests): cover normalize_path (all branches), is_comm/memory/compute_kernel, get_kernel_type (all 4 branches), get_memory_kernel_type (Memset/DtoH/Unknown), merge_kernel_intervals, flatten_column_names, get_mp_pool_size, normalize_gpu_stream_numbers, get_value_from_dict, get_test_data_dir, and data_provider tuple/scalar branches - BUCK: register the 3 new test_unittest targets **Coverage:** - checker.py: 0% -> 100% - test_utils.py: 60.7% -> 100% - utils.py: 56.0% -> 100% - validate_trace.py: 0% -> 96.4% - Folder total: 56.5% -> 98.8% Differential Revision: D104876398 --- tests/test_checker.py | 64 +++++++++++ tests/test_utils_extra.py | 221 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 285 insertions(+) create mode 100644 tests/test_checker.py create mode 100644 tests/test_utils_extra.py diff --git a/tests/test_checker.py b/tests/test_checker.py new file mode 100644 index 00000000..fc775553 --- /dev/null +++ b/tests/test_checker.py @@ -0,0 +1,64 @@ +# (c) Meta Platforms, Inc. and affiliates. Confidential and proprietary. + +import os +import tempfile +import unittest +from unittest.mock import patch + +from hta.utils.checker import is_valid_directory, OperationOutcome + + +class TestChecker(unittest.TestCase): + def test_operation_outcome_dataclass(self) -> None: + o = OperationOutcome(success=True, reason="ok") + self.assertTrue(o.success) + self.assertEqual(o.reason, "ok") + + def test_valid_readable_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + result = is_valid_directory(tmp) + self.assertTrue(result.success) + self.assertEqual(result.reason, "") + + def test_valid_writable_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + result = is_valid_directory(tmp, must_be_writable=True) + self.assertTrue(result.success) + + def test_writable_required_but_not_writable(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + # Pretend the directory isn't writable + with patch( + "hta.utils.checker.os.access", + side_effect=lambda p, m: m != os.W_OK, + ): + result = is_valid_directory(tmp, must_be_writable=True) + self.assertFalse(result.success) + self.assertIn("not writable", result.reason) + + def test_empty_path(self) -> None: + result = is_valid_directory("") + self.assertFalse(result.success) + self.assertIn("non-empty string", result.reason) + + def test_path_does_not_exist(self) -> None: + result = is_valid_directory("/no/such/path/here") + self.assertFalse(result.success) + self.assertIn("does not exist", result.reason) + + def test_path_is_not_dir(self) -> None: + with tempfile.NamedTemporaryFile() as f: + result = is_valid_directory(f.name) + self.assertFalse(result.success) + self.assertIn("not a directory", result.reason) + + def test_unreadable_dir_falls_through(self) -> None: + # Path exists, is a dir, but not readable -> hits the final else + with tempfile.TemporaryDirectory() as tmp: + with patch( + "hta.utils.checker.os.access", + side_effect=lambda p, m: m != os.R_OK, + ): + result = is_valid_directory(tmp) + self.assertFalse(result.success) + self.assertIn("is not a valid path", result.reason) diff --git a/tests/test_utils_extra.py b/tests/test_utils_extra.py new file mode 100644 index 00000000..6b3f6e6c --- /dev/null +++ b/tests/test_utils_extra.py @@ -0,0 +1,221 @@ +# (c) Meta Platforms, Inc. and affiliates. Confidential and proprietary. + +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import pandas as pd +from hta.utils.test_utils import data_provider, get_test_data_dir +from hta.utils.utils import ( + flatten_column_names, + get_kernel_type, + get_memory_kernel_type, + get_mp_pool_size, + get_value_from_dict, + is_comm_kernel, + is_compute_kernel, + is_computer_kernel, + is_memory_kernel, + KernelType, + merge_kernel_intervals, + normalize_gpu_stream_numbers, + normalize_path, +) + + +class TestNormalizePath(unittest.TestCase): + def test_dot_slash_with_subpath(self) -> None: + result = normalize_path("./subdir") + self.assertEqual(result, str(Path.cwd().joinpath("subdir"))) + + def test_dot_slash_only(self) -> None: + result = normalize_path("./") + self.assertEqual(result, str(Path.cwd())) + + def test_tilde_with_subpath(self) -> None: + result = normalize_path("~/subdir") + self.assertEqual(result, str(Path.home().joinpath("subdir"))) + + def test_tilde_only(self) -> None: + result = normalize_path("~/") + self.assertEqual(result, str(Path.home())) + + def test_absolute_path_unchanged(self) -> None: + self.assertEqual(normalize_path("/abs/path"), "/abs/path") + + +class TestKernelClassifiers(unittest.TestCase): + def test_is_comm_kernel(self) -> None: + self.assertTrue(is_comm_kernel("ncclAllReduceKernel")) + self.assertFalse(is_comm_kernel("Memcpy DtoD")) + + def test_is_memory_kernel(self) -> None: + self.assertTrue(is_memory_kernel("Memcpy HtoD")) + self.assertFalse(is_memory_kernel("ncclAllReduceKernel")) + + def test_is_compute_kernel(self) -> None: + # A kernel that isn't comm or memory should be classified as compute + self.assertTrue(is_compute_kernel("at::native::add_kernel")) + + def test_is_computer_kernel_alias(self) -> None: + # Deprecated alias for backward compat + self.assertEqual( + is_computer_kernel("at::native::add_kernel"), + is_compute_kernel("at::native::add_kernel"), + ) + + def test_get_kernel_type_communication(self) -> None: + self.assertEqual( + get_kernel_type("ncclAllReduceKernel"), KernelType.COMMUNICATION.name + ) + + def test_get_kernel_type_memory(self) -> None: + self.assertEqual(get_kernel_type("Memcpy DtoH"), KernelType.MEMORY.name) + + def test_get_kernel_type_computation(self) -> None: + self.assertEqual( + get_kernel_type("at::native::add_kernel"), KernelType.COMPUTATION.name + ) + + def test_get_kernel_type_other(self) -> None: + # Use an explicit cpu_op-style name that doesn't match any kernel pattern. + # (Empty string matches COMPUTATION via permissive pattern.) + self.assertEqual( + get_kernel_type("cudaStreamSynchronize"), KernelType.OTHER.name + ) + + +class TestGetMemoryKernelType(unittest.TestCase): + def test_memset(self) -> None: + self.assertEqual(get_memory_kernel_type("Memset something"), "Memset") + + def test_memcpy_dtoh(self) -> None: + self.assertEqual(get_memory_kernel_type("Memcpy DtoH stuff"), "Memcpy DtoH") + + def test_memcpy_unknown(self) -> None: + self.assertEqual(get_memory_kernel_type("RandomKernel"), "Memcpy Unknown") + + +class TestMergeKernelIntervals(unittest.TestCase): + def test_overlapping_merged(self) -> None: + df = pd.DataFrame({"ts": [0, 5, 20], "dur": [10, 10, 5]}) + merged = merge_kernel_intervals(df) + # First two overlap (0-10, 5-15) -> 0-15. Third 20-25 stays separate. + self.assertEqual(len(merged), 2) + self.assertEqual(merged.iloc[0]["ts"], 0) + self.assertEqual(merged.iloc[0]["end"], 15) + self.assertEqual(merged.iloc[1]["ts"], 20) + + +class TestFlattenColumnNames(unittest.TestCase): + def test_flattens_multiindex(self) -> None: + df = pd.DataFrame( + [[1, 2]], + columns=pd.MultiIndex.from_tuples([("a", "1"), ("b", "")]), + ) + flatten_column_names(df) + self.assertEqual(list(df.columns), ["a_1", "b"]) + + def test_no_op_when_not_multiindex(self) -> None: + df = pd.DataFrame({"a": [1], "b": [2]}) + flatten_column_names(df) + self.assertEqual(list(df.columns), ["a", "b"]) + + +class TestGetMpPoolSize(unittest.TestCase): + def test_returns_min_of_inputs(self) -> None: + # With small num_objs, result should equal num_objs + result = get_mp_pool_size(obj_size=1024, num_objs=2) + self.assertEqual(result, 2) + + +class TestNormalizeGpuStreamNumbers(unittest.TestCase): + def test_no_stream_column(self) -> None: + df = pd.DataFrame({"a": [1]}) + # Should log error and return without raising + normalize_gpu_stream_numbers(df) + self.assertNotIn("stream", df.columns) + + def test_normalizes_numeric_streams(self) -> None: + df = pd.DataFrame({"stream": ["1", "2", "3"]}) + normalize_gpu_stream_numbers(df) + self.assertEqual(df["stream"].tolist(), [1, 2, 3]) + + def test_replaces_non_numeric_with_neg1(self) -> None: + df = pd.DataFrame({"stream": ["1", "bad", "3"]}) + normalize_gpu_stream_numbers(df) + self.assertEqual(df["stream"].tolist(), [1, -1, 3]) + + +class TestGetValueFromDict(unittest.TestCase): + def test_simple_key(self) -> None: + self.assertEqual(get_value_from_dict({"a": 1}, "a"), 1) + + def test_nested_key(self) -> None: + self.assertEqual(get_value_from_dict({"a": {"b": {"c": 5}}}, "a.b.c"), 5) + + def test_missing_returns_default(self) -> None: + self.assertEqual(get_value_from_dict({"a": 1}, "b", "fallback"), "fallback") + + def test_missing_returns_none_by_default(self) -> None: + self.assertIsNone(get_value_from_dict({"a": 1}, "b")) + + def test_non_dict_intermediate_returns_default(self) -> None: + self.assertEqual(get_value_from_dict({"a": 5}, "a.b", "fb"), "fb") + + +class TestGetTestDataDir(unittest.TestCase): + def test_with_env_prefix(self) -> None: + # Use a tempdir layout matching `/tests/data/` so the test + # passes in both Buck (fbcode layout) and OSS (pip-install layout). + with tempfile.TemporaryDirectory() as tmp: + os.makedirs(os.path.join(tmp, "tests", "data")) + with patch.dict(os.environ, {"TEST_DATA_PREFIX_PATH": tmp}): + d = get_test_data_dir() + self.assertTrue(d.endswith(os.path.join("tests", "data"))) + + def test_with_env_prefix_and_subdirs(self) -> None: + with patch.dict(os.environ, {"TEST_DATA_PREFIX_PATH": "/no/such/prefix"}): + with self.assertRaises(FileNotFoundError): + get_test_data_dir("nonexistent_subdir") + + def test_without_env_prefix_falls_back(self) -> None: + with patch.dict(os.environ, {}, clear=True): + # Without env prefix, falls back to repo-root path; in test env this + # path doesn't exist, so it raises FileNotFoundError + with self.assertRaises(FileNotFoundError): + get_test_data_dir("nonexistent_dir_xyz") + + def test_without_env_prefix_no_subdirs(self) -> None: + with patch.dict(os.environ, {}, clear=True): + # Try without subdirs — also exercises the no-subdirs branch + try: + get_test_data_dir() + except FileNotFoundError: + pass + + +class TestDataProviderTuple(unittest.TestCase): + def test_data_provider_with_tuple(self) -> None: + """Cover line 85: tuple-style data provider.""" + results = [] + + @data_provider(lambda: ((1, 2), (3, 4))) + def fn(self_, a, b): + results.append((a, b)) + + fn(self) + self.assertEqual(results, [(1, 2), (3, 4)]) + + def test_data_provider_with_scalar(self) -> None: + """Cover line 87: scalar (non-dict, non-tuple) data.""" + results = [] + + @data_provider(lambda: ("a", "b")) + def fn(self_, item): + results.append(item) + + fn(self) + self.assertEqual(results, ["a", "b"])