Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/common/file_operate.py: 86%
191 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-21 15:37 +0800
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# ----------------------------------------------------------------------------
19import csv
20import os
21import shutil
22import configparser
24from common.log import log_error, log_debug
25from common.const import CONFIG_TABLE_FILE
27__all__ = ["FileOperate", "MOVE_MODE", "COPY_MODE"]
29MOVE_MODE = 'm'
30COPY_MODE = 'c'
31ENCODE_UTF_8 = "utf-8"
34class FileOperate:
36 @staticmethod
37 def check_file(file_path):
38 if not file_path:
39 return False
40 return os.path.isfile(file_path)
42 @staticmethod
43 def check_dir(dir_path):
44 if not dir_path:
45 return False
46 return os.path.isdir(dir_path)
48 @staticmethod
49 def check_exists(path):
50 if not path:
51 return False
52 return os.path.exists(path)
54 @staticmethod
55 def check_emtpy(path):
56 if not path:
57 return True
58 if os.path.exists(path) and os.path.isdir(path):
59 return not os.listdir(path)
60 return True
62 @staticmethod
63 def check_access(path, mode=os.F_OK):
64 if not path:
65 return False
66 return os.access(path, mode) # mode: F_OK, R_OK, W_OK, X_OK
68 @staticmethod
69 def remove_file(file_path):
70 if file_path and os.path.exists(file_path):
71 os.remove(file_path)
73 @staticmethod
74 def create_dir(dir_path, exist_ok=False):
75 try:
76 os.makedirs(dir_path, mode=0o750, exist_ok=exist_ok)
77 return True
78 except OSError as e:
79 log_debug(f"Failed to create directory {dir_path}, error is {e}")
80 return False
82 @staticmethod
83 def remove_dir(dir_path):
84 if not dir_path or not os.access(dir_path, os.F_OK):
85 log_debug("dir: {0} is not exist, do not need to remove.".format(dir_path))
86 return False
87 if not os.access(dir_path, os.W_OK):
88 log_debug("dir: {0} is not access to write, can not remove.".format(dir_path))
89 return False
90 shutil.rmtree(dir_path)
91 return True
93 @staticmethod
94 def walk_dir(dir_path):
95 if not dir_path or not os.access(dir_path, os.R_OK):
96 return False
97 f = os.walk(dir_path)
98 return f
100 @staticmethod
101 def list_dir(dir_path):
102 if not os.access(dir_path, os.R_OK):
103 return False
104 f = os.listdir(dir_path)
105 return f
107 @staticmethod
108 def write_file(file_path, info):
109 if not file_path:
110 return
111 file_dir = os.path.split(file_path)[0]
112 if not os.path.exists(file_dir) and not FileOperate.create_dir(file_dir):
113 log_error("Create path directory: \"{}\" failed in write file.".format(file_dir))
114 return
115 with open(file_path, mode="w", encoding=ENCODE_UTF_8) as f:
116 f.write(info)
118 @staticmethod
119 def append_write_file(file_path, info):
120 file_dir = os.path.split(file_path)[0]
121 if not os.path.exists(file_dir) and not FileOperate.create_dir(file_dir):
122 log_error("Create path directory: \"{}\" failed in write file.".format(file_dir))
123 return
124 with open(file_path, mode="a", encoding=ENCODE_UTF_8) as f:
125 f.write(info)
127 @staticmethod
128 def read_file(file_path):
129 if file_path.endswith(".ini"):
130 cf = configparser.ConfigParser()
131 cf.read(file_path, encoding=ENCODE_UTF_8)
132 return cf
133 elif file_path.endswith(".csv"):
134 csv_buf = []
135 with open(file_path, mode="r", encoding=ENCODE_UTF_8) as f:
136 reader = csv.reader(f)
137 for row in reader:
138 csv_buf.append(row)
139 return csv_buf
140 else:
141 with open(file_path, mode="r", encoding=ENCODE_UTF_8) as f:
142 file_buf = f.read()
143 return file_buf
145 @staticmethod
146 def delete_dirs(dir_list):
147 if not dir_list:
148 return
149 for inter_dir in dir_list:
150 if inter_dir and os.path.exists(inter_dir):
151 if not FileOperate.remove_dir(inter_dir):
152 log_error("Delete intermediate: \"{}\" failed in asys clean work.".format(inter_dir))
154 @staticmethod
155 def copy_file_to_dir(source_file_path, target_dir_path):
156 if not os.path.exists(source_file_path) or not os.access(source_file_path, os.R_OK) or \
157 not os.path.isfile(source_file_path):
158 return False
159 if not os.path.exists(target_dir_path):
160 os.makedirs(target_dir_path)
161 shutil.copy(source_file_path, target_dir_path)
162 return True
164 @staticmethod
165 def copy_dir(source_dir_path, target_dir_path):
166 if not os.path.exists(source_dir_path) or not os.access(source_dir_path, os.R_OK) or \
167 not os.path.isdir(source_dir_path):
168 return False
169 if os.path.relpath(source_dir_path, target_dir_path).endswith(".."):
170 log_error("The output directory cannot be in the data directory.")
171 return False
172 shutil.copytree(source_dir_path, target_dir_path)
173 return True
175 @staticmethod
176 def move_file_to_dir(source_file_path, target_dir_path):
177 if not os.path.exists(source_file_path) or not os.access(source_file_path, os.R_OK) or \
178 not os.path.isfile(source_file_path):
179 return False
180 if not os.path.exists(target_dir_path):
181 os.makedirs(target_dir_path)
182 shutil.move(source_file_path, target_dir_path)
183 return True
185 @staticmethod
186 def move_dir(source_dir_path, target_dir_path):
187 if not os.path.exists(source_dir_path) or not os.access(source_dir_path, os.R_OK) or \
188 not os.path.isdir(source_dir_path):
189 return False
190 if os.path.exists(target_dir_path):
191 shutil.rmtree(target_dir_path)
192 shutil.move(source_dir_path, target_dir_path)
193 return True
195 @staticmethod
196 def collect_file_to_dir(source_file_path, target_dir_path, mode):
197 if mode == MOVE_MODE: # move mode
198 return FileOperate.move_file_to_dir(source_file_path, target_dir_path)
199 elif mode == COPY_MODE: # copy mode
200 return FileOperate.copy_file_to_dir(source_file_path, target_dir_path)
201 else:
202 log_error("Unknown mode in collect file.")
203 return False
205 @staticmethod
206 def collect_dir(source_dir_path, target_dir_path, mode):
207 if mode == MOVE_MODE: # move mode
208 return FileOperate.move_dir(source_dir_path, target_dir_path)
209 elif mode == COPY_MODE: # copy mode
210 return FileOperate.copy_dir(source_dir_path, target_dir_path)
211 else:
212 log_error("Unknown mode in collect directory.")
213 return False
215 @staticmethod
216 def check_valid_dir(dir_path):
217 if not (os.path.exists(dir_path) and os.path.isdir(dir_path) and os.access(dir_path, os.R_OK)):
218 return False
219 if len(os.listdir(dir_path)) == 0:
220 return False
221 return True
223 def read_config(self):
224 if not os.path.isfile(CONFIG_TABLE_FILE):
225 log_error(f"Error: The file {CONFIG_TABLE_FILE} does not exist, please check env.")
226 return {}
227 try:
228 return self._read_config()
229 except PermissionError:
230 log_error(f"Error: Permission denied for file {CONFIG_TABLE_FILE}.")
231 return {}
232 except (csv.Error, IndexError):
233 log_error(f"Error: {CONFIG_TABLE_FILE} format or content is error.")
234 return {}
236 @staticmethod
237 def _read_config():
238 """读取config配置清单并解析成字典"""
239 config_table = {}
240 with open(CONFIG_TABLE_FILE, newline='') as f:
241 data = csv.reader(f)
242 _, cfg_get, cfg_set, cfg_restore = next(data)
243 for row in data:
244 config_table[row[0]] = {
245 cfg_get: row[1].split(","),
246 cfg_set: row[2].split(","),
247 cfg_restore: row[3].split(",")
248 }
249 return config_table