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 shutil
  21 import stat
  22 import time
  23
  24 import pmrule
  25 import pmdb
  26 import pmfile
  27 import util
  28 from util import vprint
  29
  30 class pmtest(object):
  31     """Test object
  32     """
  33
  34     def __init__(self, name, root):
  35         self.name = name
  36         self.testname = os.path.basename(name).replace('.py', '')
  37         self.root = root
  38         self.cachepkgs = True
  39
  40     def __str__(self):
  41         return "name = %s\n" \
  42                "testname = %s\n" \
  43                "root = %s" % (self.name, self.testname, self.root)
  44
  45     def addpkg2db(self, treename, pkg):
  46         if not treename in self.db:
  47             self.db[treename] = pmdb.pmdb(treename, self.root)
  48         self.db[treename].pkgs.append(pkg)
  49
  50     def addpkg(self, pkg):
  51         self.localpkgs.append(pkg)
  52
  53     def findpkg(self, name, version, allow_local=False):
  54         """Find a package object matching the name and version specified in
  55         either sync databases or the local package collection. The local database
  56         is allowed to match if allow_local is True."""
  57         for db in self.db.itervalues():
  58             if db.is_local and not allow_local:
  59                 continue
  60             pkg = db.getpkg(name)
  61             if pkg and pkg.version == version:
  62                 return pkg
  63         for pkg in self.localpkgs:
  64             if pkg.name == name and pkg.version == version:
  65                 return pkg
  66
  67         return None
  68
  69     def addrule(self, rulename):
  70         rule = pmrule.pmrule(rulename)
  71         self.rules.append(rule)
  72
  73     def load(self):
  74         # Reset test parameters
  75         self.result = {
  76             "success": 0,
  77             "fail": 0
  78         }
  79         self.args = ""
  80         self.retcode = 0
  81         self.db = {
  82             "local": pmdb.pmdb("local", self.root)
  83         }
  84         self.localpkgs = []
  85         self.createlocalpkgs = False
  86         self.filesystem = []
  87
  88         self.description = ""
  89         self.option = {}
  90
  91         # Test rules
  92         self.rules = []
  93         self.files = []
  94         self.expectfailure = False
  95
  96         if os.path.isfile(self.name):
  97             # all tests expect this to be available
  98             from pmpkg import pmpkg
  99             execfile(self.name)
 100         else:
 101             raise IOError("file %s does not exist!" % self.name)
 102
 103     def generate(self):
 104         print "==> Generating test environment"
 105
 106         # Cleanup leftover files from a previous test session
 107         if os.path.isdir(self.root):
 108             shutil.rmtree(self.root)
 109         vprint("\t%s" % self.root)
 110
 111         # Create directory structure
 112         vprint("    Creating directory structure:")
 113         dbdir = os.path.join(self.root, util.PM_SYNCDBPATH)
 114         cachedir = os.path.join(self.root, util.PM_CACHEDIR)
 115         syncdir = os.path.join(self.root, util.SYNCREPO)
 116         tmpdir = os.path.join(self.root, util.TMPDIR)
 117         logdir = os.path.join(self.root, os.path.dirname(util.LOGFILE))
 118         etcdir = os.path.join(self.root, os.path.dirname(util.PACCONF))
 119         bindir = os.path.join(self.root, "bin")
 120         sys_dirs = [dbdir, cachedir, syncdir, tmpdir, logdir, etcdir, bindir]
 121         for sys_dir in sys_dirs:
 122             if not os.path.isdir(sys_dir):
 123                 vprint("\t%s" % sys_dir[len(self.root)+1:])
 124                 os.makedirs(sys_dir, 0755)
 125         # Only the dynamically linked binary is needed for fakechroot
 126         shutil.copy("/bin/sh", bindir)
 127
 128         # Configuration file
 129         vprint("    Creating configuration file")
 130         util.mkcfgfile(util.PACCONF, self.root, self.option, self.db)
 131
 132         # Creating packages
 133         vprint("    Creating package archives")
 134         for pkg in self.localpkgs:
 135             vprint("\t%s" % os.path.join(util.TMPDIR, pkg.filename()))
 136             pkg.makepkg(tmpdir)
 137         for key, value in self.db.iteritems():
 138             if key == "local" and not self.createlocalpkgs:
 139                 continue
 140             for pkg in value.pkgs:
 141                 vprint("\t%s" % os.path.join(util.PM_CACHEDIR, pkg.filename()))
 142                 if self.cachepkgs:
 143                     pkg.makepkg(cachedir)
 144                 else:
 145                     pkg.makepkg(os.path.join(syncdir, value.treename))
 146                 pkg.md5sum = util.getmd5sum(pkg.path)
 147                 pkg.csize = os.stat(pkg.path)[stat.ST_SIZE]
 148
 149         # Creating sync database archives
 150         vprint("    Creating databases")
 151         for key, value in self.db.iteritems():
 152             vprint("\t" + value.treename)
 153             value.generate()
 154
 155         # Filesystem
 156         vprint("    Populating file system")
 157         for pkg in self.db["local"].pkgs:
 158             vprint("\tinstalling %s" % pkg.fullname())
 159             pkg.install_package(self.root)
 160         for f in self.filesystem:
 161             vprint("\t%s" % f)
 162             util.mkfile(self.root, f, f)
 163             path = os.path.join(self.root, f)
 164             if os.path.isfile(path):
 165                 os.utime(path, (355, 355))
 166
 167         # Done.
 168         vprint("    Taking a snapshot of the file system")
 169         for roots, dirs, files in os.walk(self.root):
 170             for i in files:
 171                 filename = os.path.join(roots, i)
 172                 f = pmfile.PacmanFile(self.root, filename.replace(self.root + "/", ""))
 173                 self.files.append(f)
 174                 vprint("\t%s" % f.name)
 175
 176     def run(self, pacman):
 177         if os.path.isfile(util.PM_LOCK):
 178             print "\tERROR: another pacman session is on-going -- skipping"
 179             return
 180
 181         print "==> Running test"
 182         vprint("\tpacman %s" % self.args)
 183
 184         cmd = [""]
 185         if os.geteuid() != 0:
 186             fakeroot = util.which("fakeroot")
 187             if not fakeroot:
 188                 print "WARNING: fakeroot not found!"
 189             else:
 190                 cmd.append("fakeroot")
 191
 192             fakechroot = util.which("fakechroot")
 193             if fakechroot:
 194                 cmd.append("fakechroot")
 195
 196         if pacman["gdb"]:
 197             cmd.append("libtool execute gdb --args")
 198         if pacman["valgrind"]:
 199             cmd.append("valgrind -q --tool=memcheck --leak-check=full --show-reachable=yes --suppressions=%s/valgrind.supp" % os.getcwd())
 200         cmd.append("\"%s\" --config=\"%s\" --root=\"%s\" --dbpath=\"%s\" --cachedir=\"%s\"" \
 201                    % (pacman["bin"],
 202                        os.path.join(self.root, util.PACCONF),
 203                        self.root,
 204                        os.path.join(self.root, util.PM_DBPATH),
 205                        os.path.join(self.root, util.PM_CACHEDIR)))
 206         if not pacman["manual-confirm"]:
 207             cmd.append("--noconfirm")
 208         if pacman["debug"]:
 209             cmd.append("--debug=%s" % pacman["debug"])
 210         cmd.append("%s" % self.args)
 211         if not pacman["gdb"] and not pacman["valgrind"] and not pacman["nolog"]:
 212             cmd.append(">\"%s\" 2>&1" % os.path.join(self.root, util.LOGFILE))
 213         vprint("\trunning: %s" % " ".join(cmd))
 214
 215         # Change to the tmp dir before running pacman, so that local package
 216         # archives are made available more easily.
 217         curdir = os.getcwd()
 218         tmpdir = os.path.join(self.root, util.TMPDIR)
 219         os.chdir(tmpdir)
 220
 221         time_start = time.time()
 222         self.retcode = os.system(" ".join(cmd))
 223         time_end = time.time()
 224         vprint("\ttime elapsed: %ds" % (time_end - time_start))
 225
 226         if self.retcode == None:
 227             self.retcode = 0
 228         else:
 229             self.retcode /= 256
 230         vprint("\tretcode = %s" % self.retcode)
 231         os.chdir(curdir)
 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: