#!/usr/bin/python
'''Test crash-digger.'''

# Copyright (C) 2007 - 2009 Canonical Ltd.
# Author: Martin Pitt <martin.pitt@ubuntu.com>
# 
# 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, time, signal

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')
        open(crashdb_conf, 'w').write('''default = 'memory'
databases = { 'memory': { 
    'impl': 'memory', 'distro': 'Testux', 'dummy_data': '1' }
}''')

        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')
        open(self.apport_retrace, 'w').write('''#!/bin/sh
echo "$@" >> %s''' % self.apport_retrace_log)
        os.chmod(self.apport_retrace, 0755)

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

        os.environ['APPORT_CRASHDB_CONF'] = crashdb_conf
        os.environ['PYTHONPATH'] = '.'
        os.environ['PATH'] = '%s:%s' % ('./bin', os.environ.get('PATH', ''))

    def tearDown(self):
        shutil.rmtree(self.workdir)

    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, err)

    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')

        retrace_log = open(self.apport_retrace_log).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))

    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')
        open(self.apport_retrace, 'w').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, 0755)

        (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')

        retrace_log = open(self.apport_retrace_log).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')
        open(self.apport_retrace, 'w').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, 0755)

        (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)

        retrace_log = open(self.apport_retrace_log).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'''

        open(self.apport_retrace, 'w').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)

tl = unittest.TestLoader()
tests_all = unittest.TestSuite((
    tl.loadTestsFromName('__main__')
))
unittest.TextTestRunner(verbosity=2).run(tests_all)