RPM Package Manager, CVS Repository
http://rpm5.org/cvs/
____________________________________________________________________________
Server: rpm5.org Name: Jeff Johnson
Root: /v/rpm/cvs Email: j...@rpm5.org
Module: mancoosi Date: 13-Apr-2011 13:30:25
Branch: HEAD Handle: 2011041311302500
Modified files:
mancoosi/framework rpmtest.py
Log:
- sanity.
Summary:
Revision Changes Path
1.6 +672 -231 mancoosi/framework/rpmtest.py
____________________________________________________________________________
patch -p0 <<'@@ .'
Index: mancoosi/framework/rpmtest.py
============================================================================
$ cvs diff -u -r1.5 -r1.6 rpmtest.py
--- mancoosi/framework/rpmtest.py 5 Apr 2011 11:07:57 -0000 1.5
+++ mancoosi/framework/rpmtest.py 13 Apr 2011 11:30:25 -0000 1.6
@@ -15,26 +15,40 @@
import subprocess
import csvwriter
+from copy import deepcopy
from termcolor import colored
# Globals
-initial_contents = {} # PKGNAME: [(FILE, mtime),...]
-initial_versions = {} # PKGNAME: (NEVRA, INSTALLTID)
+#initial_versions = {} # PKGNAME: (NVRA, INSTALLTID)
+
+# pkg_up_contents = {}
+# pkg_up_versions = {}
+# current_killtime = 0
+# rollback_list = []
+# upgrade_list = []
+# sgl_pkg_NEVRA = []
+# mainrelease_dir = './repo/cm14-main-release/'
+# rpm_version = 0
bin_rpm = "/bin/rpm"
var_lib_rpm = '/var/lib/rpm/'
usr_bin_dbrecover = '/usr/bin/db_recover'
usr_bin_dbstat = '/usr/bin/db_stat'
+qf_NVRA = "--qf=%{NAME}-%{VERSION}-%{RELEASE}.%{ARCH}"
+qf_NEV = "--qf='%{NAME}-%{EPOCH}:%{VERSION}'"
+qf_NEVR = "--qf='%{NAME}-%{EPOCH}:%{VERSION}-%{RELEASE}'"
+qf_INSTALLTID = "--qf='%{INSTALLTID}'"
+
+# -- Careful: E:V:R not E:V-R
+qf_EVR = "--qf='%{EPOCH}:%{VERSION}:%{RELEASE}'"
+
class Uut:
- def __init__(self, pkgset, niters=10, maxdelay=3.0, **kwargs):
- self.pkgset = pkgset
- self.npkgs = len(self.pkgset)
- self.pkgs = [p for (p, fn) in pkgset]
- print 'pkgs: ' + str(self.pkgs)
- self.rpms = [fn for (p, fn) in pkgset]
- print 'rpms: ' + str(self.rpms)
+ def __init__(self, pkgstr, pkgdir, niters=10, maxdelay=3.0, **kwargs):
+
+ self.pkgstr = pkgstr
+ self.pkgdir = pkgdir
if niters:
self.niters = niters
if maxdelay:
@@ -43,26 +57,94 @@
self.query_maxdelay = maxdelay
self.verify_maxdelay = maxdelay
self.install_maxdelay = maxdelay
+
+ self.contents = {} # PKGNAME: [(FILE,mtime), ...]
+ self.initial_sizes = {} # PKGNAME: (NVRA, fileSize(file))
+ self.to_preinstall = []
+ self.to_install = []
+
+ # Build a package set by globbing the pkgstr CSV list
+ self.ts = rpm.TransactionSet()
+ for pkgname in self.pkgstr.split(','):
+ pat = self.pkgdir + os.path.sep + pkgname + '-'
+ if string.find(pkgname, '-') < 0:
+ pat += '[0-9]'
+ pat += '*.rpm'
+ flist = glob.glob(pat)
+ for file in flist:
+ NVRA = self.pkgNVRA(file)
+ if NVRA not in self.initial_sizes.keys():
+ self.contents[NVRA] = [(f, fileTimestamp(f))
+ for f in listPkgContent(self, NVRA)]
+ self.initial_sizes[NVRA] = fileSize(file)
+ self.to_install.append((NVRA, file))
+ del self.ts
self.ts = None
+
+ # Initialize other items from the package set
+ self.pkgset = self.to_install
+ self.npkgs = len(self.pkgset)
+ self.pkgs = [p for (p, fn) in self.pkgset]
+ self.rpms = [fn for (p, fn) in self.pkgset]
+ self.rpm_version = self.getVersion()
+
self.killtimes = []
+# pkg_kill_times = {} # {pkgname, killtimes[]}
+ self.total_kill_time = 0
+
+ self.success_cases_fs = {}
+ self.failure_cases_fs = {}
+ self.num_fs_sgl_pkg_failures = {}
+ self.num_fs_sgl_pkg_successes = {}
+
+# success_cases = {}
+# failure_cases = {}
+# pkg_upgraded = {}
+# pkg_not_upgraded = {}
+# num_pkgs_upgraded = 0
+# num_pkgs_not_upgraded = 0
+
+# num_iters_pkg_upgrade = 100
+# mpl_pkg_ud = './mpl_pkg/updates/'
+# mpl_pkg_bd = './mpl_pkg/baseline/'
+
+# num_iters_pkg_upgrade = 10
+# fs_bd = './fs_pkg/baseline/'
+# fs_ud = './fs_pkg/updates/'
+
+# num_iters_pkg_upgrade = 10
+# uni_bd = './uni_pkg/baseline/'
+# uni_ud = './uni_pkg/updates/'
+
+# num_iters_pkg_upgrade = 25
+# sgl_pkg_ud = './sgl_pkg/updates/'
+# sgl_pkg_bd = './sgl_pkg/baseline/'
self.stats = {}
self.stats['tests_run'] = 0
- self.stats['tests_success'] = 0
- self.stats['tests_updated'] = 0
- self.stats['tests_nochange'] = 0
- self.stats['version_wrong'] = 0 # XXX unused
- self.stats['files_wrong'] = 0 # XXX unused
+# self.stats['version_wrong'] = 0 # XXX unused
+# self.stats['files_wrong'] = 0 # XXX unused
+# self.stats['files_modified'] = 0 # XXX unused
self.stats['stale_lock'] = 0
self.stats['recover_failed'] = 0
self.stats['script_error'] = 0
self.stats['deps_missing'] = 0
- self.stats['file_truncated'] = 0
+ self.stats['ldconfig_file_truncated'] = 0
self.stats['ldconfig_file_empty'] = 0
self.stats['config_rpmnew'] = 0
self.stats['config_rpmsave'] = 0
+ self.stats['transaction_failures'] = 0
+ self.stats['transaction_successes'] = 0
+ self.stats['transaction_all_rollback'] = 0
+ self.stats['transaction_all_succeeded'] = 0
+ self.stats['num_sgl_pkg_failures'] = 0
+ self.stats['num_sgl_pkg_successes'] = 0
+# self.stats['total_kill_time'] = 0
+ self.stats['transaction_fs_failures'] = 0
+ self.stats['transaction_fs_successes'] = 0
+
self.warning_pat = '.*warning: (.*)$'
self.error_pat = '.*error: (.*)$'
@@ -91,13 +173,34 @@
# openldap-servers
'Usage: service -[Rfshv] SERVICE ARGUMENTS',
]
-
+
def bad(self, args):
print colored(args, 'red')
+ def failed(self, arg):
+ self.bad('\tFAILED: %s' % arg)
+
def good(self, args):
print colored(args, 'green')
+ def passed(self, arg):
+ self.good('\tPASSED: %s' % arg)
+
+ def blue(self, args):
+ print colored(args, 'blue')
+
+ def white_on_red(self, args):
+ print colored(args, 'white', 'on_red')
+
+ def white_on_green(self, args):
+ print colored(args, 'white', 'on_green')
+
+ def white_on_blue(self, args):
+ print colored(args, 'white', 'on_blue')
+
+ def red_on_blue(self, args):
+ print colored(args, 'red', 'on_blue')
+
# Run a command.
def run(self, args):
# print 'Running: ' + str(args)
@@ -196,14 +299,29 @@
xx = True
spew = self.filter(stdout)
if spew:
- print '=== ' + operation + '(%s) stdout ===' % pkg
+ print '=== %s(%s) stdout ===' % (operation, pkg)
print spew
spew = self.filter(stderr)
if spew:
- print '=== ' + operation + '(%s) stderr ===' % pkg
+ print '=== %s(%s) stderr ===' % (operation, pkg)
print spew
return xx
+ # Return rpm version
+ def getVersion(self):
+ args = [bin_rpm, '--version']
+ (stdout, stderr) = self.communicate(args)
+ rpmver = 0
+ if not (stdout == None):
+ retlist = stdout.split(" ")
+ if retlist[2].startswith('4.6.0'):
+ rpmver = 4.6
+ if retlist[2].startswith('4.8.'):
+ rpmver = 4.8
+ elif retlist[2].startswith('5.3.'):
+ rpmver = 5.3
+ return rpmver
+
# Erase the pkg set.
def erase_pkgset(self):
xx = True
@@ -229,29 +347,26 @@
bgn = time.time()
args = [bin_rpm, '-U' ] + self.rpms
xx = self.run(args)
- self.install_maxdelay = 300 + (time.time() - bgn) * 1000.
+ self.install_maxdelay = 20 + (time.time() - bgn) * 1000.
bgn = time.time()
args = [bin_rpm, '-q' ] + self.pkgs
xx = self.run(args)
- self.query_maxdelay = 100 + (time.time() - bgn) * 1000.
+ self.query_maxdelay = 20 + (time.time() - bgn) * 1000.
bgn = time.time()
args = [bin_rpm, '-V' ] + self.pkgs
xx = self.run(args)
- self.verify_maxdelay = 500 + (time.time() - bgn) * 1000.
+ self.verify_maxdelay = 20 + (time.time() - bgn) * 1000.
bgn = time.time()
args = [bin_rpm, '-e' ] + self.pkgs
xx = self.run(args)
- self.erase_maxdelay = 300 + (time.time() - bgn) * 1000.
+ self.erase_maxdelay = 20 + (time.time() - bgn) * 1000.
return xx
def rollback(self, rollback_list):
- print "--> rollback(" + str(rollback_list) + ")"
-
- # rollback_list must be really a list !!
if not self.install_pkgset():
self.bad('Rollback failed!')
@@ -277,32 +392,84 @@
xx = True
return xx
- def queryByPkgName(self, name):
- if not self.ts:
- self.ts = rpm.TransactionSet()
- match = self.ts.dbMatch('name', name)
- h = None;
- if match:
- for h in match:
- break # Assumes single match
- return h
-
- def installTID(self, name):
- h = self.queryByPkgName(name)
- res = '' # None instead?
- if h:
- res = h[rpm.RPMTAG_INSTALLTID]
- return res
+#========================================================
+ def query(self, opts, pkg):
+ args = [bin_rpm, opts, pkg]
+ (stdout, stderr) = self.communicate(args)
+ return stdout
- def printNEVRA(self, name):
- h = self.queryByPkgName(name)
- res = '' # None instead?
- if h:
- E = (str(h['epoch']) + ':' if h['epoch'] else '')
- res = '%s-%s%s-%s.%s' % (h['name'], E, h['version'],
- h['release'], h['arch'])
- return res
+# def queryByPkgName(self, NVRA):
+# args = [bin_rpm, "-q", qf_NEVR, NVRA]
+# (stdout, stderr) = self.communicate(args)
+# print 'Output= %s' % stdout
+# if len(stdout) <= 0:
+# print 'No NEVRA details for this package'
+# return None
+# print 'NEVRA details found for the package installed on the system'
+# stdout = stdout.replace("'","")
+# eva_list = stdout.split(':')
+# print [eva_list[0], eva_list[1], eva_list[2]]
+# return (eva_list[0], eva_list[1], eva_list[2])
+
+#========================================================
+ # Return the NVRA from a *.rpm package file.
+ def pkgNVRA(self, fn):
+ args = [bin_rpm, "-qp", qf_NVRA, fn]
+ (stdout, stderr) = self.communicate(args)
+ NVRA = stdout
+ return NVRA
+
+# def queryPkgEVRA(fn):
+# args = [bin_rpm, "-qp", qf_EVR, fn"]
+# (stdout, stderr) = self.communicate(args)
+# print 'Output update pkg= %s' % stdout
+# if len(stdout) <= 0:
+# print 'No NEVRA details for this package'
+# print 'Error, we just passed this package directly from the list on
the filesystem'
+# return None
+# print 'NEVRA details found for package in update path repository'
+# stdout = stdout.replace("'","")
+# eva_list = stdout.split(':')
+# print [eva_list[0], eva_list[1], eva_list[2]]
+# return (eva_list[0], eva_list[1], eva_list[2])
+#========================================================
+ def installTID(self, NVRA):
+ args = [bin_rpm, "-q", qf_INSTALLTID, NVRA]
+ (stdout, stderr) = self.communicate(args)
+ if len(stdout) <= 0:
+ stdout = None
+ return stdout
+
+# def installTID_pkg(self, fn):
+# args = [bin_rpm, "-qp", qf_INSTALLTID, fn"]
+# (stdout, stderr) = self.communicate(args)
+# print 'Output update pkg= %s' % stdout
+# if len(stdout) <= 0:
+# stdout = None
+#========================================================
+# def printNEVRA(self, NVRA):
+# args = [bin_rpm, "-q", qf_NEVR, NVRA]
+# (stdout, stderr) = self.communicate(args)
+# if len(stdout) <= 0:
+# stdout = None
+# else:
+# stdout = stdout.replace("(none)","")
+# stdout = stdout.replace("-:", "-") #epoch fix
+# print 'Output update pkg= %s' % stdout
+# return stdout
+
+# def printNEVRA_pkg(self, fn):
+# args = [bin_rpm, "-qp", qf_NEVR, fn] #removed %Arch because of
installed/non installed behaviour
+# (stdout, stderr) = self.communicate(args)
+# if len(stdout) <= 0:
+# stdout = None
+# else:
+# stdout = stdout.replace("(none)","")
+# stdout = stdout.replace("-:", "-") #epoch fix
+# print 'Output update pkg= %s' % stdout
+# return stdout
+#========================================================
# Start with known pre-conditions.
def prenatal(self, opts):
# XXX necessary?
@@ -338,10 +505,241 @@
else:
res = True
return res
+#========================================================
+ # Return count of pkgs with NVRA installed
+ def countNVRA(self, NVRA):
+ args = [bin_rpm, "-q", NVRA]
+ (stdout, stderr) = self.communicate(args)
+ res = 0
+ if len(stdout) > 0:
+ lines = stdout.split("\n")
+ res = len(lines)
+ if not len(lines[res-1]): # chomp
+ res = res -1
+ return res
+#========================================================
+ def checkResultsSinglePkg(self, NVRA):
+ result = False
+ res = self.countNVRA(NVRA)
+ if res > 1: # Multiple instances installed
+ self.failed(NVRA)
+ elif res == 0: # Not installed
+ self.failed(NVRA)
+ else: # Single instance installed
+ self.passed(NVRA)
+ result = True
+ return result
+#========================================================
+ def checkResults(self):
+
+ # Initialize the return tuple
+ transaction_succeeded = False
+ total_pkgs = self.npkgs
+ successes = 0
+ failures = 0
+ all_upgraded = False
+ all_rolled_back = False
+ version_wrong = 0
+ files_wrong = 0
+ files_modified = 0
+
+ pkgs_upgraded = []
+ pkgs_rolled_back = []
+ pkgs_with_diff_NVRA = []
+
+ print 'Single package --'
+ to_return = [transaction_succeeded, total_pkgs, successes, failures,
all_upgraded, all_rolled_back, version_wrong, files_wrong, files_modified]
+ for p, file in self.pkgset:
+ if self.checkResultsSinglePkg(p):
+ successes += 1
+ else:
+ failures += 1
+
+ print ' Total(%d) = Passed(%d) + Failed(%d)' % (total_pkgs,
successes, failures)
+
+ # FIXME: For install/erase testing with strict preconditions (as now),
+ # assume "worked" if there are no "missing" tests with unknown results.
+ if (successes+failures) == total_pkgs:
+ transaction_succeeded = True
+ all_upgraded = True
+ all_rolled_back = True
+
+# del pkgs_upgraded[:]
+# del pkgs_rolled_back[:] #clear the lists
+#
+# # FIXME: needs pkg_up_versions
+# for p, file in self.pkgset:
+# if self.printNEVRA(p) == pkg_up_versions[p][0]: # check against
what we expected to install
+# pkgs_upgraded.append(p) #add pkg to successfully
upgraded/installed
+# elif self.printNEVRA(p) == initial_versions[p][0]:
+# pkgs_rolled_back.append(p)
+# else:
+# pkgs_with_diff_NVRA.append(p)
+# uut.white_on_red('A package with a different version to either
intended to be installed or initial made it on to the system??!')
+
+# if len(pkgs_upgraded) == total_pkgs:
+# transaction_succeeded = True
+# all_upgraded = True
+# uut.white_on_green('Transaction succeeded, all packages upgraded')
+# elif len(pkgs_rolled_back) == total_pkgs:
+# transaction_succeeded = True
+# all_rolled_back = True
+# uut.white_on_green('Transaction succeeded, all packages roll\'ed
back')
+# else:
+# transaction_succeeded = False
+# uut.white_on_red('Transaction failed. Some packages were upgraded,
others not. %d packages upgraded, %d packages rolled back or never installed' %
(len(pkgs_upgraded), len(pkgs_rolled_back)))
+
+ to_return = [transaction_succeeded, total_pkgs, successes, failures,
all_upgraded, all_rolled_back, version_wrong, files_wrong, files_modified]
+
+ return to_return
+#========================================================
+ def check_fs_consistency(self, NVRA):
+ contents = self.contents[NVRA]
+ pkg_up = {}
+ # FIXME: needs pkg_up_contents
+# pkg_up = deepcopy(pkg_up_contents[NVRA])
+ #contents: List of (file, timestamp) tuples
+ num_files = 0
+ num_files_modified = 0
+ num_files_not_modified = 0
+ num_files_not_existing = 0
+ num_files_with_current_time = 0
+
+ num_files_original_only = 0
+ num_files_upgrade_only = 0
+
+ contained_in_original_only = []
+ contained_in_both = []
+ contained_in_upgrade_only = []
+
+ for f in contents:
+ if f in pkg_up:
+ contained_in_both.append((f[0], f[1]))
+ pkg_up.remove(f)
+ else:
+ contained_in_original_only.append((f[0],f[1]))
+
+ for g in pkg_up:
+ if g in contents:
+ print 'Error - The element should have been found in the
first loop'
+ else:
+ contained_in_upgrade_only.append((g[0],g[1]))
+
+# print 'Number of files in package only %d' %
len(contained_in_original_only)
+# print 'Number of files in both original and upgrade package = %d' %
len(contained_in_both)
+# print 'Number of files in only the upgrade package = %d' %
len(contained_in_upgrade_only)
+
+ for f in contained_in_both:
+
+ num_files += 1
+ if os.path.exists(f[0]):
+ if fileTimestamp(f[0]) > f[1]:
+ print 'File %s was modified' % f[0]
+ original_filetime = datetime.datetime.fromtimestamp(f[1])
+ updated_filetime =
datetime.datetime.fromtimestamp(fileTimestamp(f[0]))
+ print updated_filetime
+
+ if ((datetime.datetime.now()- updated_filetime).seconds
< 3600 ) and ((datetime.datetime.now() - updated_filetime).days == 0 ):
#(updated_filetime - original_filetime) > timedelta(0)) and (
+ num_files_with_current_time += 1
+ else:
+ num_files_modified += 1
+ else:
+ #uut.bad('File %s not touched' % f[0])
+ num_files_not_modified += 1
+ else:
+ #file no longer exists on the system. Package should not
remove files unless denoted by scripts. Could check scripts at this stage
+ #uut.bad('Non-existing file %s ' % f[0])
+ num_files_not_existing += 1
+ #return False
+
+ for f in contained_in_original_only:
+ if os.path.exists(f[0]):
+ num_files_original_only += 1
+
+ for f in contained_in_upgrade_only:
+ if os.path.exists(f[0]):
+ num_files_upgrade_only += 1
+
+ old_pkg_removed = False
+ new_pkg_installed = False
+ set = False
+
+ if not len(contained_in_original_only) == 0:
+ set = True
+ if num_files_original_only == len(contained_in_original_only):
+ uut.white_on_blue('All files only in original package are
present on system')
+ old_pkg_removed = False
+ else:
+ uut.red_on_blue('All files only in original package are not
present on system')
+ old_pkg_removed = True
+
+ if not len(contained_in_upgrade_only) == 0:
+ set = True
+ if num_files_upgrade_only == len(contained_in_upgrade_only):
+ uut.white_on_blue('All files only in upgrade package are
present on system')
+ new_pkg_installed = True
+ else:
+ uut.red_on_blue('All files only in upgrade package are not
present on system')
+ new_pkg_installed = True
+
+ rm_old_inst_new = False
+ left_in_orginal_state = False
+
+ if set:
+ if new_pkg_installed and old_pkg_removed:
+ uut.white_on_green('File-system confirms full upgrade')
+ rm_old_inst_new = True
+ if not new_pkg_installed and not old_pkg_removed:
+ uut.white_on_green('File-system suggests that the package
was not upgraded')
+ left_in_original_state = True
+
+ be_pedantic = False #check that old files were removed for an
upgrade and that no new files installed for left in original state
+
+ if num_files == num_files_modified + num_files_not_existing +
num_files_with_current_time:
+ if num_files_with_current_time > 0:
+ uut.white_on_green('All files were upgraded, %d brought to
current time' % (num_files_with_current_time))
+ if be_pedantic and not rm_old_inst_new:
+ return False
+ else:
+ return True
+ elif num_files == num_files_not_modified + num_files_not_existing +
num_files_with_current_time:
+ if num_files_with_current_time > 0:
+ uut.white_on_green('Files were left untouched but %d brought
to current time' % num_files_with_current_time)
+ if be_pedantic and not left_in_original_state:
+ return False
+ else:
+ return True
+ else:
+ uut.white_on_red('%d files modified, %d files with the current
time, %d files not modified, with %d non-existing files' % (num_files_modified,
num_files_with_current_time, num_files_not_modified, num_files_not_existing))
+ return False
+#========================================================
+ def checkFSResults(self):
+
+ self.success_cases_fs = {}
+ self.failure_cases_fs = {}
+
+ print 'File system --'
+ for p, file in self.pkgset:
+ if self.check_fs_consistency(p):
+ self.passed(p)
+ if self.success_cases_fs.get(p, 0) == 0:
+ self.success_cases_fs[p] = 0
+ self.success_cases_fs[p] += 1
+ else:
+ self.failed(p)
+ if self.failure_cases_fs.get(p, 0) == 0:
+ self.failure_cases_fs[p] = 0
+ self.failure_cases_fs[p] += 1
+# xx = self.verify('-Vp', file)
+ xx = self.verify('-V', p)
+ return self.success_cases_fs
+#========================================================
# Post-mortem test results analysis.
def postmortem(self, opts):
+ res = True
+
was_updated = 0
not_updated = 0
@@ -363,41 +761,69 @@
args = ['/bin/rm', '-f', var_lib_rpm + '__db*']
xx = self.run(args)
- for (p, file) in self.pkgset:
- iniNEVRA = initial_versions[p]
- curNEVRA = self.printNEVRA(p)
- if iniNEVRA and iniNEVRA != curNEVRA:
- was_updated += 1
- else:
- not_updated += 1
- xx = self.verify('-V', p)
+ # Reopen the rpmdb
+ if self.ts:
+ del self.ts
+ self.ts = None
+ self.ts = rpm.TransactionSet()
+
+ last_kill_time = 0
+ if self.killtimes[-1] != None:
+ last_kill_time = self.killtimes[-1]
+ self.total_kill_time += last_kill_time
+ else:
+ self.bad('No kill time was set up')
+
+ # Apply single package checks
+ mpl_pkg_results = self.checkResults()
+ # Apply FS checks
+ mpl_pkg_fs_results = self.checkFSResults()
+
+ # Accumulate statistics
+ self.stats['num_sgl_pkg_failures'] += mpl_pkg_results[3]
+ self.stats['num_sgl_pkg_successes'] += mpl_pkg_results[2]
+
+ num_successes_fs = 0
+ for p in self.pkgset:
+ if mpl_pkg_fs_results.get(p[0],0) == 0:
+ pass
+ else:
+ if self.num_fs_sgl_pkg_successes.get(p[0],0) == 0:
+ self.num_fs_sgl_pkg_successes[p[0]] = 0
+ self.num_fs_sgl_pkg_successes[p[0]] += 1
+ num_successes_fs += 1
+
+ if num_successes_fs == len(self.pkgset):
+ self.stats['transaction_fs_successes'] += 1
+ self.white_on_green('File-system tests succeeded for all packages
in txn')
+ else:
+ self.stats['transaction_fs_failures'] += 1
+ num_fs_failures = len(self.pkgset) - num_successes_fs
+ self.white_on_red('File-system tests failed for %d packages in txn'
% (num_fs_failures))
+
+ if mpl_pkg_results[0] == False:
+ self.stats['transaction_failures'] += 1
+ self.bad('Transaction did not succeed')
+ self.bad('%d packages were in a reliable state out of %d packages' %
(mpl_pkg_results[2],mpl_pkg_results[1]))
+ elif mpl_pkg_results[0] == True:
+ self.stats['transaction_successes'] += 1
+ self.good('Transaction was left in a reliable state')
+ self.bad('%d packages were in a reliable state out of %d packages' %
(mpl_pkg_results[2],mpl_pkg_results[1]))
+ if mpl_pkg_results[4] == True:
+ self.white_on_green('All packages were upgraded for this
transaction')
+ self.stats['transaction_all_succeeded'] += 1
+ elif mpl_pkg_results[5] == True:
+ self.white_on_red('All packages were rolled back to the initial
state or were not installed')
+ self.stats['transaction_all_rollback'] += 1
+ # Close the rpmdb
del self.ts
self.ts = None
- if was_updated > 0 and not_updated > 0:
- res = False
- else:
- if was_updated == 0:
- self.stats['tests_nochange'] += 1
- else:
- self.stats['tests_updated'] += 1
-
- res = True
- self.stats['tests_success'] += 1
-
- print '''Within this transaction %d pkgs were updated and %d were
not updated''' \
- % (was_updated, not_updated)
-
- # print '''Results:\n\t %d out of %d Packages failed the database
test\n\t
- # %d out of %d failed the timestamps test''' % (version_wrong,
self.npkgs,files_wrong,
- # .... self.npkgs)
- # return [prettyPrintPkgList(pkgs), str(self.npkgs), str(version_wrong),
str(files_wrong), str(self.killtimes[-1])]
-
return res
-
-def listPkgContent(ts, name):
- match = ts.dbMatch('name', name)
+#========================================================
+def listPkgContent(uut, NVRA):
+ match = uut.ts.dbMatch('name', NVRA)
h = None;
if match:
for h in match:
@@ -406,41 +832,55 @@
for file in h[rpm.RPMTAG_FILENAMES]:
yield file
+#def listPkgContent_pkg(uut, fn):
+# args = [bin_rpm, "-qlp", fn]
+# (stdout, stderr) = uut.communicate(args)
+# to_return = []
+# if len(stdout) <= 0:
+# print 'No Files in this package?'
+# elif not stdout.find("not installed") == -1:
+# print 'Cannot query package as it is not installed or is a virtual
package'
+# elif not stdout.find("contains no files") == -1:
+# print 'Package is a virtual package with no files itself'
+# else:
+# to_return = stdout.split("\n")
+# for sentence in to_return:
+# if sentence == '':
+# to_return.remove('')
+# return to_return
+
+#def rpmRunAndParse(NVRA, args):
+# (stdout, stderr) = uut.communicate(args)
+# if stdout.find('not installed') == -1:
+# print 'Package %s is not found in the RPMDb' % NVRA
+# stdout = None
+# return stdout
+
+#def rpmDbMatch(NVRA):
+# args = [bin_rpm, '-q', qf_NEV, NVRA]
+# return rpmRunAndParse(name, args).split('\n')
+
+#def rpmlistPkgContent(NVRA):
+# args = [bin_rpm, '-ql', NVRA]
+# return rpmRunAndParse(NVRA, args).split('\n')
-def fileTimestamp(filePath):
- try:
- timestamp = os.stat(filePath).st_mtime
- return timestamp
- except OSError:
-
- # uut.bad("Filepath %s not found" % filePath)
-
- pass
-
-
-def dumpResultsToCsv(results, filename):
- print "--> dumpResultsToCsv(results, filename)"
- outfile = open(filename, 'w')
+def dumpResultsToCsv(results, fn):
+ outfile = open(fn, 'w')
writer = csvwriter.UnicodeWriter(f=outfile)
writer.writerows(results)
-
-def pkgNameFromNEVRA(nevra):
- list = nevra.split('-')
+def pkgNameFromNEVRA(NEVRA):
+ list = NEVRA.split('-')
res = '-'.join(list[0:-2])
- print "<-- pkgNameFromNEVRA(" + nevra + ") res " + res
return res
-
-def splitNEVR(nevra):
- list = nevra.split('-')
+def splitNEVR(NEVRA):
+ list = NEVRA.split('-')
res = ('-'.join(list[0:-2]), list[-2], list[-1])
- print "<-- splitNEVR(" + nevra + ") res " + res
return res
-
-def printNVRA(ts, name):
- match = ts.dbMatch('name', name)
+def printNVRA(uut, name):
+ match = uut.ts.dbMatch('name', name)
h = None;
if match:
for h in match:
@@ -450,86 +890,72 @@
res = '%s-%s-%s.%s' % (h['name'], h['version'], h['release'],
h['arch'])
return res
+def fileTimestamp(filePath):
+ try:
+ timestamp = os.stat(filePath).st_mtime
+ return timestamp
+ except OSError:
+ # uut.bad("Filepath %s not found" % filePath)
+ pass
-def doublePkgEntry(ts, name):
- match = ts.dbMatch('name', name)
- res = match.count()
- print "<-- doublePkgEntry(ts, " + name + ") res " + str(res)
- return res
-
-
-def checkResultsSinglePkg(uut, pkg_test):
- print 'Single Pkg test for upgrading %s, waiting time %f seg.' \
- % (pkg_test, uut.killtimes[-1])
- res = True
-
- uut.run([usr_bin_dbstat, '-Cl', '-h', var_lib_rpm])
- uut.run(['/bin/rm', '-f', var_lib_rpm + '__db*'])
-
- # After all the messup caused by (failed) installs we can't reuse the ts
- ts = rpm.TransactionSet()
- res = doublePkgEntry(ts, pkg_test)
- del ts
- if res > 1:
- uut.bad('Two versions detected for package %s' % pkg_test)
- res = False
- elif res == 0:
- uut.good('No version of package %s was found' % pkg_test)
- else:
- uut.good('Just one version of package %s was found' % pkg_test)
-
- print "<-- checkResultsSinglePkg(uut, " + str(pkg_test) + ") res " +
str(res)
- return res
-
-
-def check_fs_consistency(pkgname):
- contents = initial_contents[pkgname]
-
- # contents: List of (file, timestamp) tuples
-
- for f in contents:
- if os.path.exists(f[0]):
- if fileTimestamp(f[0]) > f[1]:
- print 'File was modified'
- else:
- uut.bad('File %s not touched' % f[0])
- res = False
- else:
- uut.bad('Non-existing file %s ' % f[0])
- res = False
- else:
-
- res = True
- print "<-- check_fs_consistency(" + pkgname + ") res " + str(res)
- return res
-
-
-def readRpmHeader(ts, filename):
- fd = os.open(filename, os.O_RDONLY)
- ts.setVSFlags(rpm.RPMVSF_NODSA | rpm._RPMVSF_NOSIGNATURES) # Ignore
signatures, please
- h = ts.hdrFromFdno(fd)
- os.close(fd)
- print "<-- readRpmHeader(ts, " + filename + ")"
- return h
-
-
-def compareWithInstalledVersion(uut, ts, pkgfile):
- h = readRpmHeader(ts, pkgfile)
- pkg_ds = h.dsOfHeader()
- inst_h = uut.queryByPkgName(h['name'])
- inst_ds = inst_h.dsOfHeader()
- pkg_evr = pkg_ds.EVR()
- instr_evr = inst_ds.EVR()
-
- if pkg_evr == instr_evr:
- res = 0
- elif pkg_evr > instr_evr:
- res = 1
- else:
- res = -1
- print "<-- compareWithInstalledVersion(ts, " + pkgfile + ") res " +
str(res)
- return res
+def fileSize(filePath):
+ try:
+ filesize = os.stat(filePath).st_size
+ return filesize
+ except OSError:
+ # uut.bad("Filepath %s not found" % filePath)
+ pass
+#def compareWithInstalledVersion(ts, pkgname, pkgfile):
+# print "--------------------------------------------------------"
+# print "--------------------------------------------------------"
+# print pkgfile
+# pkg_evr = queryPkgEVRA(pkgfile)
+# inst_evr = queryByPkgName(ts, pkgname)
+# more_recent = False
+#
+# npkg_evr0 = (re.findall(r'\d+',pkg_evr[0]))
+# npkg_evr1 = (re.findall(r'\d+',pkg_evr[1]))
+# npkg_evr2 = (re.findall(r'\d+',pkg_evr[2]))
+# ninst_evr0 = (re.findall(r'\d+',inst_evr[0]))
+# ninst_evr1 = (re.findall(r'\d+',inst_evr[1]))
+# ninst_evr2 = (re.findall(r'\d+',inst_evr[2]))
+#
+# if (npkg_evr0 > ninst_evr0):
+# more_recent = True
+# uut.white_on_blue('Package epoch is more recent')
+# elif ((npkg_evr0 == ninst_evr0) and (npkg_evr1 > ninst_evr1)):
+# more_recent = True
+# uut.white_on_blue('Package version is more recent')
+# elif ((npkg_evr0 == ninst_evr0) and (npkg_evr1 == ninst_evr1) and
(npkg_evr2 > ninst_evr2)):
+# more_recent = True
+# uut.white_on_blue('Package release is more recent')
+# else:
+# more_recent = False
+# uut.white_on_green('Package version is same as installed version')
+# return 0
+#
+#
+# if more_recent == True:
+# return 1
+# else:
+# print 'Should never have reached here'
+# return -1
+
+#def compareEVRShellMode(pkgfile):
+# pass
+
+
+#def checkSglPkgUpgraded(uut, NVRA):
+## for p, file in pkgs:
+# p = NVRA
+# if uut.printNEVRA(p) == pkg_up_versions[p][0]: # check against what we
expected to install
+# return 1
+# elif uut.printNEVRA(p) == initial_versions[p][0]:
+# return 0
+# else:
+# uut.white_on_red('A package with a different version to either
intended to be installed or initial made it on to the system??!')
+# return -1
def prettyPrintPkgList(list):
out = ''
@@ -540,22 +966,22 @@
if __name__ == '__main__':
- parser = \
- optparse.OptionParser(usage='Usage: %prog [options] PackageDir
[MaxDelay] [#Iters]...'
- , version='%prog-0.1',
- conflict_handler='resolve')
+ parser = optparse.OptionParser(
+ usage='Usage: %prog [options] PackageDir [MaxDelay] [#Iters]...',
+ version='%prog-0.1',
+ conflict_handler = 'resolve')
parser.add_option('-s', '--singlepkg', action='store_true',
- dest='singlepkg',
- help='Run test for single package transactions')
- parser.add_option('-d', '--debug', action='store_true', dest='debug'
- ,
- help='Dont actually run the tests, just print what
would be done'
- )
+ dest='singlepkg',
+ help='Run test for single package transactions')
+ parser.add_option('-d', '--debug', action='store_true',
+ dest='debug',
+ help='Dont actually run the tests, just print what would be done')
+# parser.add_option('-f', '--filesize', action='store_true',
dest='filesize', help='''Run the filesize vs success test''')
+# parser.add_option('-u', '--uniform', action='store_true',
dest='uniform', help='''Perform the test of error injection time vs
successes''')
parser.add_option('-p', '--package-set', type='int',
- dest='package_set',
- help='''Choose which package set to use for the
multi-package test: [1-10] '''
- )
+ dest='package_set',
+ help='''Choose which package set to use for the multi-package test: 1-
Emacs, 2- IM Packages, 3- Video Packages, 4- Wine, 5- Firefox''')
(opts, args) = parser.parse_args()
@@ -563,10 +989,6 @@
parser.print_help()
sys.exit(-1)
- to_install = []
- to_upgrade = []
- to_upgrade_multi = set()
-
output_file = time.strftime('%d%m%Y-%H%M%S', time.gmtime()) + '.csv'
global_results = [['List of packages', '# packages', 'Failures DB',
'Failures timestamps', 'Kill interval (sec)']]
@@ -588,40 +1010,28 @@
if opts.package_set <= 0:
opts.package_set = 1
- ts = rpm.TransactionSet()
- for pkgname in package_sets[opts.package_set - 1].split(','):
- pat = args[0] + os.path.sep + pkgname + '-'
- if string.find(pkgname, '-') < 0:
- pat += '[0-9]'
- pat += '*.rpm'
- flist = glob.glob(pat)
- for file in flist:
- if pkgname not in initial_versions.keys():
- initial_contents[pkgname] = [(f, fileTimestamp(f))
- for f in listPkgContent(ts, pkgname)]
- initial_versions[pkgname] = printNVRA(ts, pkgname)
- to_upgrade_multi.add((pkgname, file))
- del ts
+ pkgdir = args[0]
+ n_iters = (int(args[2]) if len(args) > 2 else 100)
+ max_delay = (args[1] if len(args) > 1 else 3000.)
+ pkgstr = package_sets[opts.package_set - 1]
+
+ uut = Uut(pkgstr, pkgdir, niters=n_iters, maxdelay=max_delay)
if opts.debug:
print 'Selected %d Packages for Upgrades: \n%s' \
- % (len(to_upgrade), str(to_upgrade))
- print 'Selected %d Packages for New Install: \n%s' \
- % (len(to_install), str(to_install))
- for p in to_upgrade:
+ % (len(uut.to_preinstall), str(uut.to_preinstall))
+ print 'Selected %d Packages for Install/Erase: \n%s' \
+ % (len(uut.to_install), str(uut.to_install))
+ for p in uut.to_install:
print 'Dry run of filesystem checks for ' + p[0]
- print str(check_fs_consistency(p[0]))
+ print str(uut.check_fs_consistency(p[0]))
sys.exit(0)
- n_iters = (int(args[2]) if len(args) > 2 else 10)
- max_delay = (args[1] if len(args) > 1 else 3000.)
-
- uut = Uut(to_upgrade_multi, niters=n_iters, maxdelay=max_delay)
if uut.npkgs <= 0:
sys.exit(0)
if opts.singlepkg:
- pkg_test = random.choice(to_upgrade) # Same-pkg test
+ pkg_test = random.choice(uut.to_install) # Same-pkg test
else: # Calibrate maxdelay from actual install/query/erase times
xx = uut.calibrate()
@@ -639,18 +1049,49 @@
xx = uut.postmortem('-ev')
uut.stats['tests_run'] += 1
+ if uut.to_preinstall:
+
+ # Pre-install the packages needed for an upgrade test
+ pkgset = uut.pkgset
+ npkgs = uut.npkgs
+ pkgs = uut.pkgs
+ rpms = uut.rpms
+
+ uut.pkgset = uut.to_preinstall
+ uut.npkgs = len(uut.pkgset)
+ uut.pkgs = [p for (p, fn) in uut.pkgset]
+ uut.rpms = [fn for (p, fn) in uut.pkgset]
+
+ xx = uut.prenatal('-ev')
+
+ uut.pkgset = pkgset
+ uut.npkgs = npkgs
+ uut.pkgs = pkgs
+ uut.rpms = rpms
+
+ # Run the upgrade test
+ xx = uut.test_case('-Uvh')
+ xx = uut.postmortem('-Uvh')
+ uut.stats['tests_run'] += 1
+
+ uut.white_on_blue(str(uut.stats));
+
# Final clean-up
xx = uut.prenatal('-Uvh')
- print 'Test aggregate results: %d/%d pkgs' \
- % (uut.stats['tests_success'], uut.stats['tests_run'])
-
- print 'Multiple package test details: %d transactions to completion /%d
initial state' \
- % (uut.stats['tests_updated'], uut.stats['tests_nochange'])
+ print 'For %d tests iterations on the pkgset: %s' % (uut.niters,
str(uut.pkgs))
- print 'Average killtimes: %f' % (sum(uut.killtimes) / len(uut.killtimes))
+ print 'These were the results'
+ print 'The average killtime was %f' % (uut.total_kill_time / uut.niters)
#/len(uut.pkgset)
+ uut.good('Number of successful transactions was %d' %
uut.stats['transaction_successes'])
+ if uut.stats['transaction_successes'] > 0:
+ uut.white_on_red('Of the successes, %d were all packages being left in
their initial state' % uut.stats['transaction_all_rollback'])
+ uut.white_on_green('%d were package transactions that fully completed'
% uut.stats['transaction_all_succeeded'])
+ uut.bad('Number of transactions that failed was %d' %
uut.stats['transaction_failures'])
+ uut.bad('Number of txn that failed fs tests was %d' %
uut.stats['transaction_fs_failures'])
+ uut.good('Number of txn that passed fs tests was %d' %
uut.stats['transaction_fs_successes'])
+ uut.blue('Number of individual package successes was %d and number of
failures was %d' % (uut.stats['num_sgl_pkg_successes'],
uut.stats['num_sgl_pkg_failures']))
- print "Detected failure cases:"
- print uut.stats
+ uut.white_on_blue(str(uut.stats));
# dumpResultsToCsv(global_results, output_file)
@@ .
______________________________________________________________________
RPM Package Manager http://rpm5.org
CVS Sources Repository rpm-cvs@rpm5.org
