Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/data_dump/data_dump_collect.py: 95%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------- 

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

5# 

6# Licensed under the Apache License, Version 2.0 (the "License"); 

7# you may not use this file except in compliance with the License. 

8# You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, software 

13# distributed under the License is distributed on an "AS IS" BASIS, 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

15# See the License for the specific language governing permissions and 

16# limitations under the License. 

17# ---------------------------------------------------------------------------- 

18 

19import os 

20 

21from params import ParamDict 

22from common import consts 

23from common import FileOperate as f 

24from common.file_operate import MOVE_MODE, COPY_MODE 

25from common import log_debug, log_warning 

26from drv import EnvVarName 

27 

28__all__ = ["collect_data_dump"] 

29 

30 

31def _get_collect_path(): 

32 collect_data_dump_path_list = [] 

33 # asys launch 

34 if ParamDict().get_command() == consts.launch_cmd: 

35 collect_data_dump_path_list.append( 

36 os.path.join(ParamDict().asys_output_timestamp_dir, "npu_collect_intermediates") 

37 ) 

38 return collect_data_dump_path_list 

39 

40 # asys collect 

41 task_dir = ParamDict().get_arg("task_dir") 

42 if task_dir: 

43 collect_data_dump_path_list.append(task_dir) 

44 env_var = EnvVarName() 

45 if env_var.npu_collect_path: 

46 log_debug("Get env NPU_COLLECT_PATH successfully, add NPU_COLLECT_PATH to dump collect path.") 

47 collect_data_dump_path_list.append(env_var.npu_collect_path) 

48 

49 if env_var.work_path: 

50 log_debug("Get env ASCEND_WORK_PATH successfully, add ASCEND_WORK_PATH to dump collect path.") 

51 collect_data_dump_path_list.append(env_var.work_path) 

52 

53 collect_data_dump_path_list.append(env_var.current_path) 

54 

55 return collect_data_dump_path_list 

56 

57 

58def get_source_dir(): 

59 for data_dump_path in _get_collect_path(): 

60 data_dump_source_dir = os.path.join(data_dump_path, "extra-info", "data-dump") 

61 if data_dump_source_dir and f.check_dir(data_dump_source_dir): 

62 log_debug("Data-dump source check success, path={}.".format(data_dump_source_dir)) 

63 return data_dump_source_dir 

64 return None 

65 

66 

67def collect_data_dump(output_root_path): 

68 data_dump_source_dir = get_source_dir() 

69 if data_dump_source_dir is None: 

70 log_warning("Data-dump collect failed.") 

71 return 

72 

73 data_dump_target_dir = os.path.join(output_root_path, "dfx", "data-dump") 

74 if ParamDict().get_command() == consts.launch_cmd: 

75 ret = f.collect_dir(data_dump_source_dir, data_dump_target_dir, MOVE_MODE) 

76 else: 

77 ret = f.collect_dir(data_dump_source_dir, data_dump_target_dir, COPY_MODE) 

78 if not ret: 

79 log_warning("Data-dump collect failed, copy {} dir failed.".format(data_dump_source_dir))