« back to all changes in this revision
Viewing changes to test/test_crashdb.py
- browse files at revision 2161
- compare with another revision
- download diff
- download tarball
- view history from revision 2161
- Committer: Martin Pitt
- Date: 2012-02-23 10:31:55 UTC
- Revision ID: martin.pitt@canonical.com-20120223103155-m7p4is95x27vzm8n
Move all test suites out of the code modules into test/test_<module>.py. This avoids having to load it every time the program runs, and also allows running the tests against the installed version of Apport.
- files added:
- test/test_backend-apt-dpkg.py
- files modified:
- NEWS
added
removed
test/test_crashdb.py
1
# coding: UTF-8
2
import unittest, tempfile, shutil, os.path, copy
3
4
import apport
5
from apport.crashdb_impl.memory import CrashDatabase
6
7
class T(unittest.TestCase):
8
def setUp(self):
9
self.workdir = tempfile.mkdtemp()
10
self.dupdb_dir = os.path.join(self.workdir, 'dupdb')
11
self.crashes = CrashDatabase(None, {'dummy_data': '1',
12
'dupdb_url': self.dupdb_dir})
13
14
self.assertEqual(self.crashes.get_comment_url(self.crashes.download(0),
15
0), 'http://foo.bugs.example.com/0')
16
17
# test-suite internal consistency check: Python signatures are
18
# indeed equal and exist
19
assert self.crashes.download(3).crash_signature(), \
20
'test-suite internal check: Python crash sigs exist'
21
self.assertEqual(self.crashes.download(3).crash_signature(),
22
self.crashes.download(4).crash_signature())
23
24
# we should have 5 crashes
25
self.assertEqual(self.crashes.latest_id(), 4)
26
27
def tearDown(self):
28
shutil.rmtree(self.workdir)
29
30
def test_no_dummy_data(self):
31
'''No dummy data is added by default'''
32
33
self.crashes = CrashDatabase(None, {})
34
self.assertEqual(self.crashes.latest_id(), -1)
35
self.assertRaises(IndexError, self.crashes.download, 0)
36
37
def test_retrace_markers(self):
38
'''Bookkeeping in retraced and dupchecked bugs'''
39
40
self.assertEqual(self.crashes.get_unretraced(), set([0, 1, 2]))
41
self.assertEqual(self.crashes.get_dup_unchecked(), set([3, 4]))
42
43
def test_dynamic_crashdb_conf(self):
44
'''Dynamic code in crashdb.conf'''
45
46
# use our dummy crashdb
47
crashdb_conf = tempfile.NamedTemporaryFile()
48
crashdb_conf.write(b'''default = 'testsuite'
49
50
def get_dyn():
51
return str(2 + 2)
52
53
def get_dyn_name():
54
return 'on_the' + 'fly'
55
56
databases = {
57
'testsuite': {
58
'impl': 'memory',
59
'dyn_option': get_dyn(),
60
},
61
get_dyn_name(): {
62
'impl': 'memory',
63
'whoami': 'dynname',
64
}
65
}
66
''')
67
crashdb_conf.flush()
68
69
db = apport.crashdb.get_crashdb(None, None, crashdb_conf.name)
70
self.assertEqual(db.options['dyn_option'], '4')
71
db = apport.crashdb.get_crashdb(None, 'on_thefly', crashdb_conf.name)
72
self.assertFalse('dyn_opion' in db.options)
73
self.assertEqual(db.options['whoami'], 'dynname')
74
75
#
76
# Test memory.py implementation
77
#
78
79
def test_submit(self):
80
'''Crash uploading and downloading'''
81
82
# setUp() already checks upload() and get_comment_url()
83
r = self.crashes.download(0)
84
self.assertEqual(r['SourcePackage'], 'foo')
85
self.assertEqual(r['Package'], 'libfoo1 1.2-3')
86
self.assertEqual(self.crashes.reports[0]['dup_of'], None)
87
88
self.assertRaises(IndexError, self.crashes.download, 5)
89
90
def test_get_affected_packages(self):
91
self.assertEqual(self.crashes.get_affected_packages(0), ['foo'])
92
self.assertEqual(self.crashes.get_affected_packages(1), ['foo'])
93
self.assertEqual(self.crashes.get_affected_packages(2), ['bar'])
94
self.assertEqual(self.crashes.get_affected_packages(3), ['pygoo'])
95
96
def test_update(self):
97
'''update()'''
98
99
r = apport.Report()
100
r['Package'] = 'new'
101
r['FooBar'] = 'Bogus'
102
r['StacktraceTop'] = 'Fresh!'
103
104
self.crashes.update(1, r, 'muhaha')
105
self.assertEqual(self.crashes.reports[1]['comment'], 'muhaha')
106
self.assertEqual(self.crashes.download(1)['Package'], 'new')
107
self.assertEqual(self.crashes.download(1)['StacktraceTop'], 'Fresh!')
108
self.assertEqual(self.crashes.download(1)['FooBar'], 'Bogus')
109
110
self.assertRaises(IndexError, self.crashes.update, 5, None, '')
111
112
def test_update_filter(self):
113
'''update() with key_filter'''
114
115
r = apport.Report()
116
r['Package'] = 'new'
117
r['FooBar'] = 'Bogus'
118
r['StacktraceTop'] = 'Fresh!'
119
120
self.crashes.update(1, r, 'muhaha', key_filter=['FooBar', 'StacktraceTop'])
121
self.assertEqual(self.crashes.reports[1]['comment'], 'muhaha')
122
self.assertEqual(self.crashes.download(1)['Package'], 'libfoo1 1.2-4')
123
self.assertEqual(self.crashes.download(1)['StacktraceTop'], 'Fresh!')
124
self.assertEqual(self.crashes.download(1)['FooBar'], 'Bogus')
125
126
self.assertRaises(IndexError, self.crashes.update, 5, None, '')
127
128
def test_update_traces(self):
129
'''update_traces()'''
130
131
r = apport.Report()
132
r['Package'] = 'new'
133
r['FooBar'] = 'Bogus'
134
r['StacktraceTop'] = 'Fresh!'
135
136
self.crashes.update_traces(1, r, 'muhaha')
137
self.assertEqual(self.crashes.reports[1]['comment'], 'muhaha')
138
self.assertEqual(self.crashes.download(1)['Package'], 'libfoo1 1.2-4')
139
self.assertEqual(self.crashes.download(1)['StacktraceTop'], 'Fresh!')
140
self.assertFalse('FooBar' in self.crashes.download(1))
141
142
self.assertRaises(IndexError, self.crashes.update_traces, 5, None)
143
144
def test_get_distro_release(self):
145
'''get_distro_release()'''
146
147
self.assertEqual(self.crashes.get_distro_release(0), 'FooLinux Pi/2')
148
149
def test_status(self):
150
'''get_unfixed(), get_fixed_version(), duplicate_of(), close_duplicate()'''
151
152
self.assertEqual(self.crashes.get_unfixed(), set([0, 1, 2, 3, 4]))
153
self.assertEqual(self.crashes.get_fixed_version(0), None)
154
self.assertEqual(self.crashes.get_fixed_version(1), None)
155
self.assertEqual(self.crashes.get_fixed_version(3), None)
156
157
self.assertEqual(self.crashes.duplicate_of(0), None)
158
self.assertEqual(self.crashes.duplicate_of(1), None)
159
self.crashes.close_duplicate({}, 1, 0)
160
self.assertEqual(self.crashes.duplicate_of(0), None)
161
self.assertEqual(self.crashes.duplicate_of(1), 0)
162
163
self.assertEqual(self.crashes.get_unfixed(), set([0, 2, 3, 4]))
164
self.assertEqual(self.crashes.get_fixed_version(1), 'invalid')
165
166
self.assertEqual(self.crashes.get_fixed_version(99), 'invalid')
167
168
def test_mark_regression(self):
169
'''mark_regression()'''
170
171
self.crashes.reports[3]['fixed_version'] = '4.1'
172
173
self.crashes.mark_regression(4, 3)
174
self.assertEqual(self.crashes.reports[4]['comment'],
175
'regression, already fixed in #3')
176
self.assertEqual(self.crashes.duplicate_of(3), None)
177
self.assertEqual(self.crashes.duplicate_of(4), None)
178
179
#
180
# Test crash duplication detection API of crashdb.py
181
#
182
183
def test_duplicate_db_fixed(self):
184
'''duplicate_db_fixed()'''
185
186
self.crashes.init_duplicate_db(':memory:')
187
self.assertEqual(self.crashes.check_duplicate(0), None)
188
189
self.assertEqual(self.crashes._duplicate_db_dump(),
190
{self.crashes.download(0).crash_signature(): (0, None)})
191
192
self.crashes.duplicate_db_fixed(0, '42')
193
194
self.assertEqual(self.crashes._duplicate_db_dump(),
195
{self.crashes.download(0).crash_signature(): (0, '42')})
196
197
def test_duplicate_db_remove(self):
198
'''duplicate_db_remove()'''
199
200
# db not yet initialized
201
self.assertRaises(AssertionError, self.crashes.check_duplicate, 0)
202
203
self.crashes.init_duplicate_db(':memory:')
204
205
self.assertEqual(self.crashes.check_duplicate(0), None)
206
self.assertEqual(self.crashes.check_duplicate(2), None)
207
208
# invalid ID (raising KeyError is *hard*, so it's not done)
209
self.crashes.duplicate_db_remove(99)
210
211
# nevertheless, this should not change the DB
212
self.assertEqual(self.crashes._duplicate_db_dump(),
213
{self.crashes.download(0).crash_signature(): (0, None),
214
self.crashes.download(2).crash_signature(): (2, None)})
215
216
# valid ID
217
self.crashes.duplicate_db_remove(2)
218
219
# check DB consistency
220
self.assertEqual(self.crashes._duplicate_db_dump(),
221
{self.crashes.download(0).crash_signature(): (0, None)})
222
223
def test_check_duplicate(self):
224
'''check_duplicate() and known()'''
225
226
# db not yet initialized
227
self.assertRaises(AssertionError, self.crashes.check_duplicate, 0,
228
self.crashes.download(0))
229
self.assertRaises(AssertionError, self.crashes.check_duplicate, 0)
230
231
self.crashes.init_duplicate_db(':memory:')
232
233
self.assertEqual(self.crashes._duplicate_db_dump(), {})
234
235
# ID#0 -> no dup
236
self.assertEqual(self.crashes.known(self.crashes.download(0)), None)
237
self.assertEqual(self.crashes.check_duplicate(0), None)
238
# can't be known before publishing DB
239
self.assertEqual(self.crashes.known(self.crashes.download(0)), None)
240
self.crashes.duplicate_db_publish(self.dupdb_dir)
241
self.assertEqual(self.crashes.known(self.crashes.download(0)),
242
'http://foo.bugs.example.com/0')
243
244
# bug is not a duplicate of itself, when reprocessed
245
self.assertEqual(self.crashes.check_duplicate(0), None)
246
247
# ID#1 -> dup of #0
248
self.crashes.duplicate_db_publish(self.dupdb_dir)
249
self.assertEqual(self.crashes.known(self.crashes.download(1)),
250
'http://foo.bugs.example.com/0')
251
self.assertEqual(self.crashes.check_duplicate(1), (0, None))
252
253
# ID#2 is unrelated, no dup
254
self.crashes.duplicate_db_publish(self.dupdb_dir)
255
self.assertEqual(self.crashes.known(self.crashes.download(2)), None)
256
self.assertEqual(self.crashes.check_duplicate(2), None)
257
self.crashes.duplicate_db_publish(self.dupdb_dir)
258
self.assertEqual(self.crashes.known(self.crashes.download(2)),
259
'http://bar.bugs.example.com/2')
260
261
# ID#3: no dup, master of ID#4
262
self.assertEqual(self.crashes.check_duplicate(3), None)
263
264
# ID#4: dup of ID#3
265
self.assertEqual(self.crashes.check_duplicate(4), (3, None))
266
# not marked as regression
267
self.assertFalse('comment' in self.crashes.reports[3])
268
269
# check DB consistency; #1 and #4 are dupes and do not appear
270
self.assertEqual(self.crashes._duplicate_db_dump(),
271
{self.crashes.download(0).crash_signature(): (0, None),
272
self.crashes.download(2).crash_signature(): (2, None),
273
self.crashes.download(3).crash_signature(): (3, None)})
274
275
# now mark the python crash as fixed
276
self.crashes.reports[3]['fixed_version'] = '4.1'
277
278
# ID#4 is dup of ID#3, but happend in version 5 -> regression
279
self.crashes.close_duplicate(self.crashes.download(4), 4, None) # reset
280
self.assertEqual(self.crashes.check_duplicate(4), None)
281
self.assertEqual(self.crashes.duplicate_of(4), None)
282
self.assertEqual(self.crashes.reports[4]['comment'], 'regression, already fixed in #3')
283
284
# check DB consistency; ID#3 should now be updated to be fixed in 4.1,
285
# and as 4 is a regression, appear as a new crash
286
self.assertEqual(self.crashes._duplicate_db_dump(),
287
{self.crashes.download(0).crash_signature(): (0, None),
288
self.crashes.download(2).crash_signature(): (2, None),
289
self.crashes.download(3).crash_signature(): (3, '4.1'),
290
self.crashes.download(4).crash_signature(): (4, None)})
291
292
# add two more Python crash dups and verify that they are dup'ed
293
# to the correct ID
294
r = copy.copy(self.crashes.download(3))
295
self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
296
'http://pygoo.bugs.example.com/5')
297
self.assertEqual(self.crashes.check_duplicate(5), (3, '4.1'))
298
self.assertEqual(self.crashes.duplicate_of(5), 3)
299
# not marked as regression, happened earlier than #3
300
self.assertFalse('comment' in self.crashes.reports[5])
301
302
r = copy.copy(self.crashes.download(3))
303
r['Package'] = 'python-goo 5.1'
304
self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
305
'http://pygoo.bugs.example.com/6')
306
self.assertEqual(self.crashes.check_duplicate(6), (4, None))
307
self.assertEqual(self.crashes.duplicate_of(6), 4)
308
# not marked as regression, as it's now a dupe of new master bug 4
309
self.assertFalse('comment' in self.crashes.reports[6])
310
311
# check DB consistency; #5 and #6 are dupes of #3 and #4, so no new
312
# entries
313
self.assertEqual(self.crashes._duplicate_db_dump(),
314
{self.crashes.download(0).crash_signature(): (0, None),
315
self.crashes.download(2).crash_signature(): (2, None),
316
self.crashes.download(3).crash_signature(): (3, '4.1'),
317
self.crashes.download(4).crash_signature(): (4, None)})
318
319
# check with unknown fixed version
320
self.crashes.reports[3]['fixed_version'] = ''
321
self.crashes.duplicate_db_fixed(3, '')
322
323
r = copy.copy(self.crashes.download(3))
324
r['Package'] = 'python-goo 5.1'
325
self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
326
'http://pygoo.bugs.example.com/7')
327
self.assertEqual(self.crashes.check_duplicate(7), (3, ''))
328
# not marked as regression
329
self.assertFalse('comment' in self.crashes.reports[6])
330
331
# final consistency check
332
self.assertEqual(self.crashes._duplicate_db_dump(),
333
{self.crashes.download(0).crash_signature(): (0, None),
334
self.crashes.download(2).crash_signature(): (2, None),
335
self.crashes.download(3).crash_signature(): (3, ''),
336
self.crashes.download(4).crash_signature(): (4, None)})
337
338
def test_check_duplicate_utf8(self):
339
'''check_duplicate() with UTF-8 strings'''
340
341
# assertion failure, with UTF-8 strings
342
r = apport.Report()
343
r['Package'] = 'bash 5'
344
r['SourcePackage'] = 'bash'
345
r['DistroRelease'] = 'Testux 2.2'
346
r['ExecutablePath'] = '/bin/bash'
347
r['Signal'] = '6'
348
r['AssertionMessage'] = 'Afirmação x != 0'
349
self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
350
'http://bash.bugs.example.com/5')
351
self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
352
'http://bash.bugs.example.com/6')
353
354
self.crashes.init_duplicate_db(':memory:')
355
self.assertEqual(self.crashes.check_duplicate(5), None)
356
self.assertEqual(self.crashes.check_duplicate(6), (5, None))
357
358
def test_check_duplicate_custom_signature(self):
359
'''check_duplicate() with custom DuplicateSignature: field'''
360
361
r = apport.Report()
362
r['SourcePackage'] = 'bash'
363
r['Package'] = 'bash 5'
364
r['DuplicateSignature'] = 'Code42Blue'
365
self.assertEqual(self.crashes.get_comment_url(r, self.crashes.upload(r)),
366
'http://bash.bugs.example.com/5')
367
368
self.crashes.init_duplicate_db(':memory:')
369
self.assertEqual(self.crashes.check_duplicate(5), None)
370
371
self.assertEqual(self.crashes._duplicate_db_dump(), {'Code42Blue': (5, None)})
372
373
# this one has a standard crash_signature
374
self.assertEqual(self.crashes.check_duplicate(0), None)
375
# ... but DuplicateSignature wins
376
self.crashes.download(0)['DuplicateSignature'] = 'Code42Blue'
377
self.assertEqual(self.crashes.check_duplicate(0), (5, None))
378
379
self.crashes.download(1)['DuplicateSignature'] = 'CodeRed'
380
self.assertEqual(self.crashes.check_duplicate(1), None)
381
self.assertEqual(self.crashes._duplicate_db_dump(),
382
{'Code42Blue': (5, None), 'CodeRed': (1, None),
383
self.crashes.download(0).crash_signature(): (0, None)})
384
385
def test_check_duplicate_report_arg(self):
386
'''check_duplicate() with explicitly passing report'''
387
388
self.crashes.init_duplicate_db(':memory:')
389
390
# ID#0 -> no dup
391
self.assertEqual(self.crashes.check_duplicate(0), None)
392
393
# ID#2 is unrelated, no dup
394
self.assertEqual(self.crashes.check_duplicate(2), None)
395
396
# report from ID#1 is a dup of #0
397
self.assertEqual(self.crashes.check_duplicate(2,
398
self.crashes.download(1)), (0, None))
399
400
def test_known_address_sig(self):
401
'''known() for address signatures'''
402
403
self.crashes.init_duplicate_db(':memory:')
404
405
r = apport.Report()
406
r['SourcePackage'] = 'bash'
407
r['Package'] = 'bash 5'
408
r['ExecutablePath'] = '/bin/bash'
409
r['Signal'] = '11'
410
r['ProcMaps'] = '''
411
00400000-004df000 r-xp 00000000 08:02 1044485 /bin/bash
412
7f491fa8f000-7f491fc24000 r-xp 00000000 08:02 522605 /lib/x86_64-linux-gnu/libc-2.13.so
413
'''
414
415
r['Stacktrace'] = '''
416
#0 0x00007f491fac5687 in kill ()
417
#1 0x000000000042eb76 in ?? ()
418
#2 0x00000000004324d8 in ??
419
#3 0x00000000004707e3 in parse_and_execute ()
420
#4 0x000000000041d703 in _start ()
421
'''
422
423
self.assertNotEqual(r.crash_signature_addresses(), None)
424
self.crashes.duplicate_db_publish(self.dupdb_dir)
425
self.assertEqual(self.crashes.known(r), None)
426
r_id = self.crashes.upload(r)
427
self.assertEqual(self.crashes.check_duplicate(r_id), None)
428
self.crashes.duplicate_db_publish(self.dupdb_dir)
429
self.assertEqual(self.crashes.known(r),
430
self.crashes.get_comment_url(r, r_id))
431
432
# another report with same address signature
433
r2 = apport.Report()
434
r2['SourcePackage'] = 'bash'
435
r2['Package'] = 'bash 5'
436
r2['ExecutablePath'] = '/bin/bash'
437
r2['Signal'] = '11'
438
439
r2['ProcMaps'] = '''
440
00400000-004df000 r-xp 00000000 08:02 1044485 /bin/bash
441
5f491fa8f000-5f491fc24000 r-xp 00000000 08:02 522605 /lib/x86_64-linux-gnu/libc-2.13.so
442
'''
443
444
r2['Stacktrace'] = '''
445
#0 0x00005f491fac5687 in kill ()
446
#1 0x000000000042eb76 in ?? ()
447
#2 0x00000000004324d8 in ??
448
#3 0x00000000004707e3 in parse_and_execute ()
449
#4 0x000000000041d703 in _start ()
450
'''
451
452
self.assertEqual(r.crash_signature_addresses(),
453
r2.crash_signature_addresses())
454
455
# DB knows about this already
456
self.crashes.duplicate_db_publish(self.dupdb_dir)
457
self.assertEqual(self.crashes.known(r2),
458
self.crashes.get_comment_url(r, r_id))
459
460
# if it gets uploaded anyway, duplicate it properly
461
r2_id = self.crashes.upload(r2)
462
self.assertEqual(self.crashes.check_duplicate(r2_id), (r_id, None))
463
464
# different address signature
465
r3 = apport.Report()
466
r3['SourcePackage'] = 'bash'
467
r3['Package'] = 'bash 5'
468
r3['ExecutablePath'] = '/bin/bash'
469
r3['Signal'] = '11'
470
471
r3['ProcMaps'] = '''
472
00400000-004df000 r-xp 00000000 08:02 1044485 /bin/bash
473
5f491fa8f000-5f491fc24000 r-xp 00000000 08:02 522605 /lib/x86_64-linux-gnu/libc-2.13.so
474
'''
475
476
r3['Stacktrace'] = '''
477
#0 0x00005f491fac5687 in kill ()
478
#1 0x000000000042eb76 in ?? ()
479
#2 0x0000000000432401 in ??
480
#3 0x00000000004707e3 in parse_and_execute ()
481
#4 0x000000000041d703 in _start ()
482
'''
483
self.assertNotEqual(r.crash_signature_addresses(),
484
r3.crash_signature_addresses())
485
self.crashes.duplicate_db_publish(self.dupdb_dir)
486
self.assertEqual(self.crashes.known(r3), None)
487
488
# pretend that we went through retracing and r and r3 are actually
489
# dupes; temporarily add a signature here to convince check_duplicate()
490
self.crashes.init_duplicate_db(':memory:')
491
r['DuplicateSignature'] = 'moo'
492
r3['DuplicateSignature'] = 'moo'
493
r_id = self.crashes.upload(r)
494
self.assertEqual(self.crashes.check_duplicate(r_id), None)
495
r3_id = self.crashes.upload(r3)
496
self.assertEqual(self.crashes.check_duplicate(r3_id), (r_id, None))
497
del r['DuplicateSignature']
498
del r3['DuplicateSignature']
499
500
# now both r and r3 address sigs should be known as r_id
501
self.crashes.duplicate_db_publish(self.dupdb_dir)
502
self.assertEqual(self.crashes.known(r),
503
self.crashes.get_comment_url(r, r_id))
504
self.assertEqual(self.crashes.known(r3),
505
self.crashes.get_comment_url(r3, r_id))
506
507
# changing ID also works on address signatures
508
self.crashes.duplicate_db_change_master_id(r_id, r3_id)
509
self.crashes.duplicate_db_publish(self.dupdb_dir)
510
self.assertEqual(self.crashes.known(r),
511
self.crashes.get_comment_url(r, r3_id))
512
self.assertEqual(self.crashes.known(r3),
513
self.crashes.get_comment_url(r3, r3_id))
514
515
# removing an ID also works for address signatures
516
self.crashes.duplicate_db_remove(r3_id)
517
self.crashes.duplicate_db_publish(self.dupdb_dir)
518
self.assertEqual(self.crashes.known(r), None)
519
self.assertEqual(self.crashes.known(r3), None)
520
521
self.assertEqual(self.crashes._duplicate_db_dump(), {})
522
523
def test_change_master_id(self):
524
'''duplicate_db_change_master_id()'''
525
526
# db not yet initialized
527
self.assertRaises(AssertionError, self.crashes.check_duplicate, 0)
528
529
self.crashes.init_duplicate_db(':memory:')
530
531
self.assertEqual(self.crashes.check_duplicate(0), None)
532
self.assertEqual(self.crashes.check_duplicate(2), None)
533
534
# check DB consistency
535
self.assertEqual(self.crashes._duplicate_db_dump(),
536
{self.crashes.download(0).crash_signature(): (0, None),
537
self.crashes.download(2).crash_signature(): (2, None)})
538
539
# invalid ID (raising KeyError is *hard*, so it's not done)
540
self.crashes.duplicate_db_change_master_id(5, 99)
541
542
# nevertheless, this should not change the DB
543
self.assertEqual(self.crashes._duplicate_db_dump(),
544
{self.crashes.download(0).crash_signature(): (0, None),
545
self.crashes.download(2).crash_signature(): (2, None)})
546
547
# valid ID
548
self.crashes.duplicate_db_change_master_id(2, 99)
549
550
# check DB consistency
551
self.assertEqual(self.crashes._duplicate_db_dump(),
552
{self.crashes.download(0).crash_signature(): (0, None),
553
self.crashes.download(2).crash_signature(): (99, None)})
554
555
def test_db_corruption(self):
556
'''Detection of DB file corruption'''
557
558
try:
559
(fd, db) = tempfile.mkstemp()
560
os.close(fd)
561
self.crashes.init_duplicate_db(db)
562
self.assertEqual(self.crashes.check_duplicate(0), None)
563
self.assertEqual(self.crashes._duplicate_db_dump(),
564
{self.crashes.download(0).crash_signature(): (0, None)})
565
self.crashes.duplicate_db_fixed(0, '42')
566
self.assertEqual(self.crashes._duplicate_db_dump(),
567
{self.crashes.download(0).crash_signature(): (0, '42')})
568
569
del self.crashes
570
571
# damage file
572
f = open(db, 'r+')
573
f.truncate(os.path.getsize(db)*2/3)
574
f.close()
575
576
self.crashes = CrashDatabase(None, {})
577
self.assertRaises(Exception, self.crashes.init_duplicate_db, db)
578
579
finally:
580
os.unlink(db)
581
582
if __name__ == '__main__':
583
unittest.main()
Loggerhead is a web-based interface for Breezy
Version: 2.0.1