test/pacman/pmtest.py

   1 #! /usr/bin/python2
   2 #
   3 #  Copyright (c) 2006 by Aurelien Foret <orelien@chez.com>
   4 #
   5 #  This program is free software; you can redistribute it and/or modify
   6 #  it under the terms of the GNU General Public License as published by
   7 #  the Free Software Foundation; either version 2 of the License, or
   8 #  (at your option) any later version.
   9 #
  10 #  This program is distributed in the hope that it will be useful,
  11 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #  GNU General Public License for more details.
  14 #
  15 #  You should have received a copy of the GNU General Public License
  16 #  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18
  19 import os
  20 import shlex
  21 import shutil
  22 import stat
  23 import subprocess
  24 import time
  25
  26 import pmrule
  27 import pmdb
  28 import pmfile
  29 import util
  30 from util import vprint
  31
  32 class pmtest(object):
  33     """Test object
  34     """
  35
  36     def __init__(self, name, root):
  37         self.name = name
  38         self.testname = os.path.basename(name).replace('.py', '')
  39         self.root = root
  40         self.cachepkgs = True
  41
  42     def __str__(self):
  43         return "name = %s\n" \
  44                "testname = %s\n" \
  45                "root = %s" % (self.name, self.testname, self.root)
  46
  47     def addpkg2db(self, treename, pkg):
  48         if not treename in self.db:
  49             self.db[treename] = pmdb.pmdb(treename, self.root)
  50         self.db[treename].pkgs.append(pkg)
  51
  52     def addpkg(self, pkg):
  53         self.localpkgs.append(pkg)
  54
  55     def findpkg(self, name, version, allow_local=False):
  56         """Find a package object matching the name and version specified in
  57         either sync databases or the local package collection. The local database
  58         is allowed to match if allow_local is True."""
  59         for db in self.db.itervalues():
  60             if db.is_local and not allow_local:
  61                 continue
  62             pkg = db.getpkg(name)
  63             if pkg and pkg.version == version:
  64                 return pkg
  65         for pkg in self.localpkgs:
  66             if pkg.name == name and pkg.version == version:
  67                 return pkg
  68
  69         return None
  70
  71     def addrule(self, rulename):
  72         rule = pmrule.pmrule(rulename)
  73         self.rules.append(rule)
  74
  75     def load(self):
  76         # Reset test parameters
  77         self.result = {
  78             "success": 0,
  79             "fail": 0
  80         }
  81         self.args = ""
  82         self.retcode = 0
  83         self.db = {
  84             "local": pmdb.pmdb("local", self.root)
  85         }
  86         self.localpkgs = []
  87         self.createlocalpkgs = False
  88         self.filesystem = []
  89
  90         self.description = ""
  91         self.option = {}
  92
  93         # Test rules
  94         self.rules = []
  95         self.files = []
  96         self.expectfailure = False
  97
  98         if os.path.isfile(self.name):
  99             # all tests expect this to be available
 100             from pmpkg import pmpkg
 101             execfile(self.name)
 102         else:
 103             raise IOError("file %s does not exist!" % self.name)
 104
 105     def generate(self):
 106         print "==> Generating test environment"
 107
 108         # Cleanup leftover files from a previous test session
 109         if os.path.isdir(self.root):
 110             shutil.rmtree(self.root)
 111         vprint("\t%s" % self.root)
 112
 113         # Create directory structure
 114         vprint("    Creating directory structure:")
 115         dbdir = os.path.join(self.root, util.PM_SYNCDBPATH)
 116         cachedir = os.path.join(self.root, util.PM_CACHEDIR)
 117         syncdir = os.path.join(self.root, util.SYNCREPO)
 118         tmpdir = os.path.join(self.root, util.TMPDIR)
 119         logdir = os.path.join(self.root, os.path.dirname(util.LOGFILE))
 120         etcdir = os.path.join(self.root, os.path.dirname(util.PACCONF))
 121         bindir = os.path.join(self.root, "bin")
 122         sbindir = os.path.join(self.root, "sbin")
 123         sys_dirs = [dbdir, cachedir, syncdir, tmpdir, logdir, etcdir, bindir, sbindir]
 124         for sys_dir in sys_dirs:
 125             if not os.path.isdir(sys_dir):
 126                 vprint("\t%s" % sys_dir[len(self.root)+1:])
 127                 os.makedirs(sys_dir, 0755)
 128         # Only the dynamically linked binary is needed for fakechroot
 129         shutil.copy("/bin/sh", bindir)
 130         shutil.copy(os.path.join(util.SELFPATH, "ldconfig.stub"),
 131             os.path.join(sbindir, "ldconfig"))
 132         ld_so_conf = open(os.path.join(etcdir, "ld.so.conf"), "w")
 133         ld_so_conf.close()
 134
 135         # Configuration file
 136         vprint("    Creating configuration file")
 137         util.mkcfgfile(util.PACCONF, self.root, self.option, self.db)
 138
 139         # Creating packages
 140         vprint("    Creating package archives")
 141         for pkg in self.localpkgs:
 142             vprint("\t%s" % os.path.join(util.TMPDIR, pkg.filename()))
 143             pkg.makepkg(tmpdir)
 144         for key, value in self.db.iteritems():
 145             if key == "local" and not self.createlocalpkgs:
 146                 continue
 147             for pkg in value.pkgs:
 148                 vprint("\t%s" % os.path.join(util.PM_CACHEDIR, pkg.filename()))
 149                 if self.cachepkgs:
 150                     pkg.makepkg(cachedir)
 151                 else:
 152                     pkg.makepkg(os.path.join(syncdir, value.treename))
 153                 pkg.md5sum = util.getmd5sum(pkg.path)
 154                 pkg.csize = os.stat(pkg.path)[stat.ST_SIZE]
 155
 156         # Creating sync database archives
 157         vprint("    Creating databases")
 158         for key, value in self.db.iteritems():
 159             vprint("\t" + value.treename)
 160             value.generate()
 161
 162         # Filesystem
 163         vprint("    Populating file system")
 164         for pkg in self.db["local"].pkgs:
 165             vprint("\tinstalling %s" % pkg.fullname())
 166             pkg.install_package(self.root)
 167         for f in self.filesystem:
 168             vprint("\t%s" % f)
 169             util.mkfile(self.root, f, f)
 170             path = os.path.join(self.root, f)
 171             if os.path.isfile(path):
 172                 os.utime(path, (355, 355))
 173
 174         # Done.
 175         vprint("    Taking a snapshot of the file system")
 176         for roots, dirs, files in os.walk(self.root):
 177             for i in files:
 178                 filename = os.path.join(roots, i)
 179                 f = pmfile.PacmanFile(self.root, filename.replace(self.root + "/", ""))
 180                 self.files.append(f)
 181                 vprint("\t%s" % f.name)
 182
 183     def run(self, pacman):
 184         if os.path.isfile(util.PM_LOCK):
 185             print "\tERROR: another pacman session is on-going -- skipping"
 186             return
 187
 188         print "==> Running test"
 189         vprint("\tpacman %s" % self.args)
 190
 191         cmd = []
 192         if os.geteuid() != 0:
 193             fakeroot = util.which("fakeroot")
 194             if not fakeroot:
 195                 print "WARNING: fakeroot not found!"
 196             else:
 197                 cmd.append("fakeroot")
 198
 199             fakechroot = util.which("fakechroot")
 200             if fakechroot:
 201                 cmd.append("fakechroot")
 202
 203         if pacman["gdb"]:
 204             cmd.extend(["libtool", "execute", "gdb", "--args"])
 205         if pacman["valgrind"]:
 206             cmd.extend(["libtool", "execute", "valgrind", "-q",
 207                 "--tool=memcheck", "--leak-check=full",
 208                 "--show-reachable=yes", "--suppressions=%s/valgrind.supp" % os.getcwd()])
 209         cmd.extend([pacman["bin"],
 210             "--config", os.path.join(self.root, util.PACCONF),
 211             "--root", self.root,
 212             "--dbpath", os.path.join(self.root, util.PM_DBPATH),
 213             "--cachedir", os.path.join(self.root, util.PM_CACHEDIR)])
 214         if not pacman["manual-confirm"]:
 215             cmd.append("--noconfirm")
 216         if pacman["debug"]:
 217             cmd.append("--debug=%s" % pacman["debug"])
 218         cmd.extend(shlex.split(self.args))
 219         if not (pacman["gdb"] or pacman["valgrind"] or pacman["nolog"]):
 220             output = open(os.path.join(self.root, util.LOGFILE), 'w')
 221         else:
 222             output = None
 223         vprint("\trunning: %s" % " ".join(cmd))
 224
 225         # Change to the tmp dir before running pacman, so that local package
 226         # archives are made available more easily.
 227         time_start = time.time()
 228         self.retcode = subprocess.call(cmd, stdout=output, stderr=output,
 229                 cwd=os.path.join(self.root, util.TMPDIR))
 230         time_end = time.time()
 231         vprint("\ttime elapsed: %.2fs" % (time_end - time_start))
 232
 233         if output:
 234             output.close()
 235
 236         vprint("\tretcode = %s" % self.retcode)
 237
 238         # Check if the lock is still there
 239         if os.path.isfile(util.PM_LOCK):
 240             print "\tERROR: %s not removed" % util.PM_LOCK
 241             os.unlink(util.PM_LOCK)
 242         # Look for a core file
 243         if os.path.isfile(os.path.join(self.root, util.TMPDIR, "core")):
 244             print "\tERROR: pacman dumped a core file"
 245
 246     def check(self):
 247         print "==> Checking rules"
 248
 249         for i in self.rules:
 250             success = i.check(self)
 251             if success == 1:
 252                 msg = " OK "
 253                 self.result["success"] += 1
 254             elif success == 0:
 255                 msg = "FAIL"
 256                 self.result["fail"] += 1
 257             else:
 258                 msg = "SKIP"
 259             print "\t[%s] %s" % (msg, i)
 260
 261 # vim: set ts=4 sw=4 et: