path: root/test/gvfs-test

#!/usr/bin/python3

__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>'
__copyright__ = '(C) 2012 Canonical Ltd.'
__license__ = 'LGPL v2 or later'

# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General
# Public License along with this library; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
# Boston, MA 02110-1301, USA.

import os
import os.path
import sys
import unittest
import subprocess
import tempfile
import tarfile
import zipfile
import time
import shutil
import fcntl
import re
import locale
from glob import glob

in_testbed = os.path.exists('/home/gvfs_sandbox_marker')
samba_running = subprocess.call(['pidof', 'smbd'], stdout=subprocess.PIPE) == 0
have_httpd = subprocess.call(['which', 'apachectl'], stdout=subprocess.PIPE) == 0
sshd_path = subprocess.check_output(['which', 'sshd'], universal_newlines=True).strip()

local_ip = subprocess.check_output("ip -4 addr | sed -nr '/127\.0\.0/ n; "
                                   "/inet / {  s/^.*inet ([0-9.]+).*$/\\1/; p; q  }'",
                                   shell=True, universal_newlines=True)

my_dir = os.path.dirname(os.path.abspath(__file__))

# we need this flag to check if we can test error messages
locale.setlocale(locale.LC_ALL, '')
lc = locale.getlocale(locale.LC_MESSAGES)[0]
english_messages = not lc or lc.startswith('en_')

# http://sg.danny.cz/sg/sdebug26.html
PTYPE_DISK = 0 
PTYPE_CDROM = 5 

class GvfsTestCase(unittest.TestCase):
    '''Gvfs tests base class.

    Provide some utility functions and a temporary work dir.
    '''
    def setUp(self):
        self.workdir = tempfile.mkdtemp()

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

    def program_code_out_err(self, argv):
        '''Return (exitcode, stdout, stderr) from a program call.'''

        prog = subprocess.Popen(argv, stdout=subprocess.PIPE,
                stderr=subprocess.PIPE, universal_newlines=True)
        (out, err) = prog.communicate()
        return (prog.returncode, out, err)

    def program_out_err(self, argv):
        '''Return (stdout, stderr) from a program call.'''

        (code, out, err) = self.program_code_out_err(argv)
        self.assertEqual(code, 0, err)
        return (out, err)

    def program_out_success(self, argv):
        '''Return stdout from a successful program call.'''

        (out, err) = self.program_out_err(argv)
        self.assertEqual(err, '', err)
        return out

    @classmethod
    def root_command(klass, command):
        '''Run a shell command string as root.

        This only works when running under gvfs-testbed.

        Return (code, stdout, stderr).
        '''
        assert in_testbed, 'root_command() only works under gvfs-testbed'

        rootsh = subprocess.Popen(['./rootsh'], stdin=subprocess.PIPE,
                                  stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                  universal_newlines=True)
        # set reasonable path that includes /sbin
        rootsh.stdin.write('export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n')

        (out, err) = rootsh.communicate(command.encode('UTF-8'))
        return (rootsh.returncode, out, err)

    @classmethod
    def root_command_success(klass, command):
        '''root_command() for commands that should succeed without errors.'''

        (code, out, err) = klass.root_command(command)
        if code != 0:
            raise SystemError('command "%s" failed with code %i:\n%s' % (code, err))
        if err:
            raise SystemError('command "%s" produced error:\n%s' % err)

    def unmount(self, uri):
        self.program_out_success(['gvfs-mount', '-u', uri])

        timeout = 5
        while timeout > 0:
            (out, err) = self.program_out_err(['gvfs-mount', '-li'])
            if 'Mount(0)' not in out:
                break
            timeout -= 1
            time.sleep(1)
        else:
            self.fail('gvfs-mount -u %s failed' % uri)

    @classmethod
    def quote(klass, path):
        '''Quote a path for GIO URLs'''

        return path.replace('%', '%25').replace('/', '%2F').replace(':', '%3A')

    def wait_for_gvfs_mount_user_prompt(self, popen):
        '''Wait for a gvfs-mount Popen process to show an User auth prompt'''

        empty_timeout = 50
        while True:
            r = popen.stdout.read(1000)
            if not r:
                empty_timeout -= 1
                self.assertGreater(empty_timeout, 0,
                                   'timed out waiting for auth prompt')

            if b'User' in r or b'Domain' in r:
                break
            time.sleep(0.1)

class Programs(GvfsTestCase):
    '''Test gvfs-* programs'''

    def test_gvfs_info_filesystem(self):
        '''gvfs-info --filesystem'''

        out = self.program_out_success(['gvfs-info', '-f', '/'])
        self.assertTrue('filesystem::size:' in out, out)
        self.assertTrue('filesystem::type:' in out, out)

class ArchiveMounter(GvfsTestCase):
    def test_tar(self):
        '''archive:// for tar'''

        tar_path = os.path.join(self.workdir, 'stuff.tar')
        tf = tarfile.open(tar_path, 'w')
        tf.add(__file__, 'gvfs-test.py')
        tf.close()

        self.do_test_for_archive(tar_path)

    def test_tar_gz(self):
        '''archive:// for tar.gz'''

        tar_path = os.path.join(self.workdir, 'stuff.tar.gz')
        tf = tarfile.open(tar_path, 'w:gz')
        tf.add(__file__, 'gvfs-test.py')
        tf.close()

        self.do_test_for_archive(tar_path)

    def test_tar_bz2(self):
        '''archive:// for tar.bz2'''

        tar_path = os.path.join(self.workdir, 'stuff.tar.bz2')
        tf = tarfile.open(tar_path, 'w:bz2')
        tf.add(__file__, 'gvfs-test.py')
        tf.close()

        self.do_test_for_archive(tar_path)

    def test_zip(self):
        '''archive:// for .zip'''

        zip_path = os.path.join(self.workdir, 'stuff.zip')
        zf = zipfile.ZipFile(zip_path, 'w')
        zf.write(__file__, 'gvfs-test.py')
        zf.close()

        self.do_test_for_archive(zip_path)

    def test_iso_rr(self):
        '''archive:// for RockRidge .iso'''

        shutil.copy(__file__, os.path.join(self.workdir, 'gvfs-test.py'))
        subprocess.check_call(['genisoimage', '-R', '-quiet', '-o', 'stuff.iso', 'gvfs-test.py'],
                             cwd=self.workdir)
        self.do_test_for_archive(os.path.join(self.workdir, 'stuff.iso'))

    def test_iso_joliet(self):
        '''archive:// for Joliet .iso'''

        shutil.copy(__file__, os.path.join(self.workdir, 'gvfs-test.py'))
        iso_path = os.path.join(self.workdir, 'stuff.iso')
        subprocess.check_call(['genisoimage', '-JR', '-quiet', '-o', 'stuff.iso', 'gvfs-test.py'],
                             cwd=self.workdir)
        self.do_test_for_archive(os.path.join(self.workdir, 'stuff.iso'))

    def do_test_for_archive(self, path):
        # mount it; yes, gvfs expects double quoting
        uri = 'archive://' + self.quote(self.quote('file://' + path))
        subprocess.check_call(['gvfs-mount', uri])

        # appears in gvfs-mount list
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        try:
            self.assertTrue('Mount(0)' in out, out)
            self.assertTrue('%s -> %s' % (os.path.basename(path), uri) in out, out)

            # check gvfs-info
            out = self.program_out_success(['gvfs-info', uri])
            self.assertTrue('standard::content-type: inode/directory' in out, out)
            self.assertTrue('access::can-read: TRUE' in out, out)

            # check gvfs-cat
            out = self.program_out_success(['gvfs-cat', uri + '/gvfs-test.py'])
            with open(__file__) as f:
                self.assertEqual(out, f.read())
        finally:
            self.unmount(uri)

@unittest.skipUnless(os.getenv('XDG_RUNTIME_DIR'), 'No $XDG_RUNTIME_DIR available')
class Sftp(GvfsTestCase):
    def setUp(self):
        '''Run ssh server'''

        super().setUp()

        self.assertTrue(os.path.exists(os.path.expanduser('~/.ssh/id_rsa')),
                        'This test needs an existing ~/.ssh/id_rsa')

        # find sftp-server
        for dir in ['/usr/local/lib/openssh',
                    '/usr/lib/openssh',
                    '/usr/local/libexec/openssh',
                    '/usr/libexec/openssh']:
            sftp_server = os.path.join(dir, 'sftp-server')
            if os.path.exists(sftp_server):
                break
        else:
            self.fail('Cannot locate OpenSSH sftp-server program, please update tests for your distribution')

        # look for authorized_keys in a temporary dir, to avoid having to mess
        # with the actual user files when not calling this through gvfs-testbed
        # (unfortunately ssh doesn't consider $HOME); NB we cannot use
        # self.workdir as ssh refuses files in /tmp.
        self.authorized_keys = os.path.join(os.environ['XDG_RUNTIME_DIR'],
                                            'gvfs_test_authorized_keys')

        # generate sshd configuration; note that we must ensure that the
        # private key is not world-readable, so we need to copy it
        shutil.copy(os.path.join(my_dir, 'files', 'ssh_host_rsa_key'), self.workdir)
        os.chmod(os.path.join(self.workdir, 'ssh_host_rsa_key'), 0o600)
        shutil.copy(os.path.join(my_dir, 'files', 'ssh_host_rsa_key.pub'), self.workdir)
        self.sshd_config = os.path.join(self.workdir, 'sshd_config')
        with open(self.sshd_config, 'w') as f:
            f.write('''Port 22222
HostKey %(workdir)s/ssh_host_rsa_key
UsePrivilegeSeparation no
AuthorizedKeysFile %(authorized_keys)s
UsePam no
Subsystem sftp %(sftp_server)s
''' % {'workdir': self.workdir,
       'sftp_server': sftp_server,
       'authorized_keys': self.authorized_keys,
      })

        self.sshd = subprocess.Popen([sshd_path, '-Dde', '-f', self.sshd_config],
                                     universal_newlines=True,
                                     stderr=subprocess.PIPE)

    def tearDown(self):
        if self.sshd.returncode is None:
            self.sshd.terminate()
            self.sshd.wait()
        super().tearDown()

    def run(self, result=None):
        '''Show sshd log output on failed tests'''

        if result:
            orig_err_fail = result.errors + result.failures
        super().run(result)
        if result and result.errors + result.failures > orig_err_fail:
            print('\n----- sshd log -----\n%s\n------\n' % self.sshd.stderr.read())

    def test_rsa(self):
        '''sftp://localhost with RSA authentication'''

        # accept our key for localhost logins
        shutil.copy(os.path.expanduser('~/.ssh/id_rsa.pub'), self.authorized_keys)

        # mount it
        uri = 'sftp://localhost:22222'
        subprocess.check_call(['gvfs-mount', uri])

        self.do_mount_check(uri)

    @unittest.skipUnless(local_ip, 'not having any non-localhost IP')
    def test_unknown_host(self):
        '''sftp:// with RSA authentication for unknown host'''

        # accept our key for localhost logins
        shutil.copy(os.path.expanduser('~/.ssh/id_rsa.pub'), self.authorized_keys)

        # try to mount it; should fail as it's an unknown host
        uri = 'sftp://%s:22222' % local_ip
        (code, out, err) = self.program_code_out_err(['gvfs-mount', uri])

        self.assertNotEqual(code, 0)
        # there is nothing in our testbed which would show or answer the
        # dialog
        if english_messages:
            self.assertTrue('Login dialog cancelled' in err, err)

    def do_mount_check(self, uri):
        with open('stuff.txt', 'w') as f:
            f.write('moo!')

        # appears in gvfs-mount list
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        try:
            self.assertRegex(out, 'Mount\(\d+\):.*localhost -> %s' % uri)

            # check gvfs-info
            out = self.program_out_success(['gvfs-info', uri])
            self.assertRegex(out, 'display name: / .* localhost')
            self.assertTrue('type: directory' in out, out)
            self.assertTrue('access::can-read: TRUE' in out, out)

            # check gvfs-ls
            out = self.program_out_success(['gvfs-ls', uri + '/home'])
            self.assertTrue('%s\n' % os.environ['USER'] in out, out)

            # check gvfs-cat
            out = self.program_out_success(['gvfs-cat', uri + '/etc/passwd'])
            self.assertTrue('root:' in out, out)
        finally:
            self.unmount(uri)

class Ftp(GvfsTestCase):
    def setUp(self):
        '''Launch FTP server'''

        super().setUp()
        with open(os.path.join(self.workdir, 'myfile.txt'), 'w') as f:
            f.write('hello world\n')
        os.mkdir(os.path.join(self.workdir, 'mydir'))
        secret_path = os.path.join(self.workdir, 'mydir', 'onlyme.txt')
        with open(secret_path, 'w') as f:
            f.write('secret\n')
        os.chmod(secret_path, 0o600)

        self.ftpd = subprocess.Popen(['twistd', '-n', 'ftp', '-p', '2121',
                                      '-r', self.workdir,
                                      '--auth', 'memory:testuser:pwd1'],
                                     stdout=subprocess.PIPE)
        # give ftp server some time to start up
        time.sleep(0.5)
 
    def tearDown(self):
        '''Shut down FTP server'''

        self.ftpd.terminate()
        self.ftpd.wait()
        super().tearDown()

    def test_anonymous(self):
        '''ftp:// anonymous'''

        uri = 'ftp://anonymous@localhost:2121'
        subprocess.check_call(['gvfs-mount', uri])

        self.do_mount_check(uri, True)

    def test_authenticated(self):
        '''ftp:// authenticated'''

        uri = 'ftp://localhost:2121'
        mount = subprocess.Popen(['gvfs-mount', uri],
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)

        # wrong user name
        self.wait_for_gvfs_mount_user_prompt(mount)
        mount.stdin.write(b'eve\nh4ck\n')

        # wrong password name
        self.wait_for_gvfs_mount_user_prompt(mount)
        mount.stdin.write(b'testuser\nh4ck\n')

        # correct credentials
        self.wait_for_gvfs_mount_user_prompt(mount)
        (out, err) = mount.communicate(b'testuser\npwd1\n')
        self.assertEqual(mount.returncode, 0)
        self.assertEqual(err, b'')

        # in test bed, there is nothing interesting in /home/testuser/, and
        # without the test bed we do not know what's in the folder, so skip
        # gvfs-ls check
        self.do_mount_check(uri, False)

    def do_mount_check(self, uri, check_contents):
        # appears in gvfs-mount list
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        try:
            self.assertRegex(out, 'Mount\(\d+\):.* -> ftp://([a-z0-9]+@)?localhost:2121')

            # check gvfs-info
            out = self.program_out_success(['gvfs-info', uri])
            self.assertRegex(out, 'display name: / .* localhost', out)
            self.assertTrue('type: directory' in out, out)

            # check gvfs-ls
            if check_contents:
                out = self.program_out_success(['gvfs-ls', uri])
                self.assertEqual(set(out.split()), set(['myfile.txt', 'mydir']))
                out = self.program_out_success(['gvfs-ls', uri + '/mydir'])
                self.assertEqual(out, 'onlyme.txt\n')

                # check gvfs-cat
                out = self.program_out_success(['gvfs-cat', uri + '/myfile.txt'])
                self.assertEqual(out, 'hello world\n')
        finally:
            self.unmount(uri)

@unittest.skipUnless(in_testbed or 'LIBSMB_PROG' in os.environ,
                     'not running under gvfs-testbed or LIBSMB_PROG="nc localhost 1445"')
class Smb(GvfsTestCase):
    def setUp(self):
        '''start local smbd as user if we are not in test bed'''
        super().setUp()

        # create a few test files
        if in_testbed:
            pubdir = os.path.expanduser('~/public')
            privdir = os.path.expanduser('~/private')
        else:
            pubdir = os.path.join(self.workdir, 'public')
            privdir = os.path.join(self.workdir, 'private')
        if not os.path.exists(pubdir):
            # only run this once
            os.mkdir(pubdir)
            os.makedirs(os.path.join(privdir, 'mydir'))
            with open(os.path.join(pubdir, 'myfile.txt'), 'w') as f:
                f.write('hello world\n')
            secret_path = os.path.join(privdir, 'mydir', 'onlyme.txt')
            with open(secret_path, 'w') as f:
                f.write('secret\n')
            os.chmod(secret_path, 0o600)

        if in_testbed:
            return

        # smbpasswd file with password "foo"
        smbpasswd = os.path.join(self.workdir, 'smbpasswd')
        with open(smbpasswd, 'w') as f:
            f.write(os.environ['USER'])
            f.write(':2:XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX:AC8E657F83DF82BEEA5D43BDAF7800CC:[U          ]:LCT-507C14C7:\n')
        
        # create local samba configuration
        smbdir = os.path.join(self.workdir, 'samba')
        os.mkdir(smbdir, 0o755)
        smbconf = os.path.join(self.workdir, 'smb.conf')
        with open(smbconf, 'w') as f:
            f.write('''[global]
workgroup = TESTGROUP
interfaces = lo 127.0.0.0/8
smb ports = 1445
log level = 2
map to guest = Bad User
passdb backend = smbpasswd
smb passwd file = %(workdir)s/smbpasswd
lock directory = %(workdir)s/samba
state directory = %(workdir)s/samba
cache directory = %(workdir)s/samba
pid directory = %(workdir)s/samba
private directory = %(workdir)s/samba
ncalrpc dir = %(workdir)s/samba

[public]
  path = %(workdir)s/public
  guest ok = yes

[private]
  path = %(workdir)s/private
  read only = no
''' % {'workdir': self.workdir})

        # start smbd
        self.smbd = subprocess.Popen(['smbd', '-iFS', '-s', smbconf],
                                     universal_newlines=True,
                                     stdout=subprocess.PIPE)
        time.sleep(1)

    def tearDown(self):
        # stop smbd
        if hasattr(self, 'smbd') and self.smbd.returncode is None:
            self.smbd.terminate()
            self.smbd.wait()
        super().tearDown()

    def run(self, result=None):
        '''Show smbd log output on failed tests'''

        if result:
            orig_err_fail = result.errors + result.failures
        super().run(result)
        if hasattr(self, 'smbd'):
            if result and result.errors + result.failures > orig_err_fail:
                print('\n----- smbd log -----\n%s\n------\n' % self.smbd.stdout.read())

    def test_anonymous(self):
        '''smb:// anonymous'''

        uri = 'smb://%s/public' % os.uname()[1]

        # ensure that this does not ask for any credentials
        mount = subprocess.Popen(['gvfs-mount', uri])
        timeout = 50
        while timeout > 0:
            time.sleep(0.1)
            timeout -= 1
            if mount.poll() is not None:
                self.assertEqual(mount.returncode, 0, 'gvfs-mount %s failed' % uri)
                break
        else:
            mount.terminate()
            self.fail('timed out waiting for gvfs-mount %s' % uri)
        mount.wait()

        self.do_mount_check(uri, False)

    def test_authenticated(self):
        '''smb:// authenticated'''

        uri = 'smb://%s@%s/private' % (os.environ['USER'], os.uname()[1])
        mount = subprocess.Popen(['gvfs-mount', uri],
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)

        # correct credentials
        self.wait_for_gvfs_mount_user_prompt(mount)
        # default domain, password
        (out, err) = mount.communicate(b'\nfoo\n')
        self.assertEqual(mount.returncode, 0, err)
        #self.assertEqual(err, b'') # we get some warnings

        self.do_mount_check(uri, True)

    def do_mount_check(self, uri, writable):
        sharename = uri.split('/')[-1]

        # appears in gvfs-mount list
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        try:
            self.assertRegex(out, 'Mount\(0\): %s .* smb://.*/%s' % (sharename, sharename))

            # check gvfs-info
            out = self.program_out_success(['gvfs-info', uri])
            self.assertTrue('display name: ' + sharename in out, out)
            self.assertTrue('type: directory' in out, out)

            # check gvfs-ls and gvfs-cat
            out = self.program_out_success(['gvfs-ls', uri])
            if sharename == 'public':
                self.assertEqual(out, 'myfile.txt\n')

                out = self.program_out_success(['gvfs-cat', uri + '/myfile.txt'])
                self.assertEqual(out, 'hello world\n')
            else:
                self.assertEqual(out, 'mydir\n')
                self.assertEqual(self.program_out_success(['gvfs-ls', uri + '/mydir']),
                                 'onlyme.txt\n')

                out = self.program_out_success(['gvfs-cat', uri + '/mydir/onlyme.txt'])
                self.assertEqual(out, 'secret\n')

            if writable:
                # should be writable
                self.program_out_success(['gvfs-copy', '/etc/passwd', uri + '/newfile.txt'])
                out = self.program_out_success(['gvfs-cat', uri + '/newfile.txt'])
                with open('/etc/passwd') as f:
                    self.assertEqual(out, f.read())
            else:
                # should not be writable
                (code, out, err) = self.program_code_out_err(
                    ['gvfs-copy', '/etc/passwd', uri + '/newfile.txt'])
                self.assertNotEqual(code, 0)
                self.assertEqual(out, '')
                self.assertNotEqual(err, '')
        finally:
            self.unmount(uri)

@unittest.skipUnless(in_testbed, 'not running under gvfs-testbed')
@unittest.skipIf(os.path.exists('/sys/module/scsi_debug'), 'scsi_debug is already loaded')
class Drive(GvfsTestCase):
    @classmethod
    def setUpClass(klass):
        '''Load scsi_debug and put a simple .iso into it'''

        # generate a test .iso
        test_iso = 'test.iso'
        subprocess.check_call(['genisoimage', '-R', '-quiet', '-V', 'bogus-cd', '-o',
                               test_iso, '/etc/passwd', __file__])

        # we cannot write to a scsi_debug CD drive, so write it into it in hard
        # disk mode
        klass.root_command_success('modprobe scsi_debug add_host=0 dev_size_mb=64')
        dev = klass.create_host(PTYPE_DISK)

        # put test.iso onto disk
        klass.root_command_success('cat %s > /dev/%s; sync' % (test_iso, dev))

        # leave the actual device creation to the individual tests; all devices
        # created henceforth will default to the test.iso contents
        klass.remove_device(dev)

        while klass.get_devices():
            time.sleep(0.2)

    @classmethod
    def tearDownClass(klass):
        for dev in klass.get_devices():
            klass.remove_device(dev)

        # remove scsi_debug; might need a few tries while being busy
        timeout = 10
        while timeout > 0:
            (code, out, err) = klass.root_command('rmmod -v scsi_debug')
            if code == 0:
                break
            if 'in use' in err:
                time.sleep(0.2)
            else:
                break
        if code != 0:
            raise SystemError('cannot rmmod scsi_debug: ' + err)

    @classmethod
    def get_devices(klass):
        '''Return current set of device names from scsi_debug'''

        devs = []
        for dir in glob('/sys/bus/pseudo/drivers/scsi_debug/adapter*/host*/target*/*:*/block'):
            try:
                devs += os.listdir(dir)
            except OSError:
                # TOCTOU, might change underneath us
                pass
        return set(devs)

    @classmethod
    def create_host(klass, ptype):
        '''Create a new SCSI host.

        Return device name.
        '''
        orig_devs = klass.get_devices()
        klass.root_command_success('echo %i > /sys/bus/pseudo/drivers/scsi_debug/ptype' % ptype)
        klass.root_command_success('echo 1 > /sys/bus/pseudo/drivers/scsi_debug/add_host')

        timeout = 1000
        while timeout >= 0:
            devs = klass.get_devices()
            if devs - orig_devs:
                break
            time.sleep(0.2)
            timeout -= 1
        else:
            raise SystemError('timed out waiting for new device')

        new_devs = devs - orig_devs
        assert len(new_devs) == 1
        return new_devs.pop()

    @classmethod
    def remove_device(klass, device):
        '''Remove given device name.'''

        klass.root_command_success('echo 1 > /sys/block/%s/device/delete' % device)

    def setUp(self):
        self.mock_polkit = None

        self.monitor = subprocess.Popen(['gvfs-mount', '-oi'],
                                        stdout=subprocess.PIPE)
        # set monitor stdout to non-blocking
        fl = fcntl.fcntl(self.monitor.stdout, fcntl.F_GETFL)
        fcntl.fcntl(self.monitor.stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)

        # wait until monitor is ready
        while 'Monitoring events' not in self.get_monitor_output():
            time.sleep(0.1)

    def tearDown(self):
        self.monitor.terminate()
        self.monitor.wait()
        self.stop_polkit()

    def test_cdrom(self):
        '''drive mount: cdrom'''

        dev = self.create_host(PTYPE_CDROM)

        # check that gvfs monitor picks up the new drive
        out = self.get_monitor_output()
        self.assertRegex(out, 'Drive connected:\s+.*CD')
        self.assertRegex(out, 'unix-device:.*/dev/%s' % dev)
        self.assertTrue('has_media=1' in out, out)

        self.assertRegex(out, 'Volume added:\s+.*bogus-cd')
        self.assertRegex(out, "label:\s+'bogus-cd")
        self.assertTrue('can_mount=1' in out, out)
        self.assertTrue('should_automount=1' in out, out)
        self.assertRegex(out, 'themed icons:.*media-optical')

        # user is not on any local session in the sandbox, so mounting ought to
        # fail
        (code, out, err) = self.program_code_out_err(['gvfs-mount', '-d', '/dev/' + dev])
        self.assertNotEqual(code, 0)
        self.assertRegex(err, 'Not authorized')

        # tell polkit to do allow removable (but not internal) storage
        self.start_polkit(['org.freedesktop.udisks2.filesystem-mount'])

        # now mounting should succeed
        (out, err) = self.program_out_err(['gvfs-mount', '-d', '/dev/' + dev])

        # should appear as Mount
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        self.assertEqual(err.strip(), '')
        match = re.search('Mount\(\d+\): bogus-cd -> (file://.*/media/.*/bogus-cd)', out)
        self.assertTrue(match, 'no Mount found in gvfs-mount -li output:\n' + out)

        # unmount it again
        self.unmount(match.group(1))

    def test_system_partition(self):
        '''drive mount: system partition'''

        dev = self.create_host(PTYPE_DISK)

        # check that gvfs monitor picks up the new drive
        out = self.get_monitor_output()
        self.assertRegex(out, 'Drive connected:\s+.*Disk')
        self.assertRegex(out, 'unix-device:.*/dev/%s' % dev)
        self.assertTrue('has_media=1' in out, out)

        self.assertRegex(out, 'Volume added:\s+.*bogus-cd')
        self.assertRegex(out, "label:\s+'bogus-cd")
        self.assertTrue('should_automount=0' in out, out)
        self.assertRegex(out, 'themed icons:.*harddisk')

        # should fail with only allowing the user to mount removable storage
        self.start_polkit(['org.freedesktop.udisks2.filesystem-mount'])
        (code, out, err) = self.program_code_out_err(['gvfs-mount', '-d', '/dev/' + dev])
        self.assertNotEqual(code, 0)
        self.assertRegex(err, 'Not authorized')

        # should succeed with allowing the user to mount system storage
        self.start_polkit(['org.freedesktop.udisks2.filesystem-mount-system'])
        (out, err) = self.program_out_err(['gvfs-mount', '-d', '/dev/' + dev])

        # should appear as Mount
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        self.assertEqual(err.strip(), '')
        match = re.search('Mount\(\d+\): bogus-cd -> (file://.*/media/.*/bogus-cd)', out)
        self.assertTrue(match, 'no Mount found in gvfs-mount -li output:\n' + out)

        # unmount it again
        self.unmount(match.group(1))

    def test_media_player(self):
        '''drive mount: media player'''

        def cleanup():
            rootsh = subprocess.Popen(['./rootsh'], stdin=subprocess.PIPE)
            rootsh.communicate(b'''rm /run/udev/rules.d/40-scsi_debug-fake-mediaplayer.rules
pkill --signal HUP udevd || pkill --signal HUP systemd-udevd                              
''')

        # create udev rule to turn it into a music player
        self.addCleanup(cleanup)
        rootsh = subprocess.Popen(['./rootsh'], stdin=subprocess.PIPE)
        rootsh.communicate(b'''export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
echo 'SUBSYSTEM=="block", ATTRS{model}=="scsi_debug*", ENV{ID_MEDIA_PLAYER}="MockTune"' > /run/udev/rules.d/40-scsi_debug-fake-mediaplayer.rules
sync
pkill --signal HUP udevd || pkill --signal HUP systemd-udevd
''')

        dev = self.create_host(PTYPE_DISK)

        # check that gvfs monitor picks up the new volume
        out = self.get_monitor_output()
        self.assertRegex(out, 'Volume added:\s+.*bogus-cd')
        self.assertRegex(out, "label:\s+'bogus-cd")
        self.assertTrue('should_automount=0' in out, out)
        self.assertRegex(out, 'themed icons:.*harddisk')

        # mount it
        self.start_polkit(['org.freedesktop.udisks2.filesystem-mount-system'])
        (out, err) = self.program_out_err(['gvfs-mount', '-d', '/dev/' + dev])

        # should appear as Mount
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        self.assertEqual(err.strip(), '')
        match = re.search('Mount\(\d+\): bogus-cd -> (file://.*/media/.*/bogus-cd)', out)
        self.assertTrue(match, 'no Mount found in gvfs-mount -li output:\n' + out)

        # should have media player content
        self.assertRegex(out, 'x_content_types:.*x-content/audio-player')

        # unmount it again
        self.unmount(match.group(1))

    def get_monitor_output(self):
        '''Wait for gvfs monitor to output something, and return it'''

        empty_timeout = 50
        while True:
            out = self.monitor.stdout.readall()
            if out:
                break
            else:
                empty_timeout -= 1
                self.assertGreater(empty_timeout, 0,
                                   'timed out waiting for monitor output')

            time.sleep(0.1)

        # wait a bit more to see whether we catch some stragglers
        time.sleep(0.2)
        out2 = self.monitor.stdout.readall()
        if out2:
            out += out2

        return out.decode()

    def start_polkit(self, actions):
        '''Start mock polkit with list of allowed actions.'''

        self.stop_polkit()
        self.mock_polkit = subprocess.Popen(['./rootsh'],
                                            stdin=subprocess.PIPE)
        self.mock_polkit.stdin.write(('set -e\n/home/test_polkitd.py -r -a %s\n'
                                      % ','.join(actions)).encode('ASCII'))
        # wait until it started up
        if actions:
            timeout = 50
            while timeout > 0:
                try:
                    out = subprocess.check_output(['pkcheck', '--action-id', actions[0], '--process', '1'],
                                                  stderr=subprocess.PIPE)
                    if b'test=test' in out:
                        break
                except subprocess.CalledProcessError:
                    pass

                time.sleep(0.1)
                timeout -= 1
            else:
                self.fail('timed out waiting for test_polkitd.py')
        else:
            # we can only cross fingers here, as we do not have an action to verify
            time.sleep(0.5)

        self.assertEqual(self.mock_polkit.poll(), None,
                         'mock polkitd unexpectedly terminated')

    def stop_polkit(self):
        '''Stop mock polkit, if it is running.'''

        if self.mock_polkit:
            # for some reason, terminating the shell doesn't terminate the
            # polkitd running in it, so kill that separately
            self.root_command('kill `pidof -x /home/test_polkitd.py`')
            self.mock_polkit.terminate()
            self.mock_polkit.wait()
            self.mock_polkit = None

@unittest.skipUnless(have_httpd, 'Apache httpd not installed')
class Dav(GvfsTestCase):
    '''Test WebDAV backend'''

    @classmethod
    def setUpClass(klass):
        '''Set up Apache httpd sandbox'''

        klass.mod_dir = klass.get_httpd_module_dir()
        klass.httpd_sandbox = tempfile.mkdtemp()

        klass.public_dir = os.path.join(klass.httpd_sandbox, 'public')
        os.mkdir(klass.public_dir)
        with open(os.path.join(klass.public_dir, 'hello.txt'), 'w') as f:
            f.write('hi\n')

        klass.secret_dir = os.path.join(klass.httpd_sandbox, 'secret')
        os.mkdir(klass.secret_dir)
        with open(os.path.join(klass.secret_dir, 'restricted.txt'), 'w') as f:
            f.write('dont tell anyone\n')

        # test:s3kr1t
        with open(os.path.join(klass.httpd_sandbox, 'htpasswd'), 'w') as f:
            f.write('test:$apr1$t0B4mfkT$Tr8ip333/ZR/7xrRBuxjI.\n')

        # some distros have some extra modules which we need to load
        modules = ''
        for m in ['authn_core', 'authz_core', 'authz_user', 'auth_basic',
                  'authn_file', 'mpm_prefork', 'unixd', 'dav', 'dav_fs', 'ssl']:
            if os.path.exists(os.path.join(klass.mod_dir, 'mod_%s.so' % m)):
                modules += 'LoadModule %s_module %s/mod_%s.so\n' % (m, klass.mod_dir, m)

        with open(os.path.join(klass.httpd_sandbox, 'apache2.conf'), 'w') as f:
            f.write('''Listen localhost:8088
Listen localhost:4443
%(modules)s

DocumentRoot .
ServerName localhost
PidFile apache.pid
LogLevel debug
ErrorLog error_log
DAVLockDB DAVLock

<VirtualHost localhost:4443>
  SSLEngine on
  SSLCertificateFile %(mydir)s/files/testcert.pem
  SSLCertificateKeyFile %(mydir)s/files/testcert.pem
</VirtualHost>

<Directory %(root)s/public>
  Dav On
</Directory>

<Directory %(root)s/secret>
  Dav On
  AuthType Basic
  AuthName DAV
  AuthUserFile htpasswd
  Require valid-user
</Directory>
''' % {'mod_dir': klass.mod_dir,
       'root': klass.httpd_sandbox,
       'modules': modules,
       'mydir': my_dir})

        # start server
        try:
            subprocess.check_call(['apachectl', '-d', klass.httpd_sandbox, '-f', 'apache2.conf', '-k', 'start'])
        except subprocess.CalledProcessError as e:
            error_log = os.path.join(klass.httpd_sandbox, 'error_log')
            if os.path.exists(error_log):
                with open(error_log) as f:
                    print('---- apache http error log ----\n%s\n---------\n' % f.read())
            raise

    @classmethod
    def tearDownClass(klass):
        '''Stop httpd server and remove sandbox'''

        subprocess.call(['apachectl', '-d', klass.httpd_sandbox, '-f', 'apache2.conf', '-k', 'stop'])
        shutil.rmtree(klass.httpd_sandbox)

    @classmethod
    def get_httpd_module_dir(klass):
        '''Return module directory for Apache httpd.

        Unfortunately this is highly distro/platform specific, so try to
        determine it from apxs2 or apachectl.
        '''
        # if we have apxs2 installed, use this
        try:
            apxs2 = subprocess.Popen(['apxs2', '-q', 'LIBEXECDIR'],
                                     stdout=subprocess.PIPE,
                                     stderr=subprocess.PIPE,
                                     universal_newlines=True)
            out = apxs2.communicate()[0].strip()
            assert apxs2.returncode == 0, 'apxs2 -V failed'
            return out
        except OSError:
            # Look for apxs instead
            try:
                apxs2 = subprocess.Popen(['apxs', '-q', 'LIBEXECDIR'],
                                         stdout=subprocess.PIPE,
                                         stderr=subprocess.PIPE,
                                         universal_newlines=True)
                out = apxs2.communicate()[0].strip()
                assert apxs2.returncode == 0, 'apxs2 -V failed'
                return out
            except OSError:
                print('[no apxs2, falling back]')
                pass

        # fall back to looking for modules in HTTPD_ROOT/modules/
        ctl = subprocess.Popen(['apachectl', '-V'],
                               stdout=subprocess.PIPE,
                               universal_newlines=True)
        out = ctl.communicate()[0]
        assert ctl.returncode == 0, 'apachectl -V failed'
        m = re.search('\sHTTPD_ROOT="([^"]+)"\s', out)
        assert m, 'apachectl -V does not show HTTPD_ROOT'
        mod_dir = os.path.join(m.group(1), 'modules')
        assert os.path.isdir(mod_dir), \
            '%s does not exist, cannot determine httpd module path' % mod_dir
        return mod_dir

    def test_http_noauth(self):
        '''dav://localhost without credentials'''

        uri = 'dav://localhost:8088/public'
        subprocess.check_call(['gvfs-mount', uri])
        self.do_mount_check(uri, 'hello.txt', 'hi\n')

    def test_https_noauth(self):
        '''davs://localhost without credentials'''

        uri = 'davs://localhost:4443/public'
        subprocess.check_call(['gvfs-mount', uri])
        self.do_mount_check(uri, 'hello.txt', 'hi\n')

    def test_http_auth(self):
        '''dav://localhost with credentials'''

        uri = 'dav://localhost:8088/secret'

        mount = subprocess.Popen(['gvfs-mount', uri],
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)

        # wrong password
        self.wait_for_gvfs_mount_user_prompt(mount)
        mount.stdin.write(b'test\nh4ck\n')

        # correct password
        (out, err) = mount.communicate(b's3kr1t\n')
        self.assertEqual(mount.returncode, 0)
        self.assertEqual(err, b'')

        self.do_mount_check(uri, 'restricted.txt', 'dont tell anyone\n')

    def test_https_auth(self):
        '''davs://localhost with credentials'''

        uri = 'davs://localhost:4443/secret'

        mount = subprocess.Popen(['gvfs-mount', uri],
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)

        # wrong password
        self.wait_for_gvfs_mount_user_prompt(mount)
        mount.stdin.write(b'test\nh4ck\n')

        # correct password
        (out, err) = mount.communicate(b's3kr1t\n')
        self.assertEqual(mount.returncode, 0)
        self.assertEqual(err, b'')

        self.do_mount_check(uri, 'restricted.txt', 'dont tell anyone\n')

    def do_mount_check(self, uri, testfile, content):
        # appears in gvfs-mount list
        (out, err) = self.program_out_err(['gvfs-mount', '-li'])
        try:
            self.assertRegex(out, 'Mount\(\d+\):.* -> davs?://([a-z0-9]+@)?localhost')

            # check gvfs-info
            out = self.program_out_success(['gvfs-info', uri])
            self.assertRegex(out, 'id::filesystem: dav')
            self.assertTrue('type: directory' in out, out)

            # check gvfs-ls
            out = self.program_out_success(['gvfs-ls', uri])
            self.assertEqual(out.strip(), testfile)

            # check gvfs-cat
            out = self.program_out_success(['gvfs-cat', uri + '/' + testfile])
            self.assertEqual(out, content)

            # create a new file
            self.program_out_success(['gvfs-copy', uri + '/' + testfile, uri + '/foo'])
            out = self.program_out_success(['gvfs-cat', uri + '/foo'])
            self.assertEqual(out, content)

            # remove it again
            self.program_out_success(['gvfs-rm', uri + '/foo'])
            out = self.program_out_success(['gvfs-ls', uri])
            self.assertFalse('foo' in out.split(), out)
        finally:
            self.unmount(uri)


if __name__ == '__main__':
    # do not break tests due to translations
    try:
        del os.environ['LANGUAGE']
    except KeyError:
        pass
    os.environ['LC_ALL'] = 'C'
    unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout, verbosity=2))