pactest/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, write to the Free Software
  17 #  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
  18 #  USA.
  19
  20
  21 import os
  22 import os.path
  23 import shutil
  24 import time
  25
  26 import pmrule
  27 import pmdb
  28 import pmfile
  29 from pmpkg import pmpkg
  30 from util import *
  31
  32
  33 class pmtest:
  34     """Test object
  35     """
  36
  37     def __init__(self, name, root):
  38         self.name = name
  39         self.testname = os.path.basename(name).replace('.py', '')
  40         self.root = root
  41         self.cachepkgs = True
  42
  43     def __str__(self):
  44         return "name = %s\n" \
  45                "testname = %s\n" \
  46                "root = %s" % (self.name, self.testname, self.root)
  47
  48     def addpkg2db(self, treename, pkg):
  49         """
  50         """
  51         if not treename in self.db:
  52             self.db[treename] = pmdb.pmdb(treename, os.path.join(self.root, PM_DBPATH))
  53         self.db[treename].pkgs.append(pkg)
  54
  55     def addpkg(self, pkg):
  56         """
  57         """
  58         self.localpkgs.append(pkg)
  59
  60     def addrule(self, rulename):
  61         """
  62         """
  63         rule = pmrule.pmrule(rulename)
  64         self.rules.append(rule)
  65
  66     def load(self):
  67         """
  68         """
  69
  70         # Reset test parameters
  71         self.result = {
  72             "success": 0,
  73             "fail": 0
  74         }
  75         self.args = ""
  76         self.retcode = 0
  77         self.db = {
  78             "local": pmdb.pmdb("local", os.path.join(self.root, PM_DBPATH))
  79         }
  80         self.localpkgs = []
  81         self.filesystem = []
  82
  83         self.description = ""
  84         self.option = {
  85             "noupgrade": [],
  86             "ignorepkg": [],
  87             "ignoregroup": [],
  88             "noextract": []
  89         }
  90
  91         # Test rules
  92         self.rules = []
  93         self.files = []
  94
  95         if os.path.isfile(self.name):
  96             execfile(self.name)
  97         else:
  98             err("file %s does not exist!" % self.name)
  99
 100     def generate(self):
 101         """
 102         """
 103
 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, PM_DBPATH)
 114         cachedir = os.path.join(self.root, PM_CACHEDIR)
 115         syncdir = os.path.join(self.root, SYNCREPO)
 116         tmpdir = os.path.join(self.root, TMPDIR)
 117         logdir = os.path.join(self.root, os.path.dirname(LOGFILE))
 118         etcdir = os.path.join(self.root, os.path.dirname(PACCONF))
 119         for dir in [dbdir, cachedir, syncdir, tmpdir, logdir, etcdir]:
 120             if not os.path.isdir(dir):
 121                 vprint("\t%s" % dir[len(self.root)+1:])
 122                 os.makedirs(dir, 0755)
 123
 124         # Configuration file
 125         vprint("    Creating configuration file")
 126         vprint("\t%s" % PACCONF)
 127         mkcfgfile(PACCONF, self.root, self.option, self.db)
 128
 129         # Creating packages
 130         vprint("    Creating package archives")
 131         for pkg in self.localpkgs:
 132             vprint("\t%s" % os.path.join(TMPDIR, pkg.filename()))
 133             pkg.makepkg(tmpdir)
 134         for key, value in self.db.iteritems():
 135             if key == "local": continue
 136             for pkg in value.pkgs:
 137                 vprint("\t%s" % os.path.join(PM_CACHEDIR, pkg.filename()))
 138                 if self.cachepkgs:
 139                     pkg.makepkg(cachedir)
 140                 else:
 141                     pkg.makepkg(os.path.join(syncdir, value.treename))
 142                 pkg.md5sum = getmd5sum(pkg.path)
 143                 pkg.csize = os.stat(pkg.path)[stat.ST_SIZE]
 144
 145         # Populating databases
 146         vprint("    Populating databases")
 147         for key, value in self.db.iteritems():
 148             for pkg in value.pkgs:
 149                 vprint("\t%s/%s" % (key, pkg.fullname()))
 150                 if key == "local":
 151                     pkg.installdate = time.ctime()
 152                 value.db_write(pkg)
 153
 154         # Creating sync database archives
 155         vprint("    Creating sync database archives")
 156         for key, value in self.db.iteritems():
 157             if key == "local": continue
 158             archive = value.treename + PM_EXT_DB
 159             vprint("\t" + os.path.join(SYNCREPO, archive))
 160             value.gensync(os.path.join(syncdir, value.treename))
 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_files(self.root)
 167         for f in self.filesystem:
 168             vprint("\t%s" % f)
 169             mkfile(os.path.join(self.root, f), f)
 170
 171         # Done.
 172         vprint("    Taking a snapshot of the file system")
 173         for roots, dirs, files in os.walk(self.root):
 174             for i in files:
 175                 filename = os.path.join(roots, i)
 176                 f = pmfile.pmfile(self.root, filename.replace(self.root + "/", ""))
 177                 self.files.append(f)
 178                 vprint("\t%s" % f.name)
 179
 180     def run(self, pacman):
 181         """
 182         """
 183
 184         if os.path.isfile(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 = ["fakeroot"]
 192         if pacman["gdb"]:
 193             cmd.append("libtool gdb --args")
 194         if pacman["valgrind"]:
 195             cmd.append("valgrind --tool=memcheck --leak-check=full --show-reachable=yes")
 196         cmd.append("%s --config=%s --root=%s --dbpath=%s --cachedir=%s" \
 197                    % (pacman["bin"],
 198                        os.path.join(self.root, PACCONF),
 199                        self.root,
 200                        os.path.join(self.root, PM_DBPATH),
 201                        os.path.join(self.root, PM_CACHEDIR)))
 202         if not pacman["manual-confirm"]:
 203             cmd.append("--noconfirm")
 204         if pacman["debug"]:
 205             cmd.append("--debug=%s" % pacman["debug"])
 206         cmd.append("%s" % self.args)
 207         if not pacman["gdb"] and not pacman["valgrind"] and not pacman["nolog"]:
 208             cmd.append(">%s 2>&1" % os.path.join(self.root, LOGFILE))
 209         vprint("\trunning: %s" % " ".join(cmd))
 210
 211         # Change to the tmp dir before running pacman, so that local package
 212         # archives are made available more easily.
 213         curdir = os.getcwd()
 214         tmpdir = os.path.join(self.root, TMPDIR)
 215         os.chdir(tmpdir)
 216
 217         t0 = time.time()
 218         self.retcode = os.system(" ".join(cmd))
 219         t1 = time.time()
 220         vprint("\ttime elapsed: %ds" % (t1-t0))
 221
 222         if self.retcode == None:
 223             self.retcode = 0
 224         else:
 225             self.retcode /= 256
 226         vprint("\tretcode = %s" % self.retcode)
 227         os.chdir(curdir)
 228
 229         # Check if pacman failed because of bad permissions
 230         if self.retcode and not pacman["nolog"] \
 231            and grep(os.path.join(self.root, LOGFILE),
 232                     "you cannot perform this operation unless you are root"):
 233             print "\tERROR: pacman support for fakeroot is not disabled"
 234         # Check if the lock is still there
 235         if os.path.isfile(PM_LOCK):
 236             print "\tERROR: %s not removed" % PM_LOCK
 237             os.unlink(PM_LOCK)
 238         # Look for a core file
 239         if os.path.isfile(os.path.join(self.root, TMPDIR, "core")):
 240             print "\tERROR: pacman dumped a core file"
 241
 242     def check(self):
 243         """
 244         """
 245
 246         print "==> Checking rules"
 247
 248         for i in self.rules:
 249             success = i.check(self.root, self.retcode, self.db["local"], self.files)
 250             if success == 1:
 251                 msg = " OK "
 252                 self.result["success"] += 1
 253             elif success == 0:
 254                 msg = "FAIL"
 255                 self.result["fail"] += 1
 256             else:
 257                 msg = "SKIP"
 258             print "\t[%s] %s" % (msg, i.rule)
 259             i.result = success
 260
 261
 262 if __name__ == "__main__":
 263     test = pmtest("test1", "./root")
 264     print test
 265 # vim: set ts=4 sw=4 et: