'''Test crash-digger'''

# Copyright (C) 2007 - 2009 Canonical Ltd.
# Author: Martin Pitt <[email protected]>
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version. See http://www.gnu.org/copyleft/gpl.html for
# the full text of the license.

import unittest, subprocess, tempfile, os, shutil, os.path

import apport.fileutils


class T(unittest.TestCase):
def setUp(self):
'''Set up dummy config dir, crashdb.conf, and apport-retrace'''

self.workdir = tempfile.mkdtemp()

crashdb_conf = os.path.join(self.workdir, 'crashdb.conf')
with open(crashdb_conf, 'w') as f:
f.write('''default = 'memory'
databases = {
'memory': {'impl': 'memory', 'distro': 'Testux', 'dummy_data': '1',
'dupdb_url': '%s'},
'empty': {'impl': 'memory', 'distro': 'Foonux'},
}''' % os.path.join(self.workdir, 'dupdb'))

self.config_dir = os.path.join(self.workdir, 'config')
os.mkdir(self.config_dir)
os.mkdir(os.path.join(self.config_dir, 'Testux 1.0'))
os.mkdir(os.path.join(self.config_dir, 'Testux 2.2'))

self.apport_retrace_log = os.path.join(self.workdir, 'apport-retrace.log')

self.apport_retrace = os.path.join(self.workdir, 'apport-retrace')
with open(self.apport_retrace, 'w') as f:
f.write('''#!/bin/sh
echo "$@" >> %s''' % self.apport_retrace_log)
os.chmod(self.apport_retrace, 0o755)

self.lock_file = os.path.join(self.workdir, 'lock')

os.environ['APPORT_CRASHDB_CONF'] = crashdb_conf
os.environ['PYTHONPATH'] = '.'

self.orig_report_dir = apport.fileutils.report_dir
apport.fileutils.report_dir = os.path.join(self.workdir, 'crashes')
os.mkdir(apport.fileutils.report_dir)
os.environ['APPORT_REPORT_DIR'] = apport.fileutils.report_dir

def tearDown(self):
shutil.rmtree(self.workdir)
apport.fileutils.report_dir = self.orig_report_dir

def call(self, args):
'''Call crash-digger with given arguments.

Return a pair (stdout, stderr).
'''
s = subprocess.Popen(['crash-digger', '--apport-retrace',
self.apport_retrace] + args, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(out, err) = s.communicate()
return (out.decode('UTF-8'), err.decode('UTF-8'))

def test_crashes(self):
'''Crash retracing'''

(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file])
self.assertEqual(err, '', 'no error messages:\n' + err)
self.assertTrue("Available releases: ['Testux 1.0', 'Testux 2.2']" in out)
self.assertTrue('retracing #0' in out)
self.assertTrue('retracing #1' in out)
self.assertTrue('retracing #2' in out)
self.assertTrue('crash is release FooLinux Pi/2 which does not have a config available' in out)
self.assertFalse('failed with status' in out)
self.assertFalse('#3' in out, 'dupcheck crashes are not retraced')
self.assertFalse('#4' in out, 'dupcheck crashes are not retraced')

with open(self.apport_retrace_log) as f:
retrace_log = f.read()
self.assertEqual(len(retrace_log.splitlines()), 2)
self.assertFalse('dup.db -v 0\n' in retrace_log)
self.assertTrue('dup.db -v 1\n' in retrace_log)
self.assertTrue('dup.db -v 2\n' in retrace_log)
self.assertFalse(os.path.exists(self.lock_file))

self.assertFalse(os.path.isdir(os.path.join(self.workdir, 'dupdb', 'sig')))

def test_crashes_error(self):
'''Crash retracing if apport-retrace fails on bug #1'''

# make apport-retrace fail on bug 1
os.rename(self.apport_retrace, self.apport_retrace + '.bak')
with open(self.apport_retrace, 'w') as f:
f.write('''#!/bin/sh
echo "$@" >> %s
while [ -n "$2" ]; do shift; done
if [ "$1" = 1 ]; then
echo "cannot frobnicate bug" >&2
exit 1
fi
''' % self.apport_retrace_log)
os.chmod(self.apport_retrace, 0o755)

(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file])
self.assertTrue('Traceback' in err)
self.assertTrue('SystemError: retracing #1 failed' in err)
self.assertTrue("Available releases: ['Testux 1.0', 'Testux 2.2']" in out)
self.assertTrue('retracing #0' in out)
self.assertTrue('retracing #1' in out)
self.assertFalse('retracing #2' in out, 'should not continue after errors')
self.assertTrue('crash is release FooLinux Pi/2 which does not have a config available' in out)
self.assertFalse('#0 failed with status' in out)
self.assertTrue('#1 failed with status: 1' in out)
self.assertFalse('#3' in out, 'dupcheck crashes are not retraced')
self.assertFalse('#4' in out, 'dupcheck crashes are not retraced')

with open(self.apport_retrace_log) as f:
retrace_log = f.read()
self.assertEqual(len(retrace_log.splitlines()), 1)
self.assertFalse('dup.db -v 0\n' in retrace_log)
self.assertTrue('dup.db -v 1\n' in retrace_log)
# stops after failing #1
self.assertFalse('dup.db -v 2\n' in retrace_log)
self.assertTrue(os.path.exists(self.lock_file))

os.rename(self.apport_retrace + '.bak', self.apport_retrace)

# subsequent start should not do anything until the lock file is cleaned up
(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file])
self.assertEqual(out, '')
self.assertEqual(err, '')

os.unlink(self.lock_file)

# now it should run again
(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file])
self.assertTrue('retracing #2' in out)
self.assertEqual(err, '', 'no error messages:\n' + err)
self.assertFalse(os.path.exists(self.lock_file))

def test_crashes_transient_error(self):
'''Crash retracing if apport-retrace reports a transient error'''

# make apport-retrace fail on bug 1
os.rename(self.apport_retrace, self.apport_retrace + '.bak')
with open(self.apport_retrace, 'w') as f:
f.write('''#!/bin/sh
echo "$@" >> %s
while [ -n "$2" ]; do shift; done
if [ "$1" = 1 ]; then
echo "cannot frobnicate crash db" >&2
exit 99
fi
''' % self.apport_retrace_log)
os.chmod(self.apport_retrace, 0o755)

(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file])
self.assertTrue("Available releases: ['Testux 1.0', 'Testux 2.2']" in out)
self.assertTrue('retracing #0' in out)
self.assertTrue('retracing #1' in out)
self.assertFalse('retracing #2' in out, 'should not continue after errors')
self.assertTrue('transient error reported; halting' in out)

with open(self.apport_retrace_log) as f:
retrace_log = f.read()
self.assertTrue('dup.db -v 1\n' in retrace_log)
# stops after failing #1
self.assertFalse('dup.db -v 2\n' in retrace_log)

self.assertFalse(os.path.exists(self.lock_file))

def test_dupcheck(self):
'''Duplicate checking'''

(out, err) = self.call(['-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vDl', self.lock_file])
self.assertEqual(err, '', 'no error messages:\n' + err)
self.assertFalse('#1' in out, 'signal crashes are not retraced')
self.assertFalse('#2' in out, 'signal crashes are not retraced')
self.assertTrue('checking #3 for duplicate' in out)
self.assertTrue('checking #4 for duplicate' in out)
self.assertTrue('Report is a duplicate of #3 (not fixed yet)' in out)
self.assertFalse(os.path.exists(self.apport_retrace_log))
self.assertFalse(os.path.exists(self.lock_file))

def test_stderr_redirection(self):
'''apport-retrace's stderr is redirected to stdout'''

with open(self.apport_retrace, 'w') as f:
f.write('''#!/bin/sh
echo ApportRetraceError >&2''')
(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file])
self.assertEqual(err, '', 'no error messages:\n' + err)
self.assertTrue('ApportRetraceError' in out)

def test_publish_db(self):
'''Duplicate database publishing'''

(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero', '-d',
os.path.join(self.workdir, 'dup.db'), '-vl', self.lock_file,
'--publish-db', os.path.join(self.workdir, 'dupdb')])
self.assertEqual(err, '', 'no error messages:\n' + err)
self.assertTrue('retracing #0' in out)

self.assertTrue(os.path.isdir(os.path.join(self.workdir, 'dupdb', 'sig')))

def test_alternate_crashdb(self):
'''Alternate crash database name'''

# existing DB "empty" has no crashes
(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero',
'-vl', self.lock_file, '--crash-db', 'empty'])
self.assertEqual(err, '', 'no error messages:\n' + err)
self.assertFalse('retracing #' in out)
self.assertFalse('crash is' in out)
self.assertFalse('failed with status' in out)

# nonexisting DB
(out, err) = self.call(['-c', self.config_dir, '-a', '/dev/zero',
'-vl', self.lock_file, '--crash-db', 'nonexisting'])
self.assertEqual(out, '', 'no output messages:\n' + out)
self.assertFalse('Traceback' in err, err)
self.assertTrue('nonexisting' in err, err)

unittest.main()