Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/common/file_operate.py: 86%
193 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-19 17:46 +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 file_dir and 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 if not file_path:
121 return
122 file_dir = os.path.split(file_path)[0]
123 if file_dir and not os.path.exists(file_dir) and not FileOperate.create_dir(file_dir):
124 log_error("Create path directory: \"{}\" failed in write file.".format(file_dir))
125 return
126 with open(file_path, mode="a", encoding=ENCODE_UTF_8) as f:
127 f.write(info)
129 @staticmethod
130 def read_file(file_path):
131 if file_path.endswith(".ini"):
132 cf = configparser.ConfigParser()
133 cf.read(file_path, encoding=ENCODE_UTF_8)
134 return cf
135 elif file_path.endswith(".csv"):
136 csv_buf = []
137 with open(file_path, mode="r", encoding=ENCODE_UTF_8) as f:
138 reader = csv.reader(f)
139 for row in reader:
140 csv_buf.append(row)
141 return csv_buf
142 else:
143 with open(file_path, mode="r", encoding=ENCODE_UTF_8) as f:
144 file_buf = f.read()
145 return file_buf
147 @staticmethod
148 def delete_dirs(dir_list):
149 if not dir_list:
150 return
151 for inter_dir in dir_list:
152 if inter_dir and os.path.exists(inter_dir):
153 if not FileOperate.remove_dir(inter_dir):
154 log_error("Delete intermediate: \"{}\" failed in asys clean work.".format(inter_dir))
156 @staticmethod
157 def copy_file_to_dir(source_file_path, target_dir_path):
158 if not os.path.exists(source_file_path) or not os.access(source_file_path, os.R_OK) or \
159 not os.path.isfile(source_file_path):
160 return False
161 if not os.path.exists(target_dir_path):
162 os.makedirs(target_dir_path)
163 shutil.copy(source_file_path, target_dir_path)
164 return True
166 @staticmethod
167 def copy_dir(source_dir_path, target_dir_path):
168 if not os.path.exists(source_dir_path) or not os.access(source_dir_path, os.R_OK) or \
169 not os.path.isdir(source_dir_path):
170 return False
171 if os.path.relpath(source_dir_path, target_dir_path).endswith(".."):
172 log_error("The output directory cannot be in the data directory.")
173 return False
174 shutil.copytree(source_dir_path, target_dir_path)
175 return True
177 @staticmethod
178 def move_file_to_dir(source_file_path, target_dir_path):
179 if not os.path.exists(source_file_path) or not os.access(source_file_path, os.R_OK) or \
180 not os.path.isfile(source_file_path):
181 return False
182 if not os.path.exists(target_dir_path):
183 os.makedirs(target_dir_path)
184 shutil.move(source_file_path, target_dir_path)
185 return True
187 @staticmethod
188 def move_dir(source_dir_path, target_dir_path):
189 if not os.path.exists(source_dir_path) or not os.access(source_dir_path, os.R_OK) or \
190 not os.path.isdir(source_dir_path):
191 return False
192 if os.path.exists(target_dir_path):
193 shutil.rmtree(target_dir_path)
194 shutil.move(source_dir_path, target_dir_path)
195 return True
197 @staticmethod
198 def collect_file_to_dir(source_file_path, target_dir_path, mode):
199 if mode == MOVE_MODE: # move mode
200 return FileOperate.move_file_to_dir(source_file_path, target_dir_path)
201 elif mode == COPY_MODE: # copy mode
202 return FileOperate.copy_file_to_dir(source_file_path, target_dir_path)
203 else:
204 log_error("Unknown mode in collect file.")
205 return False
207 @staticmethod
208 def collect_dir(source_dir_path, target_dir_path, mode):
209 if mode == MOVE_MODE: # move mode
210 return FileOperate.move_dir(source_dir_path, target_dir_path)
211 elif mode == COPY_MODE: # copy mode
212 return FileOperate.copy_dir(source_dir_path, target_dir_path)
213 else:
214 log_error("Unknown mode in collect directory.")
215 return False
217 @staticmethod
218 def check_valid_dir(dir_path):
219 if not (os.path.exists(dir_path) and os.path.isdir(dir_path) and os.access(dir_path, os.R_OK)):
220 return False
221 if len(os.listdir(dir_path)) == 0:
222 return False
223 return True
225 def read_config(self):
226 if not os.path.isfile(CONFIG_TABLE_FILE):
227 log_error(f"Error: The file {CONFIG_TABLE_FILE} does not exist, please check env.")
228 return {}
229 try:
230 return self._read_config()
231 except PermissionError:
232 log_error(f"Error: Permission denied for file {CONFIG_TABLE_FILE}.")
233 return {}
234 except (csv.Error, IndexError):
235 log_error(f"Error: {CONFIG_TABLE_FILE} format or content is error.")
236 return {}
238 @staticmethod
239 def _read_config():
240 """读取config配置清单并解析成字典"""
241 config_table = {}
242 with open(CONFIG_TABLE_FILE, newline='') as f:
243 data = csv.reader(f)
244 _, cfg_get, cfg_set, cfg_restore = next(data)
245 for row in data:
246 config_table[row[0]] = {
247 cfg_get: row[1].split(","),
248 cfg_set: row[2].split(","),
249 cfg_restore: row[3].split(",")
250 }
251 return config_table