test/pacman/pmtest.py

   1 #! /usr/bin/python
   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         sys_dirs = [dbdir, cachedir, syncdir, tmpdir, logdir, etcdir, bindir]
 123         for sys_dir in sys_dirs:
 124             if not os.path.isdir(sys_dir):
 125                 vprint("\t%s" % sys_dir[len(self.root)+1:])
 126                 os.makedirs(sys_dir, 0755)
 127         # Only the dynamically linked binary is needed for fakechroot
 128         shutil.copy("/bin/sh", bindir)
 129
 130         # Configuration file
 131         vprint("    Creating configuration file")
 132         util.mkcfgfile(util.PACCONF, self.root, self.option, self.db)
 133
 134         # Creating packages
 135         vprint("    Creating package archives")
 136         for pkg in self.localpkgs:
 137             vprint("\t%s" % os.path.join(util.TMPDIR, pkg.filename()))
 138             pkg.makepkg(tmpdir)
 139         for key, value in self.db.iteritems():
 140             if key == "local" and not self.createlocalpkgs:
 141                 continue
 142             for pkg in value.pkgs:
 143                 vprint("\t%s" % os.path.join(util.PM_CACHEDIR, pkg.filename()))
 144                 if self.cachepkgs:
 145                     pkg.makepkg(cachedir)
 146                 else:
 147                     pkg.makepkg(os.path.join(syncdir, value.treename))
 148                 pkg.md5sum = util.getmd5sum(pkg.path)
 149                 pkg.csize = os.stat(pkg.path)[stat.ST_SIZE]
 150
 151         # Creating sync database archives
 152         vprint("    Creating databases")
 153         for key, value in self.db.iteritems():
 154             vprint("\t" + value.treename)
 155             value.generate()
 156
 157         # Filesystem
 158         vprint("    Populating file system")
 159         for pkg in self.db["local"].pkgs:
 160             vprint("\tinstalling %s" % pkg.fullname())
 161             pkg.install_package(self.root)
 162         for f in self.filesystem:
 163             vprint("\t%s" % f)
 164             util.mkfile(self.root, f, f)
 165             path = os.path.join(self.root, f)
 166             if os.path.isfile(path):
 167                 os.utime(path, (355, 355))
 168
 169         # Done.
 170         vprint("    Taking a snapshot of the file system")
 171         for roots, dirs, files in os.walk(self.root):
 172             for i in files:
 173                 filename = os.path.join(roots, i)
 174                 f = pmfile.PacmanFile(self.root, filename.replace(self.root + "/", ""))
 175                 self.files.append(f)
 176                 vprint("\t%s" % f.name)
 177
 178     def run(self, pacman):
 179         if os.path.isfile(util.PM_LOCK):
 180             print "\tERROR: another pacman session is on-going -- skipping"
 181             return
 182
 183         print "==> Running test"
 184         vprint("\tpacman %s" % self.args)
 185
 186         cmd = []
 187         if os.geteuid() != 0:
 188             fakeroot = util.which("fakeroot")
 189             if not fakeroot:
 190                 print "WARNING: fakeroot not found!"
 191             else:
 192                 cmd.append("fakeroot")
 193
 194             fakechroot = util.which("fakechroot")
 195             if fakechroot:
 196                 cmd.append("fakechroot")
 197
 198         if pacman["gdb"]:
 199             cmd.extend(["libtool", "execute", "gdb", "--args"])
 200         if pacman["valgrind"]:
 201             cmd.extend(["libtool", "execute", "valgrind", "-q",
 202                 "--tool=memcheck", "--leak-check=full",
 203                 "--show-reachable=yes", "--suppressions=%s/valgrind.supp" % os.getcwd()])
 204         cmd.extend([pacman["bin"],
 205             "--config", os.path.join(self.root, util.PACCONF),
 206             "--root", self.root,
 207             "--dbpath", os.path.join(self.root, util.PM_DBPATH),
 208             "--cachedir", os.path.join(self.root, util.PM_CACHEDIR)])
 209         if not pacman["manual-confirm"]:
 210             cmd.append("--noconfirm")
 211         if pacman["debug"]:
 212             cmd.append("--debug=%s" % pacman["debug"])
 213         cmd.extend(shlex.split(self.args))
 214         if not (pacman["gdb"] or pacman["valgrind"] or pacman["nolog"]):
 215             output = open(os.path.join(self.root, util.LOGFILE), 'w')
 216         else:
 217             output = None
 218         vprint("\trunning: %s" % " ".join(cmd))
 219
 220         # Change to the tmp dir before running pacman, so that local package
 221         # archives are made available more easily.
 222         time_start = time.time()
 223         self.retcode = subprocess.call(cmd, stdout=output, stderr=output,
 224                 cwd=os.path.join(self.root, util.TMPDIR))
 225         time_end = time.time()
 226         vprint("\ttime elapsed: %.2fs" % (time_end - time_start))
 227
 228         if output:
 229             output.close()
 230
 231         vprint("\tretcode = %s" % self.retcode)
 232
 233         # Check if the lock is still there
 234         if os.path.isfile(util.PM_LOCK):
 235             print "\tERROR: %s not removed" % util.PM_LOCK
 236             os.unlink(util.PM_LOCK)
 237         # Look for a core file
 238         if os.path.isfile(os.path.join(self.root, util.TMPDIR, "core")):
 239             print "\tERROR: pacman dumped a core file"
 240
 241     def check(self):
 242         print "==> Checking rules"
 243
 244         for i in self.rules:
 245             success = i.check(self)
 246             if success == 1:
 247                 msg = " OK "
 248                 self.result["success"] += 1
 249             elif success == 0:
 250                 msg = "FAIL"
 251                 self.result["fail"] += 1
 252             else:
 253                 msg = "SKIP"
 254             print "\t[%s] %s" % (msg, i)
 255
 256 # vim: set ts=4 sw=4 et: