diff --git a/apertis_tests_lib/__init__.py b/apertis_tests_lib/__init__.py
index 109589c31fa57b72a76e8e7ae323d28fe476d4e3..4cc81457bf7f11d439256c82fc16babe448f4dfc 100644
--- a/apertis_tests_lib/__init__.py
+++ b/apertis_tests_lib/__init__.py
@@ -11,6 +11,8 @@ import unittest
import shutil
import os
+from gi.repository import GLib
+
MEDIADIR = '/usr/lib/apertis-tests/resources/media'
# define this here to make lint happy with too long lines
@@ -18,6 +20,11 @@ LONG_JPEG_NAME = '320px-European_Common_Frog_Rana_temporaria.jpg'
class ApertisTest(unittest.TestCase):
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.loop = GLib.MainLoop.new(None, False)
+ self.homedir = os.path.expanduser("~")
+
def copy_medias(self, dst):
"""
Copy media resources from apertis-tests to @dst with per media type
diff --git a/apertis_tests_lib/tracker.py b/apertis_tests_lib/tracker.py
index f0f04baa090342a01e03550d4639666edb9c4b2c..43a19600abd1d954000512668b21012537f8ab47 100644
--- a/apertis_tests_lib/tracker.py
+++ b/apertis_tests_lib/tracker.py
@@ -31,8 +31,6 @@ class TrackerIndexerMixin:
stdout=subprocess.DEVNULL,
shell=True)
- self.loop = GLib.MainLoop.new(None, False)
-
print("TrackerIndexerMixin: init done")
def miner_progress_cb(self, manager, miner, status, progress,
diff --git a/apertis_tests_lib/tumbler.py b/apertis_tests_lib/tumbler.py
new file mode 100644
index 0000000000000000000000000000000000000000..74862efe2bbd43107a5b2c0af4d48486681a0ab4
--- /dev/null
+++ b/apertis_tests_lib/tumbler.py
@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+
+# Copyright © 2015 Collabora Ltd.
+#
+# SPDX-License-Identifier: MPL-2.0
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+import hashlib
+import os
+
+from gi.repository import GLib
+from gi.repository import Gio
+
+
+class TumblerMixin:
+ def __init__(self):
+ # Monitor thumbnail creation to know when it's done. The DBus API
+ # doesn't have a method to query the initial state but we can be
+ # reasonably sure it's not thumbnailing anything at this point.
+ self.tumbler_queue = []
+ self.tumbler = Gio.DBusProxy.new_for_bus_sync(
+ Gio.BusType.SESSION,
+ Gio.DBusProxyFlags.DO_NOT_LOAD_PROPERTIES,
+ None,
+ 'org.freedesktop.thumbnails.Thumbnailer1',
+ '/org/freedesktop/thumbnails/Thumbnailer1',
+ 'org.freedesktop.thumbnails.Thumbnailer1',
+ None)
+ self.tumbler.connect('g-signal', self.__g_signal_cb)
+
+ # Spy on unicast signals that weren't meant for us, because
+ # the Thumbnailer API uses those, and we want to use them to
+ # determine when it has finished. GDBus doesn't have
+ # high-level API for this sort of nonsense so we do it the hard way.
+ match_rule = ("type=signal,"
+ "sender='org.freedesktop.thumbnails.Thumbnailer1',"
+ "eavesdrop=true")
+ conn = self.tumbler.get_connection()
+ conn.call_sync('org.freedesktop.DBus',
+ '/org/freedesktop/DBus',
+ 'org.freedesktop.DBus',
+ 'AddMatch',
+ GLib.Variant('(s)', (match_rule,)),
+ None, Gio.DBusCallFlags.NONE, -1, None)
+
+ def __g_signal_cb(self, proxy, sender_name, signal_name, parameters):
+ print('TumblerMixin: Received signal:', signal_name)
+
+ if signal_name == 'Started':
+ child = parameters.get_child_value(0)
+ self.tumbler_queue.append(child.get_uint32())
+ elif signal_name == 'Finished':
+ child = parameters.get_child_value(0)
+ self.tumbler_queue.remove(child.get_uint32())
+ elif signal_name == 'Error':
+ child = parameters.get_child_value(1)
+ uris = child.get_strv()
+ child = parameters.get_child_value(3)
+ msg = child.get_string()
+ raise Exception("Error creating thumbnail for %s: %s" %
+ ', '.join(uris), msg)
+
+ def __get_supported_cb(self, source, result):
+ self.tumbler.call_finish(result)
+ self.loop.quit()
+
+ def tumbler_drain_queue(self):
+ # Drain the DBus queue to make sure we received all signals
+ self.tumbler.call('GetSupported',
+ GLib.Variant.new_tuple(),
+ Gio.DBusCallFlags.NONE,
+ -1,
+ None,
+ self.__get_supported_cb)
+ self.loop.run()
+
+ context = self.loop.get_context()
+ while len(self.tumbler_queue) > 0:
+ context.iteration(True)
+
+ def tumbler_assert_thumbnailed(self, root, filename):
+ removable = False
+ monitor = Gio.VolumeMonitor.get()
+ mounts = monitor.get_mounts()
+ for m in mounts:
+ if m.get_root().get_path() == root:
+ removable = m.can_eject()
+ break
+
+ media_path = os.path.join(root, filename)
+
+ if removable:
+ dirname = os.path.dirname(media_path)
+ basename = os.path.basename(media_path)
+ digest = hashlib.md5(bytes(basename, encoding='UTF-8')).hexdigest()
+ thumbdir = os.path.join(dirname, '.sh_thumbnails')
+ else:
+ uri = 'file://' + media_path
+ digest = hashlib.md5(bytes(uri, encoding='UTF-8')).hexdigest()
+ thumbdir = os.path.join(self.homedir, '.cache', 'thumbnails')
+
+ path = os.path.join(thumbdir, 'normal', digest + '.png')
+ self.assertTrue(os.path.isfile(path))
diff --git a/tracker/automated/test-tracker.py b/tracker/automated/test-tracker.py
index d7eb21b5a0f213019cc78fc61d1ae41dd1e29c90..682c7083dabd24f72b006dffe17fd9b3413925c7 100755
--- a/tracker/automated/test-tracker.py
+++ b/tracker/automated/test-tracker.py
@@ -10,7 +10,6 @@
import unittest
import os
-import hashlib
import sys
from gi.repository import GLib
@@ -22,75 +21,26 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__),
os.pardir, os.pardir))
from apertis_tests_lib.tracker import TrackerIndexerMixin
from apertis_tests_lib.grilo import GriloBrowserMixin
+from apertis_tests_lib.tumbler import TumblerMixin
from apertis_tests_lib import ApertisTest
from apertis_tests_lib import LONG_JPEG_NAME
-class TrackerTest(ApertisTest, TrackerIndexerMixin, GriloBrowserMixin):
+class TrackerTest(ApertisTest, TrackerIndexerMixin, GriloBrowserMixin,
+ TumblerMixin):
def __init__(self, *args, **kwargs):
ApertisTest.__init__(self, *args, **kwargs)
TrackerIndexerMixin.__init__(self)
GriloBrowserMixin.__init__(self)
+ TumblerMixin.__init__(self)
def setUp(self):
- self.loop = GLib.MainLoop.new(None, False)
- self.homedir = os.path.expanduser("~")
self.copy_medias(self.homedir)
- # Monitor thumbnail creation to know when it's done. The DBus API
- # doesn't have a method to query the initial state but we can be
- # reasonably sure it's not thumbnailing anything at this point.
- self.tumbler_queue = []
- self.tumbler = Gio.DBusProxy.new_for_bus_sync(
- Gio.BusType.SESSION,
- Gio.DBusProxyFlags.DO_NOT_LOAD_PROPERTIES,
- None,
- 'org.freedesktop.thumbnails.Thumbnailer1',
- '/org/freedesktop/thumbnails/Thumbnailer1',
- 'org.freedesktop.thumbnails.Thumbnailer1',
- None)
- self.tumbler.connect('g-signal', self.tumbler_signal_cb)
-
- # Spy on unicast signals that weren't meant for us, because
- # the Thumbnailer API uses those, and we want to use them to
- # determine when it has finished. GDBus doesn't have
- # high-level API for this sort of nonsense so we do it the hard way.
- match_rule = ("type=signal,"
- "sender='org.freedesktop.thumbnails.Thumbnailer1',"
- "eavesdrop=true")
- conn = self.tumbler.get_connection()
- conn.call_sync('org.freedesktop.DBus',
- '/org/freedesktop/DBus',
- 'org.freedesktop.DBus',
- 'AddMatch',
- GLib.Variant('(s)', (match_rule,)),
- None, Gio.DBusCallFlags.NONE, -1, None)
-
def tearDown(self):
# Keep everything in place for further manual checks
pass
- def tumbler_signal_cb(self, proxy, sender_name, signal_name, parameters):
- print('TrackerTest: Received tumbler signal:', signal_name)
-
- if signal_name == 'Started':
- child = parameters.get_child_value(0)
- self.tumbler_queue.append(child.get_uint32())
- elif signal_name == 'Finished':
- child = parameters.get_child_value(0)
- self.tumbler_queue.remove(child.get_uint32())
- elif signal_name == 'Error':
- child = parameters.get_child_value(1)
- uris = child.get_strv()
- child = parameters.get_child_value(3)
- msg = child.get_string()
- raise Exception("Error creating thumbnail for %s: %s" %
- ', '.join(uris), msg)
-
- def tumbler_get_supported_cb(self, source, result):
- self.tumbler.call_finish(result)
- self.loop.quit()
-
def tracker_config_tests(self):
print("TrackerTest: config tests")
settings = Gio.Settings.new("org.freedesktop.Tracker.Miner.Files")
@@ -133,31 +83,13 @@ class TrackerTest(ApertisTest, TrackerIndexerMixin, GriloBrowserMixin):
self.wait(False)
self.assert_not_indexed(filename)
- def assert_has_thumbnail(self, filename, root=None):
- # Note that this is the path for local storage only, not for removable
- # devices.
- if not root:
- root = self.homedir
- uri = 'file://%s/%s' % (root, filename)
- uri_hash = hashlib.md5(bytes(uri, encoding='UTF-8')).hexdigest()
- path = '%s/.cache/thumbnails/normal/%s.png' % (self.homedir, uri_hash)
- self.assertTrue(os.path.isfile(path))
+ def assert_has_thumbnail(self, filename):
+ self.tumbler_assert_thumbnailed(self.homedir, filename)
def thumbnail_tests(self):
print("TrackerTest: thumbnail tests")
- # Drain the DBus queue to make sure we received all signals
- self.tumbler.call('GetSupported',
- GLib.Variant.new_tuple(),
- Gio.DBusCallFlags.NONE,
- -1,
- None,
- self.tumbler_get_supported_cb)
- self.loop.run()
-
- context = self.loop.get_context()
- while len(self.tumbler_queue) > 0:
- context.iteration(True)
+ self.tumbler_drain_queue()
self.assert_has_thumbnail('Documents/lorem_presentation.odp')
self.assert_has_thumbnail('Documents/lorem_spreadsheet.ods')
@@ -168,8 +100,8 @@ class TrackerTest(ApertisTest, TrackerIndexerMixin, GriloBrowserMixin):
# Make sure it thumbnailed shared medias as well. This particular OGG
# was known to have issues for thumbnailing.
- self.assert_has_thumbnail('Music/Dee_Yan-Key_-_Lockung.ogg',
- '/home/shared')
+ self.tumbler_assert_thumbnailed('/home/shared',
+ 'Music/Dee_Yan-Key_-_Lockung.ogg')
def grl_assert_common(self, medias, filename):
self.assertTrue(filename in medias)
diff --git a/tracker/manual/test-removable-device.py b/tracker/manual/test-removable-device.py
index d320c15d795a2f7c415efd61571d890aac7599bc..0c267ed323f99e80cad5c771f055552cde7a0354 100755
--- a/tracker/manual/test-removable-device.py
+++ b/tracker/manual/test-removable-device.py
@@ -19,16 +19,18 @@ from gi.repository import Gio
sys.path.insert(0, os.path.join(os.path.dirname(__file__),
os.pardir, os.pardir))
from apertis_tests_lib.tracker import TrackerIndexerMixin
+from apertis_tests_lib.tumbler import TumblerMixin
from apertis_tests_lib import ApertisTest
+from apertis_tests_lib import LONG_JPEG_NAME
-class TestRemovableDevice(ApertisTest, TrackerIndexerMixin):
+class TestRemovableDevice(ApertisTest, TrackerIndexerMixin, TumblerMixin):
def __init__(self, *args, **kwargs):
ApertisTest.__init__(self, *args, **kwargs)
TrackerIndexerMixin.__init__(self)
+ TumblerMixin.__init__(self)
def setUp(self):
- self.loop = GLib.MainLoop.new(None, False)
self.monitor = Gio.VolumeMonitor.get()
self.monitor.connect('mount-added', self.mount_added_cb)
@@ -40,6 +42,9 @@ class TestRemovableDevice(ApertisTest, TrackerIndexerMixin):
self.path = mount.get_root().get_path()
self.loop.quit()
+ def assert_has_thumbnail(self, filename):
+ self.tumbler_assert_thumbnailed(self.path, filename)
+
def test_removable_device(self):
print('Please insert storage ...')
self.loop.run()
@@ -49,5 +54,13 @@ class TestRemovableDevice(ApertisTest, TrackerIndexerMixin):
self.start()
self.assert_all_indexed(self.path)
+ self.tumbler_drain_queue()
+ self.assert_has_thumbnail('Documents/lorem_presentation.odp')
+ self.assert_has_thumbnail('Documents/lorem_spreadsheet.ods')
+ self.assert_has_thumbnail('Documents/more_lorem_ipsum.odt')
+ self.assert_has_thumbnail('Pictures/' + LONG_JPEG_NAME)
+ self.assert_has_thumbnail('Pictures/collabora-logo-big.png')
+ self.assert_has_thumbnail('Videos/big_buck_bunny_smaller.ogv')
+
if __name__ == "__main__":
unittest.main()