Back to index

apport  2.4
test_crashdb.py
Go to the documentation of this file.
00001 # coding: UTF-8
00002 import unittest, tempfile, shutil, os.path, copy
00003 
00004 import apport
00005 from apport.crashdb_impl.memory import CrashDatabase
00006 
00007 
00008 class T(unittest.TestCase):
00009     def setUp(self):
00010         self.workdir = tempfile.mkdtemp()
00011         self.dupdb_dir = os.path.join(self.workdir, 'dupdb')
00012         self.crashes = CrashDatabase(None, {'dummy_data': '1',
00013                                             'dupdb_url': 'file://' + self.dupdb_dir})
00014 
00015         self.assertEqual(self.crashes.get_comment_url(self.crashes.download(0), 0),
00016                          'http://foo.bugs.example.com/0')
00017 
00018         # test-suite internal consistency check: Python signatures are
00019         # indeed equal and exist
00020         assert self.crashes.download(3).crash_signature(), \
00021             'test-suite internal check: Python crash sigs exist'
00022         self.assertEqual(self.crashes.download(3).crash_signature(),
00023                          self.crashes.download(4).crash_signature())
00024 
00025         # we should have 5 crashes
00026         self.assertEqual(self.crashes.latest_id(), 4)
00027 
00028     def tearDown(self):
00029         shutil.rmtree(self.workdir)
00030 
00031     def test_no_dummy_data(self):
00032         '''No dummy data is added by default'''
00033 
00034         self.crashes = CrashDatabase(None, {})
00035         self.assertEqual(self.crashes.latest_id(), -1)
00036         self.assertRaises(IndexError, self.crashes.download, 0)
00037 
00038     def test_retrace_markers(self):
00039         '''Bookkeeping in retraced and dupchecked bugs'''
00040 
00041         self.assertEqual(self.crashes.get_unretraced(), set([0, 1, 2]))
00042         self.assertEqual(self.crashes.get_dup_unchecked(), set([3, 4]))
00043 
00044     def test_dynamic_crashdb_conf(self):
00045         '''Dynamic code in crashdb.conf'''
00046 
00047         # use our dummy crashdb
00048         crashdb_conf = tempfile.NamedTemporaryFile()
00049         crashdb_conf.write(b'''default = 'testsuite'
00050 
00051 def get_dyn():
00052     return str(2 + 2)
00053 
00054 def get_dyn_name():
00055     return 'on_the' + 'fly'
00056 
00057 databases = {
00058     'testsuite': {
00059         'impl': 'memory',
00060         'dyn_option': get_dyn(),
00061     },
00062     get_dyn_name(): {
00063         'impl': 'memory',
00064         'whoami': 'dynname',
00065     }
00066 }
00067 ''')
00068         crashdb_conf.flush()
00069 
00070         db = apport.crashdb.get_crashdb(None, None, crashdb_conf.name)
00071         self.assertEqual(db.options['dyn_option'], '4')
00072         db = apport.crashdb.get_crashdb(None, 'on_thefly', crashdb_conf.name)
00073         self.assertFalse('dyn_opion' in db.options)
00074         self.assertEqual(db.options['whoami'], 'dynname')
00075 
00076     def test_accepts_default(self):
00077         '''accepts(): default configuration'''
00078 
00079         # by default crash DBs accept any type
00080         self.assertTrue(self.crashes.accepts(apport.Report('Crash')))
00081         self.assertTrue(self.crashes.accepts(apport.Report('Bug')))
00082         self.assertTrue(self.crashes.accepts(apport.Report('weirdtype')))
00083 
00084     def test_accepts_problem_types(self):
00085         '''accepts(): problem_types option in crashdb.conf'''
00086 
00087         # create a crash DB with type limits
00088         crashdb_conf = tempfile.NamedTemporaryFile()
00089         crashdb_conf.write(b'''default = 'testsuite'
00090 
00091 databases = {
00092     'testsuite': {
00093         'impl': 'memory',
00094         'problem_types': ['Bug', 'Kernel'],
00095     },
00096 }
00097 ''')
00098         crashdb_conf.flush()
00099 
00100         db = apport.crashdb.get_crashdb(None, None, crashdb_conf.name)
00101 
00102         self.assertTrue(db.accepts(apport.Report('Bug')))
00103         self.assertFalse(db.accepts(apport.Report('Crash')))
00104         self.assertFalse(db.accepts(apport.Report('weirdtype')))
00105 
00106     #
00107     # Test memory.py implementation
00108     #
00109 
00110     def test_submit(self):
00111         '''Crash uploading and downloading'''
00112 
00113         # setUp() already checks upload() and get_comment_url()
00114         r = self.crashes.download(0)
00115         self.assertEqual(r['SourcePackage'], 'foo')
00116         self.assertEqual(r['Package'], 'libfoo1 1.2-3')
00117         self.assertEqual(self.crashes.reports[0]['dup_of'], None)
00118 
00119         self.assertRaises(IndexError, self.crashes.download, 5)
00120 
00121     def test_get_affected_packages(self):
00122         self.assertEqual(self.crashes.get_affected_packages(0), ['foo'])
00123         self.assertEqual(self.crashes.get_affected_packages(1), ['foo'])
00124         self.assertEqual(self.crashes.get_affected_packages(2), ['bar'])
00125         self.assertEqual(self.crashes.get_affected_packages(3), ['pygoo'])
00126 
00127     def test_update(self):
00128         '''update()'''
00129 
00130         r = apport.Report()
00131         r['Package'] = 'new'
00132         r['FooBar'] = 'Bogus'
00133         r['StacktraceTop'] = 'Fresh!'
00134 
00135         self.crashes.update(1, r, 'muhaha')
00136         self.assertEqual(self.crashes.reports[1]['comment'], 'muhaha')
00137         self.assertEqual(self.crashes.download(1)['Package'], 'new')
00138         self.assertEqual(self.crashes.download(1)['StacktraceTop'], 'Fresh!')
00139         self.assertEqual(self.crashes.download(1)['FooBar'], 'Bogus')
00140 
00141         self.assertRaises(IndexError, self.crashes.update, 5, None, '')
00142 
00143     def test_update_filter(self):
00144         '''update() with key_filter'''
00145 
00146         r = apport.Report()
00147         r['Package'] = 'new'
00148         r['FooBar'] = 'Bogus'
00149         r['StacktraceTop'] = 'Fresh!'
00150 
00151         self.crashes.update(1, r, 'muhaha', key_filter=['FooBar', 'StacktraceTop'])
00152         self.assertEqual(self.crashes.reports[1]['comment'], 'muhaha')
00153         self.assertEqual(self.crashes.download(1)['Package'], 'libfoo1 1.2-4')
00154         self.assertEqual(self.crashes.download(1)['StacktraceTop'], 'Fresh!')
00155         self.assertEqual(self.crashes.download(1)['FooBar'], 'Bogus')
00156 
00157         self.assertRaises(IndexError, self.crashes.update, 5, None, '')
00158 
00159     def test_update_traces(self):
00160         '''update_traces()'''
00161 
00162         r = apport.Report()
00163         r['Package'] = 'new'
00164         r['FooBar'] = 'Bogus'
00165         r['StacktraceTop'] = 'Fresh!'
00166 
00167         self.crashes.update_traces(1, r, 'muhaha')
00168         self.assertEqual(self.crashes.reports[1]['comment'], 'muhaha')
00169         self.assertEqual(self.crashes.download(1)['Package'], 'libfoo1 1.2-4')
00170         self.assertEqual(self.crashes.download(1)['StacktraceTop'], 'Fresh!')
00171         self.assertFalse('FooBar' in self.crashes.download(1))
00172 
00173         self.assertRaises(IndexError, self.crashes.update_traces, 5, None)
00174 
00175     def test_get_distro_release(self):
00176         '''get_distro_release()'''
00177 
00178         self.assertEqual(self.crashes.get_distro_release(0), 'FooLinux Pi/2')
00179 
00180     def test_status(self):
00181         '''get_unfixed(), get_fixed_version(), duplicate_of(), close_duplicate()'''
00182 
00183         self.assertEqual(self.crashes.get_unfixed(), set([0, 1, 2, 3, 4]))
00184         self.assertEqual(self.crashes.get_fixed_version(0), None)
00185         self.assertEqual(self.crashes.get_fixed_version(1), None)
00186         self.assertEqual(self.crashes.get_fixed_version(3), None)
00187 
00188         self.assertEqual(self.crashes.duplicate_of(0), None)
00189         self.assertEqual(self.crashes.duplicate_of(1), None)
00190         self.crashes.close_duplicate({}, 1, 0)
00191         self.assertEqual(self.crashes.duplicate_of(0), None)
00192         self.assertEqual(self.crashes.duplicate_of(1), 0)
00193 
00194         self.assertEqual(self.crashes.get_unfixed(), set([0, 2, 3, 4]))
00195         self.assertEqual(self.crashes.get_fixed_version(1), 'invalid')
00196 
00197         self.assertEqual(self.crashes.get_fixed_version(99), 'invalid')
00198 
00199     def test_mark_regression(self):
00200         '''mark_regression()'''
00201 
00202         self.crashes.reports[3]['fixed_version'] = '4.1'
00203 
00204         self.crashes.mark_regression(4, 3)
00205         self.assertEqual(self.crashes.reports[4]['comment'],
00206                          'regression, already fixed in #3')
00207         self.assertEqual(self.crashes.duplicate_of(3), None)
00208         self.assertEqual(self.crashes.duplicate_of(4), None)
00209 
00210     #
00211     # Test crash duplication detection API of crashdb.py
00212     #
00213 
00214     def test_duplicate_db_fixed(self):
00215         '''duplicate_db_fixed()'''
00216 
00217         self.crashes.init_duplicate_db(':memory:')
00218         self.assertEqual(self.crashes.check_duplicate(0), None)
00219 
00220         self.assertEqual(self.crashes._duplicate_db_dump(),
00221                          {self.crashes.download(0).crash_signature(): (0, None)})
00222 
00223         self.crashes.duplicate_db_fixed(0, '42')
00224 
00225         self.assertEqual(self.crashes._duplicate_db_dump(),
00226                          {self.crashes.download(0).crash_signature(): (0, '42')})
00227 
00228     def test_duplicate_db_remove(self):
00229         '''duplicate_db_remove()'''
00230 
00231         # db not yet initialized
00232         self.assertRaises(AssertionError, self.crashes.check_duplicate, 0)
00233 
00234         self.crashes.init_duplicate_db(':memory:')
00235 
00236         self.assertEqual(self.crashes.check_duplicate(0), None)
00237         self.assertEqual(self.crashes.check_duplicate(2), None)
00238 
00239         # invalid ID (raising KeyError is *hard*, so it's not done)
00240         self.crashes.duplicate_db_remove(99)
00241 
00242         # nevertheless, this should not change the DB
00243         self.assertEqual(self.crashes._duplicate_db_dump(),
00244                          {self.crashes.download(0).crash_signature(): (0, None),
00245                           self.crashes.download(2).crash_signature(): (2, None)})
00246 
00247         # valid ID
00248         self.crashes.duplicate_db_remove(2)
00249 
00250         # check DB consistency
00251         self.assertEqual(self.crashes._duplicate_db_dump(),
00252                          {self.crashes.download(0).crash_signature(): (0, None)})
00253 
00254     def test_check_duplicate(self):
00255         '''check_duplicate() and known()'''
00256 
00257         # db not yet initialized
00258         self.assertRaises(AssertionError, self.crashes.check_duplicate, 0,
00259                           self.crashes.download(0))
00260         self.assertRaises(AssertionError, self.crashes.check_duplicate, 0)
00261 
00262         self.crashes.init_duplicate_db(':memory:')
00263 
00264         self.assertEqual(self.crashes._duplicate_db_dump(), {})
00265 
00266         # ID#0 -> no dup
00267         self.assertEqual(self.crashes.known(self.crashes.download(0)), None)
00268         self.assertEqual(self.crashes.check_duplicate(0), None)
00269         # can't be known before publishing DB
00270         self.assertEqual(self.crashes.known(self.crashes.download(0)), None)
00271         self.crashes.duplicate_db_publish(self.dupdb_dir)
00272         self.assertEqual(self.crashes.known(self.crashes.download(0)),
00273                          'http://foo.bugs.example.com/0')
00274 
00275         # bug is not a duplicate of itself, when reprocessed
00276         self.assertEqual(self.crashes.check_duplicate(0), None)
00277 
00278         # ID#1 -> dup of #0
00279         self.crashes.duplicate_db_publish(self.dupdb_dir)
00280         self.assertEqual(self.crashes.known(self.crashes.download(1)),
00281                          'http://foo.bugs.example.com/0')
00282         self.assertEqual(self.crashes.check_duplicate(1), (0, None))
00283 
00284         # ID#2 is unrelated, no dup
00285         self.crashes.duplicate_db_publish(self.dupdb_dir)
00286         self.assertEqual(self.crashes.known(self.crashes.download(2)), None)
00287         self.assertEqual(self.crashes.check_duplicate(2), None)
00288         self.crashes.duplicate_db_publish(self.dupdb_dir)
00289         self.assertEqual(self.crashes.known(self.crashes.download(2)),
00290                          'http://bar.bugs.example.com/2')
00291 
00292         # ID#3: no dup, master of ID#4
00293         self.assertEqual(self.crashes.check_duplicate(3), None)
00294 
00295         # ID#4: dup of ID#3
00296         self.assertEqual(self.crashes.check_duplicate(4), (3, None))
00297         # not marked as regression
00298         self.assertFalse('comment' in self.crashes.reports[3])
00299 
00300         # check DB consistency; #1 and #4 are dupes and do not appear
00301         self.assertEqual(self.crashes._duplicate_db_dump(),
00302                          {self.crashes.download(0).crash_signature(): (0, None),
00303                           self.crashes.download(2).crash_signature(): (2, None),
00304                           self.crashes.download(3).crash_signature(): (3, None)})
00305 
00306         # now mark the python crash as fixed
00307         self.crashes.reports[3]['fixed_version'] = '4.1'
00308 
00309         # ID#4 is dup of ID#3, but happend in version 5 -> regression
00310         self.crashes.close_duplicate(self.crashes.download(4), 4, None)  # reset
00311         self.assertEqual(self.crashes.check_duplicate(4), None)
00312         self.assertEqual(self.crashes.duplicate_of(4), None)
00313         self.assertEqual(self.crashes.reports[4]['comment'], 'regression, already fixed in #3')
00314 
00315         # check DB consistency; ID#3 should now be updated to be fixed in 4.1,
00316         # and as 4 is a regression, appear as a new crash
00317         self.assertEqual(self.crashes._duplicate_db_dump(),
00318                          {self.crashes.download(0).crash_signature(): (0, None),
00319                           self.crashes.download(2).crash_signature(): (2, None),
00320                           self.crashes.download(3).crash_signature(): (3, '4.1'),
00321                           self.crashes.download(4).crash_signature(): (4, None)})
00322 
00323         # add two more  Python crash dups and verify that they are dup'ed
00324         # to the correct ID
00325         r = copy.copy(self.crashes.download(3))
00326         self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
00327                          'http://pygoo.bugs.example.com/5')
00328         self.assertEqual(self.crashes.check_duplicate(5), (3, '4.1'))
00329         self.assertEqual(self.crashes.duplicate_of(5), 3)
00330         # not marked as regression, happened earlier than #3
00331         self.assertFalse('comment' in self.crashes.reports[5])
00332 
00333         r = copy.copy(self.crashes.download(3))
00334         r['Package'] = 'python-goo 5.1'
00335         self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
00336                          'http://pygoo.bugs.example.com/6')
00337         self.assertEqual(self.crashes.check_duplicate(6), (4, None))
00338         self.assertEqual(self.crashes.duplicate_of(6), 4)
00339         # not marked as regression, as it's now a dupe of new master bug 4
00340         self.assertFalse('comment' in self.crashes.reports[6])
00341 
00342         # check DB consistency; #5 and #6 are dupes of #3 and #4, so no new
00343         # entries
00344         self.assertEqual(self.crashes._duplicate_db_dump(),
00345                          {self.crashes.download(0).crash_signature(): (0, None),
00346                           self.crashes.download(2).crash_signature(): (2, None),
00347                           self.crashes.download(3).crash_signature(): (3, '4.1'),
00348                           self.crashes.download(4).crash_signature(): (4, None)})
00349 
00350         # check with unknown fixed version
00351         self.crashes.reports[3]['fixed_version'] = ''
00352         self.crashes.duplicate_db_fixed(3, '')
00353 
00354         r = copy.copy(self.crashes.download(3))
00355         r['Package'] = 'python-goo 5.1'
00356         self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
00357                          'http://pygoo.bugs.example.com/7')
00358         self.assertEqual(self.crashes.check_duplicate(7), (3, ''))
00359         # not marked as regression
00360         self.assertFalse('comment' in self.crashes.reports[6])
00361 
00362         # final consistency check
00363         self.assertEqual(self.crashes._duplicate_db_dump(),
00364                          {self.crashes.download(0).crash_signature(): (0, None),
00365                           self.crashes.download(2).crash_signature(): (2, None),
00366                           self.crashes.download(3).crash_signature(): (3, ''),
00367                           self.crashes.download(4).crash_signature(): (4, None)})
00368 
00369     def test_check_duplicate_utf8(self):
00370         '''check_duplicate() with UTF-8 strings'''
00371 
00372         # assertion failure, with UTF-8 strings
00373         r = apport.Report()
00374         r['Package'] = 'bash 5'
00375         r['SourcePackage'] = 'bash'
00376         r['DistroRelease'] = 'Testux 2.2'
00377         r['ExecutablePath'] = '/bin/bash'
00378         r['Signal'] = '6'
00379         r['AssertionMessage'] = 'Afirmação x != 0'
00380         self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
00381                          'http://bash.bugs.example.com/5')
00382         self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
00383                          'http://bash.bugs.example.com/6')
00384 
00385         self.crashes.init_duplicate_db(':memory:')
00386         self.assertEqual(self.crashes.check_duplicate(5), None)
00387         self.assertEqual(self.crashes.check_duplicate(6), (5, None))
00388 
00389         self.crashes.duplicate_db_publish(self.dupdb_dir)
00390 
00391     def test_check_duplicate_custom_signature(self):
00392         '''check_duplicate() with custom DuplicateSignature: field'''
00393 
00394         r = apport.Report()
00395         r['SourcePackage'] = 'bash'
00396         r['Package'] = 'bash 5'
00397         r['DuplicateSignature'] = 'Code42Blue'
00398         self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
00399                          'http://bash.bugs.example.com/5')
00400 
00401         self.crashes.init_duplicate_db(':memory:')
00402         self.assertEqual(self.crashes.check_duplicate(5), None)
00403 
00404         self.assertEqual(self.crashes._duplicate_db_dump(), {'Code42Blue': (5, None)})
00405 
00406         # this one has a standard crash_signature
00407         self.assertEqual(self.crashes.check_duplicate(0), None)
00408         # ... but DuplicateSignature wins
00409         self.crashes.download(0)['DuplicateSignature'] = 'Code42Blue'
00410         self.assertEqual(self.crashes.check_duplicate(0), (5, None))
00411 
00412         self.crashes.download(1)['DuplicateSignature'] = 'CodeRed'
00413         self.assertEqual(self.crashes.check_duplicate(1), None)
00414         self.assertEqual(self.crashes._duplicate_db_dump(),
00415                          {'Code42Blue': (5, None), 'CodeRed': (1, None),
00416                           self.crashes.download(0).crash_signature(): (0, None)})
00417 
00418     def test_check_duplicate_report_arg(self):
00419         '''check_duplicate() with explicitly passing report'''
00420 
00421         self.crashes.init_duplicate_db(':memory:')
00422 
00423         # ID#0 -> no dup
00424         self.assertEqual(self.crashes.check_duplicate(0), None)
00425 
00426         # ID#2 is unrelated, no dup
00427         self.assertEqual(self.crashes.check_duplicate(2), None)
00428 
00429         # report from ID#1 is a dup of #0
00430         self.assertEqual(self.crashes.check_duplicate(2, self.crashes.download(1)),
00431                          (0, None))
00432 
00433     def test_check_duplicate_multiple_masters(self):
00434         '''check_duplicate() with multiple master bugs
00435 
00436         Due to the unavoidable jitter in gdb stack traces, it can happen that a
00437         bug B has the same symbolic signature as a bug S, but the same address
00438         signature as a bug A, where A and S have slightly different symbolic
00439         and address signatures and thus were not identified as duplicates. In
00440         that case we want the lowest ID to become the new master bug, and the
00441         other two duplicates.
00442         '''
00443         a = apport.Report()
00444         a['SourcePackage'] = 'bash'
00445         a['Package'] = 'bash 5'
00446         a.crash_signature = lambda: '/bin/bash:11:read:main'
00447         a.crash_signature_addresses = lambda: '/bin/bash:11:/lib/libc.so+123:/bin/bash+DEAD'
00448         self.assertEqual(self.crashes.get_comment_url(a, self.crashes.upload(a)),
00449                          'http://bash.bugs.example.com/5')
00450 
00451         s = apport.Report()
00452         s['SourcePackage'] = 'bash'
00453         s['Package'] = 'bash 5'
00454         s.crash_signature = lambda: '/bin/bash:11:__getch:read:main'
00455         s.crash_signature_addresses = lambda: '/bin/bash:11:/lib/libc.so+BEEF:/lib/libc.so+123:/bin/bash+DEAD'
00456         self.assertEqual(self.crashes.get_comment_url(s, self.crashes.upload(s)),
00457                          'http://bash.bugs.example.com/6')
00458 
00459         # same addr sig as a, same symbolic sig as s
00460         b = apport.Report()
00461         b['SourcePackage'] = 'bash'
00462         b['Package'] = 'bash 5'
00463         b.crash_signature = lambda: '/bin/bash:11:__getch:read:main'
00464         b.crash_signature_addresses = lambda: '/bin/bash:11:/lib/libc.so+123:/bin/bash+DEAD'
00465         self.assertEqual(self.crashes.get_comment_url(b, self.crashes.upload(b)),
00466                          'http://bash.bugs.example.com/7')
00467 
00468         self.crashes.init_duplicate_db(':memory:')
00469         self.assertEqual(self.crashes.check_duplicate(5, a), None)
00470 
00471         # a and s have slightly different sigs -> no dupe
00472         self.assertEqual(self.crashes.check_duplicate(6, s), None)
00473 
00474         # now throw the interesting b at it
00475         self.assertEqual(self.crashes.check_duplicate(7, b), (5, None))
00476 
00477         # s and b should now be duplicates of a
00478         self.assertEqual(self.crashes.duplicate_of(5), None)
00479         self.assertEqual(self.crashes.duplicate_of(6), 5)
00480         self.assertEqual(self.crashes.duplicate_of(7), 5)
00481 
00482         # sig DB should only have a now
00483         self.assertEqual(self.crashes._duplicate_db_dump(), {'/bin/bash:11:read:main': (5, None)})
00484 
00485         # addr DB should have both possible patterns on a
00486         self.assertEqual(self.crashes._duplicate_search_address_signature(b.crash_signature_addresses()), 5)
00487         self.assertEqual(self.crashes._duplicate_search_address_signature(s.crash_signature_addresses()), 5)
00488 
00489     def test_known_address_sig(self):
00490         '''known() for address signatures'''
00491 
00492         self.crashes.init_duplicate_db(':memory:')
00493 
00494         r = apport.Report()
00495         r['SourcePackage'] = 'bash'
00496         r['Package'] = 'bash 5'
00497         r['ExecutablePath'] = '/bin/bash'
00498         r['Signal'] = '11'
00499         r['ProcMaps'] = '''
00500 00400000-004df000 r-xp 00000000 08:02 1044485                            /bin/bash
00501 7f491fa8f000-7f491fc24000 r-xp 00000000 08:02 522605                     /lib/x86_64-linux-gnu/libc-2.13.so
00502 '''
00503 
00504         r['Stacktrace'] = '''
00505 #0  0x00007f491fac5687 in kill ()
00506 #1  0x000000000042eb76 in ?? ()
00507 #2  0x00000000004324d8 in ??
00508 #3  0x00000000004707e3 in parse_and_execute ()
00509 #4  0x000000000041d703 in _start ()
00510 '''
00511 
00512         self.assertNotEqual(r.crash_signature_addresses(), None)
00513         self.crashes.duplicate_db_publish(self.dupdb_dir)
00514         self.assertEqual(self.crashes.known(r), None)
00515         r_id = self.crashes.upload(r)
00516         self.assertEqual(self.crashes.check_duplicate(r_id), None)
00517         self.crashes.duplicate_db_publish(self.dupdb_dir)
00518         self.assertEqual(self.crashes.known(r),
00519                          self.crashes.get_comment_url(r, r_id))
00520 
00521         # another report with same address signature
00522         r2 = apport.Report()
00523         r2['SourcePackage'] = 'bash'
00524         r2['Package'] = 'bash 5'
00525         r2['ExecutablePath'] = '/bin/bash'
00526         r2['Signal'] = '11'
00527 
00528         r2['ProcMaps'] = '''
00529 00400000-004df000 r-xp 00000000 08:02 1044485                            /bin/bash
00530 5f491fa8f000-5f491fc24000 r-xp 00000000 08:02 522605                     /lib/x86_64-linux-gnu/libc-2.13.so
00531 '''
00532 
00533         r2['Stacktrace'] = '''
00534 #0  0x00005f491fac5687 in kill ()
00535 #1  0x000000000042eb76 in ?? ()
00536 #2  0x00000000004324d8 in ??
00537 #3  0x00000000004707e3 in parse_and_execute ()
00538 #4  0x000000000041d703 in _start ()
00539 '''
00540 
00541         self.assertEqual(r.crash_signature_addresses(),
00542                          r2.crash_signature_addresses())
00543 
00544         # DB knows about this already
00545         self.crashes.duplicate_db_publish(self.dupdb_dir)
00546         self.assertEqual(self.crashes.known(r2),
00547                          self.crashes.get_comment_url(r, r_id))
00548 
00549         # if it gets uploaded anyway, duplicate it properly
00550         r2_id = self.crashes.upload(r2)
00551         self.assertEqual(self.crashes.check_duplicate(r2_id), (r_id, None))
00552 
00553         # different address signature
00554         r3 = apport.Report()
00555         r3['SourcePackage'] = 'bash'
00556         r3['Package'] = 'bash 5'
00557         r3['ExecutablePath'] = '/bin/bash'
00558         r3['Signal'] = '11'
00559 
00560         r3['ProcMaps'] = '''
00561 00400000-004df000 r-xp 00000000 08:02 1044485                            /bin/bash
00562 5f491fa8f000-5f491fc24000 r-xp 00000000 08:02 522605                     /lib/x86_64-linux-gnu/libc-2.13.so
00563 '''
00564 
00565         r3['Stacktrace'] = '''
00566 #0  0x00005f491fac5687 in kill ()
00567 #1  0x000000000042eb76 in ?? ()
00568 #2  0x0000000000432401 in ??
00569 #3  0x00000000004707e3 in parse_and_execute ()
00570 #4  0x000000000041d703 in _start ()
00571 '''
00572         self.assertNotEqual(r.crash_signature_addresses(),
00573                             r3.crash_signature_addresses())
00574         self.crashes.duplicate_db_publish(self.dupdb_dir)
00575         self.assertEqual(self.crashes.known(r3), None)
00576 
00577         # pretend that we went through retracing and r and r3 are actually
00578         # dupes; temporarily add a signature here to convince check_duplicate()
00579         self.crashes.init_duplicate_db(':memory:')
00580         r['DuplicateSignature'] = 'moo'
00581         r3['DuplicateSignature'] = 'moo'
00582         r_id = self.crashes.upload(r)
00583         self.assertEqual(self.crashes.check_duplicate(r_id), None)
00584         r3_id = self.crashes.upload(r3)
00585         self.assertEqual(self.crashes.check_duplicate(r3_id), (r_id, None))
00586         del r['DuplicateSignature']
00587         del r3['DuplicateSignature']
00588 
00589         # now both r and r3 address sigs should be known as r_id
00590         self.crashes.duplicate_db_publish(self.dupdb_dir)
00591         self.assertEqual(self.crashes.known(r),
00592                          self.crashes.get_comment_url(r, r_id))
00593         self.assertEqual(self.crashes.known(r3),
00594                          self.crashes.get_comment_url(r3, r_id))
00595 
00596         # changing ID also works on address signatures
00597         self.crashes.duplicate_db_change_master_id(r_id, r3_id)
00598         self.crashes.duplicate_db_publish(self.dupdb_dir)
00599         self.assertEqual(self.crashes.known(r),
00600                          self.crashes.get_comment_url(r, r3_id))
00601         self.assertEqual(self.crashes.known(r3),
00602                          self.crashes.get_comment_url(r3, r3_id))
00603 
00604         # removing an ID also works for address signatures
00605         self.crashes.duplicate_db_remove(r3_id)
00606         self.crashes.duplicate_db_publish(self.dupdb_dir)
00607         self.assertEqual(self.crashes.known(r), None)
00608         self.assertEqual(self.crashes.known(r3), None)
00609 
00610         self.assertEqual(self.crashes._duplicate_db_dump(), {})
00611 
00612     def test_duplicate_db_publish_long_sigs(self):
00613         '''duplicate_db_publish() with very long signatures'''
00614 
00615         self.crashes.init_duplicate_db(':memory:')
00616 
00617         # give #0 a long symbolic sig which needs lots of quoting
00618         symb = self.crashes.download(0)
00619         symb.crash_signature = lambda: 's+' * 1000
00620 
00621         # and #1 a long addr sig
00622         addr = self.crashes.download(1)
00623         addr.crash_signature_addresses = lambda: '0x1+/' * 1000
00624 
00625         self.assertEqual(self.crashes.known(symb), None)
00626         self.assertEqual(self.crashes.check_duplicate(0), None)
00627         self.assertEqual(self.crashes.known(addr), None)
00628         self.assertEqual(self.crashes.check_duplicate(1), None)
00629 
00630         self.crashes.duplicate_db_publish(self.dupdb_dir)
00631         self.assertEqual(self.crashes.known(symb), 'http://foo.bugs.example.com/0')
00632         self.assertEqual(self.crashes.known(addr), 'http://foo.bugs.example.com/1')
00633 
00634     def test_change_master_id(self):
00635         '''duplicate_db_change_master_id()'''
00636 
00637         # db not yet initialized
00638         self.assertRaises(AssertionError, self.crashes.check_duplicate, 0)
00639 
00640         self.crashes.init_duplicate_db(':memory:')
00641 
00642         self.assertEqual(self.crashes.check_duplicate(0), None)
00643         self.assertEqual(self.crashes.check_duplicate(2), None)
00644 
00645         # check DB consistency
00646         self.assertEqual(self.crashes._duplicate_db_dump(),
00647                          {self.crashes.download(0).crash_signature(): (0, None),
00648                           self.crashes.download(2).crash_signature(): (2, None)})
00649 
00650         # invalid ID (raising KeyError is *hard*, so it's not done)
00651         self.crashes.duplicate_db_change_master_id(5, 99)
00652 
00653         # nevertheless, this should not change the DB
00654         self.assertEqual(self.crashes._duplicate_db_dump(),
00655                          {self.crashes.download(0).crash_signature(): (0, None),
00656                           self.crashes.download(2).crash_signature(): (2, None)})
00657 
00658         # valid ID
00659         self.crashes.duplicate_db_change_master_id(2, 99)
00660 
00661         # check DB consistency
00662         self.assertEqual(self.crashes._duplicate_db_dump(),
00663                          {self.crashes.download(0).crash_signature(): (0, None),
00664                           self.crashes.download(2).crash_signature(): (99, None)})
00665 
00666     def test_db_corruption(self):
00667         '''Detection of DB file corruption'''
00668 
00669         try:
00670             (fd, db) = tempfile.mkstemp()
00671             os.close(fd)
00672             self.crashes.init_duplicate_db(db)
00673             self.assertEqual(self.crashes.check_duplicate(0), None)
00674             self.assertEqual(self.crashes._duplicate_db_dump(),
00675                              {self.crashes.download(0).crash_signature(): (0, None)})
00676             self.crashes.duplicate_db_fixed(0, '42')
00677             self.assertEqual(self.crashes._duplicate_db_dump(),
00678                              {self.crashes.download(0).crash_signature(): (0, '42')})
00679 
00680             del self.crashes
00681 
00682             # damage file
00683             f = open(db, 'r+')
00684             f.truncate(os.path.getsize(db) * 2 / 3)
00685             f.close()
00686 
00687             self.crashes = CrashDatabase(None, {})
00688             self.assertRaises(Exception, self.crashes.init_duplicate_db, db)
00689 
00690         finally:
00691             os.unlink(db)
00692 
00693 if __name__ == '__main__':
00694     unittest.main()