Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def resolve_ip(ip):
from sh import grep
if ip_hostname_cache.get(ip):
return ip_hostname_cache[ip]
hostname = ip
# tricky
if ip.startswith('192.') or ip.startswith('10.'):
hosts_files = (HOSTS_FILE,)
else:
hosts_files = (HOSTS_WAN_FILE,)
for hosts_file in hosts_files:
try:
out = grep('^[^#]*%s[^0-9]' % ip, hosts_file, E=True)
if out:
hostname = out.split()[-1]
break
except Exception:
pass
ip_hostname_cache[ip] = hostname
return hostname
def parse_memory_usage(self):
"""
Parse memory usage information.
"""
st = sh.grep(sh.sysctl('-a'), 'mem')
vm = sh.vm_stat()
pattern = re.compile(':[\s]+')
st_lines = st.split('\n')
st_rows = dict()
for line in st_lines:
t = line.strip()
if t:
row = pattern.split(t)
st_rows[row[0]] = int(row[1])
vm_lines = vm.split('\n')
vm_rows = dict()
for i, line in enumerate(vm_lines):
# ignore header
if i == 0:
continue
t = line.strip()
def remote_url(git, remote, original="origin"):
origin_url = sh.grep(git.remote("-v"), original).split()[1]
repo_name = origin_url.rsplit("/", 1)[1]
return "https://github.com/{}/{}".format(remote, repo_name)
def __init__(self):
"""Loads dynamic libraries"""
try:
if 'LIBCASM' in os.environ:
libcasm_path = os.environ['LIBCASM']
else:
casm_path = find_executable('ccasm')
if platform == 'darwin':
libcasm_path = sh.grep(sh.otool('-L', casm_path), 'libcasm').split()[0]
libcasm_path = libcasm_path.replace('@loader_path', dirname(casm_path))
else:
libcasm_path = sh.grep(sh.ldd(casm_path), 'libcasm').split()[2]
libccasm_path = libcasm_path.replace('libcasm', 'libccasm')
self.lib_casm = ctypes.CDLL(libcasm_path, mode=ctypes.RTLD_GLOBAL)
self.lib_ccasm = ctypes.CDLL(libccasm_path, mode=ctypes.RTLD_GLOBAL)
except Exception as e:
print("\n~~~ Error loading casm libraries ~~~")
if 'LIBCASM' in os.environ:
libcasm_path = os.environ['LIBCASM']
print("Looking for libcasm at LIBCASM:", libcasm_path)
if not os.path.exists(libcasm_path):
print("File does not exist")
print("Install CASM if it is not installed, or update your PATH, or set LIBCASM to the location of libcasm.")
else:
def is_routing_up():
try:
sh.grep(sh.route(), 'default')
return True
except sh.ErrorReturnCode_1:
return False
def check_upgrade():
server_file = curl(PIFM_HOST + '/client_agent/pifm_agent.py')
server_sum = awk(
md5sum(
grep(server_file, '-v', '^PIFM_HOST')
), '{print $1}'
)
local_sum = awk(
md5sum(
grep('-v', '^PIFM_HOST', OUR_SCRIPT)
), '{print $1}'
)
if str(server_sum) != str(local_sum):
logging.info(
"server: {server}, local: {local}, should update.".format(
server=server_sum,
local=local_sum
)
)
def adb_devices():
outputs = sh.grep(sh.adb("devices"), "^[A-Za-z0-9]\+[[:space:]]\+device$")
raw_lists = sh.cut(outputs, "-f1")
return adb_split_stdout(raw_lists)
def check_if_package_is_installed(self, package):
"""checks if a package is installed
:param string package: package name to check
:rtype: `bool` representing whether package is installed or not
"""
lgr.debug('Checking if {0} is installed'.format(package))
r = sh.grep(sh.rpm('-qa', _ok_code=[0, 1]), package)
if r.exit_code == 0:
lgr.debug('{0} is installed'.format(package))
return True
else:
lgr.error('{0} is not installed'.format(package))
return False