Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _do_execshell(cmd, b_printcmd=True, timeout=None):
"""
do execshell
"""
if timeout is not None and timeout < 0:
raise cup.err.ShellException(
'timeout should be None or >= 0'
)
if b_printcmd is True:
print('To exec cmd:{0}'.format(cmd))
shellexec = ShellExec()
return shellexec.run(cmd, timeout)
log.info('test xxx')
log.critical('test critical')
"""
loggerman = _LoggerMan()
if not loggerman.is_initalized():
# loggerman._setlogger(logging.getLogger(loggername))
loggerman._setlogger(logging.getLogger())
if os.path.exists(logfile) is False:
if platforms.is_linux():
os.mknod(logfile)
else:
with open(logfile, 'w+') as fhandle:
fhandle.write('----Windows File Creation ----\n')
elif os.path.isfile(logfile) is False:
raise err.LoggerException(
'The log file exists. But it\'s not regular file'
)
loggerman._config_filelogger(
loglevel, logfile, logtype, maxlogsize, bprint_console, gen_wf
) # too many arg pylint: disable=w0212
info('-' * 20 + 'Log Initialized Successfully' + '-' * 20)
global G_INITED_LOGGER
G_INITED_LOGGER.append(loggername)
else:
print('[{0}:{1}] init_comlog has been already initalized'.format(
_file(1), _line(1))
)
return
@decorators.needlinux
def _real_rmrf(fpath, safemode):
"""
real rmrf
"""
if safemode:
if os.path.normpath(os.path.abspath(fpath)) == '/':
raise err.ShellException('cannot rmtree root / under safemode')
if os.path.isfile(fpath):
os.unlink(fpath)
else:
shutil.rmtree(fpath)
return _real_rmrf(fpath, safemode)
self._fpath = fpath
self._locktype = locktype
self._fhandle = None
try:
# if FILELOCK_EXCLUSIVE == locktype:
# self._fhandle = os.open(
# self._fpath, os.O_CREAT|os.O_EXCL|os.O_RDWR
# )
# else:
self._fhandle = os.open(
self._fpath, os.O_CREAT | os.O_RDWR
)
except IOError as error:
raise err.LockFileError(error)
except OSError as error:
raise err.LockFileError(error)
except Exception as error:
raise err.LockFileError(
'catch unkown error type:{0}'.format(error)
)
def terminate(self, times=15):
"""
asynchrously terminate the thread.
Return True if the termination is successful or the thread is already
stopped. Return False, otherwise.
:times:
retry times until call for failure.
"""
cnt = 0
while self.isAlive():
self.raise_exc(cup.err.ThreadTermException)
time.sleep(1)
cnt += 1
if cnt > times:
return False
return True
def backtrace_error(msg, back_trace_len=0):
"""
error msg with backtarce support
"""
try:
msg = _log_file_func_info(msg, back_trace_len)
loggerman = _LoggerMan()
loggerman._getlogger().error(msg)
except err.LoggerException:
return
except Exception as error:
_fail_handle(msg, error)
def do_recv_data(self, data, data_len):
"""
push data into the recving_msg queue
network read should be in 1 thread only.
"""
if self._recving_msg is None:
raise cup.err.NotInitialized('self._recving_msg')
try:
ret = self._recving_msg.push_data(data)
except IndexError as error:
log.warn('index error/msg len error happened:{0}'.format(error))
log.warn(traceback.format_exc())
log.warn('receive a msg that cannot handle, close the socket')
self.to_destroy()
return
if ret < 0:
log.warn(
'receive an wrong socket msg, to close the peer:{0}'.format(
self.get_peerinfo()
)
)
self.to_destroy()
self._conn.cleanup_error_context(self)
# Authors: Guannan Ma (@mythmgn),
"""
:desc:
Const variable for internal use. Use it when you know
what this _const really means
"""
from __future__ import print_function
import cup
from cup import version
# pylint:disable = R0903
class _const(object):
"""
internal const class
"""
class ConstError(cup.err.BaseCupException):
"""
const error
"""
def __init__(self, msg=''):
msg = 'Cup const error: %s.' % msg
cup.err.BaseCupException.__init__(self, msg)
def __setattr__(self, key, value):
if not key.isupper():
raise self.ConstError('Const value shoule be upper')
if key in self.__dict__:
raise self.ConstError('Const value cannot be changed')
self.__dict__[key] = value
# you can access CUP const like below:
# from cup import const
def wrapper(self, *args, **kwargs):
"""
internal wrapper for wrap_exceptions
"""
try:
return fun(self, *args, **kwargs)
except EnvironmentError as error:
# ENOENT (no such file or directory) gets raised on open().
# ESRCH (no such process) can get raised on read() if
# process is gone in meantime.
err = sys.exc_info()[1]
if err.errno in (errno.ENOENT, errno.ESRCH):
# pylint: disable=W0212
raise cup.err.NoSuchProcess(self._pid, self._process_name)
if err.errno in (errno.EPERM, errno.EACCES):
raise cup.err.ResException('EPERM or EACCES')
raise error
return wrapper
@classmethod
def get_async_run_status(cls, async_content):
"""
get the process status of executing async cmd
:return:
None if the process has finished.
Otherwise, return a object of linux.Process(async_pid)
"""
try:
async_process = linux.Process(async_content.pid)
res = async_process.get_process_status()
except err.NoSuchProcess:
res = None
return res