source: yum/trunk/test/testbase.py@ 6

Last change on this file since 6 was 2, checked in by Yuri Dario, 15 years ago

Initial import for vendor code.
  • Property svn:eol-style set to native
File size: 15.0 KB
Line 
1import os
2import sys
3import unittest
4
5import settestpath
6import logging
7import yum.logginglevels as logginglevels
8
9new_behavior = "NEW_BEHAVIOR" in os.environ.keys()
10
11from yum import YumBase
12from yum import transactioninfo
13from yum import packages
14from yum import packageSack
15from yum.constants import TS_INSTALL_STATES, TS_REMOVE_STATES
16from cli import YumBaseCli
17from yum.rpmsack import RPMDBPackageSack as _rpmdbsack
18import inspect
19from rpmUtils import arch
20
21#############################################################
22### Helper classes ##########################################
23#############################################################
24
25# Dummy translation wrapper
26def _(msg):
27 return msg
28
29class FakeConf(object):
30
31 def __init__(self):
32 self.installonlypkgs = ['kernel']
33 self.exclude = []
34 self.debuglevel = 8
35 self.obsoletes = True
36 self.exactarch = False
37 self.exactarchlist = []
38 self.installroot = '/'
39 self.tsflags = []
40 self.installonly_limit = 0
41 self.skip_broken = False
42 self.disable_excludes = []
43 self.multilib_policy = 'best'
44 self.persistdir = '/should-not-exist-bad-test!'
45 self.showdupesfromrepos = False
46 self.uid = 0
47
48class FakeRepo(object):
49
50 def __init__(self, id=None,sack=None):
51 self.id = id
52 self.sack = sack
53 self.cost = 1000
54
55 def __cmp__(self, other):
56 """ Sort base class repos. by alphanumeric on their id, also
57 see __cmp__ in YumRepository(). """
58 if self.id > other.id:
59 return 1
60 elif self.id < other.id:
61 return -1
62 else:
63 return 0
64
65class FakePackage(packages.YumAvailablePackage):
66
67 def __init__(self, name, version='1.0', release='1', epoch='0', arch='noarch', repo=None):
68 if repo is None:
69 repo = FakeRepo()
70 print "creating empty repo for %s-%s:%s-%s.%s " % (name, epoch,
71 version, release,
72 arch)
73 packages.YumAvailablePackage.__init__(self, repo)
74
75 self.name = name
76 self.version = version
77 self.ver = version
78 self.release = release
79 self.rel = release
80 self.epoch = epoch
81 self.arch = arch
82 self.pkgtup = (self.name, self.arch, self.epoch, self.version, self.release)
83 self.yumdb_info = {}
84
85 self.prco['provides'].append((name, 'EQ', (epoch, version, release)))
86
87 # Just a unique integer
88 self.id = self.__hash__()
89 self.pkgKey = self.__hash__()
90
91 def addProvides(self, name, flag=None, evr=(None, None, None)):
92 self.prco['provides'].append((name, flag, evr))
93 def addRequires(self, name, flag=None, evr=(None, None, None)):
94 self.prco['requires'].append((name, flag, evr))
95 def addConflicts(self, name, flag=None, evr=(None, None, None)):
96 self.prco['conflicts'].append((name, flag, evr))
97 def addObsoletes(self, name, flag=None, evr=(None, None, None)):
98 self.prco['obsoletes'].append((name, flag, evr))
99 def addFile(self, name, ftype='file'):
100 self.files[ftype].append(name)
101
102class _Container(object):
103 pass
104
105
106class DepSolveProgressCallBack:
107 """provides text output callback functions for Dependency Solver callback"""
108
109 def __init__(self):
110 """requires yum-cli log and errorlog functions as arguments"""
111 self.verbose_logger = logging.getLogger("yum.verbose.cli")
112 self.loops = 0
113
114 def pkgAdded(self, pkgtup, mode):
115 modedict = { 'i': _('installed'),
116 'u': _('updated'),
117 'o': _('obsoleted'),
118 'e': _('erased')}
119 (n, a, e, v, r) = pkgtup
120 modeterm = modedict[mode]
121 self.verbose_logger.log(logginglevels.INFO_2,
122 _('---> Package %s.%s %s:%s-%s set to be %s'), n, a, e, v, r,
123 modeterm)
124
125 def start(self):
126 self.loops += 1
127
128 def tscheck(self):
129 self.verbose_logger.log(logginglevels.INFO_2, _('--> Running transaction check'))
130
131 def restartLoop(self):
132 self.loops += 1
133 self.verbose_logger.log(logginglevels.INFO_2,
134 _('--> Restarting Dependency Resolution with new changes.'))
135 self.verbose_logger.debug('---> Loop Number: %d', self.loops)
136
137 def end(self):
138 self.verbose_logger.log(logginglevels.INFO_2,
139 _('--> Finished Dependency Resolution'))
140
141
142 def procReq(self, name, formatted_req):
143 self.verbose_logger.log(logginglevels.INFO_2,
144 _('--> Processing Dependency: %s for package: %s'), formatted_req,
145 name)
146
147
148 def unresolved(self, msg):
149 self.verbose_logger.log(logginglevels.INFO_2, _('--> Unresolved Dependency: %s'),
150 msg)
151
152
153 def procConflict(self, name, confname):
154 self.verbose_logger.log(logginglevels.INFO_2,
155 _('--> Processing Conflict: %s conflicts %s'), name, confname)
156
157 def transactionPopulation(self):
158 self.verbose_logger.log(logginglevels.INFO_2, _('--> Populating transaction set '
159 'with selected packages. Please wait.'))
160
161 def downloadHeader(self, name):
162 self.verbose_logger.log(logginglevels.INFO_2, _('---> Downloading header for %s '
163 'to pack into transaction set.'), name)
164
165#######################################################################
166### Abstract super class for test cases ###############################
167#######################################################################
168
169class _DepsolveTestsBase(unittest.TestCase):
170
171 res = {0 : 'empty', 2 : 'ok', 1 : 'err'}
172
173 def __init__(self, methodName='runTest'):
174 unittest.TestCase.__init__(self, methodName)
175 self.pkgs = _Container()
176 self.buildPkgs(self.pkgs)
177
178 def setUp(self):
179 pass
180 def tearDown(self):
181 pass
182
183 @staticmethod
184 def buildPkgs(pkgs, *args):
185 """Overload this staticmethod to create pkpgs that are used in several
186 test cases. It gets called from __init__ with self.pkgs as first parameter.
187 It is a staticmethod so you can call .buildPkgs() from other Tests to share
188 buildPkg code (inheritance doesn't work here, because we don't want to
189 inherit the test cases, too).
190 """
191 pass
192
193 def assertResult(self, pkgs, optional_pkgs=[]):
194 """Check if "system" contains the given pkgs. pkgs must be present,
195 optional_pkgs may be. Any other pkgs result in an error. Pkgs are
196 present if they are in the rpmdb and are not REMOVEd or they
197 are INSTALLed.
198 """
199 errors = ["Unexpected result after depsolving: \n\n"]
200 pkgs = set(pkgs)
201 optional_pkgs = set(optional_pkgs)
202 installed = set()
203
204 for pkg in self.rpmdb:
205 # got removed
206 if self.tsInfo.getMembersWithState(pkg.pkgtup, TS_REMOVE_STATES):
207 if pkg in pkgs:
208 errors.append("Package %s was removed!\n" % pkg)
209 else: # still installed
210 if pkg not in pkgs and pkg not in optional_pkgs:
211 errors.append("Package %s was not removed!\n" % pkg)
212 installed.add(pkg)
213
214 for txmbr in self.tsInfo.getMembersWithState(output_states=TS_INSTALL_STATES):
215 installed.add(txmbr.po)
216 if txmbr.po not in pkgs and txmbr.po not in optional_pkgs:
217 errors.append("Package %s was installed!\n" % txmbr.po)
218 for pkg in pkgs - installed:
219 errors.append("Package %s was not installed!\n" % pkg)
220
221 if len(errors) > 1:
222 errors.append("\nTest case was:\n\n")
223 errors.extend(inspect.getsource(inspect.stack()[1][0].f_code))
224 errors.append("\n")
225 self.fail("".join(errors))
226
227class FakeRpmDb(packageSack.PackageSack):
228 '''
229 We use a PackagePack for a Fake rpmdb insted of the normal
230 RPMDBPackageSack, getProvides works a little different on
231 unversioned requirements so we have to overload an add some
232 extra checkcode.
233 '''
234 def __init__(self):
235 packageSack.PackageSack.__init__(self)
236
237 # Need to mock out rpmdb caching... copy&paste. Gack.
238 def returnConflictPackages(self):
239 ret = []
240 for pkg in self.returnPackages():
241 if len(pkg.conflicts):
242 ret.append(pkg)
243 return ret
244 def fileRequiresData(self):
245 installedFileRequires = {}
246 installedUnresolvedFileRequires = set()
247 resolved = set()
248 for pkg in self.returnPackages():
249 for name, flag, evr in pkg.requires:
250 if not name.startswith('/'):
251 continue
252 installedFileRequires.setdefault(pkg.pkgtup, []).append(name)
253 if name not in resolved:
254 dep = self.getProvides(name, flag, evr)
255 resolved.add(name)
256 if not dep:
257 installedUnresolvedFileRequires.add(name)
258
259 fileRequires = set()
260 for fnames in installedFileRequires.itervalues():
261 fileRequires.update(fnames)
262 installedFileProviders = {}
263 for fname in fileRequires:
264 pkgtups = [pkg.pkgtup for pkg in self.getProvides(fname)]
265 installedFileProviders[fname] = pkgtups
266
267 ret = (installedFileRequires, installedUnresolvedFileRequires,
268 installedFileProviders)
269
270 return ret
271 def transactionCacheFileRequires(self, installedFileRequires,
272 installedUnresolvedFileRequires,
273 installedFileProvides,
274 problems):
275 return
276 def transactionCacheConflictPackages(self, pkgs):
277 return
278 def transactionResultVersion(self, rpmdbv):
279 return
280 def transactionReset(self):
281 return
282
283 def getProvides(self, name, flags=None, version=(None, None, None)):
284 """return dict { packages -> list of matching provides }"""
285 self._checkIndexes(failure='build')
286 result = { }
287 # convert flags & version for unversioned reqirements
288 if not version:
289 version=(None, None, None)
290 if flags == '0':
291 flags=None
292 for po in self.provides.get(name, []):
293 hits = po.matchingPrcos('provides', (name, flags, version))
294 if hits:
295 result[po] = hits
296 if name[0] == '/':
297 hit = (name, None, (None, None, None))
298 for po in self.searchFiles(name):
299 result.setdefault(po, []).append(hit)
300 return result
301
302
303#######################################################################
304### Derive Tests from these classes or unittest.TestCase ##############
305#######################################################################
306
307class DepsolveTests(_DepsolveTestsBase):
308 """Run depsolver on an manually set up transaction.
309 You can add pkgs to self.rpmdb or self.tsInfo. See
310 yum/transactioninfo.py for details.
311 A typical test case looks like:
312
313 def testInstallPackageRequireInstalled(self):
314 po = FakePackage('zsh', '1', '1', None, 'i386')
315 po.addRequires('zip', 'EQ', (None, '1.3', '2'))
316 self.tsInfo.addInstall(po)
317
318 ipo = FakePackage('zip', '1.3', '2', None, 'i386')
319 self.rpmdb.addPackage(ipo)
320
321 result, msg = self.resolveCode()
322 self.assertEquals('ok', result, msg)
323 self.assertResult((po, ipo))
324 """
325
326 def setUp(self):
327 """ Called at the start of each test. """
328 _DepsolveTestsBase.setUp(self)
329 self.tsInfo = transactioninfo.TransactionData()
330 self.tsInfo.debug = 1
331 self.rpmdb = FakeRpmDb()
332 self.xsack = packageSack.PackageSack()
333 self.repo = FakeRepo("installed")
334 # XXX this side-affect is hacky:
335 self.tsInfo.setDatabases(self.rpmdb, self.xsack)
336
337 def resetTsInfo(self):
338 self.tsInfo = transactioninfo.TransactionData()
339
340
341 def resolveCode(self):
342 solver = YumBase()
343 solver.conf = FakeConf()
344 solver.arch.setup_arch('x86_64')
345 solver.tsInfo = solver._tsInfo = self.tsInfo
346 solver.rpmdb = self.rpmdb
347 solver.pkgSack = self.xsack
348
349 for po in self.rpmdb:
350 po.repoid = po.repo.id = "installed"
351 for po in self.xsack:
352 if po.repo.id is None:
353 po.repo.id = "TestRepository"
354 po.repoid = po.repo.id
355 for txmbr in self.tsInfo:
356 if txmbr.ts_state in ('u', 'i'):
357 if txmbr.po.repo.id is None:
358 txmbr.po.repo.id = "TestRepository"
359 txmbr.po.repoid = txmbr.po.repo.id
360 else:
361 txmbr.po.repoid = txmbr.po.repo.id = "installed"
362
363 result, msg = solver.resolveDeps()
364 return (self.res[result], msg)
365
366class OperationsTests(_DepsolveTestsBase):
367 """Run a yum command (install, update, remove, ...) in a given set of installed
368 and available pkgs. Typical test case looks like:
369
370 def testUpdate(self):
371 p = self.pkgs
372 res, msg = self.runOperation(['update'], [p.installed], [p.update])
373 self.assert_(res=='ok', msg)
374 self.assertResult((p.update,))
375
376 To avoid creating the same pkgs over and over again overload the staticmethod
377 buildPkgs. It gets called from __init__ with self.pkgs as first parameter.
378 As it is a static method you can call .buildPkgs() from other Tests to share
379 buildPkg code.
380 """
381
382 def runOperation(self, args, installed=[], available=[],
383 confs={}):
384 """Sets up and runs the depsolver. args[0] must be a valid yum command
385 ("install", "update", ...). It might be followed by pkg names as on the
386 yum command line. The pkg objects in installed are added to self.rpmdb and
387 those in available to self.xsack which is the repository to resolve
388 requirements from.
389 """
390 depsolver = YumBaseCli()
391 depsolver.arch.setup_arch('x86_64')
392 self.rpmdb = depsolver.rpmdb = FakeRpmDb()
393 self.xsack = depsolver._pkgSack = packageSack.PackageSack()
394 self.repo = depsolver.repo = FakeRepo("installed")
395 depsolver.conf = FakeConf()
396 for conf in confs:
397 setattr(depsolver.conf, conf, confs[conf])
398 # We are running nosetest, so we want to see some yum output
399 # if a testcase if failing
400 depsolver.doLoggingSetup(9,9)
401 self.depsolver = depsolver
402
403 for po in installed:
404 po.repoid = po.repo.id = "installed"
405 self.depsolver.rpmdb.addPackage(po)
406 for po in available:
407 if po.repo.id is None:
408 po.repo.id = "TestRepository"
409 po.repoid = po.repo.id
410 self.depsolver._pkgSack.addPackage(po)
411
412 self.depsolver.basecmd = args[0]
413 self.depsolver.extcmds = args[1:]
414 res, msg = self.depsolver.doCommands()
415 self.tsInfo = depsolver.tsInfo
416 if res!=2:
417 return res, msg
418 res, msg = self.depsolver.buildTransaction()
419 return self.res[res], msg
Note: See TracBrowser for help on using the repository browser.