summaryrefslogtreecommitdiff
path: root/portato/backend/catapult
diff options
context:
space:
mode:
Diffstat (limited to 'portato/backend/catapult')
-rw-r--r--portato/backend/catapult/__init__.py19
-rw-r--r--portato/backend/catapult/package.py156
-rw-r--r--portato/backend/catapult/system.py252
3 files changed, 0 insertions, 427 deletions
diff --git a/portato/backend/catapult/__init__.py b/portato/backend/catapult/__init__.py
deleted file mode 100644
index 348b941..0000000
--- a/portato/backend/catapult/__init__.py
+++ /dev/null
@@ -1,19 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# File: portato/backend/catapult/__init__.py
-# This file is part of the Portato-Project, a graphical portage-frontend.
-#
-# Copyright (C) 2007 René 'Necoro' Neumann
-# This is free software. You may redistribute copies of it under the terms of
-# the GNU General Public License version 2.
-# There is NO WARRANTY, to the extent permitted by law.
-#
-# Written by René 'Necoro' Neumann <necoro@necoro.net>
-
-from __future__ import absolute_import
-
-from dbus.mainloop.glib import DBusGMainLoop
-DBusGMainLoop(set_as_default=True)
-
-from .system import CatapultSystem
-from .package import CatapultPackage
diff --git a/portato/backend/catapult/package.py b/portato/backend/catapult/package.py
deleted file mode 100644
index 2e4f471..0000000
--- a/portato/backend/catapult/package.py
+++ /dev/null
@@ -1,156 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# File: portato/backend/catapult/package.py
-# This file is part of the Portato-Project, a graphical portage-frontend.
-#
-# Copyright (C) 2007-2008 René 'Necoro' Neumann
-# This is free software. You may redistribute copies of it under the terms of
-# the GNU General Public License version 2.
-# There is NO WARRANTY, to the extent permitted by law.
-#
-# Written by René 'Necoro' Neumann <necoro@necoro.net>
-
-from __future__ import absolute_import, with_statement
-
-from ..package import Package
-from .. import flags
-from .. import system
-from ..exceptions import BlockedException, PackageNotFoundException
-from ...helper import debug, unique_array
-
-import dbus
-import catapult
-
-import os.path
-
-class CatapultPackage(Package):
-
- bus = dbus.SessionBus()
- dbus_object = bus.get_object(catapult.get_dbus_address(catapult.DEFAULT), catapult.CATAPULT_PACKAGE_BUS, follow_name_owner_changes = True)
- proxy = dbus.Interface(dbus_object, catapult.CATAPULT_PACKAGE_IFACE)
-
- _expand = {}
-
- def _new_flags (self):
- flags = self.get_new_use_flags()
-
- nflags = []
-
- for flag in flags:
- if flag[0] == "~":
- nflags.append((flag[1:], True))
- else:
- nflags.append((flag, False))
-
- return nflags
-
- def use_expanded (self, flag, suggest = None):
-
- exp = self._expand.get(flag, False)
-
- if exp is False:
- if not suggest:
- suggest = ""
- s = str(self.proxy.use_expanded(self.get_cpv(), flag, suggest))
- if not s:
- s = None
-
- self._expand[flag] = s
- return s
- else:
- return exp
-
- def get_package_path(self):
- return str(self.proxy.get_package_path(self.get_cpv()))
-
- def is_installed(self):
- return self.proxy.is_installed(self.get_cpv())
-
- def is_overlay(self):
- return self.proxy.is_in_overlay(self.get_cpv())
-
- def get_overlay_path(self):
- return str(self.proxy.get_overlay_path(self.get_cpv()))
-
- def is_in_system (self):
- return self.proxy.is_in_system(self.get_cpv())
-
- def is_missing_keyword(self):
- return self.proxy.is_missing_keyword(self.get_cpv())
-
- def is_testing(self, use_keywords = False):
- if not use_keywords:
- return self.proxy.is_testing(self.get_cpv(), False)
- else:
- status = flags.new_testing_status(self.get_cpv())
- if status is None:
- return self.proxy.is_testing(self.get_cpv(), True)
- else:
- return status
-
- def is_masked (self, use_changed = True):
- if use_changed:
- status = flags.new_masking_status(self.get_cpv())
- if status != None: # we have locally changed it
- if status == "masked": return True
- elif status == "unmasked": return False
- else:
- error(_("BUG in flags.new_masking_status. It returns \'%s\'"), status)
- else: # we have not touched the status
- return self.proxy.is_masked(self.get_cpv())
- else: # we want the original portage value XXX: bug if masked by user AND by system
- if self.proxy.is_masked(self.get_cpv()):
- if not flags.is_locally_masked(self, changes = False): # assume that if it is locally masked, it is not masked by the system
- return True
-
- return False
-
- def get_masking_reason (self):
- return str(self.proxy.get_masking_reason(self.get_cpv()))
-
- def get_iuse_flags (self, installed = False, removeForced = True):
- return [str(x) for x in self.proxy.get_iuse_flags(self.get_cpv(), installed, removeForced)]
-
- def get_matched_dep_packages (self, depvar):
- return [str(x) for x in self.proxy.get_matched_dep_packages(self.get_cpv(), self._new_flags())]
-
- def get_dep_packages (self, depvar = ["RDEPEND", "PDEPEND", "DEPEND"], with_criterions = False):
- pkgs = self.proxy.get_dep_packages(self.get_cpv(), depvar, self._new_flags())
-
- if not with_criterions:
- return [str(x) for x,y in pkgs]
- else:
- return [(str(x),str(y)) for x,y in pkgs]
-
- def get_global_settings(self, key, installed = True):
- return str(self.proxy.get_global_settings(self.get_cpv(), key, installed))
-
- def get_ebuild_path(self):
- return str(self.proxy.get_ebuild_path(self.get_cpv()))
-
- def get_package_settings(self, var, tree = True):
- return str(self.proxy.get_package_settings(self.get_cpv(), var, tree))
-
- def get_installed_use_flags(self):
- return self.proxy.get_installed_use_flags(self.get_cpv())
-
- def get_actual_use_flags(self):
- return self.proxy.get_actual_use_flags(self.get_cpv(), self._new_flags())
-
- def compare_version(self, other):
- return self.proxy.compare_version(self.get_cpv(), other.get_cpv())
-
- def matches (self, criterion):
- return self.proxy.matches(self.get_cpv(), criterion)
-
- def get_files (self):
- return self.proxy.get_files(self.get_cpv())
-
- def get_name(self):
- return str(self.proxy.get_name(self.get_cpv()))
-
- def get_version(self):
- return str(self.proxy.get_version(self.get_cpv()))
-
- def get_category(self):
- return str(self.proxy.get_category(self.get_cpv()))
diff --git a/portato/backend/catapult/system.py b/portato/backend/catapult/system.py
deleted file mode 100644
index 3a3bac5..0000000
--- a/portato/backend/catapult/system.py
+++ /dev/null
@@ -1,252 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# File: portato/backend/catapult/system.py
-# This file is part of the Portato-Project, a graphical portage-frontend.
-#
-# Copyright (C) 2007-2008 René 'Necoro' Neumann
-# This is free software. You may redistribute copies of it under the terms of
-# the GNU General Public License version 2.
-# There is NO WARRANTY, to the extent permitted by law.
-#
-# Written by René 'Necoro' Neumann <necoro@necoro.net>
-
-from __future__ import absolute_import
-
-import re, os
-from threading import Event
-import dbus
-import catapult
-
-from .package import CatapultPackage
-from ..system_interface import SystemInterface
-from ...helper import debug, info, warning, unique_array
-
-class CatapultSystem (SystemInterface):
-
- def __init__ (self):
- SystemInterface.__init__(self)
-
- self.bus = dbus.SessionBus()
- # get the system
- so = self.bus.get_object(catapult.get_dbus_address(catapult.DEFAULT), catapult.CATAPULT_SYSTEM_BUS, follow_name_owner_changes = True)
- self.proxy = dbus.Interface(so, catapult.CATAPULT_SYSTEM_IFACE)
-
- def get_version (self):
- admint = dbus.Interface(self.bus.get_object(catapult.get_dbus_address(catapult.DEFAULT), catapult.CATAPULT_BUS), catapult.CATAPULT_ADMIN_IFACE)
- return "Catapult: %s v. %s" % (self.proxy.bus_name.split(".")[-1], str(admint.version()))
-
- def geneticize_list (self, list_of_packages, only_cpv = False):
- """Convertes a list of cpv's into L{backend.Package}s.
-
- @param list_of_packages: the list of packages
- @type list_of_packages: string[]
- @param only_cpv: do nothing - return the passed list
- @type only_cpv: boolean
- @returns: converted list
- @rtype: PortagePackage[]
- """
-
- if not only_cpv:
- return [CatapultPackage(x) for x in list_of_packages]
- else:
- return [str(x) for x in list_of_packages]
-
-
- def split_cpv (self, cpv):
- split = self.proxy.split_cpv(cpv)
- if all(split):
- return map(str, split)
- else:
- return None
-
- def cpv_matches (self, cpv, criterion):
- return CatapultPackage(cpv).matches(criterion)
-
- def find_best(self, list, only_cpv = False):
- if only_cpv:
- return str(self.proxy.find_best(list))
- else:
- return CatapultPackage(self.proxy.find_best(list))
-
- def find_best_match (self, search_key, masked = False, only_installed = False, only_cpv = False):
- p = self.proxy.find_best_match(search_key, masked, only_installed)
-
- if p :
- if not only_cpv:
- return CatapultPackage(p)
- else:
- return str(p)
- return None
-
- def _wrap_find(self, key, masked, set, withVersion, only_cpv):
-
- l = []
- try:
- l = self.proxy.find_packages(key, set, masked, withVersion)
- except dbus.DBusException, e:
- name, data = str(e).split("\n")[-2].split(": ")[1:]
-
- if name == catapult.CATAPULT_ERR_AMBIGUOUS_PACKAGE:
- debug("Ambigous packages: %s.", data)
- l = []
- for cp in data.split(","):
- l += self.proxy.find_packages(cp, set, masked, withVersion)
- else:
- raise
-
- return self.geneticize_list(l, not(withVersion) or only_cpv)
-
- def find_packages (self, search_key, masked = False, only_cpv = False):
- return self._wrap_find(search_key, masked, "all", True, only_cpv)
-
- def find_installed_packages (self, search_key, masked = False, only_cpv = False):
- return self._wrap_find(search_key, masked, "installed", True, only_cpv)
-
- def find_system_packages (self, only_cpv = False):
-# result = self.proxy.find_system_packages()
-# if only_cpv:
-# return result
-# else:
-# return tuple(map(self.geneticize_list, result))
- return (self._wrap_find(search_key, False, "system", True, only_cpv), [])
-
- def find_world_packages (self, only_cpv = False):
-# result = self.proxy.find_world_packages()
-# if only_cpv:
-# return result
-# else:
-# return tuple(map(self.geneticize_list, result))
- return (self._wrap_find(search_key, False, "world", True, only_cpv), [])
-
- def _wrap_find_all (self, key, masked, set, withVersion, only_cpv):
- if not key:
- key = ""
- else:
- key = "*%s*" % key
-
- l = self.proxy.find_packages("", set, masked, withVersion)
-
- if key:
- l = catapult.filter_list(key, l)
-
- return self.geneticize_list(l, not(withVersion) or only_cpv)
-
- def find_all_installed_packages (self, name = None, withVersion = True, only_cpv = False):
- return self._wrap_find_all(name, True, "installed", withVersion, only_cpv)
-
- def find_all_uninstalled_packages (self, name = None, only_cpv = False):
- return self._wrap_find_all(name, True, "uninstalled", True, only_cpv)
-
- def find_all_packages (self, name = None, withVersion = True, only_cpv = False):
- return self._wrap_find_all(name, True, "all", withVersion, only_cpv)
-
- def find_all_world_packages (self, name = None, only_cpv = False):
- return self._wrap_find_all(name, True, "world", withVersion, only_cpv)
-
- def find_all_system_packages (self, name = None, only_cpv = False):
- return self._wrap_find_all(name, True, "system", withVersion, only_cpv)
-
- def list_categories (self, name = None):
- cats = self.proxy.list_categories()
- if name:
- cats = catapult.filter_list("*%s*" % name, cats)
-
- return map(str, cats)
-
- def sort_package_list(self, pkglist):
- return self.geneticize_list(self.proxy.sort_package_list([x.get_cpv() for x in pkglist]))
-
- def reload_settings (self):
- return self.proxy.reload_settings()
-
- def update_world (self, newuse = False, deep = False):
-
- ret = []
- e = Event()
-
- def wait (list):
- ret.extend([(CatapultPackage(x), CatapultPackage(y)) for x,y in list])
- e.set()
-
- def error (ex):
- e.set()
- raise ex
-
- self.proxy.update_world(newuse, deep, {}, reply_handler = wait, error_handler = error, timeout = 300)
- e.wait()
- return ret
- # return [(CatapultPackage(x), CatapultPackage(y)) for x,y in self.proxy.update_world(newuse, deep, {}, timeout = 300)]
-
- def get_updated_packages (self):
- ret = []
- e = Event()
-
- def wait (list):
- ret.extend([CatapultPackage(x) for x in list])
- e.set()
-
- def error (ex):
- e.set()
- raise ex
-
- self.proxy.get_updated_packages(reply_handler = wait, error_handler = error, timeout = 300)
- e.wait()
- return ret
-
- def get_use_desc (self, flag, package = None):
- if not package:
- package = ""
- return str(self.proxy.get_use_desc(flag, package))
-
- def get_global_settings(self, key):
- return str(self.proxy.get_global_settings(key))
-
- def new_package (self, cpv):
- return CatapultPackage(cpv)
-
- def get_config_path (self):
- return str(self.proxy.get_config_path())
-
- def get_sync_command (self):
- return [str(x) for x in self.proxy.get_sync_command()]
-
- def get_merge_command (self):
- return [str(x) for x in self.proxy.get_merge_command()]
-
- def get_oneshot_option (self):
- return [str(x) for x in self.proxy.get_oneshot_option()]
-
- def get_newuse_option (self):
- return [str(x) for x in self.proxy.get_newuse_option()]
-
- def get_deep_option (self):
- return [str(x) for x in self.proxy.get_deep_option()]
-
- def get_update_option (self):
- return [str(x) for x in self.proxy.get_update_option()]
-
- def get_pretend_option (self):
- return [str(x) for x in self.proxy.get_pretend_option()]
-
- def get_unmerge_option (self):
- return [str(x) for x in self.proxy.get_unmerge_option()]
-
- def get_environment (self):
- default_opts = self.get_global_settings("EMERGE_DEFAULT_OPTS")
- opts = dict(os.environ)
-
- if default_opts:
- opt_list = default_opts.split()
- changed = False
-
- for option in ["--ask", "-a", "--pretend", "-p"]:
- if option in opt_list:
- opt_list.remove(option)
- changed = True
-
- if changed:
- opts.update(EMERGE_DEFAULT_OPTS = " ".join(opt_list))
-
- opts.update(TERM = "xterm")
-
- return opts