#!/usr/bin/python3

# this testsuite is part of autopkgtest
# autopkgtest is a tool for testing Debian binary packages
#
# autopkgtest is Copyright (C) 2006-2013 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.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#
# See the file CREDITS for a full list of credits information (often
# installed as /usr/share/doc/autopkgtest/CREDITS).

import sys
import os
import re
import subprocess
import unittest
import tempfile
import shutil
import fnmatch
from glob import glob

test_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(test_dir)

# backwards compat shim for Python 3.1
if not hasattr(unittest.TestCase, 'assertRegex'):
    unittest.TestCase.assertRegex = unittest.TestCase.assertRegexpMatches


class AdtTestCase(unittest.TestCase):
    '''Base class for adt-run tests'''

    def __init__(self, virt, *args, **kwargs):
        super(AdtTestCase, self).__init__(*args, **kwargs)
        self.adt_run_path = os.path.join(root_dir, 'run-from-checkout')
        self.virt = 'adt-virt-' + virt

    def setUp(self):
        self.workdir = tempfile.mkdtemp(prefix='test.adt-run.')
        self.addCleanup(shutil.rmtree, self.workdir)
        self.cwd = os.getcwd()
        self.addCleanup(os.chdir, self.cwd)

        temp_home = os.path.join(self.workdir, 'home')
        shutil.copytree(os.path.join(test_dir, 'home'), temp_home)
        os.chmod(os.path.join(temp_home, '.ssh', 'id_rsa'), 0o600)
        os.environ['HOME'] = temp_home

    def build_src(self, test_control, test_scripts):
        '''Create source package tree with given tests.

         @test_control: contents of debian/tests/control
         @test_scripts: map of test name (in debian/tests/) to file contents

        Return path to the source tree.
        '''
        srcdir = os.path.join(self.workdir, 'testpkg')
        shutil.copytree(os.path.join(test_dir, 'testpkg'), srcdir, symlinks=True)
        if test_control:
            dtdir = os.path.join(srcdir, 'debian', 'tests')
            os.mkdir(dtdir)
            with open(os.path.join(dtdir, 'control'), 'w', encoding='UTF-8') as f:
                f.write(test_control)
            for name, contents in test_scripts.items():
                with open(os.path.join(dtdir, name), 'w', encoding='UTF-8') as f:
                    f.write(contents)

        return srcdir

    def build_dsc(self, test_control, test_scripts):
        '''Create source package dsc with given tests.

         @test_control: contents of debian/tests/control
         @test_scripts: map of test name (in debian/tests/) to file contents

        Return path to the dsc.
        '''
        srcdir = self.build_src(test_control, test_scripts)
        dbp = subprocess.Popen(['dpkg-buildpackage', '-S', '-us', '-uc'],
                               stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               cwd=srcdir)
        out, err = dbp.communicate()
        self.assertEqual(dbp.returncode, 0, err)
        return os.path.join(os.path.dirname(srcdir), 'testpkg_1.dsc')

    def adt_run(self, args, virt_args=[], env=None):
        '''Run adt-run with given arguments with configured virt runner.

         @args: command line args of adt-run, excluding "adt-run" itself; "---
                adt-virt-XXX" will be appended automatically (called from the
                source tree)

        Return a tuple (exit_code, stdout, stderr).
        '''
        # run adt command
        adt = subprocess.Popen([self.adt_run_path] + args +
                               ['---', self.virt] + virt_args,
                               stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               env=env)
        (out, err) = adt.communicate()
        return (adt.returncode, out.decode('UTF-8'), err.decode('UTF-8'))


class NullRunnerNoRoot(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(NullRunnerNoRoot, self).__init__('null', *args, **kwargs)

    def test_dev_stdouterr_access(self):
        '''write to /dev/stdout and /dev/stderr in a test'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr',
                           {'pass': '#!/bin/sh\necho I am fine >/dev/stdout\n'
                            'echo SomeDebug >/dev/stderr\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout/err
        self.assertIn('I am fine\n', out)
        # should show test stderr
        self.assertIn('\nSomeDebug\n', err)

    def test_summary(self):
        '''--summary option'''

        p = self.build_src('Tests: good bad\nDepends:\n',
                           {'good': '#!/bin/sh\necho happy\n',
                            'bad': '#!/bin/sh\nexit 1'})

        summary = os.path.join(self.workdir, 'summary.log')

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--summary=' + summary])

        # test results
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'good\s+PASS', out)
        self.assertRegex(out, 'bad\s+FAIL non-zero exit status 1', out)

        # check summary file
        with open(summary) as f:
            self.assertEqual(f.read(), '''good                 PASS
bad                  FAIL non-zero exit status 1
''')

    def test_tree_norestrictions_nobuild_fail_on_stderr_and_exit(self):
        '''source tree, no build, no restrictions, test fails with stderr+exit'''

        p = self.build_src('Tests: senz\nDepends:\n',
                           {'senz': '#!/bin/sh\necho I am sick >&2\nexit 7\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        # test should fail
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'senz\s+FAIL non-zero exit status 7')

        # should show test stderr, but inline
        self.assertRegex(err, 'stderr [ -]+\nI am sick', err)

    def test_tree_allow_stderr_nobuild_success(self):
        '''source tree, no build, allow-stderr, test success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\nRestrictions: allow-stderr',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout/err
        self.assertIn('babble\n', out)
        self.assertRegex(err, '-+\nI am fine\nadt-run \[[0-9: -]+\]: test pass: --')
        # but not complain about stderr
        self.assertNotIn(' stderr ', err)

        # should log package version
        self.assertIn('testing package testpkg version 1\n', err)

        # no restricted dependencies functionality as they are already installed
        self.assertNotIn('will only work for some packages', err)

    def test_unicode(self):
        '''Unicode test output'''

        p = self.build_src('Tests: se\nDepends:\n',
                           {'se': '#!/bin/sh\necho ‘a♩’; echo fancy ‴ʎɔuɐɟ″!>&2'})

        summary = os.path.join(self.workdir, 'summary.log')

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--summary=' + summary])
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'se\s+FAIL stderr: fancy ‴ʎɔuɐɟ″!\n')

        # should show test stdout/stderr
        self.assertIn('‘a♩’\n', out)
        self.assertRegex(err, 'stderr [ -]+\nfancy ‴ʎɔuɐɟ″!\n')

        with open(summary, encoding='UTF-8') as f:
            self.assertEqual(f.read(), 'se                   FAIL stderr: fancy ‴ʎɔuɐɟ″!\n')

    def test_isolation(self):
        '''isolation restrictions'''

        p = self.build_src('Tests: ic\nDepends:\nRestrictions: isolation-container\n\n'
                           'Tests: im\nDepends:\nRestrictions: isolation-machine\n',
                           {'ic': '#!/bin/sh\necho container ok',
                            'im': '#!/bin/sh\necho machine ok'})

        (code, out, err) = self.adt_run(['-B', '--built-tree=' + p])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'ic\s+PASS', out)
        self.assertRegex(out, 'im\s+PASS', out)
        self.assertIn('container ok\n', out)
        self.assertIn('machine ok\n', out)

    def test_no_tests_dir(self):
        '''package without debian/tests/'''

        p = self.build_src(None, None)
        (code, out, err) = self.adt_run(['--unbuilt-tree=' + p])
        self.assertEqual(code, 8, err)
        self.assertRegex(out, 'SKIP no tests in this package', out)

    @unittest.skipIf(os.getuid() == 0, 'needs to run as user')
    @unittest.skipIf(os.path.exists('/usr/bin/dotty'),
                     'needs graphviz uninstalled')
    def test_tmp_install(self):
        '''temp dir unpack of test dependencies'''

        p = self.build_src('Tests: t\nDepends: graphviz, gir1.2-json-1.0 (>= 0.14), python3-gi, a|b,',
                           {'t': '#!/bin/sh\ndotty -V 2>&1 || true\npython3 -c "from gi.repository import Json; print(Json)"'})

        (code, out, err) = self.adt_run(['-B', p + '//'])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't\s+PASS', out)

        # should show test stdout
        self.assertIn('dotty version ', out)
        try:
            from gi.repository import Json
            Json  # pyflakes
            # already installed on the system
            self.assertRegex(out, 'DynamicModule.*Json.* from .*/usr/lib/.*girepository')
        except ImportError:
            # should use from local unpack dir
            self.assertRegex(out, 'DynamicModule.*Json.* from .*/tmp/adt-run.*/deps/usr/lib')
        # no stderr
        self.assertNotIn(' stderr ', err)

        # downloads dependencies
        self.assertIn('libcgraph', err)
        self.assertIn('libcgraph', err)

        # warn about restricted functionality
        self.assertRegex(err, 'WARNING.*cannot be handled.* a | b')
        self.assertRegex(err, 'WARNING.*will only work for some packages')

    @unittest.skipIf(os.getuid() == 0, 'needs to run as user')
    def test_tmp_install_nonexisting_pkg(self):
        '''temp dir unpack of nonexisting test dependency'''

        p = self.build_src('Tests: t\nDepends: nosuchpackage',
                           {'t': '#!/bin/sh\nfalse'})

        (code, out, err) = self.adt_run(['-B', p + '//'])
        self.assertEqual(code, 20, err)

        self.assertRegex(err, 'E: .*nosuchpackage')
        self.assertIn('test dependencies are unsatisfiable', err)

    @unittest.skipIf(os.getuid() == 0, 'needs to run as user')
    @unittest.skipIf(os.path.exists('/usr/bin/dotty'),
                     'needs graphviz uninstalled')
    def test_tmp_install_no_such_version(self):
        '''temp dir unpack of test dependency with unsatisfiable version'''

        p = self.build_src('Tests: t\nDepends: graphviz (>= 4:999)',
                           {'t': '#!/bin/sh\nfalse'})

        (code, out, err) = self.adt_run(['-B', p + '//'])
        self.assertEqual(code, 20, err)

        self.assertIn('test dependency graphviz (>= 4:999) is unsatisfiable: available version ', err)

    def test_test_command(self):
        '''Test-Command: instead of Tests:'''

        p = self.build_src('Test-Command: echo "Some Stdout"\nDepends:\n\n'
                           'Test-Command: echo "Some StdErr" >&2; sleep 0.5; echo done\nDepends:\n'
                           'Restrictions: allow-stderr\n\n'
                           'Test-Command: echo "More Stderr" >&2; echo hello > $ADT_ARTIFACTS/world.txt\nDepends:\n',
                           {})

        outdir = os.path.join(self.workdir, 'out')
        os.mkdir(outdir)

        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p,
                                         '-o', outdir, '-d'])
        # two pass, one fails
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'done\n')
        self.assertRegex(out, 'command1\s+PASS')
        self.assertRegex(out, 'command2\s+PASS')
        self.assertRegex(out, 'command3\s+FAIL stderr: More Stderr')
        self.assertRegex(err, 'stderr [ -]+\nMore Stderr\n')
        # shows commands
        self.assertIn('test command1: echo "Some Stdout"\n', err)
        self.assertIn('test command2: echo "Some StdErr" >&2; sleep 0.5; echo done\n', err)
        self.assertIn('test command3: echo "More Stderr" >&2; echo hello > $ADT_ARTIFACTS/world.txt\n', err)

        # check artifacts
        with open(os.path.join(outdir, 'command1-stdout')) as f:
            self.assertEqual(f.read(), 'Some Stdout\n')
        with open(os.path.join(outdir, 'command2-stdout')) as f:
            self.assertEqual(f.read(), 'done\n')
        with open(os.path.join(outdir, 'command2-stderr')) as f:
            self.assertEqual(f.read(), 'Some StdErr\n')
        with open(os.path.join(outdir, 'command3-stderr')) as f:
            self.assertEqual(f.read(), 'More Stderr\n')
        with open(os.path.join(outdir, 'artifacts', 'world.txt')) as f:
            self.assertEqual(f.read(), 'hello\n')

    def test_apt_source_error(self):
        '''apt-source for nonexisting package'''

        (code, out, err) = self.adt_run(['no.such-package'])

        # test should succeed
        self.assertEqual(code, 12, err)
        self.assertRegex(err, 'no.such-package')
        self.assertNotIn('PASS', out)
        self.assertNotIn('[----', out)

    def test_undeletable_files(self):
        '''source tree has undeletable files'''

        p = self.build_src('Tests: t1 t2\nDepends:\n',
                           {'t1': '#!/bin/sh\necho eins\n',
                            't2': '#!/bin/sh\necho zwei\n'})
        os.mkdir(os.path.join(p, 'data'))
        with open(os.path.join(p, 'data', 'data.txt'), 'w') as f:
            f.write('data\n')
        os.chmod(os.path.join(p, 'data'), 0o555)

        (code, out, err) = self.adt_run(['--built-tree=' + p])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't1\s+PASS', out)
        self.assertRegex(out, 't2\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '^eins\n')
        self.assertRegex(out, '\nzwei\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # chown it back so that we can clean up
        os.chmod(os.path.join(p, 'data'), 0o755)


@unittest.skipIf(os.getuid() > 0,
                 'null runner tests need to run as root')
class NullRunner(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(NullRunner, self).__init__('null', *args, **kwargs)

    def test_tree_norestrictions_nobuild_success(self):
        '''source tree, no build, no restrictions, test success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\n',
                           {'pass': '#!/bin/sh\necho I am fine\n'})

        (code, out, err) = self.adt_run(['-d', '-B', '--unbuilt-tree=' + p])
        # print('----- out ----\n%s\n----- err ----\n%s\n----' % (out, err))
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        self.assertIn('processing dependency coreutils', err)
        # should show test stdout
        self.assertRegex(out, '(^|\n)I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_tree_norestrictions_nobuild_fail_on_exit(self):
        '''source tree, no build, no restrictions, test fails with non-zero'''

        p = self.build_src('Tests: nz\nDepends: coreutils\n',
                           {'nz': '#!/bin/sh\necho I am sick\nexit 7'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        # test should fail
        self.assertEqual(code, 4)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')

        # should show test stdout
        self.assertRegex(out, '(^|\n)I am sick\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_tree_norestrictions_nobuild_fail_on_stderr(self):
        '''source tree, no build, no restrictions, test fails with stderr'''

        p = self.build_src('Tests: se\nDepends: coreutils\n',
                           {'se': '#!/bin/sh\necho I am sick >&2\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        # test should fail
        self.assertEqual(code, 4)
        self.assertRegex(out, '(^|\n)se\s+FAIL stderr: I am sick\n')

        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick', err)

    def test_tree_allow_stderr_nobuild_fail_on_exit(self):
        '''source tree, no build, allow-stderr, test fails with non-zero'''

        p = self.build_src('Tests: nz\nDepends: coreutils\nRestrictions: allow-stderr',
                           {'nz': '#!/bin/sh\necho I am fine >&2\necho babble\nexit 7'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        # test should fail
        self.assertEqual(code, 4)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')

        # should show test stdout/err inline
        self.assertRegex(out, '(^|\n)babble\n')
        self.assertRegex(err, '---+\nI am fine\nadt-run', err)
        # but not complain about stderr
        self.assertNotIn(' stderr ', err)

    def test_tree_build_needed_success(self):
        '''source tree, build-needed restriction, test success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\n./test_built | grep -q "built script OK"\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p])
        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should not have any test stdout/stderr
        self.assertNotIn(' stderr ', err)

        # should build package
        self.assertIn('dh build', err)

    def test_dsc_norestrictions_nobuild_success(self):
        '''dsc, no build, no restrictions, test success'''

        p = self.build_dsc('Tests: pass\nDepends: coreutils\n',
                           {'pass': '#!/bin/sh\necho I am fine\n'})

        (code, out, err) = self.adt_run(['-d', '--no-built-binaries', p])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertIn('coreutils', err)
        self.assertRegex(out, 'pass\s+PASS', out)

        self.assertIn('processing dependency coreutils', err)
        # should show test stdout
        self.assertRegex(out, '(^|\n)I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should log package version
        self.assertIn('testing package testpkg version 1\n', err)

    def test_dsc_build_needed_success(self):
        '''dsc, build-needed restriction, test success'''

        p = self.build_dsc('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\n./test_built | grep -q "built script OK"\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', p])
        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should build package
        self.assertIn('dh build', err)

    def test_tmpdir_for_other_users(self):
        '''$TMPDIR is accessible to non-root users'''

        prev_mask = os.umask(0o077)
        self.addCleanup(os.umask, prev_mask)
        p = self.build_src('Tests: t\nDepends: coreutils\nRestrictions: needs-root\n',
                           {'t': '''#!/bin/sh -e
echo hello > ${TMPDIR:=/tmp}/rootowned.txt
su -s /bin/sh -c "echo hello > $TMPDIR/world.txt" nobody
if su -s /bin/sh -c "echo break > $TMPDIR/rootowned.txt" nobody 2>/dev/null; then
    exit 1
fi
rm $TMPDIR/rootowned.txt $TMPDIR/world.txt
'''})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p])
        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_user(self):
        '''--user option'''

        p = self.build_src('Tests: t\nDepends:\n',
                           {'t': '''#!/bin/sh -e
                                    echo world > ${TMPDIR:=/tmp}/hello.txt
                                    cat $TMPDIR/hello.txt
                                    rm $TMPDIR/hello.txt
                                    stat -c %U .
                                    stat -c %U debian
                                    stat -c %U Makefile
                                    echo "USER: $USER"
                                    whoami'''})

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--user=nobody'])
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cat and whoami
        self.assertIn('world\nnobody\nnobody\nnobody\nUSER: nobody\nnobody\n', out)

    def test_tree_output_dir(self):
        '''source tree, explicit --output-dir

        This also covers using upper-case lettes in test names.
        '''
        p = self.build_src('Tests: sP sF\nDepends: coreutils\n\n'
                           'Tests: bP\nDepends: coreutils\nRestrictions: build-needed',
                           {'sP': '#!/bin/sh\n./test_static\n',
                            'sF': '#!/bin/sh\necho kaputt >&2',
                            'bP': '#!/bin/sh\n./test_built'})

        outdir = os.path.join(self.workdir, 'out')
        os.mkdir(outdir)
        # put some cruft into it, to check that it gets cleaned up
        with open(os.path.join(outdir, 'cruft.txt'), 'w') as f:
            f.write('hello world')

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--output-dir=' + outdir])

        # test results
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'sP\s+PASS', out)
        self.assertRegex(out, 'sF\s+FAIL stderr: kaputt', out)
        self.assertRegex(out, 'bP\s+PASS', out)

        # should show test stdout and stderr
        self.assertRegex(out, '(^|\n)static script OK\n')
        self.assertIn('\nbuilt script OK\n', out)
        self.assertRegex(err, 'stderr [ -]+\nkaputt', err)

        # should build package
        self.assertIn('dh build', err)

        # check outdir test stdout/err
        with open(os.path.join(outdir, 'sP-stdout')) as f:
            self.assertEqual(f.read(), 'static script OK\n')
        self.assertFalse(os.path.exists(
            os.path.join(outdir, 'sP-stderr')))
        with open(os.path.join(outdir, 'bP-stdout')) as f:
            self.assertEqual(f.read(), 'built script OK\n')
        self.assertFalse(os.path.exists(
            os.path.join(outdir, 'bP-stderr')))
        self.assertFalse(os.path.exists(
            os.path.join(outdir, 'sF-stdout')))
        with open(os.path.join(outdir, 'sF-stderr')) as f:
            self.assertEqual(f.read(), 'kaputt\n')

        # check outdir log
        with open(os.path.join(outdir, 'log')) as f:
            contents = f.read()
        self.assertIn('build needed for tests', contents)
        self.assertIn('dh build', contents)
        self.assertRegex(contents, 'sF\s+FAIL stderr: kaputt')
        self.assertIn('testing package testpkg version 1\n', contents)

        # check summary
        with open(os.path.join(outdir, 'summary')) as f:
            self.assertRegex(f.read(), '^sP\s+PASS\nsF\s+FAIL stderr: kaputt\nbP\s+PASS$')

        # check test package version
        with open(os.path.join(outdir, 'testpkg-version')) as f:
            contents = f.read()
        self.assertEqual(contents, 'testpkg 1\n')

        # check recorded package lists
        with open(os.path.join(outdir, 'testbed-packages')) as f:
            contents = f.read()
            lines = contents.splitlines()
            self.assertGreater(len(lines), 10)
            self.assertRegex(lines[0], '^[0-9a-z.-]+\t[0-9a-z.~-]+')
            self.assertRegex(lines[1], '^[0-9a-z.-]+\t[0-9a-z.~-]+')
            self.assertIn('bash\t', contents)

        # test don't pull in any additional dependencies
        for t in ['sP', 'sF', 'bP']:
            self.assertEqual(os.path.getsize(os.path.join(
                outdir, '%s-packages' % t)), 0)

        # check for cruft in outdir
        # --no-built-binaries, we don't expect any debs
        files = [i for i in os.listdir(outdir)
                 if not fnmatch.fnmatch(i, '[sb]*-std*') and
                 not fnmatch.fnmatch(i, '[sb]*-packages')]
        self.assertEqual(set(files), set(['log', 'summary', 'testpkg-version', 'testbed-packages']))

    def test_timeout(self):
        '''handling test timeout'''

        p = self.build_dsc('Tests: totest\nDepends:\n',
                           {'totest': '#!/bin/sh\necho start\nsleep 10\necho after_sleep\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--timeout-test=3', p])
        # test should time out
        self.assertEqual(code, 16, err)
        self.assertIn("timed out on command", err)
        self.assertNotIn('totest', out)

        # should show test stdout
        self.assertIn('start', out)
        # but not let the test finish
        self.assertNotIn('after_sleep', out)

    def test_timeout_no_output(self):
        '''handling test timeout for test without any output'''

        p = self.build_dsc('Tests: totest\nDepends:\n',
                           {'totest': '#!/bin/sh\nsleep 10\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--timeout-test=3', p])
        # test should time out
        self.assertEqual(code, 16, err)
        self.assertIn("timed out on command", err)
        self.assertNotIn('totest', out)

    def test_timeout_long_test(self):
        '''long-running test with custom timeouts

        This verifies that the right timeout is being used for tests.
        '''
        p = self.build_dsc('Tests: p\nDepends:\nRestrictions: build-needed',
                           {'p': '#!/bin/sh\necho start\nsleep 5\n./test_built\n'})

        (code, out, err) = self.adt_run(['-B', '--timeout-test=6',
                                         '--timeout-build=20', p])
        # test should not time out
        self.assertEqual(code, 0, err)
        self.assertNotIn('timed out', err)

        # should show test stdout
        self.assertIn('start\n', out)
        self.assertIn('built script OK\n', out)

    def test_timeout_long_build(self):
        '''long-running build with custom timeouts

        This verifies that the right timeout is being used for builds.
        '''
        p = self.build_src('Tests: p\nDepends:\nRestrictions: build-needed',
                           {'p': '#!/bin/sh\necho start\n\n./test_built\n'})

        # make build take 4s
        subprocess.check_call(['sed', '-i', '/^build:/ s/$/\\n\\tsleep 4/',
                               os.path.join(p, 'Makefile')])

        (code, out, err) = self.adt_run(['--timeout-test=1', '--timeout-build=30',
                                         '-B', '--unbuilt-tree', p])
        # should build package
        self.assertIn('dh build', err)

        # test should not time out
        self.assertEqual(code, 0, err)
        self.assertNotIn('timeout', err)

        # should show test stdout
        self.assertIn('start\n', out)
        self.assertIn('built script OK\n', out)

    def test_timeout_long_build_fail(self):
        '''long-running build times out

        This verifies that the right timeout is being used for builds.
        '''
        p = self.build_src('Tests: p\nDepends:\nRestrictions: build-needed',
                           {'p': '#!/bin/sh\necho start\n\n./test_built\n'})

        # make build take 4s
        subprocess.check_call(['sed', '-i', '/^build:/ s/$/\\n\\tsleep 4/',
                               os.path.join(p, 'Makefile')])

        (code, out, err) = self.adt_run(['--timeout-test=1', '--timeout-build=6',
                                         '-B', '--unbuilt-tree', p])
        # should start building package
        self.assertIn('dh build', err)

        # build should time out
        self.assertEqual(code, 16, err)
        self.assertIn('timed out', err)

        # should not start tests
        self.assertNotIn('start\n', out)

    def test_logfile_success(self):
        '''--log-file option, success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\n',
                           {'pass': '#!/bin/sh\necho I am fine\n'})

        logfile = os.path.join(self.workdir, 'adt.log')
        (code, out, err) = self.adt_run(['-d', '--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--log-file=' + logfile])
        # test should succeed
        self.assertEqual(code, 0, err)

        with open(logfile) as f:
            log = f.read()
        self.assertIn('coreutils', err)
        self.assertIn('coreutils', log)
        self.assertRegex(out, 'pass\s+PASS')
        self.assertRegex(log, 'pass\s+PASS')

        self.assertIn('processing dependency coreutils', err)
        self.assertIn('processing dependency coreutils', log)
        # should show test stdout
        self.assertRegex(out, '(^|\n)I am fine\n')
        self.assertRegex(log, '(^|\n)I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)
        self.assertNotIn(' stderr ', log)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_logfile_failure(self):
        '''--log-file option, failure'''

        p = self.build_src('Tests: nz\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'nz': '#!/bin/sh\n./test_built\necho I am sick >&2\nexit 7'})

        logfile = os.path.join(self.workdir, 'adt.log')
        (code, out, err) = self.adt_run(['-d', '--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--log-file=' + logfile])
        with open(logfile) as f:
            log = f.read()

        # test should fail
        self.assertEqual(code, 4)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')
        self.assertRegex(log, 'nz\s+FAIL non-zero exit status 7')

        self.assertIn('processing dependency coreutils', err)
        self.assertIn('processing dependency coreutils', log)

        # should build package
        self.assertIn('dh build', err)
        self.assertIn('dh build', log)

        # should show test stdout
        self.assertRegex(out, '(^|\n)built script OK\n')
        self.assertRegex(log, '(^|\n)built script OK\n')
        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')
        self.assertRegex(log, 'stderr [ -]+\nI am sick\n')

    def test_breaks_testbed(self):
        '''breaks-testbed restriction'''

        p = self.build_src('Tests: rambo\nDepends:\nRestrictions: needs-root breaks-testbed',
                           {'rambo': '#!/bin/sh\ntouch /var/tmp/zap\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p])
        # test should be skipped as null runner doesn't provide revert
        self.assertEqual(code, 2, err)
        self.assertRegex(out, 'rambo\s+SKIP Test breaks testbed')
        self.assertNotIn(out, 'no tests')
        self.assertFalse(os.path.exists('/var/tmp/zap'))

    def test_tree_apply_patches(self):
        '''source tree, 3.0 (quilt) patches get applied'''

        p = self.build_src('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\ndebian/testpkg/usr/bin/test_built\n'
                            'debian/testpkg/usr/bin/test_static\n'})

        # add patch
        patchdir = os.path.join(p, 'debian', 'patches')
        os.mkdir(patchdir)
        with open(os.path.join(patchdir, '01_hack.patch'), 'w') as f:
            f.write('''--- testpkg.orig/test_static
+++ testpkg/test_static
@@ -1,2 +1,2 @@
 #!/bin/sh
-echo "static script OK"
+echo "static patched script OK"
''')
        with open(os.path.join(patchdir, 'series'), 'w') as f:
            f.write('01_hack.patch')

        # turn into 3.0 (quilt) source
        dsrcdir = os.path.join(p, 'debian', 'source')
        os.mkdir(dsrcdir)
        with open(os.path.join(dsrcdir, 'format'), 'w') as f:
            f.write('3.0 (quilt)\n')

        # run tests, should apply unapplied patches
        (code, out, err) = self.adt_run(['-B', '-d', '--unbuilt-tree=' + p])

        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should have patched source
        self.assertRegex(err, 'dpkg-source:.*01_hack.patch')
        self.assertIn('built script OK\nstatic patched script OK\n', out)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should build package
        self.assertIn('dh build', err)

    def test_default_locale(self):
        '''tests have default locale C.UTF-8'''

        p = self.build_src('Tests: r\nDepends:\nRestrictions: needs-root\n\n'
                           'Tests: u\nDepends:\n',
                           {'r': '#!/bin/sh -e\necho root $LANG',
                            'u': '#!/bin/sh -e\necho user $LANG'})

        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p])
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 'r\s+PASS', out)
        self.assertRegex(out, 'u\s+PASS', out)

        # has expected locales
        self.assertIn('root C.UTF-8\n', out)
        self.assertIn('user C.UTF-8\n', out)

    def test_specify_locale(self):
        '''tests have specified locale'''

        p = self.build_src('Tests: r\nDepends:\nRestrictions: needs-root\n\n'
                           'Tests: u\nDepends:\n',
                           {'r': '#!/bin/sh -e\necho root $LANG',
                            'u': '#!/bin/sh -e\necho user $LANG'})

        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p,
                                         '--set-lang=ab_CD.UTF-8'])
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 'r\s+PASS', out)
        self.assertRegex(out, 'u\s+PASS', out)

        # has expected locales
        self.assertIn('root ab_CD.UTF-8\n', out)
        self.assertIn('user ab_CD.UTF-8\n', out)


@unittest.skipIf('cowdancer' in os.environ.get('LD_PRELOAD', ''),
                 'chroot tests do not work under cowdancer')
@unittest.skipIf(os.getuid() > 0,
                 'chroot runner needs to run as root')
class ChrootRunner(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(ChrootRunner, self).__init__('chroot', *args, **kwargs)

    def setUp(self):
        super(ChrootRunner, self).setUp()

        def install_file(path):
            destdir = self.chroot + '/' + os.path.dirname(path)
            if not os.path.exists(destdir):
                os.makedirs(destdir)
            if os.path.isfile(path):
                shutil.copy(path, destdir)
            else:
                subprocess.check_call(['cp', '-a', path, destdir])

        def install_elf(path):
            install_file(path)
            out = subprocess.check_output(['ldd', path], universal_newlines=True)
            libs = set()
            for lib in re.finditer('/[^ ]+', out):
                libs.add(lib.group(0))
            for lib in libs:
                install_file(lib)

        # build a mini-chroot
        self.chroot = os.path.join(self.workdir, 'chroot')

        install_file('/usr/bin/which')
        install_elf('/bin/bash')
        install_elf('/bin/sh')
        install_elf('/bin/ls')
        install_elf('/bin/cat')
        install_elf('/bin/rm')
        install_elf('/bin/cp')
        install_elf('/bin/mkdir')
        install_elf('/bin/chmod')
        install_elf('/bin/chown')
        install_elf('/bin/mktemp')
        install_elf('/bin/tar')
        install_elf('/bin/sleep')
        install_elf('/bin/sed')
        install_elf('/bin/readlink')
        install_elf('/bin/grep')
        install_elf('/usr/bin/test')
        install_elf('/usr/bin/awk')
        install_elf('/usr/bin/env')
        install_elf('/usr/bin/tee')
        install_elf('/usr/bin/touch')

        # necessary bind mounts
        dev_dir = os.path.join(self.chroot, 'dev')
        os.mkdir(dev_dir)
        subprocess.check_call(['mount', '-o', 'bind', '/dev', dev_dir])
        proc_dir = os.path.join(self.chroot, 'proc')
        os.mkdir(proc_dir)
        subprocess.check_call(['mount', '-o', 'bind', '/proc', proc_dir])

        # some fakes
        for cmd in ['dpkg', 'dpkg-query', 'apt-get', 'apt-key', 'apt-cache']:
            p = os.path.join(self.chroot, 'usr', 'bin', cmd)
            with open(p, 'w') as f:
                f.write('#!/bin/sh\n')
                if cmd == 'dpkg':
                    f.write('if [ "$1" = "--print-architecture" ]; then echo megacpu; exit; fi\n')
                f.write('echo "fake-%s: $@"\n' % cmd)
                if cmd == 'apt-get':
                    f.write('if [ "$1" = source ]; then cp -r /aptget-src $3-1; fi\n')
            os.chmod(p, 0o755)

        p = os.path.join(self.chroot, 'tmp')
        os.mkdir(p)
        os.chmod(p, 0o177)
        os.makedirs(os.path.join(self.chroot, 'etc', 'apt', 'sources.list.d'))

    def tearDown(self):
        # these sometimes fail on EBUSY
        subprocess.call('for p in %(c)s/dev %(c)s/proc; do while mountpoint $p >/dev/null; do umount $p; sleep 0.1; done; done' %
                        {'c': self.chroot}, shell=True)
        subprocess.call('for i in `seq 10`; do rm -rf %s && break; sleep 0.5; done' % self.chroot, shell=True)

    def test_tree_norestrictions_nobuild_success(self):
        '''source tree, no build, no restrictions, test success'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root\n',
                           {'pass': '#!/bin/sh\necho I am fine\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.chroot])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '^I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

        self.assertNotIn('@@@@@@ test bed setup', err)

        # should log package version
        self.assertIn('testing package testpkg version 1\n', err)

    def test_tree_norestrictions_nobuild_fail_on_exit(self):
        '''source tree, no build, no restrictions, test fails with non-zero'''

        p = self.build_src('Tests: nz\nDepends: fancypkg\nRestrictions: needs-root\n',
                           {'nz': '#!/bin/sh\necho I am sick\nexit 7'})

        (code, out, err) = self.adt_run(['-d', '--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.chroot])
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')

        # should show test stdout
        self.assertRegex(out, '(^|\n)I am sick\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        self.assertIn('processing dependency fancypkg', err)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_tree_norestrictions_nobuild_fail_on_stderr(self):
        '''source tree, no build, no restrictions, test fails with stderr'''

        p = self.build_src('Tests: se\nDepends:\nRestrictions: needs-root\n',
                           {'se': '#!/bin/sh\necho I am sick >&2'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.chroot])
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'se\s+FAIL stderr: I am sick\n')

        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_tree_norestrictions_nobuild_fail_on_stderr_and_exit(self):
        '''source tree, no build, no restrictions, test fails with stderr+exit'''

        p = self.build_src('Tests: senz\nDepends:\nRestrictions: needs-root\n',
                           {'senz': '#!/bin/sh\necho I am sick >&2\nexit 7\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        [self.chroot])
        # test should fail
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'senz\s+FAIL non-zero exit status 7')

        # should show test stderr separately (no real-time output for chroot)
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_tree_allow_stderr_nobuild_success(self):
        '''source tree, no build, allow-stderr, test success'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr needs-root',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        [self.chroot])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS')

        # should show test stdout/err
        self.assertRegex(out, '^babble\n')
        self.assertIn('\nI am fine\n', err)

    def test_fancy_deps(self):
        '''wrapped and versioned test dependencies'''

        p = self.build_src('''Tests: pass
Depends: fancypkg,
         coreutils | vanilla (>= 10),
         chocolate,
Restrictions: needs-root
''', {'pass': '#!/bin/sh\ntrue'})

        (code, out, err) = self.adt_run(['-d', '-B', '--unbuilt-tree=' + p],
                                        [self.chroot])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS')

        self.assertIn('processing dependency fancypkg\n', err)
        self.assertIn('processing dependency coreutils | vanilla (>= 10)\n', err)
        self.assertIn('processing dependency chocolate\n', err)

    def test_build_deps(self):
        '''test depends on build dependencies'''

        p = self.build_src('''Tests: pass
Depends: @, testdep1, @builddeps@, testdep2,
# blabla ☺
 testdep3
Restrictions: needs-root
''', {'pass': '#!/bin/sh\ntrue'})

        # add extra build dependencies to testpkg
        subprocess.check_call(['sed', '-i', '/^Build-Depends:/ s/:.*/: bdep1 , bdep2,\\n bdep3, # moo\\n#comment\\n bdep4\\n'
                                            'Build-Depends-Indep: bdep5/',
                               os.path.join(p, 'debian', 'control')])

        # run this under C locale to test that UTF-8 debian/control is still
        # handled correctly
        (code, out, err) = self.adt_run(['-d', '-B', '--unbuilt-tree=' + p],
                                        [self.chroot], env={'LC_ALL': 'C'})
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS')

        self.assertIn('synthesised dependency testpkg\n', err)
        self.assertIn('processing dependency testdep1\n', err)
        self.assertIn('synthesised dependency bdep1\n', err)
        self.assertIn('synthesised dependency bdep2\n', err)
        self.assertIn('synthesised dependency bdep3\n', err)
        self.assertIn('synthesised dependency bdep4\n', err)
        self.assertIn('synthesised dependency bdep5\n', err)
        self.assertIn('synthesised dependency build-essential\n', err)
        self.assertIn('processing dependency testdep2\n', err)

    def test_logfile(self):
        '''--log-file option'''

        p = self.build_src('Tests: nz\nDepends: coreutils\nRestrictions: needs-root\n',
                           {'nz': '#!/bin/sh\n./test_static\necho I am sick >&2\nexit 7'})

        logfile = os.path.join(self.workdir, 'adt.log')
        (code, out, err) = self.adt_run(['--no-built-binaries', '-d',
                                         '--unbuilt-tree=' + p,
                                         '--log-file=' + logfile],
                                        [self.chroot])
        with open(logfile) as f:
            log = f.read()

        # test should fail
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')
        self.assertRegex(log, 'nz\s+FAIL non-zero exit status 7')

        self.assertIn('processing dependency coreutils', err)
        self.assertIn('processing dependency coreutils', log)

        # should show test stdout
        self.assertRegex(out, '(^|\n)static script OK\n')
        self.assertRegex(log, '(^|\n)static script OK\n')
        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')
        self.assertRegex(log, 'stderr [ -]+\nI am sick\n')

    def test_artifacts(self):
        '''tests producing additional artifacts'''

        p = self.build_src('Tests: a1 a2 a3 a4\nDepends:\nRestrictions: needs-root\n',
                           {'a1': '#!/bin/sh -e\n[ -d "$ADT_ARTIFACTS" ]\n'
                                  'echo old > $ADT_ARTIFACTS/health.txt\n',
                            'a2': '#!/bin/sh -e\n[ -d "$ADT_ARTIFACTS" ]\n'
                                  'echo I am fine > $ADT_ARTIFACTS/health.txt\n',
                            'a3': '#!/bin/sh -e\n[ -d "$ADT_ARTIFACTS" ]\n'
                                  'mkdir $ADT_ARTIFACTS/logs\n'
                                  'echo world > $ADT_ARTIFACTS/logs/hello.txt\n',
                            'a4': '#!/bin/sh -e\n[ -d "$ADT_ARTIFACTS" ]\n'
                                  'mkdir $ADT_ARTIFACTS/logs\n'
                                  'echo 42 > $ADT_ARTIFACTS/logs/answer.txt\n'})

        outdir = os.path.join(self.workdir, 'out')

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--output-dir=' + outdir],
                                        [self.chroot])

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'a1\s+PASS', out)
        self.assertRegex(out, 'a2\s+PASS', out)
        self.assertRegex(out, 'a3\s+PASS', out)
        self.assertRegex(out, 'a4\s+PASS', out)

        # check for cruft in output dir
        files = [i for i in os.listdir(outdir)
                 if not fnmatch.fnmatch(i, 'a*-std*') and
                 not fnmatch.fnmatch(i, 'a*-packages')]
        self.assertEqual(set(files),
                         set(['log', 'artifacts', 'testpkg-version',
                              'testbed-packages', 'summary']))

        # check artifact; a2 should overwrite a1's health.txt
        with open(os.path.join(outdir, 'artifacts', 'health.txt')) as f:
            self.assertEqual(f.read(), 'I am fine\n')
        with open(os.path.join(outdir, 'artifacts', 'logs', 'hello.txt')) as f:
            self.assertEqual(f.read(), 'world\n')
        with open(os.path.join(outdir, 'artifacts', 'logs', 'answer.txt')) as f:
            self.assertEqual(f.read(), '42\n')

    def test_slash_in_test_name(self):
        '''test names must not contain /'''

        p = self.build_src('Tests: pass subdir/p\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\ntrue'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.chroot])
        # invalid test gets skipped
        self.assertEqual(code, 2, err)
        self.assertRegex(out, 'subdir/p\s+SKIP test name may not contain /.*')

        # valid test still gets run
        self.assertFalse(re.match('pass\s+SKIP', out), out)
        self.assertRegex(out, 'pass\s+PASS', out)

    def test_breaks_testbed(self):
        '''breaks-testbed restriction'''

        p = self.build_src('Tests: rambo\nDepends:\nRestrictions: needs-root breaks-testbed',
                           {'rambo': '#!/bin/sh\ntouch /zap\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.chroot])
        # test should be skipped as chroot runner doesn't provide revert
        self.assertEqual(code, 2)
        self.assertRegex(out, 'rambo\s+SKIP Test breaks testbed')
        self.assertFalse(os.path.exists(os.path.join(self.chroot, 'zap')))

    def test_unicode(self):
        '''Unicode test output'''

        p = self.build_src('Tests: se\nDepends:\nRestrictions: needs-root\n',
                           {'se': '#!/bin/sh\necho ‘a♩’; echo fancy ‴ʎɔuɐɟ″!>&2'})

        summary = os.path.join(self.workdir, 'summary.log')

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--summary=' + summary],
                                        [self.chroot])
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'se\s+FAIL stderr: fancy ‴ʎɔuɐɟ″!\n')

        # should show test stdout/stderr
        self.assertRegex(out, '‘a♩’\n')
        self.assertRegex(err, '\nfancy ‴ʎɔuɐɟ″!\n')

        with open(summary, encoding='UTF-8') as f:
            self.assertRegex(f.read(), 'se\s+FAIL stderr: fancy ‴ʎɔuɐɟ″!\n')

    def test_apt_source_no_restrictions(self):
        '''apt source, no build, no restrictions'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root\n',
                           {'pass': '#!/bin/sh\necho I am fine\n'})
        # copy that into the chroot where fake apt-get source can find it
        shutil.copytree(p, os.path.join(self.chroot, 'aptget-src'))

        (code, out, err) = self.adt_run(['testpkg'], [self.chroot])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '(^|\n)I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

        self.assertRegex(err, 'fake-apt-get: source --only-source testpkg')

    def test_setup_commands_string(self):
        '''--setup-commands with command string'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root\n',
                           {'pass': '#!/bin/sh -e\n[ -x /bin/cp_cp ]; cat /setup.log\n'})

        # not expecting a normal user for chroot
        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p,
                                         '--setup-commands', '[ -z $ADT_NORMAL_USER ];'
                                         'sleep 3; cp /bin/cp /bin/cp_cp; '
                                         'echo setup_success > /setup.log',
                                         '--setup-commands', 'cp /bin/cp /bin/cp_cp',
                                         '--timeout-short=1', '--timeout-copy=1'],
                                        [self.chroot])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '^setup_success\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        self.assertIn('@@@@@@ test bed setup', err)

    def test_setup_commands_file(self):
        '''--setup-commands with command file'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root\n',
                           {'pass': '#!/bin/sh -e\n[ -x /bin/cp_cp ]; cat /setup.log\ncat /s2.log'})

        cmds = os.path.join(self.workdir, 'setup.sh')
        with open(cmds, 'w') as f:
            f.write('cp /bin/cp /bin/cp_cp\necho setup_success > /setup.log\n')
            f.flush()
        cmds2 = os.path.join(self.workdir, 'setup2.sh')
        with open(cmds2, 'w') as f:
            f.write('echo setup2_success > /s2.log\n')
            f.flush()

        (code, out, err) = self.adt_run(['-B', '-d', '--unbuilt-tree=' + p,
                                         '--setup-commands', cmds,
                                         '--setup-commands', cmds2],
                                        [self.chroot])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '^setup_success\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        self.assertIn('@@@@@@ test bed setup', err)

    def test_copy(self):
        '''--copy'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root\n',
                           {'pass': '#!/bin/sh -e\ncp -a /mytestdata/* $ADT_ARTIFACTS\necho test_ok'})

        outdir = os.path.join(self.workdir, 'out')
        os.mkdir(outdir)

        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p, '-o', outdir,
                                         '--copy', '%s:/mytestdata/Makefile' % os.path.join(root_dir, 'Makefile'),
                                         '--copy', '%s:/mytestdata/tree' % os.path.join(test_dir, 'testpkg')],
                                        [self.chroot])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertRegex(out, '^test_ok\n')

        # has copied the file correctly
        f = os.path.join(outdir, 'artifacts', 'Makefile')
        self.assertTrue(os.path.isfile(f))
        with open(f) as fcopy:
            with open(os.path.join(root_dir, 'Makefile')) as forig:
                self.assertEqual(forig.read(), fcopy.read())

        # has copied the dir correctly
        d = os.path.join(outdir, 'artifacts', 'tree')
        self.assertTrue(os.path.isdir(d))
        subprocess.check_call(['diff', '-Nur', os.path.join(test_dir, 'testpkg'), d])

    def test_apt_pocket(self):
        '''--apt-pocket'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root\n',
                           {'pass': '#!/bin/sh -e\ntest -e /etc/apt/sources.list.d/proposed.list'})

        apt_dir = os.path.join(self.chroot, 'etc', 'apt')
        with open(os.path.join(apt_dir, 'sources.list'), 'w') as f:
            f.write('''# comment
deb http://foo.ubuntu.com/ fluffy-updates main non-free
deb-src http://foo.ubuntu.com/ fluffy-updates main non-free
deb http://foo.ubuntu.com/ fluffy main non-free
deb-src http://foo.ubuntu.com/ fluffy main non-free
deb http://bar.debian.org/ fluffy extras
deb-src http://bar.debian.org/ fluffy extras
# third-party repo
deb http://something.else.net/ fluffy addons
''')

        (code, out, err) = self.adt_run(['-B', '-d', '--unbuilt-tree=' + p,
                                         '--apt-pocket', 'proposed'],
                                        [self.chroot])

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # verify proposed.list
        with open(os.path.join(apt_dir, 'sources.list.d', 'proposed.list')) as f:
            self.assertEqual(f.read(), '''deb http://foo.ubuntu.com/ fluffy-proposed main non-free
deb-src http://foo.ubuntu.com/ fluffy-proposed main non-free
deb http://bar.debian.org/ fluffy-proposed extras
deb-src http://bar.debian.org/ fluffy-proposed extras
''')

    def test_isolation(self):
        '''isolation restrictions'''

        p = self.build_src('Tests: ic\nDepends:\nRestrictions: isolation-container\n\n'
                           'Tests: im\nDepends:\nRestrictions: isolation-machine\n',
                           {'ic': '#!/bin/sh\necho container ok',
                            'im': '#!/bin/sh\necho machine ok'})

        (code, out, err) = self.adt_run(['-B', '--built-tree=' + p], [self.chroot])
        self.assertEqual(code, 2, out + err)
        self.assertRegex(out, 'ic\s+SKIP .*container', out)
        self.assertRegex(out, 'im\s+SKIP .*machine', out)
        self.assertNotIn('ok', out)

    def test_tree_garbage(self):
        '''copied source tree contains only expected files'''

        p = self.build_src('Tests: g\nDepends:\nRestrictions: needs-root\n',
                           {'g': '#!/bin/sh\npwd\nLC_ALL=C ls .\n'})

        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p],
                                        [self.chroot])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'g\s+PASS', out)

        self.assertRegex(out, '^/tmp/adt-run.*/real-tree\n'
                         'Makefile\ndebian\ntest_static\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_no_tests_dir(self):
        '''package without debian/tests/'''

        p = self.build_src(None, None)
        (code, out, err) = self.adt_run(['-B', '--unbuilt-tree=' + p], [self.chroot])
        self.assertEqual(code, 8, err)
        self.assertRegex(out, 'SKIP no tests in this package', out)

    def test_override_control(self):
        '''custom control file path with --override-control'''

        p = self.build_src('Tests: NOTME\nDepends: nonexisting\nRestrictions: breaks-testbed\n',
                           {'pass': '#!/bin/sh\necho I am fine\n'})

        custom_control = os.path.join(self.workdir, 'adtctrl')
        with open(custom_control, 'w') as f:
            f.write('Tests: pass\nDepends:\nRestrictions: needs-root')

        (code, out, err) = self.adt_run(['-B', '--override-control=' + custom_control, p + '//'],
                                        [self.chroot])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '^I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not try anything with the bad test
        self.assertNotIn('NOTME', out)
        self.assertNotIn('NOTME', err)


@unittest.skipUnless('ADT_TEST_SCHROOT' in os.environ,
                     'Set $ADT_TEST_SCHROOT to an existing schroot')
class SchrootRunner(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(SchrootRunner, self).__init__('schroot', *args, **kwargs)
        self.schroot_name = os.environ.get('ADT_TEST_SCHROOT')

    def test_tree_norestrictions_nobuild_success(self):
        '''source tree, no build, no restrictions, test success'''

        # also test some fancy dependencies
        p = self.build_src('Tests: p1 p2\n'
                           'Depends: coreutils, aspell-doc [linux-any], cpp-doc:native, unknown [nosucharch]\n'
                           'Restrictions: needs-root\n',
                           {'p1': '#!/bin/sh\necho I am fine\n',
                            'p2': '#!/bin/sh\necho I am also fine\n'})

        (code, out, err) = self.adt_run(['-d', '--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.schroot_name])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'p1\s+PASS', out)
        self.assertRegex(out, 'p2\s+PASS', out)

        # handles expected packages
        self.assertIn('processing dependency coreutils', err)
        self.assertIn('Unpacking aspell-doc', out)
        self.assertIn('Unpacking cpp-doc', out)
        self.assertNotIn('unknown', out)

        # should show test stdout
        self.assertRegex(out, '(^|\n)I am fine\n')
        self.assertIn('\nI am also fine\n', out)
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_tree_norestrictions_nobuild_fail_on_exit(self):
        '''source tree, no build, no restrictions, test fails with non-zero'''

        p = self.build_src('Tests: nz\nDepends:\nRestrictions: needs-root\n',
                           {'nz': '#!/bin/sh\necho I am sick\nexit 7'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.schroot_name])
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')

        # should show test stdout
        self.assertRegex(out, '^I am sick\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_tree_norestrictions_nobuild_fail_on_stderr(self):
        '''source tree, no build, no restrictions, test fails with stderr'''

        p = self.build_src('Tests: se\nDepends:\nRestrictions: needs-root\n',
                           {'se': '#!/bin/sh\necho I am sick >&2'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.schroot_name])
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'se\s+FAIL stderr: I am sick\n')

        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_tree_norestrictions_nobuild_fail_on_stderr_and_exit(self):
        '''source tree, no build, no restrictions, test fails with stderr+exit'''

        p = self.build_src('Tests: senz\nDepends:\nRestrictions: needs-root\n',
                           {'senz': '#!/bin/sh\necho I am sick >&2\nexit 7\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        [self.schroot_name])
        # test should fail
        self.assertEqual(code, 4)
        self.assertRegex(out, 'senz\s+FAIL non-zero exit status 7')

        # should show test stderr separately
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_tree_allow_stderr_nobuild_success(self):
        '''source tree, no build, allow-stderr, test success'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        [self.schroot_name])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout/err
        self.assertRegex(out, '^babble\n')
        self.assertIn('\nI am fine\n', err)
        # stderr test output should be inline
        self.assertNotIn(' stderr ', err)

    def test_tree_build_needed_success(self):
        '''source tree, build-needed restriction, test success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\n./test_built | grep -q "built script OK"\necho GOOD'})
        # add Build-Depends-Indep:
        subprocess.check_call(['sed', '-i', '/^Build-Depends:/ a\Build-Depends-Indep: aspell-doc',
                               os.path.join(p, 'debian', 'control')])

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.schroot_name])
        # test should succeed
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should build package
        self.assertIn('dh build', err)

        # should show test stdout
        self.assertRegex(out, '(^|\n)GOOD\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_dsc_build_needed_success(self):
        '''dsc, build-needed restriction, test success'''

        p = self.build_dsc('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\n./test_built | grep -q "built script OK"\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', p],
                                        [self.schroot_name])
        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should build package
        self.assertIn('dh build', err)

    def test_tree_built_binaries(self):
        '''source tree, install built binaries'''

        # no "build-needed" restriction here, it should be built because we
        # need to install the package as a dependency (@)
        p = self.build_src('Tests: pass\n',
                           {'pass': '#!/bin/sh -e\n/usr/bin/test_built | grep -q "built script OK"\necho GOOD'})

        # add extra dependency to testpkg, to ensure autopkgtest does not stop
        # and ask for confirmation in apt-get
        subprocess.check_call(['sed', '-i', '/^Depends:/ s/$/, aspell-doc/',
                               os.path.join(p, 'debian', 'control')])

        (code, out, err) = self.adt_run(['--unbuilt-tree=' + p], [self.schroot_name])

        # should build and install package
        self.assertIn('dh build', err)
        self.assertRegex(out, '\npass\s+PASS', out)
        self.assertIn('Unpacking testpkg', out)
        self.assertIn('binaries/./testpkg.deb', out)
        self.assertIn('Unpacking aspell-doc', out)
        # does not install Recommends by default
        self.assertNotIn('Unpacking cpp-doc', out)
        self.assertNotIn('Unpacking cpp-doc', err)

        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should show test stdout
        self.assertRegex(out, '(^|\n)GOOD\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_binary_unbuilt_tree(self):
        '''--binary for test, unbuilt tree'''

        p = self.build_src('Tests: pass\n',
                           {'pass': '#!/bin/sh -e\n/usr/bin/test_built'})

        # build the package
        subprocess.check_call(['dpkg-buildpackage', '-b', '-us', '-uc', '-tc'],
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                              cwd=p)
        deb = os.path.join(os.path.dirname(p), 'testpkg_1_all.deb')
        self.assertTrue(os.path.exists(deb))
        # destroy debian/rules, to ensure it does not try to build the package
        # again
        with open(os.path.join(p, 'debian', 'rules'), 'w') as f:
            f.write('bwhahahahano!')

        (code, out, err) = self.adt_run(['-d', '-B', '--binary', deb, '--unbuilt-tree=' + p],
                                        [self.schroot_name])

        # should install package
        self.assertNotIn('dh build', err)
        self.assertIn('Unpacking testpkg', out)
        self.assertIn('binaries/./testpkg.deb', out)

        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should show test stdout
        self.assertRegex(out, '(^|\n)built script OK\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # our deb should still be there
        self.assertTrue(os.path.exists(deb))

    def test_binary_built_tree(self):
        '''--binary for test, built tree'''

        # we test both for the installed package as well as the built tree
        p = self.build_src('Tests: pass\n',
                           {'pass': '#!/bin/sh -e\n/usr/bin/test_built\n./test_built'})

        # build the package
        subprocess.check_call(['dpkg-buildpackage', '-b', '-us', '-uc'],
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                              cwd=p)
        deb = os.path.join(os.path.dirname(p), 'testpkg_1_all.deb')
        self.assertTrue(os.path.exists(deb))
        # destroy debian/rules, to ensure it does not try to build the package
        # again
        with open(os.path.join(p, 'debian', 'rules'), 'w') as f:
            f.write('bwhahahahano!')

        # --no-built-binaries should be implied here
        (code, out, err) = self.adt_run(['-d', deb, '--built-tree=' + p],
                                        [self.schroot_name])

        # should install package
        self.assertNotIn('dh build', err)
        self.assertIn('Unpacking testpkg', out)
        self.assertIn('binaries/./testpkg.deb', out)

        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out + err)
        self.assertEqual(code, 0, out + err)

        # should show test stdout
        self.assertRegex(out, '(^|\n)built script OK\nbuilt script OK')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # our deb should still be there
        self.assertTrue(os.path.exists(deb))

    def test_changes(self):
        '''source/binary .changes'''

        p = self.build_src('Tests: pass\n',
                           {'pass': '#!/bin/sh -e\n/usr/bin/test_built'})

        # build the package
        subprocess.check_call(['dpkg-buildpackage', '-us', '-uc'],
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                              cwd=p)
        changes = glob('%s/testpkg_1_*.changes' % os.path.dirname(p))
        self.assertEqual(len(changes), 1)
        changes = changes[0]
        (code, out, err) = self.adt_run([changes], [self.schroot_name])

        # should not rebuild
        self.assertNotIn('dh build', err)

        # should install package
        self.assertIn('Unpacking testpkg', out)
        self.assertIn('binaries/./testpkg.deb', out)

        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should show test stdout
        self.assertRegex(out, '(^|\n)built script OK\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # our .changes should still be there
        self.assertTrue(os.path.exists(changes))

    def test_logfile(self):
        '''--log-file option'''

        p = self.build_src('Tests: nz\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'nz': '#!/bin/sh\n./test_built\necho I am sick >&2\nexit 7'})

        logfile = os.path.join(self.workdir, 'adt.log')
        (code, out, err) = self.adt_run(['-d', '--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--log-file=' + logfile],
                                        [self.schroot_name])
        with open(logfile) as f:
            log = f.read()

        # test should fail
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')
        self.assertRegex(log, 'nz\s+FAIL non-zero exit status 7')

        self.assertIn('processing dependency coreutils', err)
        self.assertIn('processing dependency coreutils', log)

        # should build package
        self.assertIn('dh build', err)
        self.assertIn('dh build', log)

        # should show test stdout
        self.assertRegex(out, '(^|\n)built script OK\n')
        self.assertRegex(log, '(^|\n)built script OK\n')
        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')
        self.assertRegex(log, 'stderr [ -]+\nI am sick\n')

    def test_tree_output_dir(self):
        '''source tree, explicit --output-dir'''

        p = self.build_src('Tests: ok\n\nTests: broken\nDepends: @, aspell-doc',
                           {'ok': '#!/bin/sh\n/usr/bin/test_built',
                            'broken': '#!/bin/sh\necho kaputt >&2'})

        outdir = os.path.join(self.workdir, 'out')

        (code, out, err) = self.adt_run(['--unbuilt-tree=' + p,
                                         '--output-dir=' + outdir],
                                        [self.schroot_name])

        # test results
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'ok\s+PASS', out)
        self.assertRegex(out, 'broken\s+FAIL stderr: kaputt', out)

        # should show test stdout and stderr
        self.assertRegex(out, '(^|\n)built script OK\n')
        self.assertRegex(err, 'stderr [ -]+\nkaputt', err)

        # should build package
        self.assertIn('dh build', err)

        # check outdir test stdout/err
        with open(os.path.join(outdir, 'ok-stdout')) as f:
            self.assertEqual(f.read(), 'built script OK\n')
        self.assertFalse(os.path.exists(
            os.path.join(outdir, 'ok-stderr')))
        self.assertFalse(os.path.exists(
            os.path.join(outdir, 'broken-stdout')))
        with open(os.path.join(outdir, 'broken-stderr')) as f:
            self.assertEqual(f.read(), 'kaputt\n')

        # check outdir log
        with open(os.path.join(outdir, 'log')) as f:
            contents = f.read()
        self.assertIn('dh build', contents)
        self.assertRegex(contents, 'broken\s+FAIL stderr: kaputt')

        # check summary
        with open(os.path.join(outdir, 'summary')) as f:
            self.assertRegex(f.read(), '^ok\s+PASS\nbroken\s+FAIL stderr: kaputt$')

        # check test package version
        with open(os.path.join(outdir, 'testpkg-version')) as f:
            contents = f.read()
        self.assertEqual(contents, 'testpkg 1\n')

        # check recorded package lists
        with open(os.path.join(outdir, 'testbed-packages')) as f:
            contents = f.read()
            lines = contents.splitlines()
            self.assertGreater(len(lines), 10)
            self.assertRegex(lines[0], '^[0-9a-z.-]+\t[0-9a-z.~-]+')
            self.assertRegex(lines[1], '^[0-9a-z.-]+\t[0-9a-z.~-]+')
            self.assertIn('bash\t', contents)

        # check recorded package list
        for t in ['ok', 'broken']:
            with open(os.path.join(outdir, '%s-packages' % t)) as f:
                contents = f.read()
                self.assertIn('testpkg\t1\n', contents)
                if t == 'broken':
                    self.assertIn('aspell-doc\t', contents)
                else:
                    self.assertNotIn('aspell-doc\t', contents)
                self.assertNotIn('adt-satdep', contents)

        # check binaries
        bins = os.listdir(os.path.join(outdir, 'binaries'))
        self.assertEqual(set(bins),
                         set(['Release.gpg', 'archive-key.pgp', 'Release',
                              'Packages.gz', 'Packages', 'testpkg.deb']))

        # check for cruft in outdir
        self.assertEqual(set(os.listdir(outdir)),
                         set(['log', 'summary', 'binaries', 'testpkg-version',
                              'testbed-packages', 'ok-packages', 'ok-stdout',
                              'broken-packages', 'broken-stderr']))

    def test_user(self):
        '''--user option'''

        p = self.build_src('Tests: t\nDepends: aspell-doc\n',
                           {'t': '''#!/bin/sh -e
                                    echo world > ${TMPDIR:=/tmp}/hello.txt
                                    cat $TMPDIR/hello.txt
                                    whoami
                                    echo "USER: $USER"\n'''})

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--user=nobody'],
                                        [self.schroot_name])
        self.assertEqual(code, 0, err)

        # should install dependencies
        self.assertIn('Unpacking aspell-doc', out)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cat and whoami
        self.assertRegex(out, '(^|\n)world\nnobody\nUSER: nobody\n')

    def test_user_needs_root(self):
        '''--user option with needs-root restriction'''

        p = self.build_src('Tests: t\nDepends: aspell-doc\nRestrictions: needs-root',
                           {'t': '''#!/bin/sh -e
                                    echo world > ${TMPDIR:=/tmp}/hello.txt
                                    cat $TMPDIR/hello.txt
                                    whoami
                                    echo "USER: $USER"
                                 '''})

        (code, out, err) = self.adt_run(['--no-built-binaries',
                                         '--unbuilt-tree=' + p,
                                         '--user=nobody'],
                                        [self.schroot_name])
        self.assertEqual(code, 0, err)

        # should install dependencies
        self.assertIn('Unpacking aspell-doc', out)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cat and whoami
        self.assertRegex(out, '(^|\n)world\nroot\nUSER: root\n')

    def test_breaks_testbed(self):
        '''breaks-testbed restriction'''

        p = self.build_src('Tests: rambo\nDepends:\nRestrictions: needs-root breaks-testbed',
                           {'rambo': '#!/bin/sh\ntouch /zap\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.schroot_name])
        # test should be skipped as chroot runner doesn't provide revert
        self.assertEqual(code, 2)
        self.assertRegex(out, 'rambo\s+SKIP Test breaks testbed')

    def test_isolation(self):
        '''isolation restrictions'''

        p = self.build_src('Tests: ic\nDepends:\nRestrictions: isolation-container\n\n'
                           'Tests: im\nDepends:\nRestrictions: isolation-machine\n',
                           {'ic': '#!/bin/sh\necho container ok',
                            'im': '#!/bin/sh\necho machine ok'})

        (code, out, err) = self.adt_run(['-B', '--built-tree=' + p],
                                        [self.schroot_name])
        self.assertEqual(code, 2, out + err)
        self.assertRegex(out, 'ic\s+SKIP .*container', out)
        self.assertRegex(out, 'im\s+SKIP .*machine', out)
        self.assertNotIn('ok', out)

    def test_needs_recommends(self):
        '''needs-recommends restriction'''

        # check that testpkg's Recommends: cpp-doc gets installed
        p = self.build_src('Tests: pass\nRestrictions: needs-recommends',
                           {'pass': '#!/bin/sh -e\n/usr/bin/test_built; '
                            'test -e /usr/share/doc/cpp-doc/copyright ||'
                            'echo "cpp-doc not installed!" >&2'})

        (code, out, err) = self.adt_run(['--unbuilt-tree=' + p], [self.schroot_name])
        self.assertEqual(code, 0, out + err)
        self.assertIn('Unpacking cpp-doc', out)
        self.assertIn('built script OK', out)


@unittest.skipUnless('ADT_TEST_LXC' in os.environ,
                     'Set $ADT_TEST_LXC to an existing container')
class LxcRunner(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(LxcRunner, self).__init__('lxc', *args, **kwargs)
        self.container_name = os.environ.get('ADT_TEST_LXC')
        self.virt_args = ['--ephemeral', self.container_name]
        if os.getuid() > 0:
            self.virt_args.insert(0, '--sudo')

    def test_tree_norestrictions_nobuild_success(self):
        '''source tree, no build, no restrictions, test success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\n',
                           {'pass': '#!/bin/sh\necho I am fine; echo "SOMEVAR: >$SOMEVAR<"; echo "PATH: $PATH"\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        self.virt_args,
                                        env={'SOMEVAR': 'junk',
                                             'PATH': '/something/broken:' + os.environ.get('PATH', '')})
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '(^|\n)I am fine\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)
        # $SOMEVAR does not leak into container
        self.assertIn('\nSOMEVAR: ><\n', out)
        # modified $PATH does not leak into container
        self.assertNotIn('/something/broken', out)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_tree_norestrictions_nobuild_fail_on_exit(self):
        '''source tree, no build, no restrictions, test fails with non-zero'''

        p = self.build_src('Tests: nz\nDepends:\n',
                           {'nz': '#!/bin/sh\necho I am sick\nexit 7'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        self.virt_args)
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'nz\s+FAIL non-zero exit status 7')

        # should show test stdout
        self.assertRegex(out, '^I am sick\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_tree_norestrictions_nobuild_fail_on_stderr(self):
        '''source tree, no build, no restrictions, test fails with stderr'''

        p = self.build_src('Tests: se\nDepends:\n',
                           {'se': '#!/bin/sh\necho I am sick >&2'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        self.virt_args)
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'se\s+FAIL stderr: I am sick\n')

        # should show test stderr
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_tree_norestrictions_nobuild_fail_on_stderr_and_exit(self):
        '''source tree, no build, no restrictions, test fails with stderr+exit'''

        p = self.build_src('Tests: senz\nDepends:\n',
                           {'senz': '#!/bin/sh\necho I am sick >&2\nexit 7\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        self.virt_args)
        # test should fail
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'senz\s+FAIL non-zero exit status 7')

        # should show test stderr separately (no real-time output for chroot)
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_tree_allow_stderr_nobuild_success(self):
        '''source tree, no build, allow-stderr, test success'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        self.virt_args)
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout/err
        self.assertRegex(out, '^babble\n')
        # stderr test output should be inline
        self.assertIn('\nI am fine\n', err)
        self.assertNotIn(' stderr ', err)

    def test_tree_build_needed_success(self):
        '''source tree, build-needed restriction, test success'''

        p = self.build_src('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\n./test_built | grep -q "built script OK"\necho GOOD'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        self.virt_args)
        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should build package
        self.assertIn('dh build', err)

        # should show test stdout
        self.assertRegex(out, '(^|\n)GOOD\n')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_dsc_build_needed_success(self):
        '''dsc, build-needed restriction, test success'''

        p = self.build_dsc('Tests: pass\nDepends: coreutils\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\n./test_built | grep -q "built script OK"\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', p],
                                        self.virt_args)
        # test should succeed
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertEqual(code, 0, err)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should build package
        self.assertIn('dh build', err)

    def test_tree_built_binaries(self):
        '''source tree, install built binaries'''

        # no "build-needed" restriction here, it should be built because we
        # need to install the package as a dependency (@)
        p = self.build_src('Tests: pass\n',
                           {'pass': '#!/bin/sh -e\n/usr/bin/test_built | grep -q "built script OK"\necho GOOD'})

        # add extra dependency to testpkg, to ensure autopkgtest does not stop
        # and ask for confirmation in apt-get
        subprocess.check_call(['sed', '-i', '/^Depends:/ s/$/, aspell-doc/',
                               os.path.join(p, 'debian', 'control')])

        (code, out, err) = self.adt_run(['--unbuilt-tree=' + p],
                                        self.virt_args)

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should build and install package
        self.assertIn('dh build', err)
        self.assertIn('binaries/./testpkg.deb', out)
        self.assertRegex(out, '\.\.\./aspell-doc_\d.*_all.deb')

        # should show test stdout
        self.assertRegex(out, '(^|\n)GOOD\n', out)
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

    def test_tree_clone(self):
        '''source tree, no --ephemeral option (using clone)'''

        p = self.build_src('Tests: senz\nDepends:\n',
                           {'senz': '#!/bin/sh\necho hello\necho I am sick >&2\nexit 7\n'})

        virt_args = [a for a in self.virt_args if a != '--ephemeral']
        (code, out, err) = self.adt_run(['--no-built-binaries', '--built-tree=' + p],
                                        virt_args)
        # test should fail
        self.assertEqual(code, 4, err)
        self.assertRegex(out, 'senz\s+FAIL non-zero exit status 7')

        # should show test stdout inline
        self.assertRegex(out, '^hello\n')

        # should show test stderr separately
        self.assertRegex(err, 'stderr [ -]+\nI am sick\n')

    def test_breaks_testbed(self):
        '''breaks-testbed restriction'''

        p = self.build_src('Tests: zap boom\nDepends:\nRestrictions: needs-root breaks-testbed',
                           {'zap': '#!/bin/sh\ntouch /zap\n',
                            'boom': '#!/bin/sh\n[ ! -e /zap ]; touch /boom'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p, '-d'],
                                        self.virt_args)
        # both tests should run; the second one (boom) should not see the
        # effect of the first one (/zap existing)
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'zap\s+PASS')
        self.assertRegex(out, 'boom\s+PASS')

    def test_setup_commands(self):
        '''--setup-commands'''

        p = self.build_src('Tests: t1 t2\nDepends:\n',
                           {'t1': '#!/bin/sh -e\ncat /setup.log; echo t1ok\n',
                            't2': '#!/bin/sh -e\ncat /setup.log; echo t2ok\n'})

        # we expect a normal user for LXC
        (code, out, err) = self.adt_run(['-B', '-d', p + '//', '--setup-commands',
                                         '[ -n "$ADT_NORMAL_USER" ]; getent passwd $ADT_NORMAL_USER;'
                                         'echo setupok >> /setup.log'],
                                        self.virt_args)

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't1\s+PASS', out)
        self.assertRegex(out, 't2\s+PASS', out)

        # should show test stdout
        self.assertRegex(out, '^\w+:x:.*\nsetupok\nt1ok\n')
        self.assertIn('\nsetupok\nt2ok\n', out)
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not have rebooted for these commands
        self.assertIn('@@@@@@ test bed setup', err)
        self.assertNotIn('rebooting testbed', err)

    def test_isolation(self):
        '''isolation restrictions'''

        p = self.build_src('Tests: ic\nDepends:\nRestrictions: isolation-container\n\n'
                           'Tests: im\nDepends:\nRestrictions: isolation-machine\n',
                           {'ic': '#!/bin/sh\necho container ok',
                            'im': '#!/bin/sh\necho machine ok'})

        (code, out, err) = self.adt_run(['-B', '--built-tree=' + p], self.virt_args)
        self.assertEqual(code, 2, out + err)
        self.assertRegex(out, 'ic\s+PASS', out)
        self.assertRegex(out, 'im\s+SKIP .*machine', out)
        self.assertIn('container ok\n', out)
        self.assertNotIn('machine ok', out)

    def test_user(self):
        '''Run tests as different user'''

        p = self.build_src('Tests: t\nDepends:\nRestrictions: rw-build-tree\n',
                           {'t': '''#!/bin/sh -e
                                    echo world > $ADTTMP/hello.txt
                                    cat $ADTTMP/hello.txt
                                    mkdir -p $ADTTMP/one/subdir
                                    echo deep > $ADTTMP/one/subdir/subdir.txt
                                    cat $ADTTMP/one/subdir/subdir.txt
                                    rm -r $ADTTMP/one
                                    echo dir_is_rw > rw_flag
                                    cat rw_flag
                                    whoami'''})

        (code, out, err) = self.adt_run(['-B', p + '//', '--user=nobody'],
                                        self.virt_args)
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cats and whoami
        self.assertIn('world\ndeep\ndir_is_rw\nnobody\n', out)

    def test_no_locale_conf(self):
        '''No /etc/default/locale'''

        p = self.build_src('Test-Command: locale; echo TestOK\nDepends:\n', {})

        # we expect a normal user for LXC
        (code, out, err) = self.adt_run(['-B', p + '//', '--setup-commands',
                                         'rm -f /etc/default/locale /etc/environment'],
                                        self.virt_args)

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'command1\s+PASS')
        self.assertRegex(out, 'LC_MESSAGES="C.UTF-8"')
        self.assertRegex(out, 'TestOK')
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)


@unittest.skipUnless('ADT_TEST_QEMU' in os.environ,
                     'Set $ADT_TEST_QEMU to an existing autopkgtest QEMU image')
class QemuRunner(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(QemuRunner, self).__init__('qemu', *args, **kwargs)
        self.image = os.environ.get('ADT_TEST_QEMU')

    def test_tree_norestrictions_nobuild_success(self):
        '''source tree, no build, no restrictions, test success'''

        p = self.build_src('Tests: pass\nDepends: aspell-doc\n',
                           {'pass': '#!/bin/sh -e\necho I am fine\n'
                            '[ -d "$HOME" ] || echo "HOME not set" >&2\n'
                            '[ -n "$USER" ] && [ "$USER" != "root" ] || echo "USER invalid: $USER" >&2\n'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p],
                                        [self.image])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertIn('Unpacking aspell-doc', out)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertIn('I am fine\n', out)
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not build package
        self.assertNotIn('dh build', err)

    def test_dsc_build_needed_success(self):
        '''dsc, build-needed restriction, test success'''

        p = self.build_dsc('Tests: pass\nRestrictions: build-needed\n',
                           {'pass': '#!/bin/sh -e\ntest_built\n./test_built\n'
                            'grep ^processor /proc/cpuinfo  |wc -l\n'})

        (code, out, err) = self.adt_run([p], ['-c', '2', self.image])

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout
        self.assertIn('built script OK\nbuilt script OK\n2', out)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should build package
        self.assertIn('dh build', err)

    def test_breaks_testbed(self):
        '''breaks-testbed restriction'''

        p = self.build_src('Tests: zap boom\nDepends:\nRestrictions: needs-root breaks-testbed',
                           {'zap': '#!/bin/sh\ntouch /zap\n'
                            '[ -d "$HOME" ] || echo "HOME not set" >&2\n',
                            'boom': '#!/bin/sh\n[ ! -e /zap ]; touch /boom'})

        (code, out, err) = self.adt_run(['--no-built-binaries', '--unbuilt-tree=' + p, '-d'],
                                        [self.image])
        # both tests should run; the second one (boom) should not see the
        # effect of the first one (/zap existing)
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'zap\s+PASS')
        self.assertRegex(out, 'boom\s+PASS')

    def test_setup_commands(self):
        '''--setup-commands'''

        p = self.build_src('Tests: t1 t2\nDepends:\n',
                           {'t1': '#!/bin/sh -e\ncat /setup.log; echo t1ok\n',
                            't2': '#!/bin/sh -e\ncat /setup.log; echo t2ok\n'})

        (code, out, err) = self.adt_run(['-B', '-d', p + '//',
                                         '--setup-commands', 'echo setupok >> /setup.log'],
                                        [self.image])

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't1\s+PASS', out)
        self.assertRegex(out, 't2\s+PASS', out)

        # should show test stdout
        self.assertIn('setupok\nt1ok\n', out)
        self.assertIn('\nsetupok\nt2ok\n', out)
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not have rebooted for these commands
        self.assertIn('@@@@@@ test bed setup', err)
        self.assertNotIn('rebooting testbed', err)

    def test_setup_commands_reboot(self):
        '''--setup-commands with reboot'''

        p = self.build_src('Tests: t\nDepends:\n\nTests: r\nDepends:\nRestrictions: needs-root\n',
                           {'t': '#!/bin/sh -e\n[ -d "$HOME" ] || echo "HOME not set" >&2\n',
                            'r': '#!/bin/sh -e\n[ -d "$HOME" ] || echo "HOME not set" >&2\n'
                                 '[ "$USER" = "root" ] || echo "USER invalid: $USER" >&2\n'})

        (code, out, err) = self.adt_run(['-B', '-d', p + '//', '--setup-commands',
                                         'apt-get install --reinstall -y cron'],
                                        [self.image])

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't\s+PASS', out)
        self.assertRegex(out, 'r\s+PASS', out)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should have rebooted for these commands
        self.assertIn('@@@@@@ test bed setup', err)
        self.assertIn('rebooting testbed', err)

    def test_setup_commands_suppress_reboot(self):
        '''--setup-commands with suppressed reboot'''

        p = self.build_src('Tests: t\nDepends:\n',
                           {'t': '#!/bin/sh -e\ntest -e /run/autopkgtest_no_reboot.stamp'})

        (code, out, err) = self.adt_run(['-B', '-d', p + '//', '--setup-commands',
                                         'apt-get install --reinstall -y cron; touch /run/autopkgtest_no_reboot.stamp'],
                                        [self.image])

        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't\s+PASS', out)

        # should not have any test stderr
        self.assertNotIn(' stderr ', err)

        # should not have rebooted
        self.assertIn('@@@@@@ test bed setup', err)
        self.assertNotIn('rebooting testbed', err)

    def test_isolation(self):
        '''isolation restrictions'''

        p = self.build_src('Tests: ic\nDepends:\nRestrictions: isolation-container\n\n'
                           'Tests: im\nDepends:\nRestrictions: isolation-machine\n',
                           {'ic': '#!/bin/sh\necho container ok',
                            'im': '#!/bin/sh\necho machine ok'})

        (code, out, err) = self.adt_run(['-B', '--built-tree=' + p], [self.image])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'ic\s+PASS', out)
        self.assertRegex(out, 'im\s+PASS', out)
        self.assertIn('container ok\n', out)
        self.assertIn('\nmachine ok\n', out)

    def test_default_user(self):
        '''test runs as user by default'''

        p = self.build_src('Tests: t\nDepends:\n',
                           {'t': '''#!/bin/sh -e
                                    echo world > $ADTTMP/hello.txt
                                    cat $ADTTMP/hello.txt
                                    mkdir -p $ADTTMP/one/subdir
                                    echo deep > $ADTTMP/one/subdir/subdir.txt
                                    cat $ADTTMP/one/subdir/subdir.txt
                                    rm -r $ADTTMP/one
                                    whoami'''})

        (code, out, err) = self.adt_run(['-B', p + '//'], [self.image])
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cats
        self.assertIn('world\ndeep', out)
        # check user
        whoami = out.split('world\n', 1)[1].split('\n', 1)[0]
        self.assertNotEqual(whoami, 'root')

    def test_adt_run_user(self):
        '''adt-run --user option'''

        p = self.build_src('Tests: t\nDepends:\nRestrictions: rw-build-tree\n',
                           {'t': '''#!/bin/sh -e
                                    echo world > $ADTTMP/hello.txt
                                    cat $ADTTMP/hello.txt
                                    mkdir -p $ADTTMP/one/subdir
                                    echo deep > $ADTTMP/one/subdir/subdir.txt
                                    cat $ADTTMP/one/subdir/subdir.txt
                                    rm -r $ADTTMP/one
                                    echo dir_is_rw > rw_flag
                                    cat rw_flag
                                    whoami'''})

        (code, out, err) = self.adt_run(['-B', p + '//', '--user=nobody'], [self.image])
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cats and whoami
        self.assertIn('world\ndeep\ndir_is_rw\nnobody\n', out)

    def test_virt_user(self):
        '''adt-virt-qemu --user option'''

        p = self.build_src('Tests: t\nDepends:\n',
                           {'t': '''#!/bin/sh -e
                                    echo world > $ADTTMP/hello.txt
                                    cat $ADTTMP/hello.txt
                                    mkdir -p $ADTTMP/one/subdir
                                    echo deep > $ADTTMP/one/subdir/subdir.txt
                                    cat $ADTTMP/one/subdir/subdir.txt
                                    rm -r $ADTTMP/one
                                    whoami'''})

        (code, out, err) = self.adt_run(['-B', p + '//'], ['-u', 'nobody', self.image])
        self.assertEqual(code, 0, err)

        # test should succeed
        self.assertRegex(out, 't\s+PASS', out)

        # has output from cats and whoami
        self.assertIn('world\ndeep\nnobody\n', out)

    def test_copy_timeout(self):
        '''handling copying timeout'''

        p = self.build_src('Tests: to\nDepends:\n',
                           {'to': '#!/bin/sh\ntruncate -s 10G $ADT_ARTIFACTS/huge\n'})

        outdir = os.path.join(self.workdir, 'out')
        (code, out, err) = self.adt_run(['--output-dir=' + outdir,
                                         '--timeout-copy=1',
                                         '-B', p + '//'],
                                        [self.image])

        # test should time out
        self.assertEqual(code, 16, err)
        self.assertRegex(out, 'to\s+PASS', out)

        # should have copied parts of the file
        huge_size = os.path.getsize(os.path.join(outdir, 'artifacts', 'huge'))
        self.assertGreater(huge_size, 100000)
        self.assertLess(huge_size, 10000000000)

        # we don't want the raw exception
        self.assertNotIn("Timeout", err)
        self.assertNotIn("Traceback", err)
        # but a proper error message, with cleanup handlers
        self.assertIn("got `timeout', expected `ok...'", err)

    def test_copy_performance(self):
        '''copying files between host and testbed is fast and reliable'''

        p = self.build_src('Tests: t\nDepends:\n',
                           {'t': '#!/bin/sh\ncp -r . $ADT_ARTIFACTS/tree\n'})
        # create ~ 100 MB of fake source tree (~ 1000 files with 100 kB each)
        block = b'0123456789' * 10000
        for dirnum in range(33):
            d = os.path.join(p, str(dirnum))
            os.mkdir(d)
            for filenum in range(33):
                with open(os.path.join(d, 'src%i' % filenum), 'wb') as f:
                    f.write(block)

        # we expect at least 20 MB/s, thus give it 6 seconds
        outdir = os.path.join(self.workdir, 'out')
        (code, out, err) = self.adt_run(['--output-dir=' + outdir,
                                         '--timeout-copy=6',
                                         '-B', p + '//'],
                                        [self.image])

        # test should succeed
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 't\s+PASS', out)

        # check integrity of the copy
        subprocess.check_call(['diff', '-Nur', p,
                               os.path.join(outdir, 'artifacts', 'tree')])

    def test_apt_libpng(self):
        '''apt, libpng'''

        (code, out, err) = self.adt_run(['-d', 'libpng'], [self.image])
        # test should succeed
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'Unpacking libpng.*-dev')
        self.assertRegex(out, 'build\s+PASS', out)

        # should show test stdout
        self.assertIn('OK\n', out)
        # should not have any test stderr
        self.assertNotIn(' stderr ', err)


@unittest.skipUnless('ADT_TEST_SCHROOT_CLICK' in os.environ,
                     'Set $ADT_TEST_SCHROOT_CLICK to an existing schroot')
class SchrootClickRunner(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(SchrootClickRunner, self).__init__('schroot', *args, **kwargs)
        self.schroot_name = os.environ.get('ADT_TEST_SCHROOT_CLICK')
        self.click = os.path.join(test_dir, 'testclick_0.1_all.click')
        self.click_src = os.path.join(test_dir, 'testclick')

    def test_click_local_source_implicit(self):
        '''click package with local source, implicit arg type'''

        outdir = os.path.join(self.workdir, 'out')
        os.mkdir(outdir)
        sumfile = os.path.join(self.workdir, 'summary')

        (code, out, err) = self.adt_run(
            ['-d', '-o', outdir, '--summary-file', sumfile,
             self.click_src, self.click],
            [self.schroot_name])
        self.assertEqual(code, 4, err)

        # one test (broken) should fail, the others succeed
        with open(sumfile) as f:
            summary = f.read()
        self.assertEqual(summary, '''broken               FAIL stderr: Bad Things!
inst                 PASS
serr                 PASS
shell                PASS
simple               PASS
''')

        # shell test should give expected output
        with open(os.path.join(outdir, 'shell-stdout')) as f:
            self.assertIn('root:x:0', f.read())
        with open(os.path.join(outdir, 'broken-stderr')) as f:
            self.assertEqual(f.read(), 'Bad Things!\n')

        # inst test has additional dependency
        with open(os.path.join(outdir, 'inst-packages')) as f:
            contents = f.read()
        self.assertTrue(contents.startswith('python3-evdev'), contents)

    def test_click_local_source_explicit(self):
        '''click package with local source, explicit arg type'''

        sumfile = os.path.join(self.workdir, 'summary')

        (code, out, err) = self.adt_run(
            ['-d', '--summary-file', sumfile,
             '--click-source', self.click_src, '--click', self.click],
            [self.schroot_name])
        self.assertEqual(code, 4, err)

        # one test (broken) should fail, the others succeed
        with open(sumfile) as f:
            summary = f.read()
        self.assertEqual(summary, '''broken               FAIL stderr: Bad Things!
inst                 PASS
serr                 PASS
shell                PASS
simple               PASS
''')

    def test_click_local_source_tmp_install(self):
        '''click package with local source, temp dir deps'''

        sumfile = os.path.join(self.workdir, 'summary')

        (code, out, err) = self.adt_run(
            ['-d', '--summary-file', sumfile, '--setup-commands',
             os.path.join(root_dir, 'setup-commands', 'ro-apt'),
             self.click_src, self.click],
            [self.schroot_name])
        self.assertEqual(code, 4, err)

        # one test (broken) should fail, the others succeed
        with open(sumfile) as f:
            summary = f.read()
        self.assertEqual(summary, '''broken               FAIL stderr: Bad Things!
inst                 PASS
serr                 PASS
shell                PASS
simple               PASS
''')

        self.assertIn('python3-evdev', err)

        # warn about restricted functionality
        self.assertIn('will only work for some packages', err)

    def test_click_preinstalled_all_users(self):
        '''already installed click package for all users, temp dir deps'''

        sumfile = os.path.join(self.workdir, 'summary')

        (code, out, err) = self.adt_run(
            ['-d', '--summary-file', sumfile,
             '--copy', '%s:/root/testclick.click' % self.click,
             '--setup-commands', 'click install --all-users /root/testclick.click',
             '--setup-commands',
             os.path.join(root_dir, 'setup-commands', 'ro-apt'),
             self.click_src, '--click', 'testclick'],
            [self.schroot_name])
        self.assertEqual(code, 4, err)

        # one test (broken) should fail, the others succeed
        with open(sumfile) as f:
            summary = f.read()
        self.assertEqual(summary, '''broken               FAIL stderr: Bad Things!
inst                 PASS
serr                 PASS
shell                PASS
simple               PASS
''')

    def test_click_preinstalled_user(self):
        '''already installed click package for one user, temp dir deps'''

        sumfile = os.path.join(self.workdir, 'summary')

        (code, out, err) = self.adt_run(
            ['-d', '--summary-file', sumfile,
             '--copy', '%s:/root/testclick.click' % self.click,
             '--setup-commands', 'click install --user $ADT_NORMAL_USER /root/testclick.click',
             '--setup-commands',
             os.path.join(root_dir, 'setup-commands', 'ro-apt'),
             self.click_src, '--click', 'testclick'],
            [self.schroot_name])
        self.assertEqual(code, 4, err)

        # one test (broken) should fail, the others succeed
        with open(sumfile) as f:
            summary = f.read()
        self.assertEqual(summary, '''broken               FAIL stderr: Bad Things!
inst                 PASS
serr                 PASS
shell                PASS
simple               PASS
''')

    def test_override_control(self):
        '''custom manifest with --override-control'''

        custom_manifest = os.path.join(self.workdir, 'adtctrl')
        with open(custom_manifest, 'w') as f:
            f.write('''{"name": "testclick",
   "x-test": { "simple": "tests/simple" }
}''')

        sumfile = os.path.join(self.workdir, 'summary')

        (code, out, err) = self.adt_run(
            ['--summary-file', sumfile,
             '--override-control', custom_manifest,
             '--click-source', self.click_src, '--click', self.click],
            [self.schroot_name])
        self.assertEqual(code, 0, err)

        with open(sumfile) as f:
            summary = f.read()
        self.assertEqual(summary, 'simple               PASS\n')

    def test_setup_ubuntu_touch_session(self):
        '''setup-commands/ubuntu-touch-session'''

        p = self.build_src(
            'Tests: t\nDepends: x11-utils',
            {'t': '#!/bin/sh -e\nOUT=$(xprop -root); echo "$OUT" | grep -q _XKB_RULES\n'
                  'JOBS=$(/sbin/initctl  --user list)\n'
                  'echo "$JOBS" | grep -q "dbus start/running"\n'
                  'echo "$JOBS" | grep -q "click-user-hooks"\n'
                  # FIXME: in schroot this spits out "Error opening shm /lttng-ust-wait-*"
                  'ubuntu-app-list 2>/dev/null\n'
             })

        (code, out, err) = self.adt_run(
            ['-B', p + '//', '--setup-commands',
             os.path.join(root_dir, 'setup-commands', 'ubuntu-touch-session')],
            [self.schroot_name])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 't\s+PASS', out)


@unittest.skipUnless('ADT_TEST_LXC' in os.environ,
                     'Set $ADT_TEST_LXC to an existing container')
class SshRunnerNoScript(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(SshRunnerNoScript, self).__init__('ssh', *args, **kwargs)
        self.info = {}

    def start_container(self, install_key=True, sudo=False, sudo_nopwd=False):
        '''Set up container with SSH'''

        cmd = [os.path.join(test_dir, 'ssh-setup-lxc')]
        if install_key:
            cmd.append('-k')
        if sudo_nopwd:
            cmd.append('-S')
        elif sudo:
            cmd.append('-s')
        cmd += ['open', os.environ.get('ADT_TEST_LXC')]
        out = subprocess.check_output(cmd, universal_newlines=True)
        for l in out.splitlines():
            (k, v) = l.split('=', 1)
            self.info[k] = v

    def tearDown(self):
        if self.info:
            subprocess.call([os.path.join(test_dir, 'ssh-setup-lxc'), 'cleanup']
                            + self.info['extraopts'].split())
            self.info = {}

    def test_no_root(self):
        '''no root'''

        self.start_container()

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr\n\n'
                           'Tests: rootonly\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble\nwhoami',
                            'rootonly': '#!/bin/sh\necho NOTME>&2; exit 1'})

        (code, out, err) = self.adt_run(
            ['-d', '-B', p + '//'],
            ['-d', '-H', self.info['hostname'], '-l', self.info['login'], '-i', self.info['identity']])
        self.assertEqual(code, 2, err)
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertRegex(out, 'rootonly\s+SKIP.*needs root', out)

        # should show test stdout/err
        self.assertIn('babble\nadt_test\n', out)
        self.assertIn('\nI am fine\n', err)
        self.assertRegex(err, 'adt-run \[[0-9: -]+\]: test pass: --')

        # no root
        self.assertIn('testbed capabilities: [', err)
        self.assertNotRegex(err, 'testbed capabilities:.*root-on-testbed')

    def test_with_root(self):
        '''with root'''

        self.start_container(sudo=True)

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\necho hello\nwhoami'})

        (code, out, err) = self.adt_run(
            ['-d', '-B', p + '//'],
            ['-d', '-H', self.info['hostname'], '-l', self.info['login'],
             '-i', self.info['identity'], '-P', self.info['password']])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'pass\s+PASS', out)

        self.assertIn('hello\nroot\n', out)

        # has root
        self.assertRegex(err, 'testbed capabilities: \[.*root-on-testbed')

    def test_passwordless_sudo(self):
        '''sudo without password'''

        self.start_container(sudo_nopwd=True)

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\necho hello\nwhoami'})

        (code, out, err) = self.adt_run(
            ['-d', '-B', p + '//'],
            ['-d', '-H', self.info['hostname'], '-l', self.info['login'], '-i', self.info['identity']])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'pass\s+PASS', out)

        self.assertIn('hello\nroot\n', out)

        # has root
        self.assertRegex(err, 'testbed capabilities: \[.*root-on-testbed')

    @unittest.skip('ssh password auth is not implemented')
    def test_password(self):
        '''no root, password auth'''

        self.start_container(install_key=False)

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble'})

        (code, out, err) = self.adt_run(
            ['-d', '-B', p + '//'],
            ['-d', '-H', self.info['hostname'], '-l', self.info['login'], '-P', self.info['password']])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout/err
        self.assertIn('babble\n', out)
        self.assertRegex(err, '-+\nI am fine\nadt-run \[[0-9: -]+\]: test pass: --')

        # no root
        self.assertIn('testbed capabilities: [', err)
        self.assertNotRegex(err, 'testbed capabilities:.*root-on-testbed')


@unittest.skipUnless('ADT_TEST_LXC' in os.environ,
                     'Set $ADT_TEST_LXC to an existing container')
class SshRunnerWithScript(AdtTestCase):
    def __init__(self, *args, **kwargs):
        super(SshRunnerWithScript, self).__init__('ssh', *args, **kwargs)
        self.container_template = os.environ.get('ADT_TEST_LXC')
        self.script = os.path.join(test_dir, 'ssh-setup-lxc')

    def test_no_root(self):
        '''no root'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr\n\n'
                           'Tests: rootonly\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble\nwhoami',
                            'rootonly': '#!/bin/sh\necho NOTME>&2; exit 1'})

        (code, out, err) = self.adt_run(['-d', '-B', p + '//'],
                                        ['-ds', self.script, '--', '-k', self.container_template])
        self.assertEqual(code, 2, err)
        self.assertRegex(out, 'pass\s+PASS', out)
        self.assertRegex(out, 'rootonly\s+SKIP.*needs root', out)

        # should show test stdout/err
        self.assertIn('babble\nadt_test\n', out)
        self.assertIn('\nI am fine\n', err)
        self.assertRegex(err, 'adt-run \[[0-9: -]+\]: test pass: --')

        # no root
        self.assertIn('testbed capabilities: [', err)
        self.assertNotRegex(err, 'testbed capabilities:.*root-on-testbed')

    def test_with_root(self):
        '''with root'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\necho hello\nwhoami'})

        (code, out, err) = self.adt_run(['-d', '-B', p + '//'],
                                        ['-ds', self.script, '--', '-ks', self.container_template])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'pass\s+PASS', out)

        self.assertIn('hello\nroot\n', out)

        # has root
        self.assertRegex(err, 'testbed capabilities: \[.*root-on-testbed')

    def test_passwordless_sudo(self):
        '''sudo without password'''

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: needs-root',
                           {'pass': '#!/bin/sh\necho hello\nwhoami'})

        (code, out, err) = self.adt_run(['-d', '-B', p + '//'],
                                        ['-ds', self.script, '--', '-kS', self.container_template])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'pass\s+PASS', out)

        self.assertIn('hello\nroot\n', out)

        # has root
        self.assertRegex(err, 'testbed capabilities: \[.*root-on-testbed')

    @unittest.skip('ssh password auth is not implemented')
    def test_password(self):
        '''no root, password auth'''

        self.start_container(install_key=False)

        p = self.build_src('Tests: pass\nDepends:\nRestrictions: allow-stderr',
                           {'pass': '#!/bin/sh\necho I am fine >&2\necho babble'})

        (code, out, err) = self.adt_run(['-d', '-B', p + '//'],
                                        ['-ds', self.script, '--', self.container_template])
        self.assertEqual(code, 0, err)
        self.assertRegex(out, 'pass\s+PASS', out)

        # should show test stdout/err
        self.assertIn('babble\n', out)
        self.assertRegex(err, '-+\nI am fine\nadt-run \[[0-9: -]+\]: test pass: --')

        # no root
        self.assertIn('testbed capabilities: [', err)
        self.assertNotRegex(err, 'testbed capabilities:.*root-on-testbed')

    def test_apt_source(self):
        '''apt source'''

        (code, out, err) = self.adt_run(['libpng'],
                                        ['-s', self.script, '--', '-ks', self.container_template])
        self.assertEqual(code, 0, out + err)
        self.assertRegex(out, 'build\s+PASS', out)

    def test_click_root(self):
        '''click source, with root'''

        env = os.environ.copy()
        env['ADT_CLICK_NO_FRAMEWORK_CHECK'] = '1'
        (code, out, err) = self.adt_run(
            ['-d',
             '--setup-commands', 'apt-get install -y --no-install-recommends click',
             os.path.join(test_dir, 'testclick'),
             os.path.join(test_dir, 'testclick_0.1_all.click')],
            ['-s', self.script, '--', '-sk', self.container_template],
            env=env)
        self.assertEqual(code, 4, err)

        # one test (broken) should fail, the others succeed
        self.assertRegex(out, 'inst\s+PASS')
        self.assertRegex(out, 'serr\s+PASS')
        self.assertRegex(out, 'shell\s+PASS')
        self.assertRegex(out, 'simple\s+PASS')
        self.assertRegex(out, 'broken\s+FAIL stderr: Bad Things!')

        self.assertIn('python3-evdev', err)


if __name__ == '__main__':
    # Force encoding to UTF-8 even in non-UTF-8 locales.
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding="UTF-8", line_buffering=True)
    unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout, verbosity=2))
