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             "noextract": []
  88         }
  89
  90         # Test rules
  91         self.rules = []
  92         self.files = []
  93
  94         if os.path.isfile(self.name):
  95             execfile(self.name)
  96         else:
  97             err("file %s does not exist!" % self.name)
  98
  99     def generate(self):
 100         """
 101         """
 102
 103         print "==> Generating test environment"
 104
 105         # Cleanup leftover files from a previous test session
 106         if os.path.isdir(self.root):
 107             shutil.rmtree(self.root)
 108         vprint("\t%s" % self.root)
 109
 110         # Create directory structure
 111         vprint("    Creating directory structure:")
 112         dbdir = os.path.join(self.root, PM_DBPATH)
 113         cachedir = os.path.join(self.root, PM_CACHEDIR)
 114         syncdir = os.path.join(self.root, SYNCREPO)
 115         tmpdir = os.path.join(self.root, TMPDIR)
 116         logdir = os.path.join(self.root, os.path.dirname(LOGFILE))
 117         etcdir = os.path.join(self.root, os.path.dirname(PACCONF))
 118         for dir in [dbdir, cachedir, syncdir, tmpdir, logdir, etcdir]:
 119             if not os.path.isdir(dir):
 120                 vprint("\t%s" % dir[len(self.root)+1:])
 121                 os.makedirs(dir, 0755)
 122
 123         # Configuration file
 124         vprint("    Creating configuration file")
 125         vprint("\t%s" % PACCONF)
 126         mkcfgfile(PACCONF, self.root, self.option, self.db)
 127
 128         # Creating packages
 129         vprint("    Creating package archives")
 130         for pkg in self.localpkgs:
 131             vprint("\t%s" % os.path.join(TMPDIR, pkg.filename()))
 132             pkg.makepkg(tmpdir)
 133         for key, value in self.db.iteritems():
 134             if key == "local": continue
 135             for pkg in value.pkgs:
 136                 vprint("\t%s" % os.path.join(PM_CACHEDIR, pkg.filename()))
 137                 if self.cachepkgs:
 138                     pkg.makepkg(cachedir)
 139                 else:
 140                     pkg.makepkg(os.path.join(syncdir, value.treename))
 141                 pkg.md5sum = getmd5sum(pkg.path)
 142                 pkg.csize = os.stat(pkg.path)[stat.ST_SIZE]
 143
 144         # Populating databases
 145         vprint("    Populating databases")
 146         for key, value in self.db.iteritems():
 147             for pkg in value.pkgs:
 148                 vprint("\t%s/%s" % (key, pkg.fullname()))
 149                 if key == "local":
 150                     pkg.installdate = time.ctime()
 151                 value.db_write(pkg)
 152
 153         # Creating sync database archives
 154         vprint("    Creating sync database archives")
 155         for key, value in self.db.iteritems():
 156             if key == "local": continue
 157             archive = value.treename + PM_EXT_DB
 158             vprint("\t" + os.path.join(SYNCREPO, archive))
 159             value.gensync(os.path.join(syncdir, value.treename))
 160
 161         # Filesystem
 162         vprint("    Populating file system")
 163         for pkg in self.db["local"].pkgs:
 164             vprint("\tinstalling %s" % pkg.fullname())
 165             pkg.install_files(self.root)
 166         for f in self.filesystem:
 167             vprint("\t%s" % f)
 168             mkfile(os.path.join(self.root, f), f)
 169
 170         # Done.
 171         vprint("    Taking a snapshot of the file system")
 172         for roots, dirs, files in os.walk(self.root):
 173             for i in files:
 174                 filename = os.path.join(roots, i)
 175                 f = pmfile.pmfile(self.root, filename.replace(self.root + "/", ""))
 176                 self.files.append(f)
 177                 vprint("\t%s" % f.name)
 178
 179     def run(self, pacman):
 180         """
 181         """
 182
 183         if os.path.isfile(PM_LOCK):
 184             print "\tERROR: another pacman session is on-going -- skipping"
 185             return
 186
 187         print "==> Running test"
 188         vprint("\tpacman %s" % self.args)
 189
 190         cmd = ["fakeroot"]
 191         if pacman["gdb"]:
 192             cmd.append("libtool gdb --args")
 193         if pacman["valgrind"]:
 194             cmd.append("valgrind --tool=memcheck --leak-check=full --show-reachable=yes")
 195         cmd.append("%s --config=%s --root=%s" \
 196                    % (pacman["bin"], os.path.join(self.root, PACCONF), self.root))
 197         if not pacman["manual-confirm"]:
 198             cmd.append("--noconfirm")
 199         if pacman["debug"]:
 200             cmd.append("--debug=%s" % pacman["debug"])
 201         cmd.append("%s" % self.args)
 202         if not pacman["gdb"] and not pacman["valgrind"] and not pacman["nolog"]:
 203             cmd.append(">%s 2>&1" % os.path.join(self.root, LOGFILE))
 204         vprint("\trunning: %s" % " ".join(cmd))
 205
 206         # Change to the tmp dir before running pacman, so that local package
 207         # archives are made available more easily.
 208         curdir = os.getcwd()
 209         tmpdir = os.path.join(self.root, TMPDIR)
 210         os.chdir(tmpdir)
 211
 212         t0 = time.time()
 213         self.retcode = os.system(" ".join(cmd))
 214         t1 = time.time()
 215         vprint("\ttime elapsed: %ds" % (t1-t0))
 216
 217         if self.retcode == None:
 218             self.retcode = 0
 219         else:
 220             self.retcode /= 256
 221         vprint("\tretcode = %s" % self.retcode)
 222         os.chdir(curdir)
 223
 224         # Check if pacman failed because of bad permissions
 225         if self.retcode and not pacman["nolog"] \
 226            and grep(os.path.join(self.root, LOGFILE),
 227                     "you cannot perform this operation unless you are root"):
 228             print "\tERROR: pacman support for fakeroot is not disabled"
 229         # Check if the lock is still there
 230         if os.path.isfile(PM_LOCK):
 231             print "\tERROR: %s not removed" % PM_LOCK
 232             os.unlink(PM_LOCK)
 233         # Look for a core file
 234         if os.path.isfile(os.path.join(self.root, TMPDIR, "core")):
 235             print "\tERROR: pacman dumped a core file"
 236
 237     def check(self):
 238         """
 239         """
 240
 241         print "==> Checking rules"
 242
 243         for i in self.rules:
 244             success = i.check(self.root, self.retcode, self.db["local"], self.files)
 245             if success == 1:
 246                 msg = " OK "
 247                 self.result["success"] += 1
 248             elif success == 0:
 249                 msg = "FAIL"
 250                 self.result["fail"] += 1
 251             else:
 252                 msg = "SKIP"
 253             print "\t[%s] %s" % (msg, i.rule)
 254             i.result = success
 255
 256
 257 if __name__ == "__main__":
 258     test = pmtest("test1", "./root")
 259     print test
 260 # vim: set ts=4 sw=4 et: