path: root/pmstestsuite/cli.py

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

from __future__ import print_function

import os, os.path, shlex
import gobject

from optparse import OptionParser

from . import PV
from .library import load_library
from .repository import NewEbuildRepository
from .repository.pms_eclass import get_common_eclass_files
from .output import get_output_modules, TestResult
from .pm import get_package_managers

class PMSTestSuiteCLI(object):
	"""
	A class encapsulating the CLI routines used by the PMS Test Suite.

	>>> cli = PMSTestSuiteCLI()

	@ivar optparser: the option parser
	@type optparser: optparse.OptionParser
	@ivar library: the test library
	@type library: L{TestLibrary}
	@ivar pm: the active package manager
	@type pm: L{PackageManager}
	@ivar repository: the active ebuild repository
	@type repository: L{EbuildRepository}
	"""

	_defpm = 'portage'

	def __init__(self):
		"""
		Initialize the class. Set the option parser up.
		"""
		opt = OptionParser(version = PV)

		def add_pmopts(opt, optstr, optval, parser):
			try:
				p = parser.values.pm[-1]
			except IndexError:
				p = self._defpm
			if not hasattr(parser.values, 'pmopts'):
				setattr(parser.values, 'pmopts', {})
			parser.values.pmopts[p] = optval

		opt.add_option('-l', '--library', dest='library_name',
				help='Test library to use (default: standard)',
				default='standard')
		opt.add_option('-L', '--limit-packages', dest='limit_pkgs', type='int',
				help='Limit the number of packages merged at once')
		opt.add_option('-M', '--no-manifests', dest='no_manifests',
				help='Disable Manifest generation/updates',
				action='store_true', default=False)
		opt.add_option('-o', '--output-module', dest='outputmod',
				help='Output module to use',
				default='cli')
		opt.add_option('-O', '--output-file', dest='outputfile',
				help='File to write output to (may not be used)')
		opt.add_option('-p', '--package-manager', dest='pm',
				help='Package manager to use (can be specified multiple times)',
				action='append', default=[])
		opt.add_option('-P', '--pm-options', callback=add_pmopts,
				help='Additional options to pass to the Package Manager (shell-quoted,'
					' apply to the parser specified before it)',
				action='callback', type='string')
		opt.add_option('-r', '--repository-name', dest='repo_name',
				help='Name of the repository to store ebuilds in')
		opt.add_option('-R', '--repository-path', dest='repo_path',
				help='Path to the repository to store ebuilds in')
		opt.add_option('-t', '--tests', dest='tests',
				help='Limit the tests to be run (may be specified multiple times)',
				action='append', default=[])
		opt.add_option('-T', '--thorough', dest='thorough',
				help='Run tests thoroughly (i.e. in all possible variants)',
				action='store_true', default=False)
		opt.add_option('-U', '--unsupported-eapis', dest='undefined',
				help='Run tests in undefined-behavior EAPIs as well',
				action='store_true', default=False)
		opt.add_option('-v', '--verbose', dest='verbose',
				help='Report test results verbosely (even if they succeed)',
				action='store_true', default=False)

		self.optparser = opt

	def _start(self, prog, *args):
		"""
		Initialize the program. Parse command-line args. Instantiate classes
		encapsulating the Package Manager, test library and repository.

		@param prog: the program name (C{argv[0]})
		@type prog: string
		@param args: command line arguments (C{argv[1:]})
		@type args: strings
		"""
		opt = self.optparser
		opt.prog = os.path.basename(prog)
		(opts, args) = opt.parse_args(list(args))

		if not opts.pm:
			opts.pm = [self._defpm]
		if not hasattr(opts, 'pmopts'):
			setattr(opts, 'pmopts', {})

		if opts.repo_path and opts.repo_name:
			opt.error('--repository-path and --repository-name are mutually exclusive')

		for x in get_output_modules():
			if x.name == opts.outputmod:
				self.output = x(opts.outputfile)
				break
		else:
			opt.error('Output module not available: %s' % opts.outputmod)

		pmset = set(opts.pm)
		pms = []
		for x in get_package_managers():
			if x.name in pmset:
				try:
					pm = x.inst()
				except Exception as e:
					opt.error(e)
				pm.pm_options = shlex.split(opts.pmopts[x.name]) \
						if x.name in opts.pmopts else []
				pms.append(pm)
				pmset.remove(x.name)
		if pmset:
			opt.error('PM(s) not supported: %s' % ', '.join(pmset))

		self.pms = sorted(pms, key = lambda x: opts.pm.index(x.name))

		os.umask(0o22)

		try:
			if opts.repo_path:
				self.repository = NewEbuildRepository(opts.repo_path, 'pms-test-suite')
				for pm in self.pms:
					pm.append_repository(self.repository)
			else:
				if not opts.repo_name:
					opts.repo_name = 'pms-test-suite'
				for pm in self.pms:
					# XXX: all PMs should match it...
					self.repository = pm.get_repository(opts.repo_name)
					break
		except (EnvironmentError, KeyError, ValueError) as e:
			opt.error('Repository open failed: %s' % e)

		try:
			self.test_library = load_library(opts.library_name,
					thorough = opts.thorough, undefined = opts.undefined)
		except (ImportError, TypeError) as e:
			opt.error('Test library load failed: %s' % e)

		for pm in self.pms:
			pm.package_limit = opts.limit_pkgs
		self.update_manifests = not opts.no_manifests
		self.verbose = opts.verbose

		limit_tests = set()
		for t in opts.tests:
			limit_tests.update(t.split())
		if limit_tests:
			self.test_library.limit_tests(limit_tests)

	def _print_stage(self, text):
		print('-> [%s] %s...' % (self.pm.name, text))

	def tests_done(self):
		self.pm.reload_config()
		self._print_stage('Checking test results')
		results = {}
		for t in self.test_library:
			tr = TestResult(t, self.pm)

			if tr:
				outc = '.'
			elif tr.exception:
				outc = 'E'
				raise tr.exception
			else:
				outc = 'F'
			print(outc, end='')

			results[t] = tr
			t.clean(self.pm)
		self.results[self.pm] = results
		print('')

		if self.pm.has_pending_actions:
			self._print_stage('Unmerging test ebuilds')
			self.pm.commit(self.prepare)
		else:
			self.prepare()

	def all_done(self):
		ret = self.output(self.results, verbose = self.verbose)

		self.ret = 0 if ret else 1
		self.loop.quit()

	def start_pm(self):
		self._print_stage('Running PM')
		for t in self.test_library:
			t.start(self.pm)
		self.pm.commit(self.tests_done)

	def pre_unmerge_done(self):
		self.pm.reload_config()
		for t in self.test_library:
			t.clean(self.pm)
		if self.pm.has_pending_actions:
			print('Failed to unmerge the following test ebuilds:')
			print(' '.join(self.pm.pkg_queue))
			print('Refusing to proceed.')
			self.loop.quit()
			return
		self.start_pm()

	def prepare(self, first = False):
		try:
			self.pm = next(self.pm_iter)
		except StopIteration:
			self.all_done()
		else:
			if not first:
				self.pm.reload_config()
			for t in self.test_library:
				t.clean(self.pm)

			if self.pm.has_pending_actions:
				print('-> Unmerging already-merged test ebuilds...')
				self.pm.commit(self.pre_unmerge_done)
			else:
				self.start_pm()

	def main(self, argv):
		self._start(*argv)

		print('-> Generating ebuilds...')
		files = {}
		for t in self.test_library:
			files.update(t.get_output_files())
		if len(self.test_library) == 0:
			print('No tests found (?!), refusing to proceed.')
			return 1

		files.update(get_common_eclass_files())
		files.update(self.test_library.get_common_files())

		self.repository.write_files(files)
		if self.update_manifests:
			needs_manifests = False
			for pm in self.pms:
				needs_manifests |= pm.requires_manifests
				try:
					self.repository.remanifest(files, pm)
				except NotImplementedError:
					pass
				else:
					break
			else:
				if needs_manifests:
					print('No PM was able to do the Manifests, failing.')
					return 1

		self.pm_iter = iter(self.pms)
		self.results = {}
		self.prepare(first = True)

		self.ret = 1
		self.loop = gobject.MainLoop()
		self.loop.run()

		return self.ret