• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2021 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Test the hccl parser module."""
16import csv
17import os
18import shutil
19import tempfile
20
21from mindspore.profiler.parser.hccl_parser import HcclParser
22from tests.ut.python.profiler import PROFILER_DIR
23
24
25def get_hccl_result(file_path):
26    """
27    Get hccl result from the hccl file.
28
29    Args:
30        file_path (str): The hccl file path.
31
32    Returns:
33        list[list], the parsed hccl information.
34    """
35    result = []
36    with open(file_path, 'r') as file:
37        csv_reader = csv.reader(file)
38        for row in csv_reader:
39            result.append(row)
40    return result
41
42
43class TestHcclParser:
44    """Test the class of `HcclParser`."""
45    def setup_method(self):
46        """Initialization before test case execution."""
47        self._output_path = tempfile.mkdtemp(
48            prefix='test_hccl_parser_'
49        )
50        shutil.copyfile(os.path.join(PROFILER_DIR, 'step_trace_raw_6_detail_time.csv'),
51                        os.path.join(self._output_path, 'step_trace_raw_6_detail_time.csv'))
52        self._parser = HcclParser(os.path.join(PROFILER_DIR, 'hccl_info'), '6', '6', self._output_path)
53
54    def teardown_method(self) -> None:
55        """Clear up after test case execution."""
56        shutil.rmtree(self._output_path)
57
58    def test_parse(self):
59        """Test the parse function."""
60        expect_hccl_file = os.path.join(
61            PROFILER_DIR, 'hccl_raw_6.csv'
62        )
63        expect_result = get_hccl_result(expect_hccl_file)
64
65        self._parser.parse()
66        hccl_file = os.path.join(
67            self._output_path, 'hccl_raw_6.csv'
68        )
69        result = get_hccl_result(hccl_file)
70        assert expect_result == result
71