1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
# Maintained in full by:
# Catalyst Team <catalyst@gentoo.org>
# Release Engineering Team <releng@gentoo.org>
'''fileops.py
Performs file operations such as pack/unpack,
ensuring directories exist,... imports snakeoils osutils
functions for use throughout catalyst.
'''
import os
import shutil
from stat import ST_UID, ST_GID, ST_MODE
# NOTE: pjoin and listdir_files are imported here for export
# to other catalyst modules
# pylint: disable=unused-import
from snakeoil.osutils import (ensure_dirs as snakeoil_ensure_dirs,
pjoin, listdir_files)
# pylint: enable=unused-import
from catalyst import log
from catalyst.support import (cmd, CatalystError)
def ensure_dirs(path, gid=-1, uid=-1, mode=0o755, minimal=True,
failback=None, fatal=False):
'''Wrapper to snakeoil.osutil's ensure_dirs()
This additionally allows for failures to run
cleanup or other code and/or raise fatal errors.
:param path: directory to ensure exists on disk
:param gid: a valid GID to set any created directories to
:param uid: a valid UID to set any created directories to
:param mode: permissions to set any created directories to
:param minimal: boolean controlling whether or not the specified mode
must be enforced, or is the minimal permissions necessary. For example,
if mode=0o755, minimal=True, and a directory exists with mode 0707,
this will restore the missing group perms resulting in 757.
:param failback: function to run in the event of a failed attemp
to create the directory.
:return: True if the directory could be created/ensured to have those
permissions, False if not.
'''
succeeded = snakeoil_ensure_dirs(path, gid=gid, uid=uid, mode=mode, minimal=minimal)
if not succeeded:
if failback:
failback()
if fatal:
raise CatalystError(
"Failed to create directory: %s" % path, print_traceback=True)
return succeeded
def clear_dir(target, mode=0o755, chg_flags=False, remove=False,
clear_nondir=True):
'''Universal directory clearing function
@target: string, path to be cleared or removed
@mode: integer, desired mode to set the directory to
@chg_flags: boolean used for FreeBSD hosts
@remove: boolean, passed through to clear_dir()
@return boolean
'''
log.debug('start: %s', target)
if not target:
log.debug('no target... returning')
return False
mystat = None
if os.path.isdir(target) and not os.path.islink(target):
log.notice('Emptying directory: %s', target)
# stat the dir, delete the dir, recreate the dir and set
# the proper perms and ownership
try:
log.debug('os.stat()')
mystat = os.stat(target)
# There's no easy way to change flags recursively in python
if chg_flags and os.uname()[0] == "FreeBSD":
cmd(['chflags', '-R', 'noschg', target])
log.debug('shutil.rmtree()')
shutil.rmtree(target)
except Exception:
log.error('clear_dir failed', exc_info=True)
return False
elif os.path.exists(target):
if clear_nondir:
log.debug("Clearing (unlinking) non-directory: %s", target)
os.unlink(target)
else:
log.info('clear_dir failed: %s: is not a directory', target)
return False
if not remove:
log.debug('ensure_dirs()')
ensure_dirs(target, mode=mode)
if mystat:
os.chown(target, mystat[ST_UID], mystat[ST_GID])
os.chmod(target, mystat[ST_MODE])
log.debug('DONE, returning True')
return True
def clear_path(target):
"""Nuke |target| regardless of it being a dir or file."""
clear_dir(target, remove=True)
def move_path(src, dest):
'''Move a source target to a new destination
:param src: source path to move
:param dest: destination path to move it to
:returns: boolean
'''
log.debug('Start move_path(%s, %s)', src, dest)
if os.path.isdir(src) and not os.path.islink(src):
if os.path.exists(dest):
log.warning('Removing existing target destination: %s', dest)
if not clear_dir(dest, remove=True):
return False
log.debug('Moving source...')
try:
shutil.move(src, dest)
except Exception:
log.error('move_path failed', exc_info=True)
return False
return True
return False
|
GitWeb