#! /usr/bin/env python3

# Copyright © 2017, 2019 Red Hat, Inc

# Copyright © 2020 Canonical Ltd

#

# This program 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.1 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

# 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, see <http://www.gnu.org/licenses/>.

# Authors:

#       Christian J. Kellner <christian@kellner.me>

#       Benjamin Berg <bberg@redhat.com>

#       Marco Trevisan <marco.trevisan@canonical.com>

import unittest

import time

import subprocess

import os

import os.path

import sys

import tempfile

import glob

import pwd

import shutil

import socket

import struct

import dbusmock

import gi

gi.require_version('FPrint', '2.0')

from gi.repository import GLib, Gio, FPrint

import cairo

try:

    from subprocess import DEVNULL

except ImportError:

    DEVNULL = open(os.devnull, 'wb')

SERVICE_FILE = '/usr/share/dbus-1/system-services/net.reactivated.Fprint.service'

def get_timeout(topic='default'):

    vals = {

        'valgrind': {

            'test': 300,

            'default': 20,

            'daemon_start': 60

        },

        'asan': {

            'test': 120,

            'default': 6,

            'daemon_start': 10

        },

        'default': {

            'test': 60,

            'default': 3,

            'daemon_start': 5

        }

    }

    if os.getenv('VALGRIND') is not None:

        lut = vals['valgrind']

    elif os.getenv('ADDRESS_SANITIZER') is not None:

        lut = vals['asan']

    else:

        lut = vals['default']

    if topic not in lut:

        raise ValueError('invalid topic')

    return lut[topic]

# Copied from libfprint tests

class Connection:

    def __init__(self, addr):

        self.addr = addr

    def __enter__(self):

        self.con = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

        self.con.connect(self.addr)

        return self.con

    def __exit__(self, exc_type, exc_val, exc_tb):

        self.con.close()

        del self.con

def load_image(img):

    png = cairo.ImageSurface.create_from_png(img)

    # Cairo wants 4 byte aligned rows, so just add a few pixel if necessary

    w = png.get_width()

    h = png.get_height()

    w = (w + 3) // 4 * 4

    h = (h + 3) // 4 * 4

    img = cairo.ImageSurface(cairo.Format.A8, w, h)

    cr = cairo.Context(img)

    cr.set_source_rgba(1, 1, 1, 1)

    cr.paint()

    cr.set_source_rgba(0, 0, 0, 0)

    cr.set_operator(cairo.OPERATOR_SOURCE)

    cr.set_source_surface(png)

    cr.paint()

    return img

if hasattr(os.environ, 'TOPSRCDIR'):

    root = os.environ['TOPSRCDIR']

else:

    root = os.path.join(os.path.dirname(__file__), '..')

imgdir = os.path.join(root, 'tests', 'prints')

ctx = GLib.main_context_default()

class FPrintdTest(dbusmock.DBusTestCase):

    @staticmethod

    def path_from_service_file(sf):

        with open(SERVICE_FILE) as f:

                for line in f:

                    if not line.startswith('Exec='):

                        continue

                    return line.split('=', 1)[1].strip()

        return None

    @classmethod

    def setUpClass(cls):

        super().setUpClass()

        fprintd = None

        cls._polkitd = None

        cls._has_hotplug = FPrint.Device.find_property("removed") is not None

        if 'FPRINT_BUILD_DIR' in os.environ:

            print('Testing local build')

            build_dir = os.environ['FPRINT_BUILD_DIR']

            fprintd = os.path.join(build_dir, 'fprintd')

        elif 'UNDER_JHBUILD' in os.environ:

            print('Testing JHBuild version')

            jhbuild_prefix = os.environ['JHBUILD_PREFIX']

            fprintd = os.path.join(jhbuild_prefix, 'libexec', 'fprintd')

        else:

            print('Testing installed system binaries')

            fprintd = cls.path_from_service_file(SERVICE_FILE)

        assert fprintd is not None, 'failed to find daemon'

        cls.paths = {'daemon': fprintd }

        cls.tmpdir = tempfile.mkdtemp(prefix='libfprint-')

        cls.sockaddr = os.path.join(cls.tmpdir, 'virtual-image.socket')

        os.environ['FP_VIRTUAL_IMAGE'] = cls.sockaddr

        cls.prints = {}

        for f in glob.glob(os.path.join(imgdir, '*.png')):

            n = os.path.basename(f)[:-4]

            cls.prints[n] = load_image(f)

        cls.test_bus = Gio.TestDBus.new(Gio.TestDBusFlags.NONE)

        cls.test_bus.up()

        cls.test_bus.unset()

        addr = cls.test_bus.get_bus_address()

        os.environ['DBUS_SYSTEM_BUS_ADDRESS'] = addr

        cls.dbus = Gio.DBusConnection.new_for_address_sync(addr,

            Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION |

            Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT, None, None)

        assert cls.dbus.is_closed() == False

    @classmethod

    def tearDownClass(cls):

        cls.dbus.close()

        cls.test_bus.down()

        del cls.dbus

        del cls.test_bus

        shutil.rmtree(cls.tmpdir)

        dbusmock.DBusTestCase.tearDownClass()

    def daemon_start(self):

        timeout = get_timeout('daemon_start')  # seconds

        env = os.environ.copy()

        env['G_DEBUG'] = 'fatal-criticals'

        env['STATE_DIRECTORY'] = (self.state_dir + ':' + '/hopefully/a/state_dir_path/that/shouldnt/be/writable')

        env['RUNTIME_DIRECTORY'] = self.run_dir

        argv = [self.paths['daemon'], '-t']

        valgrind = os.getenv('VALGRIND')

        if valgrind is not None:

            argv.insert(0, 'valgrind')

            argv.insert(1, '--leak-check=full')

            if os.path.exists(valgrind):

                argv.insert(2, '--suppressions=%s' % valgrind)

            self.valgrind = True

        self.daemon = subprocess.Popen(argv,

                                       env=env,

                                       stdout=None,

                                       stderr=subprocess.STDOUT)

        self.addCleanup(self.daemon_stop)

        timeout_count = timeout * 10

        timeout_sleep = 0.1

        while timeout_count > 0:

            time.sleep(timeout_sleep)

            timeout_count -= 1

            try:

                self.manager = Gio.DBusProxy.new_sync(self.dbus,

                                                      Gio.DBusProxyFlags.DO_NOT_AUTO_START,

                                                      None,

                                                      'net.reactivated.Fprint',

                                                      '/net/reactivated/Fprint/Manager',

                                                      'net.reactivated.Fprint.Manager',

                                                      None)

                devices = self.manager.GetDevices()

                # Find the virtual device, just in case it is a local run

                # and there is another usable sensor available locally

                for path in devices:

                    dev = Gio.DBusProxy.new_sync(self.dbus,

                                                 Gio.DBusProxyFlags.DO_NOT_AUTO_START,

                                                 None,

                                                 'net.reactivated.Fprint',

                                                 path,

                                                 'net.reactivated.Fprint.Device',

                                                 None)

                    if 'Virtual image device' in str(dev.get_cached_property('name')):

                        self.device = dev

                        break

                else:

                    print('Did not find virtual device! Probably libfprint was build without the corresponding driver!')

                break

            except GLib.GError:

                pass

        else:

            timeout_time = timeout * 10 * timeout_sleep

            self.fail('daemon did not start in %d seconds' % timeout_time)

    def daemon_stop(self):

        if self.daemon:

            try:

                self.daemon.terminate()

            except OSError:

                pass

            self.daemon.wait(timeout=2)

            self.assertLess(self.daemon.returncode, 128)

            self.assertGreaterEqual(self.daemon.returncode, 0)

        self.daemon = None

    def polkitd_start(self):

        if self._polkitd:

            return

        if 'POLKITD_MOCK_PATH' in os.environ:

            polkitd_template = os.path.join(os.getenv('POLKITD_MOCK_PATH'), 'polkitd.py')

        else:

            polkitd_template = os.path.join(os.path.dirname(__file__), 'dbusmock/polkitd.py')

        print ('Using template from %s' % polkitd_template)

        self._polkitd, self._polkitd_obj = self.spawn_server_template(

            polkitd_template, {}, stdout=subprocess.PIPE)

        return self._polkitd

    def polkitd_stop(self):

        if self._polkitd is None:

            return

        self._polkitd.terminate()

        self._polkitd.wait()

    def get_current_user(self):

        return pwd.getpwuid(os.getuid()).pw_name

    def setUp(self):

        self.test_dir = tempfile.mkdtemp()

        self.addCleanup(shutil.rmtree, self.test_dir)

        self.state_dir = os.path.join(self.test_dir, 'state')

        self.run_dir = os.path.join(self.test_dir, 'run')

        os.environ['FP_DRIVERS_WHITELIST'] = 'virtual_image'

    def assertFprintError(self, fprint_error):

        return self.assertRaisesRegex(GLib.Error,

            '.*net\.reactivated\.Fprint\.Error\.{}.*'.format(fprint_error))

    # From libfprint tests

    def send_retry(self, retry_error=FPrint.DeviceRetry.TOO_SHORT):

        with Connection(self.sockaddr) as con:

            con.sendall(struct.pack('ii', -1, retry_error))

    # From libfprint tests

    def send_error(self, error=FPrint.DeviceError.GENERAL):

        with Connection(self.sockaddr) as con:

            con.sendall(struct.pack('ii', -2, error))

    # From libfprint tests

    def send_remove(self):

        with Connection(self.sockaddr) as con:

            con.sendall(struct.pack('ii', -5, 0))

    # From libfprint tests

    def send_image(self, image):

        img = self.prints[image]

        with Connection(self.sockaddr) as con:

            mem = img.get_data()

            mem = mem.tobytes()

            self.assertEqual(len(mem), img.get_width() * img.get_height())

            encoded_img = struct.pack('ii', img.get_width(), img.get_height())

            encoded_img += mem

            con.sendall(encoded_img)

    def call_device_method_async(self, method, *args):

        """ add cancellable... """

        self.device.call(method, GLib.Variant(*args),

            Gio.DBusCallFlags.NONE, -1, None, self._method_call_handler)

    def _method_call_handler(self, proxy, res):

        try:

            self._async_call_res = proxy.call_finish(res)

        except Exception as e:

            self._async_call_res = e

    def wait_for_device_reply(self):

        self._async_call_res = None

        while not self._async_call_res:

            ctx.iteration(True)

        if isinstance(self._async_call_res, Exception):

            raise self._async_call_res

    def gdbus_device_method_call_process(self, method, args=[]):

        return subprocess.Popen([

            'gdbus',

            'call',

            '--system',

            '--dest',

            self.device.get_name(),

            '--object-path',

            self.device.get_object_path(),

            '--method',

            '{}.{}'.format(self.device.get_interface_name(), method),

        ] + args, stderr=subprocess.STDOUT, stdout=subprocess.PIPE)

    def call_device_method_from_other_client(self, method, args=[]):

        try:

            proc = self.gdbus_device_method_call_process(method, args)

            proc.wait(timeout=5)

            if proc.returncode != 0:

                raise GLib.GError(proc.stdout.read())

            return proc.stdout.read()

        except subprocess.TimeoutExpired as e:

            raise GLib.GError(e.output)

class FPrintdVirtualDeviceBaseTest(FPrintdTest):

    def setUp(self):

        super().setUp()

        self.manager = None

        self.device = None

        self.polkitd_start()

        self.daemon_start()

        if self.device is None:

            self.skipTest("Need virtual_image device to run the test")

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.setusername',

                                      'net.reactivated.fprint.device.enroll',

                                      'net.reactivated.fprint.device.verify'])

        def signal_cb(proxy, sender, signal, params):

            print(signal, params)

            if signal == 'EnrollStatus':

                self._abort = params[1]

                self._last_result = params[0]

                if not self._abort and self._last_result.startswith('enroll-'):

                    # Exit wait loop, onto next enroll state (if any)

                    self._abort = True

                elif self._abort:

                    pass

                else:

                    self._abort = True

                    self._last_result = 'Unexpected signal values'

                    print('Unexpected signal values')

            elif signal == 'VerifyFingerSelected':

                pass

            elif signal == 'VerifyStatus':

                self._abort = True

                self._last_result = params[0]

                self._verify_stopped = params[1]

            else:

                self._abort = True

                self._last_result = 'Unexpected signal'

        self.g_signal_id = self.device.connect('g-signal', signal_cb)

    def tearDown(self):

        self.polkitd_stop()

        self.device.disconnect(self.g_signal_id)

        self.device = None

        self.manager = None

        super().tearDown()

    def wait_for_result(self, expected=None):

        self._abort = False

        while not self._abort:

            ctx.iteration(True)

        self.assertTrue(self._abort)

        self._abort = False

        if expected is not None:

            self.assertEqual(self._last_result, expected)

    def enroll_image(self, img, finger='right-index-finger', expected_result='enroll-completed'):

        self.device.EnrollStart('(s)', finger)

        stages = self.device.get_cached_property('num-enroll-stages').unpack()

        for stage in range(stages):

            self.send_image(img)

            if stage < stages - 1:

                self.wait_for_result('enroll-stage-passed')

            else:

                self.wait_for_result(expected_result)

        self.device.EnrollStop()

        self.assertEqual(self._last_result, expected_result)

    def enroll_multiple_images(self, images_override={}, return_index=-1):

        enroll_map = {

            'left-thumb': 'whorl',

            'right-index-finger': 'arch',

            'left-little-finger': 'loop-right',

        }

        enroll_map.update(images_override)

        for finger, print in enroll_map.items():

            self.enroll_image(print, finger=finger)

        enrolled = self.device.ListEnrolledFingers('(s)', 'testuser')

        self.assertCountEqual(enroll_map.keys(), enrolled)

        if return_index >= 0:

            return enroll_map[enrolled[return_index]]

        return (enrolled, enroll_map)

class FPrintdManagerTests(FPrintdVirtualDeviceBaseTest):

    def setUp(self):

        super().setUp()

        self._polkitd_obj.SetAllowed([''])

    def test_manager_get_devices(self):

        self.assertListEqual(self.manager.GetDevices(),

            [ self.device.get_object_path() ])

    def test_manager_get_default_device(self):

        self.assertEqual(self.manager.GetDefaultDevice(),

            self.device.get_object_path())

class FPrintdManagerPreStartTests(FPrintdTest):

    def test_manager_get_no_devices(self):

        os.environ['FP_DRIVERS_WHITELIST'] = 'hopefully_no_existing_driver'

        self.daemon_start()

        self.assertListEqual(self.manager.GetDevices(), [])

    def test_manager_get_no_default_device(self):

        os.environ['FP_DRIVERS_WHITELIST'] = 'hopefully_no_existing_driver'

        self.daemon_start()

        with self.assertFprintError('NoSuchDevice'):

            self.manager.GetDefaultDevice()

    def test_manager_get_devices_on_name_appeared(self):

        self._appeared_name = None

        def on_name_appeared(connection, name, name_owner):

            self._appeared_name = name

        def on_name_vanished(connection, name):

            self._appeared_name = 'NAME_VANISHED'

        id = Gio.bus_watch_name_on_connection(self.dbus,

            'net.reactivated.Fprint', Gio.BusNameWatcherFlags.NONE,

            on_name_appeared, on_name_vanished)

        self.addCleanup(Gio.bus_unwatch_name, id)

        self.daemon_start()

        while not self._appeared_name:

            ctx.iteration(True)

        self.assertEqual(self._appeared_name, 'net.reactivated.Fprint')

        try:

            appeared_device = self.dbus.call_sync(

                'net.reactivated.Fprint',

                '/net/reactivated/Fprint/Manager',

                'net.reactivated.Fprint.Manager',

                'GetDefaultDevice', None, None,

                Gio.DBusCallFlags.NO_AUTO_START, 500, None)

        except GLib.GError as e:

            if 'net.reactivated.Fprint.Error.NoSuchDevice' in e.message:

                self.skipTest("Need virtual_image device to run the test")

            raise(e)

        self.assertIsNotNone(appeared_device)

        [dev_path] = appeared_device

        self.assertTrue(dev_path.startswith('/net/reactivated/Fprint/Device/'))

class FPrintdVirtualDeviceTest(FPrintdVirtualDeviceBaseTest):

    def test_name_property(self):

        self.assertEqual(self.device.get_cached_property('name').unpack(),

            'Virtual image device for debugging')

    def test_enroll_stages_property(self):

        self.assertEqual(self.device.get_cached_property('num-enroll-stages').unpack(), 5)

    def test_scan_type(self):

        self.assertEqual(self.device.get_cached_property('scan-type').unpack(),

            'swipe')

    def test_allowed_claim_release_enroll(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.setusername',

                                      'net.reactivated.fprint.device.enroll'])

        self.device.Claim('(s)', 'testuser')

        self.device.Release()

    def test_allowed_claim_release_verify(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.setusername',

                                      'net.reactivated.fprint.device.verify'])

        self.device.Claim('(s)', 'testuser')

        self.device.Release()

    def test_allowed_claim_current_user(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.enroll'])

        self.device.Claim('(s)', '')

        self.device.Release()

        self.device.Claim('(s)', self.get_current_user())

        self.device.Release()

    def test_allowed_list_enrolled_fingers_empty_user(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.enroll'])

        self.device.Claim('(s)', '')

        self.enroll_image('whorl', finger='left-thumb')

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.verify'])

        self.assertEqual(self.device.ListEnrolledFingers('(s)', ''), ['left-thumb'])

        self.assertEqual(self.device.ListEnrolledFingers('(s)', self.get_current_user()), ['left-thumb'])

    def test_allowed_list_enrolled_fingers_current_user(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.enroll'])

        self.device.Claim('(s)', self.get_current_user())

        self.enroll_image('whorl', finger='right-thumb')

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.verify'])

        self.assertEqual(self.device.ListEnrolledFingers('(s)', ''), ['right-thumb'])

        self.assertEqual(self.device.ListEnrolledFingers('(s)', self.get_current_user()), ['right-thumb'])

    def test_unallowed_claim(self):

        self._polkitd_obj.SetAllowed([''])

        with self.assertFprintError('PermissionDenied'):

            self.device.Claim('(s)', 'testuser')

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.setusername'])

        with self.assertFprintError('PermissionDenied'):

            self.device.Claim('(s)', 'testuser')

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.enroll'])

        with self.assertFprintError('PermissionDenied'):

            self.device.Claim('(s)', 'testuser')

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.verify'])

        with self.assertFprintError('PermissionDenied'):

            self.device.Claim('(s)', 'testuser')

    def test_unallowed_enroll_with_verify_claim(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.verify'])

        self.device.Claim('(s)', '')

        with self.assertFprintError('PermissionDenied'):

            self.enroll_image('whorl', finger='right-thumb')

    def test_unallowed_delete_with_verify_claim(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.verify'])

        self.device.Claim('(s)', '')

        with self.assertFprintError('PermissionDenied'):

            self.device.DeleteEnrolledFingers('(s)', 'testuser')

    def test_unallowed_delete2_with_verify_claim(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.verify'])

        self.device.Claim('(s)', '')

        with self.assertFprintError('PermissionDenied'):

            self.device.DeleteEnrolledFingers2()

    def test_unallowed_verify_with_enroll_claim(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.enroll'])

        self.device.Claim('(s)', '')

        with self.assertFprintError('PermissionDenied'):

            self.device.VerifyStart('(s)', 'any')

    def test_unallowed_claim_current_user(self):

        self._polkitd_obj.SetAllowed([''])

        with self.assertFprintError('PermissionDenied'):

            self.device.Claim('(s)', '')

        with self.assertFprintError('PermissionDenied'):

            self.device.Claim('(s)', self.get_current_user())

    def test_multiple_claims(self):

        self.device.Claim('(s)', 'testuser')

        with self.assertFprintError('AlreadyInUse'):

            self.device.Claim('(s)', 'testuser')

        self.device.Release()

    def test_always_allowed_release(self):

        self.device.Claim('(s)', 'testuser')

        self._polkitd_obj.SetAllowed([''])

        self.device.Release()

    def test_unclaimed_release(self):

        with self.assertFprintError('ClaimDevice'):

            self.device.Release()

    def test_unclaimed_verify_start(self):

        with self.assertFprintError('ClaimDevice'):

            self.device.VerifyStart('(s)', 'any')

    def test_unclaimed_verify_stop(self):

        with self.assertFprintError('ClaimDevice'):

            self.device.VerifyStop()

    def test_unclaimed_enroll_start(self):

        with self.assertFprintError('ClaimDevice'):

            self.device.EnrollStart('(s)', 'left-index-finger')

    def test_unclaimed_enroll_stop(self):

        with self.assertFprintError('ClaimDevice'):

            self.device.EnrollStop()

    def test_unclaimed_delete_enrolled_fingers(self):

        self.device.DeleteEnrolledFingers('(s)', 'testuser')

    def test_unclaimed_delete_enrolled_fingers2(self):

        with self.assertFprintError('ClaimDevice'):

            self.device.DeleteEnrolledFingers2()

    def test_unclaimed_list_enrolled_fingers(self):

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.ListEnrolledFingers('(s)', 'testuser')

    def test_claim_device_open_fail(self):

        os.rename(self.tmpdir, self.tmpdir + '-moved')

        self.addCleanup(os.rename, self.tmpdir + '-moved', self.tmpdir)

        with self.assertFprintError('Internal'):

            self.device.Claim('(s)', 'testuser')

    def test_claim_from_other_client_is_released_when_vanished(self):

        self.call_device_method_from_other_client('Claim', ['testuser'])

        time.sleep(1)

        self.device.Claim('(s)', 'testuser')

        self.device.Release()

    def test_claim_disconnect(self):

        addr = os.environ['DBUS_SYSTEM_BUS_ADDRESS']

        # Get a separat bus connection

        dbus = Gio.DBusConnection.new_for_address_sync(addr,

            Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION |

            Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT, None, None)

        assert dbus.is_closed() == False

        dev_path = self.device.get_object_path()

        dev = Gio.DBusProxy.new_sync(dbus,

                                     Gio.DBusProxyFlags.DO_NOT_AUTO_START,

                                     None,

                                     'net.reactivated.Fprint',

                                     dev_path,

                                     'net.reactivated.Fprint.Device',

                                     None)

        def call_done(obj, result, user_data):

            # Ignore the callback (should be an error)

            pass

        # Do an async call to claim and immediately close

        dev.Claim('(s)', 'testuser', result_handler=call_done)

        # Ensure the call is on the wire, then close immediately

        dbus.flush_sync()

        dbus.close_sync()

        time.sleep(1)

    def test_removal_during_enroll(self):

        if not self._has_hotplug:

            self.skipTest("libfprint is too old for hotplug")

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.setusername',

                                      'net.reactivated.fprint.device.enroll'])

        self.device.Claim('(s)', 'testuser')

        self.device.EnrollStart('(s)', 'left-index-finger')

        # Now remove the device while we are enrolling, which will cause an error

        self.send_remove()

        self.wait_for_result(expected='enroll-unknown-error')

        # The device will still be there now until it is released

        devices = self.manager.GetDevices()

        self.assertIn(self.device.get_object_path(), devices)

        with self.assertFprintError('Internal'):

            self.device.Release()

        # And now it will be gone

        devices = self.manager.GetDevices()

        self.assertNotIn(self.device.get_object_path(), devices)

class FPrintdVirtualDeviceClaimedTest(FPrintdVirtualDeviceBaseTest):

    def setUp(self):

        super().setUp()

        self.device.Claim('(s)', 'testuser')

    def tearDown(self):

        self._polkitd_obj.SetAllowed(['net.reactivated.fprint.device.enroll'])

        try:

            self.device.Release()

        except GLib.GError as e:

            if not 'net.reactivated.Fprint.Error.ClaimDevice' in e.message:

                raise(e)

        super().tearDown()

    def test_wrong_finger_enroll_start(self):

        with self.assertFprintError('InvalidFingername'):

            self.device.EnrollStart('(s)', 'any')

    def test_verify_with_no_enrolled_prints(self):

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.VerifyStart('(s)', 'any')

    def test_enroll_verify_list_delete(self):

        # This test can trigger a race in older libfprint, only run if we have

        # hotplug support, which coincides with the fixed release.

        if not self._has_hotplug:

            self.skipTest("libfprint is too old for hotplug")

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.ListEnrolledFingers('(s)', 'testuser')

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.ListEnrolledFingers('(s)', 'nottestuser')

        self.enroll_image('whorl')

        self.assertTrue(os.path.exists(os.path.join(self.state_dir, 'testuser/virtual_image/0/7')))

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.ListEnrolledFingers('(s)', 'nottestuser')

        self.assertEqual(self.device.ListEnrolledFingers('(s)', 'testuser'), ['right-index-finger'])

        # Finger is enrolled, try to verify it

        self.device.VerifyStart('(s)', 'any')

        # Try a wrong print; will stop verification

        self.send_image('tented_arch')

        self.wait_for_result()

        self.assertTrue(self._verify_stopped)

        self.assertEqual(self._last_result, 'verify-no-match')

        self.device.VerifyStop()

        self.device.VerifyStart('(s)', 'any')

        # Send a retry error (swipe too short); will not stop verification

        self.send_retry()

        self.wait_for_result()

        self.assertFalse(self._verify_stopped)

        self.assertEqual(self._last_result, 'verify-swipe-too-short')

        # Try the correct print; will stop verification

        self.send_image('whorl')

        self.wait_for_result()

        self.assertTrue(self._verify_stopped)

        self.assertEqual(self._last_result, 'verify-match')

        self.device.VerifyStop()

        self.assertEqual(self.device.ListEnrolledFingers('(s)', 'testuser'), ['right-index-finger'])

        # And delete the print(s) again

        self.device.DeleteEnrolledFingers('(s)', 'testuser')

        self.assertFalse(os.path.exists(os.path.join(self.state_dir, 'testuser/virtual_image/0/7')))

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.ListEnrolledFingers('(s)', 'testuser')

    def test_enroll_delete2(self):

        self.enroll_image('whorl')

        self.assertTrue(os.path.exists(os.path.join(self.state_dir, 'testuser/virtual_image/0/7')))

        # And delete the print(s) again using the new API

        self.device.DeleteEnrolledFingers2()

        self.assertFalse(os.path.exists(os.path.join(self.state_dir, 'testuser/virtual_image/0/7')))

    def test_enroll_invalid_storage_dir(self):

        # Directory wil not exist yet

        os.makedirs(self.state_dir, mode=0o500)

        self.addCleanup(os.chmod, self.state_dir, mode=0o700)

        try:

            os.open(os.path.join(self.state_dir, "testfile"), os.O_CREAT | os.O_WRONLY)

            self.skipTest('Permissions aren\'t respected (CI environment?)')

        except PermissionError:

            pass

        self.enroll_image('whorl', expected_result='enroll-failed')

    def test_verify_invalid_storage_dir(self):

        self.enroll_image('whorl')

        os.chmod(self.state_dir, mode=0o000)

        self.addCleanup(os.chmod, self.state_dir, mode=0o700)

        try:

            os.open(os.path.join(self.state_dir, "testfile"), os.O_CREAT | os.O_WRONLY)

            self.skipTest('Permissions aren\'t respected (CI environment?)')

        except PermissionError:

            pass

        with self.assertFprintError('NoEnrolledPrints'):

            self.device.VerifyStart('(s)', 'any')

    def test_enroll_stop_cancels(self):

        self.device.EnrollStart('(s)', 'left-index-finger')

        self.device.EnrollStop()

        self.wait_for_result(expected='enroll-failed')

    def test_verify_stop_cancels(self):

        self.enroll_image('whorl')