You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

197 lines
7.9 KiB

"""EasyEngine package installation using apt-get module."""
import apt
import apt_pkg
import sys
import subprocess
from ee.core.logging import Log
from ee.core.apt_repo import EERepo
from sh import apt_get
from sh import ErrorReturnCode
class EEAptGet():
"""Generic apt-get intialisation"""
def update(self):
"""
Similar to `apt-get update`
"""
try:
with open('/var/log/ee/ee.log', 'a') as f:
proc = subprocess.Popen('apt-get update',
shell=True,
stdin=None, stdout=f,
stderr=subprocess.PIPE,
executable="/bin/bash")
proc.wait()
output, error_output = proc.communicate()
# Check what is error in error_output
if "NO_PUBKEY" in str(error_output):
# Split the output
Log.info(self, "Fixing missing GPG keys, please wait...")
error_list = str(error_output).split("\\n")
# Use a loop to add misising keys
for single_error in error_list:
if "NO_PUBKEY" in single_error:
key = single_error.rsplit(None, 1)[-1]
EERepo.add_key(self, key)
proc = subprocess.Popen('apt-get update',
shell=True,
stdin=None, stdout=f, stderr=f,
executable="/bin/bash")
proc.wait()
if proc.returncode == 0:
return True
else:
Log.info(self, Log.FAIL + "Oops Something went wrong!!")
Log.error(self, "Check logs for reason "
"`tail /var/log/ee/ee.log` & Try Again!!!")
except Exception as e:
Log.error(self, "apt-get update exited with error")
def check_upgrade(self):
"""
Similar to `apt-get upgrade`
"""
try:
check_update = subprocess.Popen(['apt-get upgrade -s | grep '
'\"^Inst\" | wc -l'],
stdout=subprocess.PIPE,
shell=True).communicate()[0]
if check_update == b'0\n':
Log.error(self, "No package updates available")
Log.info(self, "Following package updates are available:")
subprocess.Popen("apt-get -s dist-upgrade | grep \"^Inst\"",
shell=True, executable="/bin/bash",
stdout=sys.stdout).communicate()
except Exception as e:
Log.error(self, "Unable to check for packages upgrades")
def dist_upgrade(self):
"""
Similar to `apt-get upgrade`
"""
try:
with open('/var/log/ee/ee.log', 'a') as f:
proc = subprocess.Popen("DEBIAN_FRONTEND=noninteractive "
"apt-get dist-upgrade -o "
"Dpkg::Options::=\"--force-confdef\""
" -o "
"Dpkg::Options::=\"--force-confold\""
" -y ",
shell=True, stdin=None,
stdout=f, stderr=f,
executable="/bin/bash")
proc.wait()
if proc.returncode == 0:
return True
else:
Log.info(self, Log.FAIL + "Oops Something went "
"wrong!!")
Log.error(self, "Check logs for reason "
"`tail /var/log/ee/ee.log` & Try Again!!!")
except Exception as e:
Log.error(self, "Error while installing packages, "
"apt-get exited with error")
def install(self, packages):
all_packages = ' '.join(packages)
try:
with open('/var/log/ee/ee.log', 'a') as f:
proc = subprocess.Popen("DEBIAN_FRONTEND=noninteractive "
"apt-get install -o "
"Dpkg::Options::=\"--force-confdef\""
" -o "
"Dpkg::Options::=\"--force-confold\""
" -y {0}"
.format(all_packages), shell=True,
stdin=None, stdout=f, stderr=f,
executable="/bin/bash")
proc.wait()
if proc.returncode == 0:
return True
else:
Log.info(self, Log.FAIL + "Oops Something went "
"wrong!!")
Log.error(self, "Check logs for reason "
"`tail /var/log/ee/ee.log` & Try Again!!!")
except Exception as e:
Log.info(self, Log.FAIL + "Oops Something went "
"wrong!!")
Log.error(self, "Check logs for reason "
"`tail /var/log/ee/ee.log` & Try Again!!!")
def remove(self, packages, auto=False, purge=False):
all_packages = ' '.join(packages)
try:
with open('/var/log/ee/ee.log', 'a') as f:
if purge:
proc = subprocess.Popen('apt-get purge -y {0}'
.format(all_packages), shell=True,
stdin=None, stdout=f, stderr=f,
executable="/bin/bash")
else:
proc = subprocess.Popen('apt-get remove -y {0}'
.format(all_packages), shell=True,
stdin=None, stdout=f, stderr=f,
executable="/bin/bash")
proc.wait()
if proc.returncode == 0:
return True
else:
Log.info(self, Log.FAIL + "Oops Something went "
"wrong!!")
Log.error(self, "Check logs for reason "
"`tail /var/log/ee/ee.log` & Try Again!!!")
except Exception as e:
Log.error(self, "Error while installing packages, "
"apt-get exited with error")
def auto_clean(self):
"""
Similar to `apt-get autoclean`
"""
try:
orig_out = sys.stdout
sys.stdout = open(self.app.config.get('log.logging', 'file'),
encoding='utf-8', mode='a')
apt_get.autoclean("-y")
sys.stdout = orig_out
except ErrorReturnCode as e:
Log.debug(self, "{0}".format(e))
Log.error(self, "Unable to apt-get autoclean")
def auto_remove(self):
"""
Similar to `apt-get autoremove`
"""
try:
Log.debug(self, "Running apt-get autoremove")
apt_get.autoremove("-y")
except ErrorReturnCode as e:
Log.debug(self, "{0}".format(e))
Log.error(self, "Unable to apt-get autoremove")
def is_installed(self, package_name):
"""
Checks if package is available in cache and is installed or not
returns True if installed otherwise returns False
"""
apt_cache = apt.cache.Cache()
apt_cache.open()
if (package_name.strip() in apt_cache and
apt_cache[package_name.strip()].is_installed):
# apt_cache.close()
return True
# apt_cache.close()
return False