Back to index

apport  2.3
test_python_crashes.py
Go to the documentation of this file.
00001 # Test apport_python_hook.py
00002 #
00003 # Copyright (c) 2006 - 2011 Canonical Ltd.
00004 # Authors: Robert Collins <robert@ubuntu.com>
00005 #          Martin Pitt <martin.pitt@ubuntu.com>
00006 #
00007 # This program is free software; you can redistribute it and/or modify it
00008 # under the terms of the GNU General Public License as published by the
00009 # Free Software Foundation; either version 2 of the License, or (at your
00010 # option) any later version.  See http://www.gnu.org/copyleft/gpl.html for
00011 # the full text of the license.
00012 
00013 import unittest, tempfile, subprocess, os, stat, shutil, atexit
00014 
00015 temp_report_dir = tempfile.mkdtemp()
00016 os.environ['APPORT_REPORT_DIR'] = temp_report_dir
00017 atexit.register(shutil.rmtree, temp_report_dir)
00018 
00019 import apport.fileutils, problem_report
00020 
00021 
00022 class T(unittest.TestCase):
00023     def tearDown(self):
00024         for f in apport.fileutils.get_all_reports():
00025             os.unlink(f)
00026 
00027     def _test_crash(self, extracode='', scriptname=None):
00028         '''Create a test crash.'''
00029 
00030         # put the script into /var/tmp, since that isn't ignored in the
00031         # hook
00032         if scriptname:
00033             script = scriptname
00034             fd = os.open(scriptname, os.O_CREAT | os.O_WRONLY)
00035         else:
00036             (fd, script) = tempfile.mkstemp(dir='/var/tmp')
00037         try:
00038             os.write(fd, ('''#!/usr/bin/env %s
00039 import apport_python_hook
00040 apport_python_hook.install()
00041 
00042 def func(x):
00043     raise Exception(b'This should happen. \\xe2\\x99\\xa5'.decode('UTF-8'))
00044 
00045 %s
00046 func(42)
00047 ''' % (os.getenv('PYTHON', 'python3'), extracode)).encode())
00048             os.close(fd)
00049             os.chmod(script, 0o755)
00050 
00051             p = subprocess.Popen([script, 'testarg1', 'testarg2'],
00052                                  stdout=subprocess.PIPE,
00053                                  stderr=subprocess.PIPE, env=os.environ)
00054             err = p.communicate()[1].decode()
00055             self.assertEqual(p.returncode, 1,
00056                              'crashing test python program exits with failure code')
00057             self.assertTrue('This should happen.' in err, err)
00058             self.assertFalse('OSError' in err, err)
00059         finally:
00060             os.unlink(script)
00061 
00062         return script
00063 
00064     def test_general(self):
00065         '''general operation of the Python crash hook.'''
00066 
00067         script = self._test_crash()
00068 
00069         # did we get a report?
00070         reports = apport.fileutils.get_new_reports()
00071         pr = None
00072         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00073         self.assertEqual(stat.S_IMODE(os.stat(reports[0]).st_mode),
00074                          0o640, 'report has correct permissions')
00075 
00076         pr = problem_report.ProblemReport()
00077         with open(reports[0], 'rb') as f:
00078             pr.load(f)
00079 
00080         # check report contents
00081         expected_keys = ['InterpreterPath', 'PythonArgs', 'Traceback',
00082                          'ProblemType', 'ProcEnviron', 'ProcStatus',
00083                          'ProcCmdline', 'Date', 'ExecutablePath', 'ProcMaps',
00084                          'UserGroups']
00085         self.assertTrue(set(expected_keys).issubset(set(pr.keys())),
00086                         'report has necessary fields')
00087         self.assertTrue('bin/python' in pr['InterpreterPath'])
00088         self.assertEqual(pr['ExecutablePath'], script)
00089         self.assertEqual(pr['PythonArgs'], "['%s', 'testarg1', 'testarg2']" % script)
00090         self.assertTrue(pr['Traceback'].startswith('Traceback'))
00091         self.assertTrue("func\n    raise Exception(b'This should happen." in
00092                         pr['Traceback'], pr['Traceback'])
00093 
00094     def test_existing(self):
00095         '''Python crash hook overwrites seen existing files.'''
00096 
00097         script = self._test_crash()
00098 
00099         # did we get a report?
00100         reports = apport.fileutils.get_new_reports()
00101         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00102         self.assertEqual(stat.S_IMODE(os.stat(reports[0]).st_mode),
00103                          0o640, 'report has correct permissions')
00104 
00105         # touch report -> "seen" case
00106         apport.fileutils.mark_report_seen(reports[0])
00107 
00108         reports = apport.fileutils.get_new_reports()
00109         self.assertEqual(len(reports), 0)
00110 
00111         script = self._test_crash(scriptname=script)
00112         reports = apport.fileutils.get_new_reports()
00113         self.assertEqual(len(reports), 1)
00114 
00115         # "unseen" case
00116         script = self._test_crash(scriptname=script)
00117         reports = apport.fileutils.get_new_reports()
00118         self.assertEqual(len(reports), 1)
00119 
00120     def test_no_argv(self):
00121         '''with zapped sys.argv.'''
00122 
00123         self._test_crash('import sys\nsys.argv = None')
00124 
00125         # did we get a report?
00126         reports = apport.fileutils.get_new_reports()
00127         pr = None
00128         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00129         self.assertEqual(stat.S_IMODE(os.stat(reports[0]).st_mode),
00130                          0o640, 'report has correct permissions')
00131 
00132         pr = problem_report.ProblemReport()
00133         with open(reports[0], 'rb') as f:
00134             pr.load(f)
00135 
00136         # check report contents
00137         expected_keys = ['InterpreterPath', 'Traceback', 'ProblemType',
00138                          'ProcEnviron', 'ProcStatus', 'ProcCmdline', 'Date',
00139                          'ExecutablePath', 'ProcMaps', 'UserGroups']
00140         self.assertTrue(set(expected_keys).issubset(set(pr.keys())),
00141                         'report has necessary fields')
00142         self.assertTrue('bin/python' in pr['InterpreterPath'])
00143         self.assertTrue(pr['Traceback'].startswith('Traceback'))
00144 
00145     def _assert_no_reports(self):
00146         '''Assert that there are no crash reports.'''
00147 
00148         reports = apport.fileutils.get_new_reports()
00149         self.assertEqual(len(reports), 0,
00150                          'no crash reports present (cwd: %s)' % os.getcwd())
00151 
00152     def test_interactive(self):
00153         '''interactive Python sessions never generate a report.'''
00154 
00155         orig_cwd = os.getcwd()
00156         try:
00157             for d in ('/tmp', '/usr/local', '/usr'):
00158                 os.chdir(d)
00159                 p = subprocess.Popen(['python'], stdin=subprocess.PIPE,
00160                                      stdout=subprocess.PIPE, stderr=subprocess.PIPE)
00161                 (out, err) = p.communicate(b'raise ValueError')
00162                 out = out.decode()
00163                 err = err.decode()
00164                 assert p.returncode != 0
00165                 assert out == ''
00166                 assert 'ValueError' in err
00167                 self._assert_no_reports()
00168         finally:
00169             os.chdir(orig_cwd)
00170 
00171     def test_ignoring(self):
00172         '''the Python crash hook respects the ignore list.'''
00173 
00174         # put the script into /var/crash, since that isn't ignored in the
00175         # hook
00176         (fd, script) = tempfile.mkstemp(dir=apport.fileutils.report_dir)
00177         ifpath = os.path.expanduser(apport.report._ignore_file)
00178         orig_ignore_file = None
00179         try:
00180             os.write(fd, ('''#!/usr/bin/env %s
00181 import apport_python_hook
00182 apport_python_hook.install()
00183 
00184 def func(x):
00185     raise Exception('This should happen.')
00186 
00187 func(42)
00188 ''' % os.getenv('PYTHON', 'python3')).encode('ascii'))
00189             os.close(fd)
00190             os.chmod(script, 0o755)
00191 
00192             # move aside current ignore file
00193             if os.path.exists(ifpath):
00194                 orig_ignore_file = ifpath + '.apporttest'
00195                 os.rename(ifpath, orig_ignore_file)
00196 
00197             # ignore
00198             r = apport.report.Report()
00199             r['ExecutablePath'] = script
00200             r.mark_ignore()
00201             r = None
00202 
00203             p = subprocess.Popen([script, 'testarg1', 'testarg2'],
00204                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE)
00205             err = p.communicate()[1].decode()
00206             self.assertEqual(p.returncode, 1,
00207                              'crashing test python program exits with failure code')
00208             self.assertTrue('Exception: This should happen.' in err, err)
00209 
00210         finally:
00211             os.unlink(script)
00212             # clean up our ignore file
00213             if os.path.exists(ifpath):
00214                 os.unlink(ifpath)
00215             if orig_ignore_file:
00216                 os.rename(orig_ignore_file, ifpath)
00217 
00218         # did we get a report?
00219         reports = apport.fileutils.get_new_reports()
00220         self.assertEqual(len(reports), 0)
00221 
00222     def test_no_flooding(self):
00223         '''limit successive reports'''
00224 
00225         count = 0
00226         limit = 5
00227         while count < limit:
00228             self._test_crash(scriptname='/var/tmp/pytestcrash')
00229             reports = apport.fileutils.get_new_reports()
00230             if not reports:
00231                 break
00232             self.assertEqual(len(reports), 1, 'crashed Python program produced one report')
00233             apport.fileutils.mark_report_seen(reports[0])
00234             count += 1
00235 
00236         self.assertGreater(count, 1)
00237         self.assertLess(count, limit)
00238 
00239 unittest.main()