def getTasksProcForce(self) :
""" Get tasks list by bruteforcing /proc """
for i in range(1, 65535, 1) :
if(os.access("/proc/" + str(i) + "/status", os.F_OK) == True):
l = self.openProcTaskStatus("/proc/" + str(i) + "/status")
if(l != []):
self.tasks_procforce.map_tasks[int(l[0])] = [int(l[1]), int(l[2]), l[3], 0, 0]
self.tasks_procforce.list_tasks.append(int(l[0]))
if(os.access("/proc/" + str(i) + "/task", os.F_OK) == True):
for srep in os.listdir("/proc/" + str(i) + "/task"):
if(srep != l[0]):
ll = self.openProcTaskStatus("/proc/" + str(i) + "/task/" + srep + "/status")
self.tasks_procforce.map_tasks[int(ll[0])] = [int(ll[1]), int(ll[2]), ll[3], 0, 1]
self.tasks_procforce.list_tasks.append(int(ll[0]))
python类access()的实例源码
def find_ancestor_cmd_path(self, cmd, cwd):
"""Recursively check for command binary in ancestors' node_modules/.bin directories."""
node_modules_bin = path.normpath(path.join(cwd, 'node_modules/.bin/'))
binary = path.join(node_modules_bin, cmd)
if sublime.platform() == 'windows' and path.splitext(binary)[1] != '.cmd':
binary += '.cmd'
if binary and access(binary, X_OK):
return binary
parent = path.normpath(path.join(cwd, '../'))
if parent == '/' or parent == cwd:
return None
return self.find_ancestor_cmd_path(cmd, parent)
def __addTooltip(self, item, text, hideWarn=False):
self.__loadTooltip()
if not ToolTip:
if not hideWarn:
self.warn("ToolTips unavailable - check tooltip.py is in the lib folder")
else:
# turn off warnings about tooltips
if hideWarn:
myWarn = self.__pauseWarn()
var = StringVar(self.topLevel)
var.set(text)
tip = ToolTip(item, delay=500, follow_mouse=1, textvariable=var)
item.tooltip = tip
if hideWarn:
self.__resumeWarn(myWarn)
return var
#####################################
# FUNCTIONS to show pop-up dialogs
#####################################
# function to access the last made pop_up
def _clone_test_db(self, number, verbosity, keepdb=False):
source_database_name = self.connection.settings_dict['NAME']
target_database_name = self.get_test_db_clone_settings(number)['NAME']
# Forking automatically makes a copy of an in-memory database.
if not self.connection.is_in_memory_db(source_database_name):
# Erase the old test database
if os.access(target_database_name, os.F_OK):
if keepdb:
return
if verbosity >= 1:
print("Destroying old test database for alias %s..." % (
self._get_database_display_str(verbosity, target_database_name),
))
try:
os.remove(target_database_name)
except Exception as e:
sys.stderr.write("Got an error deleting the old test database: %s\n" % e)
sys.exit(2)
try:
shutil.copy(source_database_name, target_database_name)
except Exception as e:
sys.stderr.write("Got an error cloning the test database: %s\n" % e)
sys.exit(2)
def check_path_owner(path):
# If we don't have a way to check the effective uid of this process, then
# we'll just assume that we own the directory.
if not hasattr(os, "geteuid"):
return True
previous = None
while path != previous:
if os.path.lexists(path):
# Check if path is writable by current user.
if os.geteuid() == 0:
# Special handling for root user in order to handle properly
# cases where users use sudo without -H flag.
try:
path_uid = get_path_uid(path)
except OSError:
return False
return path_uid == 0
else:
return os.access(path, os.W_OK)
else:
previous, path = path, os.path.dirname(path)
def which1(program, pathstr=None):
if pathstr is None:
pathstr = os.environ["PATH"]
def is_exe(fpath):
return os.path.isfile(fpath) and os.access(fpath, os.X_OK)
fpath, fname = os.path.split(program)
if fpath:
if is_exe(program):
return program
else:
for path in pathstr.split(os.pathsep):
path = path.strip('"')
exe_file = os.path.join(path, program)
if is_exe(exe_file):
return exe_file
return None
def copy_from_host(module):
compress = module.params.get('compress')
src = module.params.get('src')
if not os.path.exists(src):
module.fail_json(msg="file not found: {}".format(src))
if not os.access(src, os.R_OK):
module.fail_json(msg="file is not readable: {}".format(src))
mode = oct(os.stat(src).st_mode & 0o777)
with open(src, 'rb') as f:
raw_data = f.read()
sha1 = hashlib.sha1(raw_data).hexdigest()
data = zlib.compress(raw_data) if compress else raw_data
module.exit_json(content=base64.b64encode(data), sha1=sha1, mode=mode,
source=src)
def execd_submodule_paths(command, execd_dir=None):
"""Generate a list of full paths to the specified command within exec_dir.
"""
for module_path in execd_module_paths(execd_dir):
path = os.path.join(module_path, command)
if os.access(path, os.X_OK) and os.path.isfile(path):
yield path
def hook_mem_access(self, emu, access, address, size, value, user_data):
if access == unicorn.UC_MEM_WRITE:
self.pprint("Write: *%#x = %#x (size = %u)"% (address, value, size), "Memory")
elif access == unicorn.UC_MEM_READ:
self.pprint("Read: *%#x (size = %u)" % (address, size), "Memory")
return
def getMappingsFromTable(self):
self._maps = []
sz = self.memory_mapping.rowCount()
for i in range(sz):
name = self.memory_mapping.item(i, 0)
if not name:
continue
name = name.text()
address = self.memory_mapping.item(i, 1)
if address:
if ishex(address.text()):
address = int(address.text(), 0x10)
else:
address = int(address.text())
size = self.memory_mapping.item(i, 2)
if size:
size = int(size.text(), 0x10) if ishex(size.text()) else int(size.text())
permission = self.memory_mapping.item(i, 3)
if permission:
permission = permission.text()
read_from_file = self.memory_mapping.item(i, 4)
if read_from_file and not os.access(read_from_file.text(), os.R_OK):
read_from_file = None
self._maps.append([name, address, size, permission, read_from_file])
return
def loadCode(self, title, filter, run_disassembler):
qFile, qFilter = QFileDialog.getOpenFileName(self, title, EXAMPLES_PATH, filter)
if not os.access(qFile, os.R_OK):
return
if run_disassembler or qFile.endswith(".raw"):
body = disassemble_file(qFile, self.arch)
self.loadFile(qFile, data=body)
else:
self.loadFile(qFile)
return
def can_exe(self, fpath):
"""
Can test if path exists and is executable
"""
if isinstance(fpath, basestring):
return isfile(fpath) and access(fpath, X_OK)
return False
def export(self):
''' Export visible layers and layer groups to CoaSprite '''
if os.path.isfile(os.path.join(self.path, self.name)):
show_error_msg('ABORTING!\nDestination is not a folder.\n {path}/{name}'.format(path=self.path, name=self.name))
return
if not os.access(self.path, os.W_OK):
show_error_msg('ABORTING!\nDestination is not a writable.\n {path}'.format(path=self.path))
return
if os.path.isdir(os.path.join(self.path, self.name)):
show_error_msg('Destination exists, I may have overwritten something in {path}/{name}'.format(path=self.path, name=self.name))
self.mkdir()
# Loop through visible layers
self.img = self.original_img.duplicate()
self.img.undo_group_start()
for layer in self.img.layers:
if layer.visible:
name = '{name}.png'.format(name=layer.name)
pdb.gimp_image_set_active_layer(self.img, layer)
# Crop and the layer position
pdb.plug_in_autocrop_layer(self.img, layer)
z = 0 - pdb.gimp_image_get_item_position(self.img, layer)
if isinstance(layer, gimp.GroupLayer):
if len(layer.children) > 0:
self.sprites.append(self.export_sprite_sheet(layer, name, layer.offsets, z))
else:
self.sprites.append(self.export_sprite(layer, name, layer.offsets, z))
self.write_json()
self.img.undo_group_end()
pdb.gimp_image_delete(self.img)
def is_writable(self, path):
result = False
while not result:
if os.path.exists(path):
result = os.access(path, os.W_OK)
break
parent = os.path.dirname(path)
if parent == path:
break
path = parent
return result
def get_cache_base(suffix=None):
"""
Return the default base location for distlib caches. If the directory does
not exist, it is created. Use the suffix provided for the base directory,
and default to '.distlib' if it isn't provided.
On Windows, if LOCALAPPDATA is defined in the environment, then it is
assumed to be a directory, and will be the parent directory of the result.
On POSIX, and on Windows if LOCALAPPDATA is not defined, the user's home
directory - using os.expanduser('~') - will be the parent directory of
the result.
The result is just the directory '.distlib' in the parent directory as
determined above, or with the name specified with ``suffix``.
"""
if suffix is None:
suffix = '.distlib'
if os.name == 'nt' and 'LOCALAPPDATA' in os.environ:
result = os.path.expandvars('$localappdata')
else:
# Assume posix, or old Windows
result = os.path.expanduser('~')
# we use 'isdir' instead of 'exists', because we want to
# fail if there's a file with that name
if os.path.isdir(result):
usable = os.access(result, os.W_OK)
if not usable:
logger.warning('Directory exists but is not writable: %s', result)
else:
try:
os.makedirs(result)
usable = True
except OSError:
logger.warning('Unable to create %s', result, exc_info=True)
usable = False
if not usable:
result = tempfile.mkdtemp()
logger.warning('Default location unusable, using %s', result)
return os.path.join(result, suffix)
def extraction_error(self):
"""Give an error message for problems extracting file(s)"""
old_exc = sys.exc_info()[1]
cache_path = self.extraction_path or get_default_cache()
tmpl = textwrap.dedent("""
Can't extract file(s) to egg cache
The following error occurred while trying to extract file(s) to the Python egg
cache:
{old_exc}
The Python egg cache directory is currently set to:
{cache_path}
Perhaps your account does not have write access to this directory? You can
change the cache directory by setting the PYTHON_EGG_CACHE environment
variable to point to an accessible directory.
""").lstrip()
err = ExtractionError(tmpl.format(**locals()))
err.manager = self
err.cache_path = cache_path
err.original_error = old_exc
raise err
def _mkstemp_inner(dir, pre, suf, flags, output_type):
"""Code common to mkstemp, TemporaryFile, and NamedTemporaryFile."""
names = _get_candidate_names()
if output_type is bytes:
names = map(_os.fsencode, names)
for seq in range(TMP_MAX):
name = next(names)
file = _os.path.join(dir, pre + name + suf)
try:
fd = _os.open(file, flags, 0o600)
except FileExistsError:
continue # try again
except PermissionError:
# This exception is thrown when a directory with the chosen name
# already exists on windows.
if (_os.name == 'nt' and _os.path.isdir(dir) and
_os.access(dir, _os.W_OK)):
continue
else:
raise
return (fd, _os.path.abspath(file))
raise FileExistsError(_errno.EEXIST,
"No usable temporary file name found")
# User visible interfaces.
def mkdtemp(suffix=None, prefix=None, dir=None):
"""User-callable function to create and return a unique temporary
directory. The return value is the pathname of the directory.
Arguments are as for mkstemp, except that the 'text' argument is
not accepted.
The directory is readable, writable, and searchable only by the
creating user.
Caller is responsible for deleting the directory when done with it.
"""
prefix, suffix, dir, output_type = _sanitize_params(prefix, suffix, dir)
names = _get_candidate_names()
if output_type is bytes:
names = map(_os.fsencode, names)
for seq in range(TMP_MAX):
name = next(names)
file = _os.path.join(dir, prefix + name + suffix)
try:
_os.mkdir(file, 0o700)
except FileExistsError:
continue # try again
except PermissionError:
# This exception is thrown when a directory with the chosen name
# already exists on windows.
if (_os.name == 'nt' and _os.path.isdir(dir) and
_os.access(dir, _os.W_OK)):
continue
else:
raise
return file
raise FileExistsError(_errno.EEXIST,
"No usable temporary directory name found")
def _clean_check(cmd, target):
"""
Run the command to download target. If the command fails, clean up before
re-raising the error.
"""
try:
subprocess.check_call(cmd)
except subprocess.CalledProcessError:
if os.access(target, os.F_OK):
os.unlink(target)
raise
def _clean_check(cmd, target):
"""
Run the command to download target. If the command fails, clean up before
re-raising the error.
"""
try:
subprocess.check_call(cmd)
except subprocess.CalledProcessError:
if os.access(target, os.F_OK):
os.unlink(target)
raise
