path: root/pmstestsuite/pm/__init__.py

#	vim:fileencoding=utf-8
# (c) 2011 Michał Górny <mgorny@gentoo.org>
# Released under the terms of the 2-clause BSD license.

"""
>>> pms = get_package_managers()
>>> 'portage' in [x.name for x in pms]
True
>>> str(pms[0]) # doctest: +ELLIPSIS
'...'
"""

from abc import abstractmethod, abstractproperty
from gentoopm.util import ABCObject
import gobject

from pmstestsuite.repository import EbuildRepository

class PackageManagerOps:
	merge = 0
	unmerge = 1

class PackageManager(ABCObject):
	"""
	Base class for various package managers support.
	"""

	pm_options = []
	package_limit = None

	_current_op = None

	def __init__(self):
		self.repo_paths = []
		self.pkg_queue = []

	@abstractproperty
	def name(self):
		"""
		Human-readable, short PM name (used in option values).

		@type: string
		"""
		pass

	@abstractproperty
	def requires_manifests(self):
		"""
		Whether the PM requires Manifests.

		@type: bool
		"""
		pass

	@classmethod
	def is_available(cls):
		"""
		Check whether a particular PM is installed and executable.

		@return: True if the PM is available, False otherwise.
		@rtype: bool
		"""
		return True

	@abstractmethod
	def remanifest(self, paths):
		"""
		Regenerate Manifests in paths.

		@param paths: directory names to regenerate manifests in
		@type paths: iterable(string)
		"""
		pass

	def append_repository(self, repo):
		"""
		Append the repository to the PM's list of repos.

		@param repo: the repository to append
		@type repo: L{EbuildRepository}
		@raise NotImplementedError: if a particular PM doesn't support
			appending repos
		"""
		self.repo_paths.append(repo.path)

	def get_repository(self, name):
		"""
		Get a repository by name, using the PM's repository list.

		@param name: name to look up
		@type name: string
		@return: a repository instance
		@rtype: L{EbuildRepository}
		@raise KeyError: when no repository with matching name exists
		"""

		r = self.repositories[name].path
		if not r:
			raise KeyError('Repository %s not found.' % name)
		return EbuildRepository(r)

	@property
	def op(self):
		return self._current_op

	@op.setter
	def op(self, op):
		if self._current_op is not None:
			if self._current_op == op:
				return
			raise Exception('Please commit the pending operation first.')
		self._current_op = op

	@op.deleter
	def op(self):
		self._current_op = None
		self.pkg_queue = []

	def _append_cpvs(self, cpvs):
		if isinstance(cpvs, basestring):
			self.pkg_queue.append(cpvs)
		else:
			self.pkg_queue.extend(cpvs)

	def merge(self, cpvs):
		"""
		Queue merging CPVs.

		@param cpvs: CPVs to merge
		@type cpvs: string/list(string)
		"""
		self.op = PackageManagerOps.merge
		self._append_cpvs(cpvs)

	def unmerge(self, cpvs):
		"""
		Queue unmerging CPVs.

		@param cpvs: CPVs to unmerge
		@type cpvs: string/list(string)
		"""
		self.op = PackageManagerOps.unmerge
		self._append_cpvs(cpvs)

	@abstractmethod
	def _spawn_merge(self, pkgs):
		"""
		Spawn PM to perform the merge of packages.
		
		@param pkgs: packages to merge
		@type pkgs: iterable(string)

		@return: the merging subprocess
		@rtype: C{subprocess.Popen}
		"""
		pass

	@abstractmethod
	def _spawn_unmerge(self, pkgs):
		"""
		Spawn PM to perform unmerge of packages.

		@param pkgs: packages to merge
		@type pkgs: iterable(string)

		@return: the merging subprocess
		@rtype: C{subprocess.Popen}
		"""
		pass

	def _subprocess_done(self, pid, exitcode, callback):
		assert(pid == self._curr_pid)

		if self.op == PackageManagerOps.merge:
			if self.package_limit and self.pkg_queue:
				self.commit(callback)
				return

		del self.op
		# need to reload the config to see updated vardb
		self.reload_config()
		callback()

	@property
	def has_pending_actions(self):
		"""
		Return whether PM has any pending actions (i.e. packages
		to merge/unmerge).

		@type: bool
		"""
		return bool(self.pkg_queue)

	def commit(self, callback):
		"""
		Perform queued operations in the background (using glib), starting PM
		multiple times if necessary. Call callback when done.

		If PM has no pending actions, the callback will be called immediately.

		@param callback: function to call when done
		@type callback: func()
		"""

		if not self.has_pending_actions:
			callback()
			return

		if self.op == PackageManagerOps.merge:
			if self.package_limit:
				pkgs = self.pkg_queue[:self.package_limit]
				del self.pkg_queue[:self.package_limit]
			else:
				pkgs = self.pkg_queue
			subp = self._spawn_merge(pkgs)
		elif self.op == PackageManagerOps.unmerge:
			subp = self._spawn_unmerge(self.pkg_queue)
		else:
			raise AssertionError('PackageManager.op unmatched')

		self._curr_pid = subp.pid
		gobject.child_watch_add(subp.pid, self._subprocess_done,
				callback)

class NonAvailPM(object):
	def __init__(self, name, pkg):
		self.name = name
		self.pkg = pkg

	def is_available(self):
		return False

class PMWrapper(object):
	""" A wrapper class which stringifies into a name of particular PM. """
	def __init__(self, pmclass):
		"""
		Instantiate the wrapper for a PM class.
		
		@param pmclass: the PM class
		@type pmclass: class(L{PackageManager})
		"""
		self._pmclass = pmclass

	@property
	def name(self):
		"""
		The name of associated PM.
		
		@type: string
		"""
		return self._pmclass.name

	def __str__(self):
		""" Return the human-readable name of associated PM. """
		if not self._pmclass.is_available():
			return '(%s)' % self._pmclass.name
		return self._pmclass.name

	def inst(self):
		"""
		Instantiate the associated PM.
		
		@return: instantiated PM
		@rtype: L{PackageManager}
		"""
		if not self._pmclass.is_available():
			raise Exception('PM is not available! Please consider installing %s.'
					% self._pmclass.pkg)
		return self._pmclass()

def get_package_managers():
	"""
	Return the list of supported Package Managers.
	
	@return: supported Package Managers
	@rtype: L{PMWrapper}
	"""

	try:
		from pmstestsuite.pm.portagepm import PortagePM
	except ImportError:
		PortagePM = NonAvailPM('portage', 'sys-apps/portage')
	try:
		from pmstestsuite.pm.pkgcorepm import PkgCorePM
	except ImportError:
		PkgCorePM = NonAvailPM('pkgcore', 'sys-apps/pkgcore')
	try:
		from pmstestsuite.pm.paludispm import PaludisPM
	except ImportError:
		PaludisPM = NonAvailPM('paludis', 'sys-apps/paludis[python]')


	return [PMWrapper(x) for x in (PortagePM, PkgCorePM, PaludisPM)]