Back to index

apport  2.4
Public Member Functions | Private Member Functions
test_python_crashes.T Class Reference

List of all members.

Public Member Functions

def tearDown
def test_general
def test_existing
def test_no_argv
def test_interactive
def test_ignoring
def test_no_flooding
def test_dbus_service_unknown_invalid
def test_dbus_service_unknown_wrongbus_notrunning
def test_dbus_service_unknown_wrongbus_running
def test_dbus_service_timeout_running

Private Member Functions

def _test_crash
def _assert_no_reports
def _load_report

Detailed Description

Definition at line 23 of file test_python_crashes.py.


Member Function Documentation

def test_python_crashes.T._assert_no_reports (   self) [private]
Assert that there are no crash reports.

Definition at line 146 of file test_python_crashes.py.

00146 
00147     def _assert_no_reports(self):
00148         '''Assert that there are no crash reports.'''
00149 
00150         reports = apport.fileutils.get_new_reports()
00151         self.assertEqual(len(reports), 0,
00152                          'no crash reports present (cwd: %s)' % os.getcwd())

Here is the call graph for this function:

Here is the caller graph for this function:

def test_python_crashes.T._load_report (   self) [private]
Ensure that there is exactly one crash report and load it

Definition at line 351 of file test_python_crashes.py.

00351 
00352     def _load_report(self):
00353         '''Ensure that there is exactly one crash report and load it'''
00354 
00355         reports = apport.fileutils.get_new_reports()
00356         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00357         pr = problem_report.ProblemReport()
00358         with open(reports[0], 'rb') as f:
00359             pr.load(f)
00360         return pr
00361 
00362 unittest.main()

Here is the call graph for this function:

Here is the caller graph for this function:

def test_python_crashes.T._test_crash (   self,
  extracode = '',
  scriptname = None 
) [private]
Create a test crash.

Definition at line 28 of file test_python_crashes.py.

00028 
00029     def _test_crash(self, extracode='', scriptname=None):
00030         '''Create a test crash.'''
00031 
00032         # put the script into /var/tmp, since that isn't ignored in the
00033         # hook
00034         if scriptname:
00035             script = scriptname
00036             fd = os.open(scriptname, os.O_CREAT | os.O_WRONLY)
00037         else:
00038             (fd, script) = tempfile.mkstemp(dir='/var/tmp')
00039         try:
00040             os.write(fd, ('''#!/usr/bin/env %s
00041 import apport_python_hook
00042 apport_python_hook.install()
00043 
00044 def func(x):
00045     raise Exception(b'This should happen. \\xe2\\x99\\xa5'.decode('UTF-8'))
00046 
00047 %s
00048 func(42)
00049 ''' % (os.getenv('PYTHON', 'python3'), extracode)).encode())
00050             os.close(fd)
00051             os.chmod(script, 0o755)
00052 
00053             p = subprocess.Popen([script, 'testarg1', 'testarg2'],
00054                                  stderr=subprocess.PIPE, env=os.environ)
00055             err = p.communicate()[1].decode()
00056             self.assertEqual(p.returncode, 1,
00057                              'crashing test python program exits with failure code')
00058             if not extracode:
00059                 self.assertTrue('This should happen.' in err, err)
00060             self.assertFalse('OSError' in err, err)
00061         finally:
00062             os.unlink(script)
00063 
00064         return script

Here is the caller graph for this function:

Definition at line 24 of file test_python_crashes.py.

00024 
00025     def tearDown(self):
00026         for f in apport.fileutils.get_all_reports():
00027             os.unlink(f)

Here is the call graph for this function:

DBus.Error.NoReply with a running service

Definition at line 282 of file test_python_crashes.py.

00282 
00283     def test_dbus_service_timeout_running(self):
00284         '''DBus.Error.NoReply with a running service'''
00285 
00286         # ensure the service is running
00287         metadata_obj = dbus.SessionBus().get_object('org.gtk.vfs.Metadata', '/org/gtk/vfs/metadata')
00288         self.assertNotEqual(metadata_obj, None)
00289 
00290         # timeout of zero will always fail with NoReply
00291         try:
00292             subprocess.call(['killall', '-STOP', 'gvfsd-metadata'])
00293             self._test_crash(extracode='''import dbus
00294 obj = dbus.SessionBus().get_object('org.gtk.vfs.Metadata', '/org/gtk/vfs/metadata')
00295 assert obj
00296 i = dbus.Interface(obj, 'org.freedesktop.DBus.Peer')
00297 i.Ping(timeout=1)
00298 ''')
00299         finally:
00300             subprocess.call(['killall', '-CONT', 'gvfsd-metadata'])
00301 
00302         # check report contents
00303         reports = apport.fileutils.get_new_reports()
00304         self.assertEqual(len(reports), 0, 'NoReply is an useless exception and should not create a report')
00305 
00306         # This is disabled for now as we cannot get the bus name from the NoReply exception
00307         #pr = self._load_report()
00308         #self.assertTrue('org.freedesktop.DBus.Error.NoReply' in pr['Traceback'], pr['Traceback'])
00309         #self.assertTrue(pr['DbusErrorAnalysis'].startswith('provided by /usr/share/dbus-1/services/gvfs-metadata.service'),
00310         #                pr['DbusErrorAnalysis'])
00311         #self.assertTrue('gvfsd-metadata is running' in pr['DbusErrorAnalysis'], pr['DbusErrorAnalysis'])
00312 
00313 # This is disabled for now as we cannot get the bus name from the NoReply exception
00314 #    def test_dbus_service_timeout_notrunning(self):
00315 #        '''DBus.Error.NoReply with a crashing method'''
00316 #
00317 #        # run our own mock service with a crashing method
00318 #        subprocess.call(['killall', 'gvfsd-metadata'])
00319 #        service = subprocess.Popen([os.getenv('PYTHON', 'python3')],
00320 #                                   stdin=subprocess.PIPE,
00321 #                                   universal_newlines=True)
00322 #        service.stdin.write('''import os
00323 #import dbus, dbus.service, dbus.mainloop.glib
00324 #from gi.repository import GLib
00325 #
00326 #class MockMetadata(dbus.service.Object):
00327 #    @dbus.service.method('com.ubuntu.Test', in_signature='', out_signature='i')
00328 #    def Crash(self):
00329 #        os.kill(os.getpid(), 5)
00330 #
00331 #dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
00332 #dbus_name = dbus.service.BusName('org.gtk.vfs.Metadata', dbus.SessionBus())
00333 #svr = MockMetadata(bus_name=dbus_name, object_path='/org/gtk/vfs/metadata')
00334 #GLib.MainLoop().run()
00335 #''')
00336 #        service.stdin.close()
00337 #        self.addCleanup(service.terminate)
00338 #        time.sleep(0.5)
00339 #
00340 #        self._test_crash(extracode='''import dbus
00341 #obj = dbus.SessionBus().get_object('org.gtk.vfs.Metadata', '/org/gtk/vfs/metadata')
00342 #assert obj
00343 #dbus.Interface(obj, 'com.ubuntu.Test').Crash()
00344 #''')
00345 #
00346 #        pr = self._load_report()
00347 #        self.assertTrue('org.freedesktop.DBus.Error.NoReply' in pr['Traceback'], pr['Traceback'])
00348 #        self.assertTrue(pr['DbusErrorAnalysis'].startswith('provided by /usr/share/dbus-1/services/gvfs-metadata.service'),
00349 #                        pr['DbusErrorAnalysis'])
00350 #        self.assertTrue('gvfsd-metadata is not running' in pr['DbusErrorAnalysis'], pr['DbusErrorAnalysis'])

Here is the call graph for this function:

DBus.Error.ServiceUnknown with an invalid name

Definition at line 240 of file test_python_crashes.py.

00240 
00241     def test_dbus_service_unknown_invalid(self):
00242         '''DBus.Error.ServiceUnknown with an invalid name'''
00243 
00244         self._test_crash(extracode='''import dbus
00245 obj = dbus.SessionBus().get_object('com.example.NotExisting', '/Foo')
00246 ''')
00247 
00248         pr = self._load_report()
00249         self.assertTrue(pr['Traceback'].startswith('Traceback'), pr['Traceback'])
00250         self.assertTrue('org.freedesktop.DBus.Error.ServiceUnknown' in pr['Traceback'], pr['Traceback'])
00251         self.assertEqual(pr['DbusErrorAnalysis'], 'no service file providing com.example.NotExisting')

Here is the call graph for this function:

DBus.Error.ServiceUnknown with a valid name on a different bus (not running)

Definition at line 252 of file test_python_crashes.py.

00252 
00253     def test_dbus_service_unknown_wrongbus_notrunning(self):
00254         '''DBus.Error.ServiceUnknown with a valid name on a different bus (not running)'''
00255 
00256         subprocess.call(['killall', 'gvfsd-metadata'])
00257         self._test_crash(extracode='''import dbus
00258 obj = dbus.SystemBus().get_object('org.gtk.vfs.Metadata', '/org/gtk/vfs/metadata')
00259 ''')
00260 
00261         pr = self._load_report()
00262         self.assertTrue('org.freedesktop.DBus.Error.ServiceUnknown' in pr['Traceback'], pr['Traceback'])
00263         self.assertTrue(pr['DbusErrorAnalysis'].startswith('provided by /usr/share/dbus-1/services/gvfs-metadata.service'),
00264                         pr['DbusErrorAnalysis'])
00265         self.assertTrue('gvfsd-metadata is not running' in pr['DbusErrorAnalysis'], pr['DbusErrorAnalysis'])

Here is the call graph for this function:

DBus.Error.ServiceUnknown with a valid name on a different bus (running)

Definition at line 266 of file test_python_crashes.py.

00266 
00267     def test_dbus_service_unknown_wrongbus_running(self):
00268         '''DBus.Error.ServiceUnknown with a valid name on a different bus (running)'''
00269 
00270         self._test_crash(extracode='''import dbus
00271 # let the service be activated, to ensure it is running
00272 obj = dbus.SessionBus().get_object('org.gtk.vfs.Metadata', '/org/gtk/vfs/metadata')
00273 assert obj
00274 obj = dbus.SystemBus().get_object('org.gtk.vfs.Metadata', '/org/gtk/vfs/metadata')
00275 ''')
00276 
00277         pr = self._load_report()
00278         self.assertTrue('org.freedesktop.DBus.Error.ServiceUnknown' in pr['Traceback'], pr['Traceback'])
00279         self.assertTrue(pr['DbusErrorAnalysis'].startswith('provided by /usr/share/dbus-1/services/gvfs-metadata.service'),
00280                         pr['DbusErrorAnalysis'])
00281         self.assertTrue('gvfsd-metadata is running' in pr['DbusErrorAnalysis'], pr['DbusErrorAnalysis'])

Here is the call graph for this function:

Python crash hook overwrites seen existing files.

Definition at line 95 of file test_python_crashes.py.

00095 
00096     def test_existing(self):
00097         '''Python crash hook overwrites seen existing files.'''
00098 
00099         script = self._test_crash()
00100 
00101         # did we get a report?
00102         reports = apport.fileutils.get_new_reports()
00103         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00104         self.assertEqual(stat.S_IMODE(os.stat(reports[0]).st_mode),
00105                          0o640, 'report has correct permissions')
00106 
00107         # touch report -> "seen" case
00108         apport.fileutils.mark_report_seen(reports[0])
00109 
00110         reports = apport.fileutils.get_new_reports()
00111         self.assertEqual(len(reports), 0)
00112 
00113         script = self._test_crash(scriptname=script)
00114         reports = apport.fileutils.get_new_reports()
00115         self.assertEqual(len(reports), 1)
00116 
00117         # "unseen" case
00118         script = self._test_crash(scriptname=script)
00119         reports = apport.fileutils.get_new_reports()
00120         self.assertEqual(len(reports), 1)

Here is the call graph for this function:

general operation of the Python crash hook.

Definition at line 65 of file test_python_crashes.py.

00065 
00066     def test_general(self):
00067         '''general operation of the Python crash hook.'''
00068 
00069         script = self._test_crash()
00070 
00071         # did we get a report?
00072         reports = apport.fileutils.get_new_reports()
00073         pr = None
00074         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00075         self.assertEqual(stat.S_IMODE(os.stat(reports[0]).st_mode),
00076                          0o640, 'report has correct permissions')
00077 
00078         pr = problem_report.ProblemReport()
00079         with open(reports[0], 'rb') as f:
00080             pr.load(f)
00081 
00082         # check report contents
00083         expected_keys = ['InterpreterPath', 'PythonArgs', 'Traceback',
00084                          'ProblemType', 'ProcEnviron', 'ProcStatus',
00085                          'ProcCmdline', 'Date', 'ExecutablePath', 'ProcMaps',
00086                          'UserGroups']
00087         self.assertTrue(set(expected_keys).issubset(set(pr.keys())),
00088                         'report has necessary fields')
00089         self.assertTrue('bin/python' in pr['InterpreterPath'])
00090         self.assertEqual(pr['ExecutablePath'], script)
00091         self.assertEqual(pr['PythonArgs'], "['%s', 'testarg1', 'testarg2']" % script)
00092         self.assertTrue(pr['Traceback'].startswith('Traceback'))
00093         self.assertTrue("func\n    raise Exception(b'This should happen." in
00094                         pr['Traceback'], pr['Traceback'])

Here is the call graph for this function:

the Python crash hook respects the ignore list.

Definition at line 172 of file test_python_crashes.py.

00172 
00173     def test_ignoring(self):
00174         '''the Python crash hook respects the ignore list.'''
00175 
00176         # put the script into /var/crash, since that isn't ignored in the
00177         # hook
00178         (fd, script) = tempfile.mkstemp(dir=apport.fileutils.report_dir)
00179         ifpath = os.path.expanduser(apport.report._ignore_file)
00180         orig_ignore_file = None
00181         try:
00182             os.write(fd, ('''#!/usr/bin/env %s
00183 import apport_python_hook
00184 apport_python_hook.install()
00185 
00186 def func(x):
00187     raise Exception('This should happen.')
00188 
00189 func(42)
00190 ''' % os.getenv('PYTHON', 'python3')).encode('ascii'))
00191             os.close(fd)
00192             os.chmod(script, 0o755)
00193 
00194             # move aside current ignore file
00195             if os.path.exists(ifpath):
00196                 orig_ignore_file = ifpath + '.apporttest'
00197                 os.rename(ifpath, orig_ignore_file)
00198 
00199             # ignore
00200             r = apport.report.Report()
00201             r['ExecutablePath'] = script
00202             r.mark_ignore()
00203             r = None
00204 
00205             p = subprocess.Popen([script, 'testarg1', 'testarg2'],
00206                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE)
00207             err = p.communicate()[1].decode()
00208             self.assertEqual(p.returncode, 1,
00209                              'crashing test python program exits with failure code')
00210             self.assertTrue('Exception: This should happen.' in err, err)
00211 
00212         finally:
00213             os.unlink(script)
00214             # clean up our ignore file
00215             if os.path.exists(ifpath):
00216                 os.unlink(ifpath)
00217             if orig_ignore_file:
00218                 os.rename(orig_ignore_file, ifpath)
00219 
00220         # did we get a report?
00221         reports = apport.fileutils.get_new_reports()
00222         self.assertEqual(len(reports), 0)

Here is the call graph for this function:

interactive Python sessions never generate a report.

Definition at line 153 of file test_python_crashes.py.

00153 
00154     def test_interactive(self):
00155         '''interactive Python sessions never generate a report.'''
00156 
00157         orig_cwd = os.getcwd()
00158         try:
00159             for d in ('/tmp', '/usr/local', '/usr'):
00160                 os.chdir(d)
00161                 p = subprocess.Popen(['python'], stdin=subprocess.PIPE,
00162                                      stdout=subprocess.PIPE, stderr=subprocess.PIPE)
00163                 (out, err) = p.communicate(b'raise ValueError')
00164                 out = out.decode()
00165                 err = err.decode()
00166                 assert p.returncode != 0
00167                 assert out == ''
00168                 assert 'ValueError' in err
00169                 self._assert_no_reports()
00170         finally:
00171             os.chdir(orig_cwd)

Here is the call graph for this function:

with zapped sys.argv.

Definition at line 121 of file test_python_crashes.py.

00121 
00122     def test_no_argv(self):
00123         '''with zapped sys.argv.'''
00124 
00125         self._test_crash('import sys\nsys.argv = None')
00126 
00127         # did we get a report?
00128         reports = apport.fileutils.get_new_reports()
00129         pr = None
00130         self.assertEqual(len(reports), 1, 'crashed Python program produced a report')
00131         self.assertEqual(stat.S_IMODE(os.stat(reports[0]).st_mode),
00132                          0o640, 'report has correct permissions')
00133 
00134         pr = problem_report.ProblemReport()
00135         with open(reports[0], 'rb') as f:
00136             pr.load(f)
00137 
00138         # check report contents
00139         expected_keys = ['InterpreterPath', 'Traceback', 'ProblemType',
00140                          'ProcEnviron', 'ProcStatus', 'ProcCmdline', 'Date',
00141                          'ExecutablePath', 'ProcMaps', 'UserGroups']
00142         self.assertTrue(set(expected_keys).issubset(set(pr.keys())),
00143                         'report has necessary fields')
00144         self.assertTrue('bin/python' in pr['InterpreterPath'])
00145         self.assertTrue(pr['Traceback'].startswith('Traceback'))

Here is the call graph for this function:

limit successive reports

Definition at line 223 of file test_python_crashes.py.

00223 
00224     def test_no_flooding(self):
00225         '''limit successive reports'''
00226 
00227         count = 0
00228         limit = 5
00229         while count < limit:
00230             self._test_crash(scriptname='/var/tmp/pytestcrash')
00231             reports = apport.fileutils.get_new_reports()
00232             if not reports:
00233                 break
00234             self.assertEqual(len(reports), 1, 'crashed Python program produced one report')
00235             apport.fileutils.mark_report_seen(reports[0])
00236             count += 1
00237 
00238         self.assertGreater(count, 1)
00239         self.assertLess(count, limit)

Here is the call graph for this function:


The documentation for this class was generated from the following file: