From ec46c2b3955d36a2a90741a48f2b426b341bbba0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tomasz=20Mi=C4=85sko?= Date: Sat, 10 Nov 2018 00:00:00 +0000 Subject: tests: Add tests for dconf utility tool --- bin/meson.build | 2 +- service/meson.build | 2 +- tests/meson.build | 14 +++ tests/test-dconf.py | 334 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 350 insertions(+), 2 deletions(-) create mode 100755 tests/test-dconf.py diff --git a/bin/meson.build b/bin/meson.build index 763a10b..6fd4ca2 100644 --- a/bin/meson.build +++ b/bin/meson.build @@ -11,7 +11,7 @@ bin_deps = [ valac.find_library('posix'), ] -executable( +dconf = executable( 'dconf', sources, include_directories: top_inc, diff --git a/service/meson.build b/service/meson.build index 7d54805..d92b982 100644 --- a/service/meson.build +++ b/service/meson.build @@ -51,7 +51,7 @@ libdconf_service_dep = declare_dependency( sources: dconf_generated, ) -executable( +dconf_service = executable( 'dconf-service', sources, include_directories: top_inc, diff --git a/tests/meson.build b/tests/meson.build index d1471b4..247ad76 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -45,3 +45,17 @@ foreach unit_test: unit_tests test(unit_test[0], exe, is_parallel: false, env: envs) endforeach + +python3 = find_program('python3', required: false) +dbus_daemon = find_program('dbus-daemon', required: false) + +if python3.found() and dbus_daemon.found() + test_dconf_py = find_program('test-dconf.py') + test( + 'dconf', + test_dconf_py, + args: [dconf.full_path(), dconf_service.full_path()] + ) +else + message('Skipping dconf test because python3 or dbus-daemon is not available') +endif diff --git a/tests/test-dconf.py b/tests/test-dconf.py new file mode 100755 index 0000000..6e804d7 --- /dev/null +++ b/tests/test-dconf.py @@ -0,0 +1,334 @@ +#!/usr/bin/env python3 +# +# Copyright © 2018 Tomasz Miąsko +# +# 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 licence, 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, see . + +import os +import subprocess +import sys +import tempfile +import time +import unittest + +from textwrap import dedent + +DAEMON_CONFIG = ''' + + + session + + unix:tmpdir=/tmp + {servicedir} + EXTERNAL + + + + + + +''' + +SERVICE_CONFIG = '''\ +[D-BUS Service] +Name={name} +Exec={exec} +''' + + +def dconf(*args, **kwargs): + argv = [dconf_exe] + argv.extend(args) + + # Setup convenient defaults: + kwargs.setdefault('check', True) + kwargs.setdefault('stdout', subprocess.PIPE) + kwargs.setdefault('universal_newlines', True) + + return subprocess.run(argv, **kwargs) + + +def dconf_read(key): + return dconf('read', key).stdout.rstrip('\n') + + +def dconf_write(key, value): + dconf('write', key, value) + + +def dconf_list(key): + lines = dconf('list', key).stdout.splitlines() + # FIXME: Change dconf to produce sorted output. + lines.sort() + return lines + + +def dconf_watch(path): + args = [dconf_exe, 'watch', path] + return subprocess.Popen(args, + stdout=subprocess.PIPE, + universal_newlines=True) + + +class DBusTest(unittest.TestCase): + + def setUp(self): + self.temporary_dir = tempfile.TemporaryDirectory() + + self.runtime_dir = os.path.join(self.temporary_dir.name, 'run') + self.config_home = os.path.join(self.temporary_dir.name, 'config') + self.dbus_dir = os.path.join(self.temporary_dir.name, 'dbus-1') + + os.mkdir(self.runtime_dir, mode=0o700) + os.mkdir(self.config_home, mode=0o700) + os.mkdir(self.dbus_dir, mode=0o700) + + os.environ['XDG_RUNTIME_DIR'] = self.runtime_dir + os.environ['XDG_CONFIG_HOME'] = self.config_home + + # Prepare dbus-daemon config. + dbus_daemon_config = os.path.join(self.dbus_dir, 'session.conf') + with open(dbus_daemon_config, 'w') as file: + file.write(DAEMON_CONFIG.format(servicedir=self.dbus_dir)) + + # Prepare service config. + name = 'ca.desrt.dconf' + path = os.path.join(self.dbus_dir, '{}.service'.format(name)) + with open(path, 'w') as file: + config = SERVICE_CONFIG.format(name=name, exec=dconf_service_exe) + file.write(config) + + # Pipe where daemon will write its address. + read_fd, write_fd = os.pipe2(0) + + args = ['dbus-daemon', + '--config-file={}'.format(dbus_daemon_config), + '--nofork', + '--print-address={}'.format(write_fd)] + + # Start daemon + self.dbus_daemon_process = subprocess.Popen(args, pass_fds=[write_fd]) + + # Close our writing end of pipe. Daemon closes its own writing end of + # pipe after printing address, so subsequent reads shouldn't block. + os.close(write_fd) + + with os.fdopen(read_fd) as f: + dbus_address = f.read().rstrip() + + # Prepare environment + os.environ['DBUS_SESSION_BUS_ADDRESS'] = dbus_address + + def tearDown(self): + # Terminate dbus-daemon. + p = self.dbus_daemon_process + try: + p.terminate() + p.wait(timeout=0.5) + except subprocess.TimeoutExpired: + p.kill() + p.wait() + + self.temporary_dir.cleanup() + + def test_read_nonexisiting(self): + """Reading missing key produces no output. """ + + self.assertEqual('', dconf_read('/key')) + + def test_write_read(self): + """Read returns previously written value.""" + + dconf('write', '/key', '0') + self.assertEqual('0', dconf_read('/key')) + + dconf('write', '/key', '"hello there"') + self.assertEqual("'hello there'", dconf_read('/key')) + + def test_list(self): + """List returns a list of names inside given directory. + + Results include both keys and subdirectories. + """ + + dconf('write', '/org/gnome/app/fullscreen', 'true') + dconf('write', '/org/gnome/terminal/profile', '"default"') + dconf('write', '/key', '42') + + self.assertEqual(['key', 'org/'], dconf_list('/')) + + self.assertEqual(['gnome/'], dconf_list('/org/')) + + def test_list_missing(self): + """List can be used successfully with non existing directories. """ + + self.assertEqual([], dconf_list('/no-existing/directory/')) + + def test_reset_key(self): + """Reset can be used to reset a value of a single key.""" + + dconf('write', '/app/width', '1024') + dconf('write', '/app/height', '768') + dconf('write', '/app/fullscreen', 'true') + + # Sanity check. + self.assertEqual(['fullscreen', 'height', 'width'], dconf_list('/app/')) + + # Reset one key after another: + dconf('reset', '/app/fullscreen') + self.assertEqual(['height', 'width'], dconf_list('/app/')) + dconf('reset', '/app/width') + self.assertEqual(['height'], dconf_list('/app/')) + dconf('reset', '/app/height') + self.assertEqual([], dconf_list('/app/')) + + def test_reset_dir(self): + """Reseting whole directory is possible with -f option. + + It is an error not to use -f when resetting a dir. + """ + + dconf('write', '/app/a', '1') + dconf('write', '/app/b', '2') + dconf('write', '/app/c/d', '3') + dconf('write', '/x', '4') + dconf('write', '/y/z', '5') + + with self.assertRaises(subprocess.CalledProcessError) as cm: + dconf('reset', '/app/', stderr=subprocess.PIPE) + self.assertRegex(cm.exception.stderr, '-f must be given') + self.assertRegex(cm.exception.stderr, 'Usage:') + + # Nothing should be removed just yet. + self.assertTrue(['a', 'b', 'c'], dconf_list('/app/')) + + # Try again with -f. + dconf('reset', '-f', '/app/') + + # /app/ should be gone now: + self.assertEqual(['x', 'y/'], dconf_list('/')) + + def test_watch(self): + """Watch reports changes made using write command. + + Only changes made inside given subdirectory should be reported. + """ + + watch_root = dconf_watch('/') + watch_org = dconf_watch('/org/') + + # Arbitrary delay to give "dconf watch" time to set-up the watch. + # In the case this turns out to be problematic, dconf tool could be + # changed to produce debug message after `dconf_client_watch_sync`, + # so that we could synchronize on its output. + + time.sleep(0.2) + + dconf('write', '/com/a', '1') + dconf('write', '/org/b', '2') + dconf('write', '/organ/c', '3') + dconf('write', '/c', '4') + + # Again, give "dconf watch" some time to pick-up changes. + + time.sleep(0.2) + + watch_root.terminate() + watch_org.terminate() + + watch_root.wait() + watch_org.wait() + + # Watch for '/' should capture all writes. + expected = '''\ + /com/a + 1 + + /org/b + 2 + + /organ/c + 3 + + /c + 4 + + ''' + self.assertEqual(dedent(expected), watch_root.stdout.read()) + + # Watch for '/org/' should capture only a subset of all writes: + expected = '''\ + /org/b + 2 + + ''' + self.assertEqual(dedent(expected), watch_org.stdout.read()) + + def test_dump_load(self): + """Checks that output produced with dump can be used with load and + vice versa. + """ + # FIXME: This test depends on: + # * order of groups + # * order of items within groups + # Change dconf to produce output in sorted order. + + keyfile = dedent('''\ + [/] + password='secret' + + [org/editor/language/c-sharp] + tab-width=8 + + [org/editor/language/c] + tab-width=2 + + [org/editor] + window-size=(1280, 977) + ''') + + # Load and dump is identity. + dconf('load', '/', input=keyfile) + self.assertEqual(dconf('dump', '/').stdout, keyfile) + + # Copy /org/ directory to /com/. + keyfile = dconf('dump', '/org/').stdout + dconf('load', '/com/', input=keyfile) + + # Verify that /org/ and /com/ are now exactly the same. + keyfile_org = dconf('dump', '/org/').stdout + keyfile_com = dconf('dump', '/com/').stdout + self.assertEqual(keyfile_org, keyfile_com) + + +if __name__ == '__main__': + # Make sure we don't pick up mandatory profile. + mandatory_profile = '/run/dconf/user/{}'.format(os.getuid()) + assert not os.path.isfile(mandatory_profile) + + # Avoid profile sourced from environment + os.environ.pop('DCONF_PROFILE', None) + + if len(sys.argv) < 3: + message = 'Usage: {} path-to-dconf path-to-dconf-service'.format( + sys.argv[0]) + raise RuntimeError(message) + + dconf_exe, dconf_service_exe = sys.argv[1:3] + del sys.argv[1:3] + + # Run tests! + unittest.main() -- cgit v1.2.1