tests/test_baseimager.py

   1 #!/usr/bin/python
   2
   3 import os
   4 import sys
   5 import rpm
   6 import glob
   7 import shutil
   8 import StringIO
   9 import subprocess
  10 import unittest
  11 from mic import plugin as pluginmgr
  12 from mic import conf as configmgr
  13 from mic import msger
  14 from mic.imager import fs
  15
  16 CWD = os.path.dirname(__file__) or '.'
  17 TEST_BASEIMGR_LOC = os.path.join(CWD, 'baseimgr_fixtures')
  18 KSCONF = os.path.join(CWD, 'baseimgr_fixtures', 'test.ks')
  19 KSBAK = os.path.join(CWD, 'baseimgr_fixtures', 'test.ks.bak')
  20 REPOURI = os.path.join(CWD, 'baseimgr_fixtures')
  21 CACHEDIR = os.path.join(CWD, 'baseimgr_fixtures', 'cache')
  22 RPMLOCK_PATH = None
  23
  24 def suite():
  25     return unittest.makeSuite(BaseImgrTest)
  26
  27 class BaseImgrTest(unittest.TestCase):
  28
  29     arch = 'i686'
  30     rootdir = "%s/rootdir" % os.getcwd()
  31     expect_pkglist = ['A', 'ABC', 'C', 'D', 'E', 'F', 'G', 'H']
  32
  33     def setUp(self):
  34         self.stdout = sys.stdout
  35         self.stream = sys.stdout
  36         msger.STREAM = StringIO.StringIO()
  37         shutil.copy2(KSCONF, KSBAK)
  38         with open(KSCONF, 'r') as f:
  39             content = f.read()
  40         content = content.replace('$$$$$$', "file://" + REPOURI)
  41         with open(KSCONF, 'w') as f:
  42             f.write(content)
  43         msger.set_loglevel('quiet')
  44
  45     def tearDown(self):
  46         sys.stdout = self.stdout
  47         msger.STREAM = self.stream
  48         shutil.copy2(KSBAK, KSCONF)
  49         shutil.rmtree (self.rootdir, ignore_errors = True)
  50         shutil.rmtree (CACHEDIR, ignore_errors = True)
  51         os.unlink(KSBAK)
  52
  53     def getMountList(self, pattern):
  54         real_mount_list = []
  55         dev_null = os.open("/dev/null", os.O_WRONLY)
  56         p = subprocess.Popen('mount', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  57
  58         for entry in p.communicate()[0].split('\n'):
  59             if entry.find(pattern) > 0:
  60                 real_mount_list.append(entry.split(' ')[2])
  61         real_mount_list.sort()
  62         os.close(dev_null)
  63         return real_mount_list
  64
  65     def getInsPkgList(self, rootdir):
  66         installed_pkgs = []
  67         ts = rpm.TransactionSet (rootdir)
  68         hrs = ts.dbMatch()
  69         for pkg in hrs:
  70             installed_pkgs.append(pkg['name'])
  71         installed_pkgs.sort()
  72         ts.closeDB()
  73
  74         return installed_pkgs
  75
  76     def BaseImager(self, backend):
  77         global RPMLOCK_PATH
  78
  79         cfgmgr = configmgr.configmgr
  80         creatoropts = cfgmgr.create
  81
  82         creatoropts["cachedir"] = CACHEDIR
  83         creatoropts["outdir"] = self.rootdir
  84         creatoropts["arch"] = self.arch
  85         creatoropts['pkgmgr'] = backend
  86         cfgmgr._ksconf =  KSCONF
  87         pkgmgr = None
  88         for (key, pcls) in pluginmgr.PluginMgr().get_plugins('backend').iteritems():
  89             if key == creatoropts['pkgmgr']:
  90                 pkgmgr = pcls
  91                 break
  92
  93         creator = fs.FsImageCreator(creatoropts, pkgmgr)
  94         creator._recording_pkgs.append('name')
  95
  96         creator.check_depend_tools()
  97
  98         # Test mount interface
  99         creator.mount(None, creatoropts["cachedir"])
 100         if RPMLOCK_PATH:
 101             os.makedirs(RPMLOCK_PATH)
 102         else:
 103             RPMLOCK_PATH = "%s/var/lib/rpm" % creator._instroot
 104         exp_mount_list = ['/sys', '/proc', '/proc/sys/fs/binfmt_misc', '/dev/pts']
 105         exp_mount_list.sort()
 106         exp_mount_list = [os.path.join(creator._instroot, pth.lstrip('/')) for pth in exp_mount_list]
 107         real_mount_list = self.getMountList(creator._instroot)
 108         self.assertEqual(real_mount_list, exp_mount_list)
 109
 110         # Test Install interface
 111         creator.install()
 112         installed_pkgs = self.getInsPkgList(creator._instroot)
 113         self.assertEqual(installed_pkgs, self.expect_pkglist)
 114
 115         # Test umount interface
 116         creator.unmount()
 117         real_mount_list = self.getMountList(creator._instroot)
 118         self.assertEqual(real_mount_list, [])
 119         # Test Packaging interface
 120         creator.package(creatoropts["outdir"])
 121         installed_pkgs = self.getInsPkgList("%s/%s" % (self.rootdir, creator.name))
 122         self.assertEqual(installed_pkgs, self.expect_pkglist)
 123
 124         creator.cleanup()
 125         # Test recore_pkgs option
 126         pkglist = ['A.i586 0.1-1', 'ABC.i586 0.1-1', 'C.i686 0.2-1',
 127                    'D.i586 0.1-1', 'E.i586 0.1-1', 'F.noarch 0.1-1',
 128                    'G.i586 0.1-1', 'H.noarch 0.1-1']
 129         f = open ("%s/%s.packages" % (self.rootdir, creator.name))
 130         real_pkglist = f.read()
 131         self.assertEqual(real_pkglist, '\n'.join(pkglist))
 132
 133 if os.getuid == 0:
 134     def testBaseImagerZypp(self):
 135         self.BaseImager('zypp')
 136
 137     def testBaseImagerYum(self):
 138         self.BaseImager('yum')
 139
 140 if __name__ == "__main__":
 141     if os.getuid() != 0:
 142         raise SystemExit("Root permission is needed")
 143     unittest.main()