diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/Makefile.am | 47 | ||||
| -rw-r--r-- | test/Makefile.in | 719 | ||||
| -rw-r--r-- | test/TestSuitePythonService.service.in | 3 | ||||
| -rw-r--r-- | test/cross-test-client.py | 393 | ||||
| -rw-r--r-- | test/cross-test-server.py | 323 | ||||
| -rw-r--r-- | test/crosstest.py | 44 | ||||
| -rw-r--r-- | test/dbus_py_test.c | 126 | ||||
| -rwxr-xr-x | test/run-test.sh | 124 | ||||
| -rwxr-xr-x | test/run-with-tmp-session-bus.sh | 75 | ||||
| -rwxr-xr-x | test/test-client.py | 587 | ||||
| -rw-r--r-- | test/test-p2p.py | 95 | ||||
| -rwxr-xr-x | test/test-service.py | 344 | ||||
| -rw-r--r-- | test/test-signals.py | 160 | ||||
| -rwxr-xr-x | test/test-standalone.py | 297 | ||||
| -rw-r--r-- | test/test-unusable-main-loop.py | 39 | ||||
| -rw-r--r-- | test/tmp-session-bus.conf.in | 21 |
16 files changed, 3397 insertions, 0 deletions
diff --git a/test/Makefile.am b/test/Makefile.am new file mode 100644 index 0000000..a40c3d1 --- /dev/null +++ b/test/Makefile.am @@ -0,0 +1,47 @@ +abs_top_srcdir = @abs_top_srcdir@ +abs_top_builddir = @abs_top_builddir@ + +EXTRA_DIST = \ + cross-test-client.py \ + cross-test-server.py \ + crosstest.py \ + run-test.sh \ + run-with-tmp-session-bus.sh \ + test-client.py \ + test-p2p.py \ + test-service.py \ + test-signals.py \ + test-standalone.py \ + test-unusable-main-loop.py \ + TestSuitePythonService.service.in \ + tmp-session-bus.conf.in + +# If you try to make this noinst, libtool helpfully gives us a static +# library, which doesn't work as a Python extension: so force the install +# target not to work here instead. +pyexec_LTLIBRARIES = dbus_py_test.la + +install: + @echo "Not installing anything from test/" + +dbus_py_test_la_CPPFLAGS = -I$(top_srcdir)/include $(DBUS_CFLAGS) \ + $(PYTHON_INCLUDES) +dbus_py_test_la_LDFLAGS = -module -avoid-version \ + $(DBUS_LIBS) +dbus_py_test_la_SOURCES = dbus_py_test.c \ + $(top_srcdir)/include/dbus-python.h + +TESTS_ENVIRONMENT = DBUS_TOP_SRCDIR="$(abs_top_srcdir)" \ + DBUS_TOP_BUILDDIR="$(abs_top_builddir)" \ + DBUS_PYTHON_VERSION='$(PACKAGE_VERSION)' \ + PYTHONPATH="$(abs_top_builddir)/test/.libs:$(abs_top_srcdir):$(abs_top_srcdir)/test:$(abs_top_builddir)/_dbus_bindings/.libs:$(abs_top_builddir)/_dbus_glib_bindings/.libs" \ + PYTHON='$(PYTHON)' + +TESTS = run-test.sh + +cross-test-server: + $(TESTS_ENVIRONMENT) $(PYTHON) $(top_srcdir)/test/cross-test-server.py +cross-test-client: + $(TESTS_ENVIRONMENT) $(PYTHON) $(top_srcdir)/test/cross-test-client.py + +.PHONY: cross-test-compile cross-test-server cross-test-client diff --git a/test/Makefile.in b/test/Makefile.in new file mode 100644 index 0000000..40c85cb --- /dev/null +++ b/test/Makefile.in @@ -0,0 +1,719 @@ +# Makefile.in generated by automake 1.11.1 from Makefile.am. +# @configure_input@ + +# Copyright (C) 1994, 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002, +# 2003, 2004, 2005, 2006, 2007, 2008, 2009 Free Software Foundation, +# Inc. +# This Makefile.in is free software; the Free Software Foundation +# gives unlimited permission to copy and/or distribute it, +# with or without modifications, as long as this notice is preserved. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY, to the extent permitted by law; without +# even the implied warranty of MERCHANTABILITY or FITNESS FOR A +# PARTICULAR PURPOSE. + +@SET_MAKE@ + +VPATH = @srcdir@ +pkgdatadir = $(datadir)/@PACKAGE@ +pkgincludedir = $(includedir)/@PACKAGE@ +pkglibdir = $(libdir)/@PACKAGE@ +pkglibexecdir = $(libexecdir)/@PACKAGE@ +am__cd = CDPATH="$${ZSH_VERSION+.}$(PATH_SEPARATOR)" && cd +install_sh_DATA = $(install_sh) -c -m 644 +install_sh_PROGRAM = $(install_sh) -c +install_sh_SCRIPT = $(install_sh) -c +INSTALL_HEADER = $(INSTALL_DATA) +transform = $(program_transform_name) +NORMAL_INSTALL = : +PRE_INSTALL = : +POST_INSTALL = : +NORMAL_UNINSTALL = : +PRE_UNINSTALL = : +POST_UNINSTALL = : +build_triplet = @build@ +host_triplet = @host@ +subdir = test +DIST_COMMON = $(srcdir)/Makefile.am $(srcdir)/Makefile.in \ + $(srcdir)/TestSuitePythonService.service.in \ + $(srcdir)/tmp-session-bus.conf.in +ACLOCAL_M4 = $(top_srcdir)/aclocal.m4 +am__aclocal_m4_deps = $(top_srcdir)/m4/am-check-pymod.m4 \ + $(top_srcdir)/m4/am-check-python-headers.m4 \ + $(top_srcdir)/m4/dbus-py-add-rst2htmlflag.m4 \ + $(top_srcdir)/m4/jh-add-cflag.m4 $(top_srcdir)/m4/libtool.m4 \ + $(top_srcdir)/m4/ltoptions.m4 $(top_srcdir)/m4/ltsugar.m4 \ + $(top_srcdir)/m4/ltversion.m4 $(top_srcdir)/m4/lt~obsolete.m4 \ + $(top_srcdir)/m4/tp-compiler-flag.m4 \ + $(top_srcdir)/m4/tp-compiler-warnings.m4 \ + $(top_srcdir)/configure.ac +am__configure_deps = $(am__aclocal_m4_deps) $(CONFIGURE_DEPENDENCIES) \ + $(ACLOCAL_M4) +mkinstalldirs = $(install_sh) -d +CONFIG_HEADER = $(top_builddir)/config.h +CONFIG_CLEAN_FILES = tmp-session-bus.conf \ + TestSuitePythonService.service +CONFIG_CLEAN_VPATH_FILES = +am__vpath_adj_setup = srcdirstrip=`echo "$(srcdir)" | sed 's|.|.|g'`; +am__vpath_adj = case $$p in \ + $(srcdir)/*) f=`echo "$$p" | sed "s|^$$srcdirstrip/||"`;; \ + *) f=$$p;; \ + esac; +am__strip_dir = f=`echo $$p | sed -e 's|^.*/||'`; +am__install_max = 40 +am__nobase_strip_setup = \ + srcdirstrip=`echo "$(srcdir)" | sed 's/[].[^$$\\*|]/\\\\&/g'` +am__nobase_strip = \ + for p in $$list; do echo "$$p"; done | sed -e "s|$$srcdirstrip/||" +am__nobase_list = $(am__nobase_strip_setup); \ + for p in $$list; do echo "$$p $$p"; done | \ + sed "s| $$srcdirstrip/| |;"' / .*\//!s/ .*/ ./; s,\( .*\)/[^/]*$$,\1,' | \ + $(AWK) 'BEGIN { files["."] = "" } { files[$$2] = files[$$2] " " $$1; \ + if (++n[$$2] == $(am__install_max)) \ + { print $$2, files[$$2]; n[$$2] = 0; files[$$2] = "" } } \ + END { for (dir in files) print dir, files[dir] }' +am__base_list = \ + sed '$$!N;$$!N;$$!N;$$!N;$$!N;$$!N;$$!N;s/\n/ /g' | \ + sed '$$!N;$$!N;$$!N;$$!N;s/\n/ /g' +am__installdirs = "$(DESTDIR)$(pyexecdir)" +LTLIBRARIES = $(pyexec_LTLIBRARIES) +dbus_py_test_la_LIBADD = +am_dbus_py_test_la_OBJECTS = dbus_py_test_la-dbus_py_test.lo +dbus_py_test_la_OBJECTS = $(am_dbus_py_test_la_OBJECTS) +AM_V_lt = $(am__v_lt_$(V)) +am__v_lt_ = $(am__v_lt_$(AM_DEFAULT_VERBOSITY)) +am__v_lt_0 = --silent +dbus_py_test_la_LINK = $(LIBTOOL) $(AM_V_lt) --tag=CC \ + $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=link $(CCLD) \ + $(AM_CFLAGS) $(CFLAGS) $(dbus_py_test_la_LDFLAGS) $(LDFLAGS) \ + -o $@ +DEFAULT_INCLUDES = -I.@am__isrc@ -I$(top_builddir) +depcomp = $(SHELL) $(top_srcdir)/depcomp +am__depfiles_maybe = depfiles +am__mv = mv -f +COMPILE = $(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(AM_CPPFLAGS) \ + $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) +LTCOMPILE = $(LIBTOOL) $(AM_V_lt) --tag=CC $(AM_LIBTOOLFLAGS) \ + $(LIBTOOLFLAGS) --mode=compile $(CC) $(DEFS) \ + $(DEFAULT_INCLUDES) $(INCLUDES) $(AM_CPPFLAGS) $(CPPFLAGS) \ + $(AM_CFLAGS) $(CFLAGS) +AM_V_CC = $(am__v_CC_$(V)) +am__v_CC_ = $(am__v_CC_$(AM_DEFAULT_VERBOSITY)) +am__v_CC_0 = @echo " CC " $@; +AM_V_at = $(am__v_at_$(V)) +am__v_at_ = $(am__v_at_$(AM_DEFAULT_VERBOSITY)) +am__v_at_0 = @ +CCLD = $(CC) +LINK = $(LIBTOOL) $(AM_V_lt) --tag=CC $(AM_LIBTOOLFLAGS) \ + $(LIBTOOLFLAGS) --mode=link $(CCLD) $(AM_CFLAGS) $(CFLAGS) \ + $(AM_LDFLAGS) $(LDFLAGS) -o $@ +AM_V_CCLD = $(am__v_CCLD_$(V)) +am__v_CCLD_ = $(am__v_CCLD_$(AM_DEFAULT_VERBOSITY)) +am__v_CCLD_0 = @echo " CCLD " $@; +AM_V_GEN = $(am__v_GEN_$(V)) +am__v_GEN_ = $(am__v_GEN_$(AM_DEFAULT_VERBOSITY)) +am__v_GEN_0 = @echo " GEN " $@; +SOURCES = $(dbus_py_test_la_SOURCES) +DIST_SOURCES = $(dbus_py_test_la_SOURCES) +ETAGS = etags +CTAGS = ctags +am__tty_colors = \ +red=; grn=; lgn=; blu=; std= +DISTFILES = $(DIST_COMMON) $(DIST_SOURCES) $(TEXINFOS) $(EXTRA_DIST) +ACLOCAL = @ACLOCAL@ +AMTAR = @AMTAR@ +AM_DEFAULT_VERBOSITY = @AM_DEFAULT_VERBOSITY@ +AR = @AR@ +AUTOCONF = @AUTOCONF@ +AUTOHEADER = @AUTOHEADER@ +AUTOMAKE = @AUTOMAKE@ +AWK = @AWK@ +CC = @CC@ +CCDEPMODE = @CCDEPMODE@ +CFLAGS = @CFLAGS@ +CPP = @CPP@ +CPPFLAGS = @CPPFLAGS@ +CYGPATH_W = @CYGPATH_W@ +DBUS_CFLAGS = @DBUS_CFLAGS@ +DBUS_GLIB_CFLAGS = @DBUS_GLIB_CFLAGS@ +DBUS_GLIB_LIBS = @DBUS_GLIB_LIBS@ +DBUS_LIBS = @DBUS_LIBS@ +DBUS_PYTHON_MAJOR_VERSION = @DBUS_PYTHON_MAJOR_VERSION@ +DBUS_PYTHON_MICRO_VERSION = @DBUS_PYTHON_MICRO_VERSION@ +DBUS_PYTHON_MINOR_VERSION = @DBUS_PYTHON_MINOR_VERSION@ +DEFS = @DEFS@ +DEPDIR = @DEPDIR@ +DSYMUTIL = @DSYMUTIL@ +DUMPBIN = @DUMPBIN@ +ECHO_C = @ECHO_C@ +ECHO_N = @ECHO_N@ +ECHO_T = @ECHO_T@ +EGREP = @EGREP@ +EPYDOC = @EPYDOC@ +EXEEXT = @EXEEXT@ +FGREP = @FGREP@ +GREP = @GREP@ +INSTALL = @INSTALL@ +INSTALL_DATA = @INSTALL_DATA@ +INSTALL_PROGRAM = @INSTALL_PROGRAM@ +INSTALL_SCRIPT = @INSTALL_SCRIPT@ +INSTALL_STRIP_PROGRAM = @INSTALL_STRIP_PROGRAM@ +LD = @LD@ +LDFLAGS = @LDFLAGS@ +LIBOBJS = @LIBOBJS@ +LIBS = @LIBS@ +LIBTOOL = @LIBTOOL@ +LIPO = @LIPO@ +LN_S = @LN_S@ +LTLIBOBJS = @LTLIBOBJS@ +MAKEINFO = @MAKEINFO@ +MKDIR_P = @MKDIR_P@ +NM = @NM@ +NMEDIT = @NMEDIT@ +OBJDUMP = @OBJDUMP@ +OBJEXT = @OBJEXT@ +OTOOL = @OTOOL@ +OTOOL64 = @OTOOL64@ +PACKAGE = @PACKAGE@ +PACKAGE_BUGREPORT = @PACKAGE_BUGREPORT@ +PACKAGE_NAME = @PACKAGE_NAME@ +PACKAGE_STRING = @PACKAGE_STRING@ +PACKAGE_TARNAME = @PACKAGE_TARNAME@ +PACKAGE_URL = @PACKAGE_URL@ +PACKAGE_VERSION = @PACKAGE_VERSION@ +PATH_SEPARATOR = @PATH_SEPARATOR@ +PKG_CONFIG = @PKG_CONFIG@ +PLATFORM = @PLATFORM@ +PYTHON = @PYTHON@ +PYTHON_EXEC_PREFIX = @PYTHON_EXEC_PREFIX@ +PYTHON_INCLUDES = @PYTHON_INCLUDES@ +PYTHON_PLATFORM = @PYTHON_PLATFORM@ +PYTHON_PREFIX = @PYTHON_PREFIX@ +PYTHON_VERSION = @PYTHON_VERSION@ +RANLIB = @RANLIB@ +RST2HTML = @RST2HTML@ +RST2HTMLFLAGS = @RST2HTMLFLAGS@ +SED = @SED@ +SET_MAKE = @SET_MAKE@ +SHELL = @SHELL@ +STRIP = @STRIP@ +VERSION = @VERSION@ +abs_builddir = @abs_builddir@ +abs_srcdir = @abs_srcdir@ +abs_top_builddir = @abs_top_builddir@ +abs_top_srcdir = @abs_top_srcdir@ +ac_ct_CC = @ac_ct_CC@ +ac_ct_DUMPBIN = @ac_ct_DUMPBIN@ +am__include = @am__include@ +am__leading_dot = @am__leading_dot@ +am__quote = @am__quote@ +am__tar = @am__tar@ +am__untar = @am__untar@ +bindir = @bindir@ +build = @build@ +build_alias = @build_alias@ +build_cpu = @build_cpu@ +build_os = @build_os@ +build_vendor = @build_vendor@ +builddir = @builddir@ +datadir = @datadir@ +datarootdir = @datarootdir@ +docdir = @docdir@ +dvidir = @dvidir@ +exec_prefix = @exec_prefix@ +host = @host@ +host_alias = @host_alias@ +host_cpu = @host_cpu@ +host_os = @host_os@ +host_vendor = @host_vendor@ +htmldir = @htmldir@ +includedir = @includedir@ +infodir = @infodir@ +install_sh = @install_sh@ +libdir = @libdir@ +libexecdir = @libexecdir@ +localedir = @localedir@ +localstatedir = @localstatedir@ +lt_ECHO = @lt_ECHO@ +mandir = @mandir@ +mkdir_p = @mkdir_p@ +oldincludedir = @oldincludedir@ +pdfdir = @pdfdir@ +pkgpyexecdir = @pkgpyexecdir@ +pkgpythondir = @pkgpythondir@ +prefix = @prefix@ +program_transform_name = @program_transform_name@ +psdir = @psdir@ +pyexecdir = @pyexecdir@ +pythondir = @pythondir@ +sbindir = @sbindir@ +sharedstatedir = @sharedstatedir@ +srcdir = @srcdir@ +sysconfdir = @sysconfdir@ +target_alias = @target_alias@ +top_build_prefix = @top_build_prefix@ +top_builddir = @top_builddir@ +top_srcdir = @top_srcdir@ +EXTRA_DIST = \ + cross-test-client.py \ + cross-test-server.py \ + crosstest.py \ + run-test.sh \ + run-with-tmp-session-bus.sh \ + test-client.py \ + test-p2p.py \ + test-service.py \ + test-signals.py \ + test-standalone.py \ + test-unusable-main-loop.py \ + TestSuitePythonService.service.in \ + tmp-session-bus.conf.in + + +# If you try to make this noinst, libtool helpfully gives us a static +# library, which doesn't work as a Python extension: so force the install +# target not to work here instead. +pyexec_LTLIBRARIES = dbus_py_test.la +dbus_py_test_la_CPPFLAGS = -I$(top_srcdir)/include $(DBUS_CFLAGS) \ + $(PYTHON_INCLUDES) + +dbus_py_test_la_LDFLAGS = -module -avoid-version \ + $(DBUS_LIBS) + +dbus_py_test_la_SOURCES = dbus_py_test.c \ + $(top_srcdir)/include/dbus-python.h + +TESTS_ENVIRONMENT = DBUS_TOP_SRCDIR="$(abs_top_srcdir)" \ + DBUS_TOP_BUILDDIR="$(abs_top_builddir)" \ + DBUS_PYTHON_VERSION='$(PACKAGE_VERSION)' \ + PYTHONPATH="$(abs_top_builddir)/test/.libs:$(abs_top_srcdir):$(abs_top_srcdir)/test:$(abs_top_builddir)/_dbus_bindings/.libs:$(abs_top_builddir)/_dbus_glib_bindings/.libs" \ + PYTHON='$(PYTHON)' + +TESTS = run-test.sh +all: all-am + +.SUFFIXES: +.SUFFIXES: .c .lo .o .obj +$(srcdir)/Makefile.in: $(srcdir)/Makefile.am $(am__configure_deps) + @for dep in $?; do \ + case '$(am__configure_deps)' in \ + *$$dep*) \ + ( cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh ) \ + && { if test -f $@; then exit 0; else break; fi; }; \ + exit 1;; \ + esac; \ + done; \ + echo ' cd $(top_srcdir) && $(AUTOMAKE) --gnu test/Makefile'; \ + $(am__cd) $(top_srcdir) && \ + $(AUTOMAKE) --gnu test/Makefile +.PRECIOUS: Makefile +Makefile: $(srcdir)/Makefile.in $(top_builddir)/config.status + @case '$?' in \ + *config.status*) \ + cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh;; \ + *) \ + echo ' cd $(top_builddir) && $(SHELL) ./config.status $(subdir)/$@ $(am__depfiles_maybe)'; \ + cd $(top_builddir) && $(SHELL) ./config.status $(subdir)/$@ $(am__depfiles_maybe);; \ + esac; + +$(top_builddir)/config.status: $(top_srcdir)/configure $(CONFIG_STATUS_DEPENDENCIES) + cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh + +$(top_srcdir)/configure: $(am__configure_deps) + cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh +$(ACLOCAL_M4): $(am__aclocal_m4_deps) + cd $(top_builddir) && $(MAKE) $(AM_MAKEFLAGS) am--refresh +$(am__aclocal_m4_deps): +tmp-session-bus.conf: $(top_builddir)/config.status $(srcdir)/tmp-session-bus.conf.in + cd $(top_builddir) && $(SHELL) ./config.status $(subdir)/$@ +TestSuitePythonService.service: $(top_builddir)/config.status $(srcdir)/TestSuitePythonService.service.in + cd $(top_builddir) && $(SHELL) ./config.status $(subdir)/$@ +install-pyexecLTLIBRARIES: $(pyexec_LTLIBRARIES) + @$(NORMAL_INSTALL) + test -z "$(pyexecdir)" || $(MKDIR_P) "$(DESTDIR)$(pyexecdir)" + @list='$(pyexec_LTLIBRARIES)'; test -n "$(pyexecdir)" || list=; \ + list2=; for p in $$list; do \ + if test -f $$p; then \ + list2="$$list2 $$p"; \ + else :; fi; \ + done; \ + test -z "$$list2" || { \ + echo " $(LIBTOOL) $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=install $(INSTALL) $(INSTALL_STRIP_FLAG) $$list2 '$(DESTDIR)$(pyexecdir)'"; \ + $(LIBTOOL) $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=install $(INSTALL) $(INSTALL_STRIP_FLAG) $$list2 "$(DESTDIR)$(pyexecdir)"; \ + } + +uninstall-pyexecLTLIBRARIES: + @$(NORMAL_UNINSTALL) + @list='$(pyexec_LTLIBRARIES)'; test -n "$(pyexecdir)" || list=; \ + for p in $$list; do \ + $(am__strip_dir) \ + echo " $(LIBTOOL) $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=uninstall rm -f '$(DESTDIR)$(pyexecdir)/$$f'"; \ + $(LIBTOOL) $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=uninstall rm -f "$(DESTDIR)$(pyexecdir)/$$f"; \ + done + +clean-pyexecLTLIBRARIES: + -test -z "$(pyexec_LTLIBRARIES)" || rm -f $(pyexec_LTLIBRARIES) + @list='$(pyexec_LTLIBRARIES)'; for p in $$list; do \ + dir="`echo $$p | sed -e 's|/[^/]*$$||'`"; \ + test "$$dir" != "$$p" || dir=.; \ + echo "rm -f \"$${dir}/so_locations\""; \ + rm -f "$${dir}/so_locations"; \ + done +dbus_py_test.la: $(dbus_py_test_la_OBJECTS) $(dbus_py_test_la_DEPENDENCIES) + $(AM_V_CCLD)$(dbus_py_test_la_LINK) -rpath $(pyexecdir) $(dbus_py_test_la_OBJECTS) $(dbus_py_test_la_LIBADD) $(LIBS) + +mostlyclean-compile: + -rm -f *.$(OBJEXT) + +distclean-compile: + -rm -f *.tab.c + +@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/dbus_py_test_la-dbus_py_test.Plo@am__quote@ + +.c.o: +@am__fastdepCC_TRUE@ $(AM_V_CC)$(COMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ $< +@am__fastdepCC_TRUE@ $(AM_V_at)$(am__mv) $(DEPDIR)/$*.Tpo $(DEPDIR)/$*.Po +@am__fastdepCC_FALSE@ $(AM_V_CC) @AM_BACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ source='$<' object='$@' libtool=no @AMDEPBACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@ +@am__fastdepCC_FALSE@ $(COMPILE) -c $< + +.c.obj: +@am__fastdepCC_TRUE@ $(AM_V_CC)$(COMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ `$(CYGPATH_W) '$<'` +@am__fastdepCC_TRUE@ $(AM_V_at)$(am__mv) $(DEPDIR)/$*.Tpo $(DEPDIR)/$*.Po +@am__fastdepCC_FALSE@ $(AM_V_CC) @AM_BACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ source='$<' object='$@' libtool=no @AMDEPBACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@ +@am__fastdepCC_FALSE@ $(COMPILE) -c `$(CYGPATH_W) '$<'` + +.c.lo: +@am__fastdepCC_TRUE@ $(AM_V_CC)$(LTCOMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ $< +@am__fastdepCC_TRUE@ $(AM_V_at)$(am__mv) $(DEPDIR)/$*.Tpo $(DEPDIR)/$*.Plo +@am__fastdepCC_FALSE@ $(AM_V_CC) @AM_BACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ source='$<' object='$@' libtool=yes @AMDEPBACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@ +@am__fastdepCC_FALSE@ $(LTCOMPILE) -c -o $@ $< + +dbus_py_test_la-dbus_py_test.lo: dbus_py_test.c +@am__fastdepCC_TRUE@ $(AM_V_CC)$(LIBTOOL) $(AM_V_lt) --tag=CC $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=compile $(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(dbus_py_test_la_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -MT dbus_py_test_la-dbus_py_test.lo -MD -MP -MF $(DEPDIR)/dbus_py_test_la-dbus_py_test.Tpo -c -o dbus_py_test_la-dbus_py_test.lo `test -f 'dbus_py_test.c' || echo '$(srcdir)/'`dbus_py_test.c +@am__fastdepCC_TRUE@ $(AM_V_at)$(am__mv) $(DEPDIR)/dbus_py_test_la-dbus_py_test.Tpo $(DEPDIR)/dbus_py_test_la-dbus_py_test.Plo +@am__fastdepCC_FALSE@ $(AM_V_CC) @AM_BACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ source='dbus_py_test.c' object='dbus_py_test_la-dbus_py_test.lo' libtool=yes @AMDEPBACKSLASH@ +@AMDEP_TRUE@@am__fastdepCC_FALSE@ DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@ +@am__fastdepCC_FALSE@ $(LIBTOOL) $(AM_V_lt) --tag=CC $(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=compile $(CC) $(DEFS) $(DEFAULT_INCLUDES) $(INCLUDES) $(dbus_py_test_la_CPPFLAGS) $(CPPFLAGS) $(AM_CFLAGS) $(CFLAGS) -c -o dbus_py_test_la-dbus_py_test.lo `test -f 'dbus_py_test.c' || echo '$(srcdir)/'`dbus_py_test.c + +mostlyclean-libtool: + -rm -f *.lo + +clean-libtool: + -rm -rf .libs _libs + +ID: $(HEADERS) $(SOURCES) $(LISP) $(TAGS_FILES) + list='$(SOURCES) $(HEADERS) $(LISP) $(TAGS_FILES)'; \ + unique=`for i in $$list; do \ + if test -f "$$i"; then echo $$i; else echo $(srcdir)/$$i; fi; \ + done | \ + $(AWK) '{ files[$$0] = 1; nonempty = 1; } \ + END { if (nonempty) { for (i in files) print i; }; }'`; \ + mkid -fID $$unique +tags: TAGS + +TAGS: $(HEADERS) $(SOURCES) $(TAGS_DEPENDENCIES) \ + $(TAGS_FILES) $(LISP) + set x; \ + here=`pwd`; \ + list='$(SOURCES) $(HEADERS) $(LISP) $(TAGS_FILES)'; \ + unique=`for i in $$list; do \ + if test -f "$$i"; then echo $$i; else echo $(srcdir)/$$i; fi; \ + done | \ + $(AWK) '{ files[$$0] = 1; nonempty = 1; } \ + END { if (nonempty) { for (i in files) print i; }; }'`; \ + shift; \ + if test -z "$(ETAGS_ARGS)$$*$$unique"; then :; else \ + test -n "$$unique" || unique=$$empty_fix; \ + if test $$# -gt 0; then \ + $(ETAGS) $(ETAGSFLAGS) $(AM_ETAGSFLAGS) $(ETAGS_ARGS) \ + "$$@" $$unique; \ + else \ + $(ETAGS) $(ETAGSFLAGS) $(AM_ETAGSFLAGS) $(ETAGS_ARGS) \ + $$unique; \ + fi; \ + fi +ctags: CTAGS +CTAGS: $(HEADERS) $(SOURCES) $(TAGS_DEPENDENCIES) \ + $(TAGS_FILES) $(LISP) + list='$(SOURCES) $(HEADERS) $(LISP) $(TAGS_FILES)'; \ + unique=`for i in $$list; do \ + if test -f "$$i"; then echo $$i; else echo $(srcdir)/$$i; fi; \ + done | \ + $(AWK) '{ files[$$0] = 1; nonempty = 1; } \ + END { if (nonempty) { for (i in files) print i; }; }'`; \ + test -z "$(CTAGS_ARGS)$$unique" \ + || $(CTAGS) $(CTAGSFLAGS) $(AM_CTAGSFLAGS) $(CTAGS_ARGS) \ + $$unique + +GTAGS: + here=`$(am__cd) $(top_builddir) && pwd` \ + && $(am__cd) $(top_srcdir) \ + && gtags -i $(GTAGS_ARGS) "$$here" + +distclean-tags: + -rm -f TAGS ID GTAGS GRTAGS GSYMS GPATH tags + +check-TESTS: $(TESTS) + @failed=0; all=0; xfail=0; xpass=0; skip=0; \ + srcdir=$(srcdir); export srcdir; \ + list=' $(TESTS) '; \ + $(am__tty_colors); \ + if test -n "$$list"; then \ + for tst in $$list; do \ + if test -f ./$$tst; then dir=./; \ + elif test -f $$tst; then dir=; \ + else dir="$(srcdir)/"; fi; \ + if $(TESTS_ENVIRONMENT) $${dir}$$tst; then \ + all=`expr $$all + 1`; \ + case " $(XFAIL_TESTS) " in \ + *[\ \ ]$$tst[\ \ ]*) \ + xpass=`expr $$xpass + 1`; \ + failed=`expr $$failed + 1`; \ + col=$$red; res=XPASS; \ + ;; \ + *) \ + col=$$grn; res=PASS; \ + ;; \ + esac; \ + elif test $$? -ne 77; then \ + all=`expr $$all + 1`; \ + case " $(XFAIL_TESTS) " in \ + *[\ \ ]$$tst[\ \ ]*) \ + xfail=`expr $$xfail + 1`; \ + col=$$lgn; res=XFAIL; \ + ;; \ + *) \ + failed=`expr $$failed + 1`; \ + col=$$red; res=FAIL; \ + ;; \ + esac; \ + else \ + skip=`expr $$skip + 1`; \ + col=$$blu; res=SKIP; \ + fi; \ + echo "$${col}$$res$${std}: $$tst"; \ + done; \ + if test "$$all" -eq 1; then \ + tests="test"; \ + All=""; \ + else \ + tests="tests"; \ + All="All "; \ + fi; \ + if test "$$failed" -eq 0; then \ + if test "$$xfail" -eq 0; then \ + banner="$$All$$all $$tests passed"; \ + else \ + if test "$$xfail" -eq 1; then failures=failure; else failures=failures; fi; \ + banner="$$All$$all $$tests behaved as expected ($$xfail expected $$failures)"; \ + fi; \ + else \ + if test "$$xpass" -eq 0; then \ + banner="$$failed of $$all $$tests failed"; \ + else \ + if test "$$xpass" -eq 1; then passes=pass; else passes=passes; fi; \ + banner="$$failed of $$all $$tests did not behave as expected ($$xpass unexpected $$passes)"; \ + fi; \ + fi; \ + dashes="$$banner"; \ + skipped=""; \ + if test "$$skip" -ne 0; then \ + if test "$$skip" -eq 1; then \ + skipped="($$skip test was not run)"; \ + else \ + skipped="($$skip tests were not run)"; \ + fi; \ + test `echo "$$skipped" | wc -c` -le `echo "$$banner" | wc -c` || \ + dashes="$$skipped"; \ + fi; \ + report=""; \ + if test "$$failed" -ne 0 && test -n "$(PACKAGE_BUGREPORT)"; then \ + report="Please report to $(PACKAGE_BUGREPORT)"; \ + test `echo "$$report" | wc -c` -le `echo "$$banner" | wc -c` || \ + dashes="$$report"; \ + fi; \ + dashes=`echo "$$dashes" | sed s/./=/g`; \ + if test "$$failed" -eq 0; then \ + echo "$$grn$$dashes"; \ + else \ + echo "$$red$$dashes"; \ + fi; \ + echo "$$banner"; \ + test -z "$$skipped" || echo "$$skipped"; \ + test -z "$$report" || echo "$$report"; \ + echo "$$dashes$$std"; \ + test "$$failed" -eq 0; \ + else :; fi + +distdir: $(DISTFILES) + @srcdirstrip=`echo "$(srcdir)" | sed 's/[].[^$$\\*]/\\\\&/g'`; \ + topsrcdirstrip=`echo "$(top_srcdir)" | sed 's/[].[^$$\\*]/\\\\&/g'`; \ + list='$(DISTFILES)'; \ + dist_files=`for file in $$list; do echo $$file; done | \ + sed -e "s|^$$srcdirstrip/||;t" \ + -e "s|^$$topsrcdirstrip/|$(top_builddir)/|;t"`; \ + case $$dist_files in \ + */*) $(MKDIR_P) `echo "$$dist_files" | \ + sed '/\//!d;s|^|$(distdir)/|;s,/[^/]*$$,,' | \ + sort -u` ;; \ + esac; \ + for file in $$dist_files; do \ + if test -f $$file || test -d $$file; then d=.; else d=$(srcdir); fi; \ + if test -d $$d/$$file; then \ + dir=`echo "/$$file" | sed -e 's,/[^/]*$$,,'`; \ + if test -d "$(distdir)/$$file"; then \ + find "$(distdir)/$$file" -type d ! -perm -700 -exec chmod u+rwx {} \;; \ + fi; \ + if test -d $(srcdir)/$$file && test $$d != $(srcdir); then \ + cp -fpR $(srcdir)/$$file "$(distdir)$$dir" || exit 1; \ + find "$(distdir)/$$file" -type d ! -perm -700 -exec chmod u+rwx {} \;; \ + fi; \ + cp -fpR $$d/$$file "$(distdir)$$dir" || exit 1; \ + else \ + test -f "$(distdir)/$$file" \ + || cp -p $$d/$$file "$(distdir)/$$file" \ + || exit 1; \ + fi; \ + done +check-am: all-am + $(MAKE) $(AM_MAKEFLAGS) check-TESTS +check: check-am +all-am: Makefile $(LTLIBRARIES) +installdirs: + for dir in "$(DESTDIR)$(pyexecdir)"; do \ + test -z "$$dir" || $(MKDIR_P) "$$dir"; \ + done +install-exec: install-exec-am +install-data: install-data-am +uninstall: uninstall-am + +install-am: all-am + @$(MAKE) $(AM_MAKEFLAGS) install-exec-am install-data-am + +installcheck: installcheck-am +install-strip: + $(MAKE) $(AM_MAKEFLAGS) INSTALL_PROGRAM="$(INSTALL_STRIP_PROGRAM)" \ + install_sh_PROGRAM="$(INSTALL_STRIP_PROGRAM)" INSTALL_STRIP_FLAG=-s \ + `test -z '$(STRIP)' || \ + echo "INSTALL_PROGRAM_ENV=STRIPPROG='$(STRIP)'"` install +mostlyclean-generic: + +clean-generic: + +distclean-generic: + -test -z "$(CONFIG_CLEAN_FILES)" || rm -f $(CONFIG_CLEAN_FILES) + -test . = "$(srcdir)" || test -z "$(CONFIG_CLEAN_VPATH_FILES)" || rm -f $(CONFIG_CLEAN_VPATH_FILES) + +maintainer-clean-generic: + @echo "This command is intended for maintainers to use" + @echo "it deletes files that may require special tools to rebuild." +clean: clean-am + +clean-am: clean-generic clean-libtool clean-pyexecLTLIBRARIES \ + mostlyclean-am + +distclean: distclean-am + -rm -rf ./$(DEPDIR) + -rm -f Makefile +distclean-am: clean-am distclean-compile distclean-generic \ + distclean-tags + +dvi: dvi-am + +dvi-am: + +html: html-am + +html-am: + +info: info-am + +info-am: + +install-data-am: + +install-dvi: install-dvi-am + +install-dvi-am: + +install-exec-am: install-pyexecLTLIBRARIES + +install-html: install-html-am + +install-html-am: + +install-info: install-info-am + +install-info-am: + +install-man: + +install-pdf: install-pdf-am + +install-pdf-am: + +install-ps: install-ps-am + +install-ps-am: + +installcheck-am: + +maintainer-clean: maintainer-clean-am + -rm -rf ./$(DEPDIR) + -rm -f Makefile +maintainer-clean-am: distclean-am maintainer-clean-generic + +mostlyclean: mostlyclean-am + +mostlyclean-am: mostlyclean-compile mostlyclean-generic \ + mostlyclean-libtool + +pdf: pdf-am + +pdf-am: + +ps: ps-am + +ps-am: + +uninstall-am: uninstall-pyexecLTLIBRARIES + +.MAKE: check-am install-am install-strip + +.PHONY: CTAGS GTAGS all all-am check check-TESTS check-am clean \ + clean-generic clean-libtool clean-pyexecLTLIBRARIES ctags \ + distclean distclean-compile distclean-generic \ + distclean-libtool distclean-tags distdir dvi dvi-am html \ + html-am info info-am install install-am install-data \ + install-data-am install-dvi install-dvi-am install-exec \ + install-exec-am install-html install-html-am install-info \ + install-info-am install-man install-pdf install-pdf-am \ + install-ps install-ps-am install-pyexecLTLIBRARIES \ + install-strip installcheck installcheck-am installdirs \ + maintainer-clean maintainer-clean-generic mostlyclean \ + mostlyclean-compile mostlyclean-generic mostlyclean-libtool \ + pdf pdf-am ps ps-am tags uninstall uninstall-am \ + uninstall-pyexecLTLIBRARIES + + +install: + @echo "Not installing anything from test/" + +cross-test-server: + $(TESTS_ENVIRONMENT) $(PYTHON) $(top_srcdir)/test/cross-test-server.py +cross-test-client: + $(TESTS_ENVIRONMENT) $(PYTHON) $(top_srcdir)/test/cross-test-client.py + +.PHONY: cross-test-compile cross-test-server cross-test-client + +# Tell versions [3.59,3.63) of GNU make to not export all variables. +# Otherwise a system limit (for SysV at least) may be exceeded. +.NOEXPORT: diff --git a/test/TestSuitePythonService.service.in b/test/TestSuitePythonService.service.in new file mode 100644 index 0000000..5abdf20 --- /dev/null +++ b/test/TestSuitePythonService.service.in @@ -0,0 +1,3 @@ +[D-BUS Service] +Name=org.freedesktop.DBus.TestSuitePythonService +Exec=/bin/bash -c "@PYTHON@ @abs_top_srcdir@/test/test-service.py" diff --git a/test/cross-test-client.py b/test/cross-test-client.py new file mode 100644 index 0000000..ee18f77 --- /dev/null +++ b/test/cross-test-client.py @@ -0,0 +1,393 @@ +# Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +from time import sleep +import logging + +import gobject + +from dbus import SessionBus, Interface, Array, Byte, Double, Boolean, ByteArray, Int16, Int32, Int64, UInt16, UInt32, UInt64, String, UTF8String, Struct, Dictionary +from dbus.service import BusName +import dbus.glib + +from crosstest import CROSS_TEST_PATH, CROSS_TEST_BUS_NAME,\ + INTERFACE_SINGLE_TESTS, INTERFACE_TESTS,\ + INTERFACE_SIGNAL_TESTS, INTERFACE_CALLBACK_TESTS,\ + SignalTestsImpl + + +logging.basicConfig() +logging.getLogger().setLevel(1) +logger = logging.getLogger('cross-test-client') + + +class Client(SignalTestsImpl): + fail_id = 0 + expected = set() + + def quit(self): + for x in self.expected: + self.fail_id += 1 + print "%s fail %d" % (x, self.fail_id) + s = "report %d: reply to %s didn't arrive" % (self.fail_id, x) + print s + logger.error(s) + logger.info("asking server to Exit") + Interface(self.obj, INTERFACE_TESTS).Exit(reply_handler=self.quit_reply_handler, error_handler=self.quit_error_handler) + # if the server doesn't reply we'll just exit anyway + gobject.timeout_add(1000, lambda: (loop.quit(), False)[1]) + + def quit_reply_handler(self): + logger.info("server says it will exit") + loop.quit() + + def quit_error_handler(self, e): + logger.error("error telling server to quit: %s %s", + e.__class__, e) + loop.quit() + + @dbus.service.method(INTERFACE_CALLBACK_TESTS, 'qd') + def Response(self, input1, input2): + logger.info("signal/callback: Response received (%r,%r)", + input1, input2) + self.expected.discard('%s.Trigger' % INTERFACE_SIGNAL_TESTS) + if (input1, input2) != (42, 23): + self.fail_id += 1 + print "%s.Trigger fail %d" % (INTERFACE_SIGNAL_TESTS, self.fail_id) + s = ("report %d: expected (42,23), got %r" + % (self.fail_id, (input1, input2))) + logger.error(s) + print s + else: + print "%s.Trigger pass" % INTERFACE_SIGNAL_TESTS + self.quit() + + def assert_method_matches(self, interface, check_fn, check_arg, member, *args): + if_obj = Interface(self.obj, interface) + method = getattr(if_obj, member) + try: + real_ret = method(*args) + except Exception, e: + self.fail_id += 1 + print "%s.%s fail %d" % (interface, member, self.fail_id) + s = ("report %d: %s.%s%r: raised %r \"%s\"" + % (self.fail_id, interface, member, args, e, e)) + print s + logger.error(s) + __import__('traceback').print_exc() + return + try: + check_fn(real_ret, check_arg) + except Exception, e: + self.fail_id += 1 + print "%s.%s fail %d" % (interface, member, self.fail_id) + s = ("report %d: %s.%s%r: %s" + % (self.fail_id, interface, member, args, e)) + print s + logger.error(s) + return + print "%s.%s pass" % (interface, member) + + def assert_method_eq(self, interface, ret, member, *args): + def equals(real_ret, exp): + if real_ret != exp: + raise AssertionError('expected %r of class %s, got %r of class %s' % (exp, exp.__class__, real_ret, real_ret.__class__)) + if real_ret != exp: + raise AssertionError('expected %r, got %r' % (exp, real_ret)) + if not isinstance(exp, (tuple, type(None))): + if real_ret.variant_level != getattr(exp, 'variant_level', 0): + raise AssertionError('expected variant_level=%d, got %r with level %d' + % (getattr(exp, 'variant_level', 0), real_ret, + real_ret.variant_level)) + if isinstance(exp, list) or isinstance(exp, tuple): + for i in xrange(len(exp)): + try: + equals(real_ret[i], exp[i]) + except AssertionError, e: + if not isinstance(e.args, tuple): + e.args = (e.args,) + e.args = e.args + ('(at position %d in sequence)' % i,) + raise e + elif isinstance(exp, dict): + for k in exp: + try: + equals(real_ret[k], exp[k]) + except AssertionError, e: + if not isinstance(e.args, tuple): + e.args = (e.args,) + e.args = e.args + ('(at key %r in dict)' % k,) + raise e + self.assert_method_matches(interface, equals, ret, member, *args) + + def assert_InvertMapping_eq(self, interface, expected, member, mapping): + def check(real_ret, exp): + for key in exp: + if key not in real_ret: + raise AssertionError('missing key %r' % key) + for key in real_ret: + if key not in exp: + raise AssertionError('unexpected key %r' % key) + got = list(real_ret[key]) + wanted = list(exp[key]) + got.sort() + wanted.sort() + if got != wanted: + raise AssertionError('expected %r => %r, got %r' + % (key, wanted, got)) + self.assert_method_matches(interface, check, expected, member, mapping) + + def triggered_cb(self, param, sender_path): + logger.info("method/signal: Triggered(%r) by %r", + param, sender_path) + self.expected.discard('%s.Trigger' % INTERFACE_TESTS) + if sender_path != '/Where/Ever': + self.fail_id += 1 + print "%s.Trigger fail %d" % (INTERFACE_TESTS, self.fail_id) + s = ("report %d: expected signal from /Where/Ever, got %r" + % (self.fail_id, sender_path)) + print s + logger.error(s) + elif param != 42: + self.fail_id += 1 + print "%s.Trigger fail %d" % (INTERFACE_TESTS, self.fail_id) + s = ("report %d: expected signal param 42, got %r" + % (self.fail_id, parameter)) + print s + logger.error(s) + else: + print "%s.Trigger pass" % INTERFACE_TESTS + + def trigger_returned_cb(self): + logger.info('method/signal: Trigger() returned') + # Callback tests + logger.info("signal/callback: Emitting signal to trigger callback") + self.expected.add('%s.Trigger' % INTERFACE_SIGNAL_TESTS) + self.Trigger(UInt16(42), 23.0) + logger.info("signal/callback: Emitting signal returned") + + def run_client(self): + bus = SessionBus() + obj = bus.get_object(CROSS_TEST_BUS_NAME, CROSS_TEST_PATH) + self.obj = obj + + self.run_synchronous_tests(obj) + + # Signal tests + logger.info("Binding signal handler for Triggered") + # FIXME: doesn't seem to work when going via the Interface method + # FIXME: should be possible to ask the proxy object for its + # bus name + bus.add_signal_receiver(self.triggered_cb, 'Triggered', + INTERFACE_SIGNAL_TESTS, + CROSS_TEST_BUS_NAME, + path_keyword='sender_path') + logger.info("method/signal: Triggering signal") + self.expected.add('%s.Trigger' % INTERFACE_TESTS) + Interface(obj, INTERFACE_TESTS).Trigger(u'/Where/Ever', dbus.UInt64(42), reply_handler=self.trigger_returned_cb, error_handler=self.trigger_error_handler) + + def trigger_error_handler(self, e): + logger.error("method/signal: %s %s", e.__class__, e) + Interface(self.obj, INTERFACE_TESTS).Exit() + self.quit() + + def run_synchronous_tests(self, obj): + # We can't test that coercion works correctly unless the server has + # sent us introspection data. Java doesn't :-/ + have_signatures = True + + # "Single tests" + if have_signatures: + self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', [1, 2, 3]) + self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', ['\x01', '\x02', '\x03']) + self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', [Byte(1), Byte(2), Byte(3)]) + self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', ByteArray('\x01\x02\x03')) + + # Main tests + self.assert_method_eq(INTERFACE_TESTS, String(u'foo', variant_level=1), 'Identity', String('foo')) + self.assert_method_eq(INTERFACE_TESTS, String(u'foo', variant_level=1), 'Identity', UTF8String('foo')) + self.assert_method_eq(INTERFACE_TESTS, Byte(42, variant_level=1), 'Identity', Byte(42)) + self.assert_method_eq(INTERFACE_TESTS, Byte(42, variant_level=23), 'Identity', Byte(42, variant_level=23)) + self.assert_method_eq(INTERFACE_TESTS, Double(42.5, variant_level=1), 'Identity', 42.5) + self.assert_method_eq(INTERFACE_TESTS, Double(-42.5, variant_level=1), 'Identity', -42.5) + + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, String(u'foo', variant_level=1), 'Identity', 'foo') + self.assert_method_eq(INTERFACE_TESTS, Byte(42, variant_level=1), 'Identity', Byte(42)) + self.assert_method_eq(INTERFACE_TESTS, Double(42.5, variant_level=1), 'Identity', Double(42.5)) + self.assert_method_eq(INTERFACE_TESTS, Double(-42.5, variant_level=1), 'Identity', -42.5) + + for i in (0, 42, 255): + self.assert_method_eq(INTERFACE_TESTS, Byte(i), 'IdentityByte', Byte(i)) + for i in (True, False): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityBool', i) + + for i in (-0x8000, 0, 42, 0x7fff): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityInt16', Int16(i)) + for i in (0, 42, 0xffff): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityUInt16', UInt16(i)) + for i in (-0x7fffffff-1, 0, 42, 0x7fffffff): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityInt32', Int32(i)) + for i in (0L, 42L, 0xffffffffL): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityUInt32', UInt32(i)) + MANY = 0x8000L * 0x10000L * 0x10000L * 0x10000L + for i in (-MANY, 0, 42, MANY-1): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityInt64', Int64(i)) + for i in (0, 42, 2*MANY - 1): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityUInt64', UInt64(i)) + + self.assert_method_eq(INTERFACE_TESTS, 42.3, 'IdentityDouble', 42.3) + for i in ('', 'foo'): + self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityString', i) + for i in (u'\xa9', '\xc2\xa9'): + self.assert_method_eq(INTERFACE_TESTS, u'\xa9', 'IdentityString', i) + + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, Byte(0x42), 'IdentityByte', '\x42') + self.assert_method_eq(INTERFACE_TESTS, True, 'IdentityBool', 42) + self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt16', 42) + self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt16', 42) + self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt32', 42) + self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt32', 42) + self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt64', 42) + self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt64', 42) + self.assert_method_eq(INTERFACE_TESTS, 42.0, 'IdentityDouble', 42) + + self.assert_method_eq(INTERFACE_TESTS, [Byte('\x01', variant_level=1), + Byte('\x02', variant_level=1), + Byte('\x03', variant_level=1)], + 'IdentityArray', + Array([Byte('\x01'), + Byte('\x02'), + Byte('\x03')], + signature='v')) + + self.assert_method_eq(INTERFACE_TESTS, [Int32(1, variant_level=1), + Int32(2, variant_level=1), + Int32(3, variant_level=1)], + 'IdentityArray', + Array([Int32(1), + Int32(2), + Int32(3)], + signature='v')) + self.assert_method_eq(INTERFACE_TESTS, [String(u'a', variant_level=1), + String(u'b', variant_level=1), + String(u'c', variant_level=1)], + 'IdentityArray', + Array([String('a'), + String('b'), + String('c')], + signature='v')) + + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, [Byte('\x01', variant_level=1), + Byte('\x02', variant_level=1), + Byte('\x03', variant_level=1)], + 'IdentityArray', + ByteArray('\x01\x02\x03')) + self.assert_method_eq(INTERFACE_TESTS, [Int32(1, variant_level=1), + Int32(2, variant_level=1), + Int32(3, variant_level=1)], + 'IdentityArray', + [Int32(1), + Int32(2), + Int32(3)]) + self.assert_method_eq(INTERFACE_TESTS, [String(u'a', variant_level=1), + String(u'b', variant_level=1), + String(u'c', variant_level=1)], + 'IdentityArray', + ['a','b','c']) + + self.assert_method_eq(INTERFACE_TESTS, + [Byte(1), Byte(2), Byte(3)], + 'IdentityByteArray', + ByteArray('\x01\x02\x03')) + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityByteArray', ['\x01', '\x02', '\x03']) + self.assert_method_eq(INTERFACE_TESTS, [False,True], 'IdentityBoolArray', [False,True]) + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, [False,True,True], 'IdentityBoolArray', [0,1,2]) + + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt16Array', [Int16(1),Int16(2),Int16(3)]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt16Array', [UInt16(1),UInt16(2),UInt16(3)]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt32Array', [Int32(1),Int32(2),Int32(3)]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt32Array', [UInt32(1),UInt32(2),UInt32(3)]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt64Array', [Int64(1),Int64(2),Int64(3)]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt64Array', [UInt64(1),UInt64(2),UInt64(3)]) + + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt16Array', [1,2,3]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt16Array', [1,2,3]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt32Array', [1,2,3]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt32Array', [1,2,3]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt64Array', [1,2,3]) + self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt64Array', [1,2,3]) + + self.assert_method_eq(INTERFACE_TESTS, [1.0,2.5,3.1], 'IdentityDoubleArray', [1.0,2.5,3.1]) + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, [1.0,2.5,3.1], 'IdentityDoubleArray', [1,2.5,3.1]) + self.assert_method_eq(INTERFACE_TESTS, ['a','b','c'], 'IdentityStringArray', ['a','b','c']) + self.assert_method_eq(INTERFACE_TESTS, 6, 'Sum', [Int32(1),Int32(2),Int32(3)]) + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, 6, 'Sum', [1,2,3]) + + self.assert_InvertMapping_eq(INTERFACE_TESTS, {'fps': ['unreal', 'quake'], 'rts': ['warcraft']}, 'InvertMapping', {'unreal': 'fps', 'quake': 'fps', 'warcraft': 'rts'}) + + self.assert_method_eq(INTERFACE_TESTS, ('a', 1, 2), 'DeStruct', ('a', UInt32(1), Int16(2))) + self.assert_method_eq(INTERFACE_TESTS, Array([String('x', variant_level=1)]), + 'Primitize', [String('x', variant_level=1)]) + self.assert_method_eq(INTERFACE_TESTS, Array([String('x', variant_level=1)]), + 'Primitize', [String('x', variant_level=23)]) + self.assert_method_eq(INTERFACE_TESTS, + Array([String('x', variant_level=1), + Byte(1, variant_level=1), + Byte(2, variant_level=1)]), + 'Primitize', + Array([String('x'), Byte(1), Byte(2)], + signature='v')) + self.assert_method_eq(INTERFACE_TESTS, + Array([String('x', variant_level=1), + Byte(1, variant_level=1), + Byte(2, variant_level=1)]), + 'Primitize', + Array([String('x'), Array([Byte(1), Byte(2)])], + signature='v')) + self.assert_method_eq(INTERFACE_TESTS, Boolean(False), 'Invert', True) + self.assert_method_eq(INTERFACE_TESTS, Boolean(True), 'Invert', False) + if have_signatures: + self.assert_method_eq(INTERFACE_TESTS, Boolean(False), 'Invert', 42) + self.assert_method_eq(INTERFACE_TESTS, Boolean(True), 'Invert', 0) + + +if __name__ == '__main__': + # FIXME: should be possible to export objects without a bus name + if 0: + client = Client(dbus.SessionBus(), '/Client') + else: + # the Java cross test's interpretation is that the client should be + # at /Test too + client = Client(dbus.SessionBus(), '/Test') + gobject.idle_add(client.run_client) + + loop = gobject.MainLoop() + logger.info("running...") + loop.run() + logger.info("main loop exited.") diff --git a/test/cross-test-server.py b/test/cross-test-server.py new file mode 100644 index 0000000..40d121a --- /dev/null +++ b/test/cross-test-server.py @@ -0,0 +1,323 @@ +# Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import logging + +import gobject + +import dbus.glib +from dbus import SessionBus +from dbus.service import BusName + +from crosstest import CROSS_TEST_PATH, CROSS_TEST_BUS_NAME, \ + INTERFACE_SINGLE_TESTS, INTERFACE_TESTS,\ + INTERFACE_CALLBACK_TESTS, INTERFACE_SIGNAL_TESTS,\ + SignalTestsImpl + + +logging.basicConfig() +logging.getLogger().setLevel(1) +logger = logging.getLogger('cross-test-server') + + +class VerboseSet(set): + def add(self, thing): + print '%s ok' % thing + set.add(self, thing) + + +objects = {} + + +tested_things = VerboseSet() +testable_things = [ + INTERFACE_SINGLE_TESTS + '.Sum', + INTERFACE_TESTS + '.Identity', + INTERFACE_TESTS + '.IdentityByte', + INTERFACE_TESTS + '.IdentityBool', + INTERFACE_TESTS + '.IdentityInt16', + INTERFACE_TESTS + '.IdentityUInt16', + INTERFACE_TESTS + '.IdentityInt32', + INTERFACE_TESTS + '.IdentityUInt32', + INTERFACE_TESTS + '.IdentityInt64', + INTERFACE_TESTS + '.IdentityUInt64', + INTERFACE_TESTS + '.IdentityDouble', + INTERFACE_TESTS + '.IdentityString', + INTERFACE_TESTS + '.IdentityArray', + INTERFACE_TESTS + '.IdentityByteArray', + INTERFACE_TESTS + '.IdentityBoolArray', + INTERFACE_TESTS + '.IdentityInt16Array', + INTERFACE_TESTS + '.IdentityUInt16Array', + INTERFACE_TESTS + '.IdentityInt32Array', + INTERFACE_TESTS + '.IdentityUInt32Array', + INTERFACE_TESTS + '.IdentityInt64Array', + INTERFACE_TESTS + '.IdentityUInt64Array', + INTERFACE_TESTS + '.IdentityDoubleArray', + INTERFACE_TESTS + '.IdentityStringArray', + INTERFACE_TESTS + '.Sum', + INTERFACE_TESTS + '.InvertMapping', + INTERFACE_TESTS + '.DeStruct', + INTERFACE_TESTS + '.Primitize', + INTERFACE_TESTS + '.Trigger', + INTERFACE_TESTS + '.Exit', + INTERFACE_TESTS + '.Invert', + INTERFACE_SIGNAL_TESTS + '.Trigger', +] + + +class SingleTestsImpl(dbus.service.Object): + + @dbus.service.method(INTERFACE_SINGLE_TESTS, 'ay', 'u') + def Sum(self, input): + tested_things.add(INTERFACE_SINGLE_TESTS + '.Sum') + u = sum(input) + logger.info('Sum of %r is %r', input, u) + return u + + +class TestsImpl(dbus.service.Object): + + def __init__(self, bus_name, service_path, exit_fn): + self._exit_fn = exit_fn + dbus.service.Object.__init__(self, bus_name, service_path) + + @dbus.service.method(INTERFACE_TESTS, 'v', 'v') + def Identity(self, input): + tested_things.add(INTERFACE_TESTS + '.Identity') + return input + + @dbus.service.method(INTERFACE_TESTS, 'y', 'y') + def IdentityByte(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityByte') + return input + + @dbus.service.method(INTERFACE_TESTS, 'b', 'b') + def IdentityBool(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityBool') + return input + + @dbus.service.method(INTERFACE_TESTS, 'n', 'n') + def IdentityInt16(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityInt16') + return input + + @dbus.service.method(INTERFACE_TESTS, 'q', 'q') + def IdentityUInt16(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityUInt16') + return input + + @dbus.service.method(INTERFACE_TESTS, 'i', 'i') + def IdentityInt32(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityInt32') + return input + + @dbus.service.method(INTERFACE_TESTS, 'u', 'u') + def IdentityUInt32(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityUInt32') + return input + + @dbus.service.method(INTERFACE_TESTS, 'x', 'x') + def IdentityInt64(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityInt64') + return input + + @dbus.service.method(INTERFACE_TESTS, 't', 't') + def IdentityUInt64(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityUInt64') + return input + + @dbus.service.method(INTERFACE_TESTS, 'd', 'd') + def IdentityDouble(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityDouble') + return input + + @dbus.service.method(INTERFACE_TESTS, 's', 's') + def IdentityString(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityString') + return input + + @dbus.service.method(INTERFACE_TESTS, 'av', 'av') + def IdentityArray(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityArray') + return input + + @dbus.service.method(INTERFACE_TESTS, 'ay', 'ay') + def IdentityByteArray(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityByteArray') + return input + + @dbus.service.method(INTERFACE_TESTS, 'ab', 'ab') + def IdentityBoolArray(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityBoolArray') + return input + + @dbus.service.method(INTERFACE_TESTS, 'an', 'an') + def IdentityInt16Array(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityInt16Array') + return input + + @dbus.service.method(INTERFACE_TESTS, 'aq', 'aq') + def IdentityUInt16Array(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityUInt16Array') + return input + + @dbus.service.method(INTERFACE_TESTS, 'ai', 'ai') + def IdentityInt32Array(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityInt32Array') + return input + + @dbus.service.method(INTERFACE_TESTS, 'au', 'au') + def IdentityUInt32Array(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityUInt32Array') + return input + + @dbus.service.method(INTERFACE_TESTS, 'ax', 'ax') + def IdentityInt64Array(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityInt64Array') + return input + + @dbus.service.method(INTERFACE_TESTS, 'at', 'at') + def IdentityUInt64Array(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityUInt64Array') + return input + + @dbus.service.method(INTERFACE_TESTS, 'ad', 'ad') + def IdentityDoubleArray(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityDoubleArray') + return input + + @dbus.service.method(INTERFACE_TESTS, 'as', 'as') + def IdentityStringArray(self, input): + tested_things.add(INTERFACE_TESTS + '.IdentityStringArray') + return input + + @dbus.service.method(INTERFACE_TESTS, 'ai', 'x') + def Sum(self, input): + tested_things.add(INTERFACE_TESTS + '.Sum') + x = sum(input) + logger.info('Sum of %r is %r', input, x) + return x + + + @dbus.service.method(INTERFACE_TESTS, 'a{ss}', 'a{sas}', utf8_strings=True) + def InvertMapping(self, input): + tested_things.add(INTERFACE_TESTS + '.InvertMapping') + output = dbus.Dictionary({}) + for k, v in input.iteritems(): + output.setdefault(v, []).append(k) + return output + + @dbus.service.method(INTERFACE_TESTS, '(sun)', 'sun') + def DeStruct(self, input): + tested_things.add(INTERFACE_TESTS + '.DeStruct') + return input + + @dbus.service.method(INTERFACE_TESTS, 'v', 'av') + def Primitize(self, input): + tested_things.add(INTERFACE_TESTS + '.Primitize') + return list(self.primitivize_helper(input)) + + def primitivize_helper(self, input): + if (isinstance(input, tuple) or isinstance(input, dbus.Struct) + or isinstance(input, list) or isinstance(input, dbus.Array)): + for x in input: + for y in self.primitivize_helper(x): + yield y + elif isinstance(input, dbus.ByteArray): + for x in input: + yield dbus.Byte(ord(x)) + elif isinstance(input, dict) or isinstance(input, dbus.Dictionary): + for x in input: + for y in self.primitivize_helper(x): + yield y + for y in self.primitivize_helper(input[x]): + yield y + elif input.variant_level > 0: + yield input.__class__(input) + else: + yield input + + @dbus.service.method(INTERFACE_TESTS, 'b', 'b') + def Invert(self, input): + tested_things.add(INTERFACE_TESTS + '.Invert') + return not input + + @dbus.service.method(INTERFACE_TESTS, 'st', '', utf8_strings=True, + connection_keyword='conn') + def Trigger(self, object, parameter, conn=None): + assert isinstance(object, str) + logger.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter, object) + tested_things.add(INTERFACE_TESTS + '.Trigger') + gobject.idle_add(lambda: self.emit_Triggered_from(conn, object, + parameter)) + + def emit_Triggered_from(self, conn, object, parameter): + assert isinstance(object, str) + logger.info('method/signal: Emitting Triggered(%r) from %r', parameter, object) + obj = objects.get(object, None) + if obj is None: + obj = SignalTestsImpl(conn, object) + objects[object] = obj + obj.Triggered(parameter) + logger.info('method/signal: Emitted Triggered') + + @dbus.service.method(INTERFACE_TESTS, '', '') + def Exit(self): + logger.info('client wants me to Exit') + tested_things.add(INTERFACE_TESTS + '.Exit') + for x in testable_things: + if x not in tested_things: + print '%s untested' % x + logger.info('will quit when idle') + gobject.idle_add(self._exit_fn) + + +class Server(SingleTestsImpl, TestsImpl, SignalTestsImpl): + + def triggered_by_client(self, parameter1, parameter2, sender, sender_path): + # Called when the client emits TestSignals.Trigger from any object. + logger.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender, sender_path, parameter1, parameter2) + tested_things.add(INTERFACE_SIGNAL_TESTS + '.Trigger') + dbus.Interface(dbus.SessionBus().get_object(sender, sender_path), + INTERFACE_CALLBACK_TESTS).Response(parameter1, parameter2) + logger.info('signal/callback: Sent Response') + + + +if __name__ == '__main__': + bus = SessionBus() + bus_name = BusName(CROSS_TEST_BUS_NAME, bus=bus) + loop = gobject.MainLoop() + obj = Server(bus_name, CROSS_TEST_PATH, loop.quit) + objects[CROSS_TEST_PATH] = obj + bus.add_signal_receiver(obj.triggered_by_client, + signal_name='Trigger', + dbus_interface=INTERFACE_SIGNAL_TESTS, + named_service=None, + path=None, + sender_keyword='sender', + path_keyword='sender_path', + utf8_strings=True) + + logger.info("running...") + loop.run() + logger.info("main loop exited.") diff --git a/test/crosstest.py b/test/crosstest.py new file mode 100644 index 0000000..0020baf --- /dev/null +++ b/test/crosstest.py @@ -0,0 +1,44 @@ +# Shared code for the cross-test. + +# Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import dbus.service + +INTERFACE_SINGLE_TESTS = 'org.freedesktop.DBus.Binding.SingleTests' +INTERFACE_TESTS = 'org.freedesktop.DBus.Binding.Tests' +INTERFACE_SIGNAL_TESTS = 'org.freedesktop.DBus.Binding.TestSignals' +INTERFACE_CALLBACK_TESTS = 'org.freedesktop.DBus.Binding.TestCallbacks' + +CROSS_TEST_PATH = '/Test' +CROSS_TEST_BUS_NAME = 'org.freedesktop.DBus.Binding.TestServer' + + +# Exported by both the client and the server +class SignalTestsImpl(dbus.service.Object): + @dbus.service.signal(INTERFACE_SIGNAL_TESTS, 't') + def Triggered(self, parameter): + pass + + @dbus.service.signal(INTERFACE_SIGNAL_TESTS, 'qd') + def Trigger(self, parameter1, parameter2): + pass diff --git a/test/dbus_py_test.c b/test/dbus_py_test.c new file mode 100644 index 0000000..5604e32 --- /dev/null +++ b/test/dbus_py_test.c @@ -0,0 +1,126 @@ +/* Test fixtures for dbus-python, based on _dbus_glib_bindings. + * + * Copyright (C) 2007 Collabora Ltd. <http://www.collabora.co.uk/> + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, copy, + * modify, merge, publish, distribute, sublicense, and/or sell copies + * of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF + * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +#include <Python.h> +#include "dbus-python.h" + +PyMODINIT_FUNC initdbus_py_test(void); + +#if defined(__GNUC__) +# if __GNUC__ >= 3 +# define UNUSED __attribute__((__unused__)) +# else +# define UNUSED /*nothing*/ +# endif +#else +# define UNUSED /*nothing*/ +#endif + +static dbus_bool_t +dbus_py_test_set_up_conn(DBusConnection *conn UNUSED, void *data UNUSED) +{ + PyErr_SetString(PyExc_ValueError, "Dummy error from UnusableMainLoop"); + return 0; +} + +static dbus_bool_t +dbus_py_test_set_up_srv(DBusServer *srv UNUSED, void *data UNUSED) +{ + PyErr_SetString(PyExc_ValueError, "Dummy error from UnusableMainLoop"); + return 0; +} + +static void +dbus_py_test_free(void *data UNUSED) +{ +} + +static PyObject * +dbus_test_native_mainloop(void) +{ + PyObject *loop = DBusPyNativeMainLoop_New4(dbus_py_test_set_up_conn, + dbus_py_test_set_up_srv, + dbus_py_test_free, + NULL); + return loop; +} + +static PyObject * +UnusableMainLoop (PyObject *always_null UNUSED, PyObject *args, PyObject *kwargs) +{ + PyObject *mainloop, *function, *result; + int set_as_default = 0; + static char *argnames[] = {"set_as_default", NULL}; + + if (PyTuple_Size(args) != 0) { + PyErr_SetString(PyExc_TypeError, "UnusableMainLoop() takes no " + "positional arguments"); + return NULL; + } + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|i", argnames, + &set_as_default)) { + return NULL; + } + + mainloop = dbus_test_native_mainloop(); + if (mainloop && set_as_default) { + if (!_dbus_bindings_module) { + PyErr_SetString(PyExc_ImportError, "_dbus_bindings not imported"); + Py_DECREF(mainloop); + return NULL; + } + function = PyObject_GetAttrString(_dbus_bindings_module, + "set_default_main_loop"); + if (!function) { + Py_DECREF(mainloop); + return NULL; + } + result = PyObject_CallFunctionObjArgs(function, mainloop, NULL); + Py_DECREF(function); + if (!result) { + Py_DECREF(mainloop); + return NULL; + } + } + return mainloop; +} + +static PyMethodDef module_functions[] = { + {"UnusableMainLoop", (PyCFunction)UnusableMainLoop, + METH_VARARGS|METH_KEYWORDS, "Return a main loop that fails to attach"}, + {NULL, NULL, 0, NULL} +}; + +PyMODINIT_FUNC +initdbus_py_test(void) +{ + PyObject *this_module; + + if (import_dbus_bindings("dbus_py_test") < 0) return; + this_module = Py_InitModule3 ("dbus_py_test", module_functions, ""); + if (!this_module) return; +} + +/* vim:set ft=c cino< sw=4 sts=4 et: */ diff --git a/test/run-test.sh b/test/run-test.sh new file mode 100755 index 0000000..45e2ed4 --- /dev/null +++ b/test/run-test.sh @@ -0,0 +1,124 @@ +#! /bin/bash + +# Copyright (C) 2006 Red Hat Inc. <http://www.redhat.com/> +# Copyright (C) 2006-2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +export DBUS_FATAL_WARNINGS=1 +ulimit -c unlimited + +function die() +{ + if ! test -z "$DBUS_SESSION_BUS_PID" ; then + echo "killing message bus $DBUS_SESSION_BUS_PID" >&2 + kill -9 "$DBUS_SESSION_BUS_PID" + fi + echo "$SCRIPTNAME: $*" >&2 + exit 1 +} + +if test -z "$PYTHON"; then + echo "Warning: \$PYTHON not set, assuming 'python'" >&2 + export PYTHON=python +fi + +if test -z "$DBUS_TOP_SRCDIR" ; then + die "Must set DBUS_TOP_SRCDIR" +fi + +if test -z "$DBUS_TOP_BUILDDIR" ; then + die "Must set DBUS_TOP_BUILDDIR" +fi + +SCRIPTNAME=$0 + +## so the tests can complain if you fail to use the script to launch them +export DBUS_TEST_PYTHON_RUN_TEST_SCRIPT=1 +# Rerun ourselves with tmp session bus if we're not already +if test -z "$DBUS_TEST_PYTHON_IN_RUN_TEST"; then + DBUS_TEST_PYTHON_IN_RUN_TEST=1 + export DBUS_TEST_PYTHON_IN_RUN_TEST + exec "$DBUS_TOP_SRCDIR"/test/run-with-tmp-session-bus.sh $SCRIPTNAME +fi + +dbus-monitor > "$DBUS_TOP_BUILDDIR"/test/monitor.log & + +echo "running test-standalone.py" +$PYTHON "$DBUS_TOP_SRCDIR"/test/test-standalone.py || die "test-standalone.py failed" + +echo "running test-unusable-main-loop.py" +$PYTHON "$DBUS_TOP_SRCDIR"/test/test-unusable-main-loop.py || die "... failed" + +#echo "running the examples" + +#$PYTHON "$DBUS_TOP_SRCDIR"/examples/example-service.py & +#$PYTHON "$DBUS_TOP_SRCDIR"/examples/example-signal-emitter.py & +#$PYTHON "$DBUS_TOP_SRCDIR"/examples/list-system-services.py --session || +# die "list-system-services.py --session failed!" +#$PYTHON "$DBUS_TOP_SRCDIR"/examples/example-async-client.py || +# die "example-async-client failed!" +#$PYTHON "$DBUS_TOP_SRCDIR"/examples/example-client.py --exit-service || +# die "example-client failed!" +#$PYTHON "$DBUS_TOP_SRCDIR"/examples/example-signal-recipient.py --exit-service || +# die "example-signal-recipient failed!" + +echo "running cross-test (for better diagnostics use mjj29's dbus-test)" + +${MAKE:-make} -s cross-test-server > "$DBUS_TOP_BUILDDIR"/test/cross-server.log& +sleep 1 +${MAKE:-make} -s cross-test-client > "$DBUS_TOP_BUILDDIR"/test/cross-client.log + +if grep . "$DBUS_TOP_BUILDDIR"/test/cross-client.log >/dev/null; then + : # OK +else + die "cross-test client produced no output" +fi +if grep . "$DBUS_TOP_BUILDDIR"/test/cross-server.log >/dev/null; then + : # OK +else + die "cross-test server produced no output" +fi + +if grep fail "$DBUS_TOP_BUILDDIR"/test/cross-client.log; then + die "^^^ Cross-test client reports failures, see test/cross-client.log" +else + echo " - cross-test client reported no failures" +fi +if grep untested "$DBUS_TOP_BUILDDIR"/test/cross-server.log; then + die "^^^ Cross-test server reports incomplete test coverage" +else + echo " - cross-test server reported no untested functions" +fi + +echo "running test-client.py" +$PYTHON "$DBUS_TOP_SRCDIR"/test/test-client.py || die "test-client.py failed" +echo "running test-signals.py" +$PYTHON "$DBUS_TOP_SRCDIR"/test/test-signals.py || die "test-signals.py failed" + +echo "running test-p2p.py" +$PYTHON "$DBUS_TOP_SRCDIR"/test/test-p2p.py || die "... failed" + +rm -f "$DBUS_TOP_BUILDDIR"/test/test-service.log +rm -f "$DBUS_TOP_BUILDDIR"/test/cross-client.log +rm -f "$DBUS_TOP_BUILDDIR"/test/cross-server.log +rm -f "$DBUS_TOP_BUILDDIR"/test/monitor.log +exit 0 diff --git a/test/run-with-tmp-session-bus.sh b/test/run-with-tmp-session-bus.sh new file mode 100755 index 0000000..e767f2d --- /dev/null +++ b/test/run-with-tmp-session-bus.sh @@ -0,0 +1,75 @@ +#! /bin/bash + +# Copyright (C) 2003-2005 Red Hat Inc. <http://www.redhat.com/> +# Copyright (C) 2005 Colin Walters +# Copyright (C) 2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +SCRIPTNAME=$0 +WRAPPED_SCRIPT=$1 +shift + +function die() +{ + if ! test -z "$DBUS_SESSION_BUS_PID" ; then + echo "killing message bus $DBUS_SESSION_BUS_PID" >&2 + kill -9 "$DBUS_SESSION_BUS_PID" + fi + echo "$SCRIPTNAME: $*" >&2 + exit 1 +} + +if test -z "$DBUS_TOP_BUILDDIR" ; then + die "Must set DBUS_TOP_BUILDDIR" +fi + +## convenient to be able to ctrl+C without leaking the message bus process +trap 'die "Received SIGINT"' SIGINT + +CONFIG_FILE="$DBUS_TOP_BUILDDIR"/test/tmp-session-bus.conf + +unset DBUS_SESSION_BUS_ADDRESS +unset DBUS_SESSION_BUS_PID + +echo "Running dbus-launch --sh-syntax --config-file=$CONFIG_FILE" >&2 +eval `dbus-launch --sh-syntax --config-file=$CONFIG_FILE` + +if test -z "$DBUS_SESSION_BUS_PID" ; then + die "Failed to launch message bus for introspection generation to run" +fi + +echo "Started bus pid $DBUS_SESSION_BUS_PID at $DBUS_SESSION_BUS_ADDRESS" >&2 + +# Execute wrapped script +echo "Running: $WRAPPED_SCRIPT $*" >&2 +"$WRAPPED_SCRIPT" "$@" || die "script \"$WRAPPED_SCRIPT\" failed" + +kill -TERM "$DBUS_SESSION_BUS_PID" \ + || die "Message bus vanished! should not have happened" \ + && echo "Killed daemon $DBUS_SESSION_BUS_PID" >&2 + +sleep 2 + +## be sure it really died +kill -9 $DBUS_SESSION_BUS_PID > /dev/null 2>&1 || true + +exit 0 diff --git a/test/test-client.py b/test/test-client.py new file mode 100755 index 0000000..753d892 --- /dev/null +++ b/test/test-client.py @@ -0,0 +1,587 @@ +#!/usr/bin/env python + +# Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/> +# Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import sys +import os +import unittest +import time +import logging +import weakref + +builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"]) +pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"]) + +import dbus +import _dbus_bindings +import gobject +import dbus.glib +import dbus.service + + +logging.basicConfig() + + +pkg = dbus.__file__ +if not pkg.startswith(pydir): + raise Exception("DBus modules (%s) are not being picked up from the package"%pkg) + +if not _dbus_bindings.__file__.startswith(builddir): + raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__) + +test_types_vals = [1, 12323231, 3.14159265, 99999999.99, + "dude", "123", "What is all the fuss about?", "gob@gob.com", + u'\\u310c\\u310e\\u3114', u'\\u0413\\u0414\\u0415', + u'\\u2200software \\u2203crack', u'\\xf4\\xe5\\xe8', + [1,2,3], ["how", "are", "you"], [1.23,2.3], [1], ["Hello"], + (1,2,3), (1,), (1,"2",3), ("2", "what"), ("you", 1.2), + {1:"a", 2:"b"}, {"a":1, "b":2}, #{"a":(1,"B")}, + {1:1.1, 2:2.2}, [[1,2,3],[2,3,4]], [["a","b"],["c","d"]], + True, False, + dbus.Int16(-10), dbus.UInt16(10), 'SENTINEL', + #([1,2,3],"c", 1.2, ["a","b","c"], {"a": (1,"v"), "b": (2,"d")}) + ] + +NAME = "org.freedesktop.DBus.TestSuitePythonService" +IFACE = "org.freedesktop.DBus.TestSuiteInterface" +OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject" + +class TestDBusBindings(unittest.TestCase): + def setUp(self): + self.bus = dbus.SessionBus() + self.remote_object = self.bus.get_object(NAME, OBJECT) + self.remote_object_follow = self.bus.get_object(NAME, OBJECT, + follow_name_owner_changes=True) + self.iface = dbus.Interface(self.remote_object, IFACE) + + def testGObject(self): + print "Testing ExportedGObject... ", + remote_gobject = self.bus.get_object(NAME, OBJECT + '/GObject') + iface = dbus.Interface(remote_gobject, IFACE) + print "introspection, ", + remote_gobject.Introspect(dbus_interface=dbus.INTROSPECTABLE_IFACE) + print "method call, ", + self.assertEquals(iface.Echo('123'), '123') + print "... OK" + + def testWeakRefs(self): + # regression test for Sugar crash caused by smcv getting weak refs + # wrong - pre-bugfix, this would segfault + bus = dbus.SessionBus(private=True) + ref = weakref.ref(bus) + self.assert_(ref() is bus) + del bus + self.assert_(ref() is None) + + def testInterfaceKeyword(self): + #test dbus_interface parameter + print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = IFACE) + print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = IFACE) + self.assert_(True) + + def testGetDBusMethod(self): + self.assertEquals(self.iface.get_dbus_method('AcceptListOfByte')('\1\2\3'), [1,2,3]) + self.assertEquals(self.remote_object.get_dbus_method('AcceptListOfByte', dbus_interface=IFACE)('\1\2\3'), [1,2,3]) + + def testCallingConventionOptions(self): + self.assertEquals(self.iface.AcceptListOfByte('\1\2\3'), [1,2,3]) + self.assertEquals(self.iface.AcceptListOfByte('\1\2\3', byte_arrays=True), '\1\2\3') + self.assertEquals(self.iface.AcceptByteArray('\1\2\3'), [1,2,3]) + self.assertEquals(self.iface.AcceptByteArray('\1\2\3', byte_arrays=True), '\1\2\3') + self.assert_(isinstance(self.iface.AcceptUTF8String('abc'), unicode)) + self.assert_(isinstance(self.iface.AcceptUTF8String('abc', utf8_strings=True), str)) + self.assert_(isinstance(self.iface.AcceptUnicodeString('abc'), unicode)) + self.assert_(isinstance(self.iface.AcceptUnicodeString('abc', utf8_strings=True), str)) + + def testIntrospection(self): + #test introspection + print "\n********* Introspection Test ************" + print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable") + print "Introspection test passed" + self.assert_(True) + + def testMultiPathIntrospection(self): + # test introspection on an object exported in multiple places + # https://bugs.freedesktop.org/show_bug.cgi?id=11794 + remote_object = self.bus.get_object(NAME, OBJECT + '/Multi1') + remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable") + remote_object = self.bus.get_object(NAME, OBJECT + '/Multi2') + remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable") + remote_object = self.bus.get_object(NAME, OBJECT + '/Multi2/3') + remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable") + self.assert_(True) + + def testPythonTypes(self): + #test sending python types and getting them back + print "\n********* Testing Python Types ***********" + + for send_val in test_types_vals: + print "Testing %s"% str(send_val) + recv_val = self.iface.Echo(send_val) + self.assertEquals(send_val, recv_val) + self.assertEquals(recv_val.variant_level, 1) + + def testMethodExtraInfoKeywords(self): + print "Testing MethodExtraInfoKeywords..." + sender, path, destination, message_cls = self.iface.MethodExtraInfoKeywords() + self.assert_(sender.startswith(':')) + self.assertEquals(path, '/org/freedesktop/DBus/TestSuitePythonObject') + # we're using the "early binding" form of get_object (without + # follow_name_owner_changes), so the destination we actually sent it + # to will be the unique name + self.assert_(destination.startswith(':')) + self.assertEquals(message_cls, 'dbus.lowlevel.MethodCallMessage') + + def testUtf8StringsSync(self): + send_val = u'foo' + recv_val = self.iface.Echo(send_val, utf8_strings=True) + self.assert_(isinstance(recv_val, str)) + self.assert_(isinstance(recv_val, dbus.UTF8String)) + recv_val = self.iface.Echo(send_val, utf8_strings=False) + self.assert_(isinstance(recv_val, unicode)) + self.assert_(isinstance(recv_val, dbus.String)) + + def testBenchmarkIntrospect(self): + print "\n********* Benchmark Introspect ************" + a = time.time() + print a + print self.iface.GetComplexArray() + b = time.time() + print b + print "Delta: %f" % (b - a) + self.assert_(True) + + def testAsyncCalls(self): + #test sending python types and getting them back async + print "\n********* Testing Async Calls ***********" + + failures = [] + main_loop = gobject.MainLoop() + + class async_check: + def __init__(self, test_controler, expected_result, do_exit, utf8): + self.expected_result = expected_result + self.do_exit = do_exit + self.utf8 = utf8 + self.test_controler = test_controler + + def callback(self, val): + try: + if self.do_exit: + main_loop.quit() + + self.test_controler.assertEquals(val, self.expected_result) + self.test_controler.assertEquals(val.variant_level, 1) + if self.utf8 and not isinstance(val, dbus.UTF8String): + failures.append('%r should have been utf8 but was not' % val) + return + elif not self.utf8 and isinstance(val, dbus.UTF8String): + failures.append('%r should not have been utf8' % val) + return + except Exception, e: + failures.append("%s:\n%s" % (e.__class__, e)) + + def error_handler(self, error): + print error + if self.do_exit: + main_loop.quit() + + failures.append('%s: %s' % (error.__class__, error)) + + last_type = test_types_vals[-1] + for send_val in test_types_vals: + print "Testing %s" % str(send_val) + utf8 = (send_val == 'gob@gob.com') + check = async_check(self, send_val, last_type == send_val, + utf8) + recv_val = self.iface.Echo(send_val, + reply_handler=check.callback, + error_handler=check.error_handler, + utf8_strings=utf8) + main_loop.run() + if failures: + self.assert_(False, failures) + + def testStrictMarshalling(self): + print "\n********* Testing strict return & signal marshalling ***********" + + # these values are the same as in the server, and the + # methods should only succeed when they are called with + # the right value number, because they have out_signature + # decorations, and return an unmatching type when called + # with a different number + values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]] + methods = [ + (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])), + (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([1])), + (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])), + # all of our test values are sequences so will marshall correctly into an array :P + (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])), + (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4])) + ] + + for (method, signal, success_values, return_values) in methods: + print "\nTrying correct behaviour of", method._method_name + for value in range(len(values)): + try: + ret = method(value) + except Exception, e: + print "%s(%r) raised %s: %s" % (method._method_name, values[value], e.__class__, e) + + # should fail if it tried to marshal the wrong type + self.assert_(value not in success_values, "%s should succeed when we ask it to return %r\n%s\n%s" % (method._method_name, values[value], e.__class__, e)) + else: + print "%s(%r) returned %r" % (method._method_name, values[value], ret) + + # should only succeed if it's the right return type + self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value])) + + # check the value is right too :D + returns = map(lambda n: values[n], return_values) + self.assert_(ret in returns, "%s should return one of %r but it returned %r instead" % (method._method_name, returns, ret)) + + print "\nTrying correct emission of", signal + for value in range(len(values)): + try: + self.iface.EmitSignal(signal, value) + except Exception, e: + print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__) + + # should fail if it tried to marshal the wrong type + self.assert_(value not in success_values, "EmitSignal(%s) should succeed when we ask it to return %r\n%s\n%s" % (signal, values[value], e.__class__, e)) + else: + print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value]) + + # should only succeed if it's the right return type + self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value])) + + # FIXME: wait for the signal here + + print + + def testInheritance(self): + print "\n********* Testing inheritance from dbus.method.Interface ***********" + ret = self.iface.CheckInheritance() + print "CheckInheritance returned %s" % ret + self.assert_(ret, "overriding CheckInheritance from TestInterface failed") + + def testAsyncMethods(self): + print "\n********* Testing asynchronous method implementation *******" + for async in (True, False): + for fail in (True, False): + try: + val = ('a', 1, False, [1,2], {1:2}) + print "calling AsynchronousMethod with %s %s %s" % (async, fail, val) + ret = self.iface.AsynchronousMethod(async, fail, val) + except Exception, e: + self.assert_(fail, '%s: %s' % (e.__class__, e)) + print "Expected failure: %s: %s" % (e.__class__, e) + else: + self.assert_(not fail, 'Expected failure but succeeded?!') + self.assertEquals(val, ret) + self.assertEquals(1, ret.variant_level) + + def testBusInstanceCaching(self): + print "\n********* Testing dbus.Bus instance sharing *********" + + # unfortunately we can't test the system bus here + # but the codepaths are the same + for (cls, type, func) in ((dbus.SessionBus, dbus.Bus.TYPE_SESSION, dbus.Bus.get_session), (dbus.StarterBus, dbus.Bus.TYPE_STARTER, dbus.Bus.get_starter)): + print "\nTesting %s:" % cls.__name__ + + share_cls = cls() + share_type = dbus.Bus(bus_type=type) + share_func = func() + + private_cls = cls(private=True) + private_type = dbus.Bus(bus_type=type, private=True) + private_func = func(private=True) + + print " - checking shared instances are the same..." + self.assert_(share_cls == share_type, '%s should equal %s' % (share_cls, share_type)) + self.assert_(share_type == share_func, '%s should equal %s' % (share_type, share_func)) + + print " - checking private instances are distinct from the shared instance..." + self.assert_(share_cls != private_cls, '%s should not equal %s' % (share_cls, private_cls)) + self.assert_(share_type != private_type, '%s should not equal %s' % (share_type, private_type)) + self.assert_(share_func != private_func, '%s should not equal %s' % (share_func, private_func)) + + print " - checking private instances are distinct from each other..." + self.assert_(private_cls != private_type, '%s should not equal %s' % (private_cls, private_type)) + self.assert_(private_type != private_func, '%s should not equal %s' % (private_type, private_func)) + self.assert_(private_func != private_cls, '%s should not equal %s' % (private_func, private_cls)) + + def testSenderName(self): + print '\n******** Testing sender name keyword ********' + myself = self.iface.WhoAmI() + print "I am", myself + + def testBusGetNameOwner(self): + ret = self.bus.get_name_owner(NAME) + self.assert_(ret.startswith(':'), ret) + + def testBusListNames(self): + ret = self.bus.list_names() + self.assert_(NAME in ret, ret) + + def testBusListActivatableNames(self): + ret = self.bus.list_activatable_names() + self.assert_(NAME in ret, ret) + + def testBusNameHasOwner(self): + self.assert_(self.bus.name_has_owner(NAME)) + self.assert_(not self.bus.name_has_owner('badger.mushroom.snake')) + + def testBusNameCreation(self): + print '\n******** Testing BusName creation ********' + test = [('org.freedesktop.DBus.Python.TestName', True), + ('org.freedesktop.DBus.Python.TestName', True), + ('org.freedesktop.DBus.Python.InvalidName&^*%$', False)] + # Do some more intelligent handling/testing of queueing vs success? + # ('org.freedesktop.DBus.TestSuitePythonService', False)] + # For some reason this actually succeeds + # ('org.freedesktop.DBus', False)] + + # make a method call to ensure the test service is active + self.iface.Echo("foo") + + names = {} + for (name, succeed) in test: + try: + print "requesting %s" % name + busname = dbus.service.BusName(name, dbus.SessionBus()) + except Exception, e: + print "%s:\n%s" % (e.__class__, e) + self.assert_(not succeed, 'did not expect registering bus name %s to fail' % name) + else: + print busname + self.assert_(succeed, 'expected registering bus name %s to fail'% name) + if name in names: + self.assert_(names[name] == busname, 'got a new instance for same name %s' % name) + print "instance of %s re-used, good!" % name + else: + names[name] = busname + + del busname + + print + + del names + + bus = dbus.Bus() + ret = bus.name_has_owner('org.freedesktop.DBus.Python.TestName') + self.assert_(not ret, 'deleting reference failed to release BusName org.freedesktop.DBus.Python.TestName') + + def testMultipleReturnWithoutSignature(self): + # https://bugs.freedesktop.org/show_bug.cgi?id=10174 + ret = self.iface.MultipleReturnWithoutSignature() + self.assert_(not isinstance(ret, dbus.Struct), repr(ret)) + self.assertEquals(ret, ('abc', 123)) + + def testListExportedChildObjects(self): + self.assert_(self.iface.TestListExportedChildObjects()) + + def testRemoveFromConnection(self): + # https://bugs.freedesktop.org/show_bug.cgi?id=10457 + self.assert_(not self.iface.HasRemovableObject()) + self.assert_(self.iface.AddRemovableObject()) + self.assert_(self.iface.HasRemovableObject()) + + removable = self.bus.get_object(NAME, OBJECT + '/RemovableObject') + iface = dbus.Interface(removable, IFACE) + self.assert_(iface.IsThere()) + self.assert_(iface.RemoveSelf()) + + self.assert_(not self.iface.HasRemovableObject()) + + # and again... + self.assert_(self.iface.AddRemovableObject()) + self.assert_(self.iface.HasRemovableObject()) + self.assert_(iface.IsThere()) + self.assert_(iface.RemoveSelf()) + self.assert_(not self.iface.HasRemovableObject()) + + def testFallbackObjectTrivial(self): + obj = self.bus.get_object(NAME, OBJECT + '/Fallback') + iface = dbus.Interface(obj, IFACE) + path, rel, unique_name = iface.TestPathAndConnKeywords() + self.assertEquals(path, OBJECT + '/Fallback') + self.assertEquals(rel, '/') + self.assertEquals(unique_name, obj.bus_name) + + def testFallbackObjectNested(self): + obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Nested') + iface = dbus.Interface(obj, IFACE) + path, rel, unique_name = iface.TestPathAndConnKeywords() + self.assertEquals(path, OBJECT + '/Fallback/Nested') + self.assertEquals(rel, '/') + self.assertEquals(unique_name, obj.bus_name) + + obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Nested/Badger/Mushroom') + iface = dbus.Interface(obj, IFACE) + path, rel, unique_name = iface.TestPathAndConnKeywords() + self.assertEquals(path, OBJECT + '/Fallback/Nested/Badger/Mushroom') + self.assertEquals(rel, '/Badger/Mushroom') + self.assertEquals(unique_name, obj.bus_name) + + def testFallbackObject(self): + obj = self.bus.get_object(NAME, OBJECT + '/Fallback/Badger/Mushroom') + iface = dbus.Interface(obj, IFACE) + path, rel, unique_name = iface.TestPathAndConnKeywords() + self.assertEquals(path, OBJECT + '/Fallback/Badger/Mushroom') + self.assertEquals(rel, '/Badger/Mushroom') + self.assertEquals(unique_name, obj.bus_name) + + def testTimeoutSync(self): + self.assert_(self.iface.BlockFor500ms(timeout=1.0) is None) + self.assertRaises(dbus.DBusException, + lambda: self.iface.BlockFor500ms(timeout=0.25)) + + def testAsyncRaise(self): + self.assertRaises(dbus.DBusException, self.iface.AsyncRaise) + try: + self.iface.AsyncRaise() + except dbus.DBusException, e: + self.assert_(e.get_dbus_name() == + 'org.freedesktop.bugzilla.bug12403', + e.get_dbus_name()) + else: + self.assert_(False) + + def testClosePrivateBus(self): + # fd.o #12096 + dbus.Bus(private=True).close() + + def testTimeoutAsyncClient(self): + loop = gobject.MainLoop() + passes = [] + fails = [] + def correctly_returned(): + passes.append('1000') + if len(passes) + len(fails) >= 2: + loop.quit() + def correctly_failed(exc): + passes.append('250') + if len(passes) + len(fails) >= 2: + loop.quit() + def incorrectly_returned(): + fails.append('250') + if len(passes) + len(fails) >= 2: + loop.quit() + def incorrectly_failed(exc): + fails.append('1000') + if len(passes) + len(fails) >= 2: + loop.quit() + self.iface.BlockFor500ms(timeout=1.0, + reply_handler=correctly_returned, + error_handler=incorrectly_failed) + self.iface.BlockFor500ms(timeout=0.25, + reply_handler=incorrectly_returned, + error_handler=correctly_failed) + loop.run() + self.assertEquals(passes, ['250', '1000']) + self.assertEquals(fails, []) + + def testTimeoutAsyncService(self): + self.assert_(self.iface.AsyncWait500ms(timeout=1.0) is None) + self.assertRaises(dbus.DBusException, + lambda: self.iface.AsyncWait500ms(timeout=0.25)) + + def testExceptions(self): + #self.assertRaises(dbus.DBusException, + # lambda: self.iface.RaiseValueError) + #self.assertRaises(dbus.DBusException, + # lambda: self.iface.RaiseDBusExceptionNoTraceback) + #self.assertRaises(dbus.DBusException, + # lambda: self.iface.RaiseDBusExceptionWithTraceback) + + try: + self.iface.RaiseValueError() + except Exception, e: + self.assert_(isinstance(e, dbus.DBusException), e.__class__) + self.assert_('.ValueError: Traceback ' in str(e), + 'Wanted a traceback but got:\n"""%s"""' % str(e)) + else: + raise AssertionError('Wanted an exception') + + try: + self.iface.RaiseDBusExceptionNoTraceback() + except Exception, e: + self.assert_(isinstance(e, dbus.DBusException), e.__class__) + self.assertEquals(str(e), + 'com.example.Networking.ServerError: ' + 'Server not responding') + else: + raise AssertionError('Wanted an exception') + + try: + self.iface.RaiseDBusExceptionWithTraceback() + except Exception, e: + self.assert_(isinstance(e, dbus.DBusException), e.__class__) + self.assert_(str(e).startswith('com.example.Misc.RealityFailure: ' + 'Traceback '), + 'Wanted a traceback but got:\n%s' % str(e)) + else: + raise AssertionError('Wanted an exception') + +""" Remove this for now +class TestDBusPythonToGLibBindings(unittest.TestCase): + def setUp(self): + self.bus = dbus.SessionBus() + self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuiteGLibService", "/org/freedesktop/DBus/Tests/MyTestObject") + self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.Tests.MyObject") + + def testIntrospection(self): + #test introspection + print "\n********* Introspection Test ************" + print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable") + print "Introspection test passed" + self.assert_(True) + + def testCalls(self): + print "\n********* Call Test ************" + result = self.iface.ManyArgs(1000, 'Hello GLib', 2) + print result + self.assert_(result == [2002.0, 'HELLO GLIB']) + + arg0 = {"Dude": 1, "john": "palmieri", "python": 2.4} + result = self.iface.ManyStringify(arg0) + print result + + print "Call test passed" + self.assert_(True) + + def testPythonTypes(self): + print "\n********* Testing Python Types ***********" + + for send_val in test_types_vals: + print "Testing %s"% str(send_val) + recv_val = self.iface.EchoVariant(send_val) + self.assertEquals(send_val, recv_val.object) +""" +if __name__ == '__main__': + gobject.threads_init() + dbus.glib.init_threads() + + unittest.main() diff --git a/test/test-p2p.py b/test/test-p2p.py new file mode 100644 index 0000000..8494f23 --- /dev/null +++ b/test/test-p2p.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python + +# Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/> +# Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + + +import sys +import os +import unittest +import time +import logging +import weakref + +builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"]) +pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"]) + +import dbus +import _dbus_bindings +import gobject +import dbus.glib +import dbus.service + + +logging.basicConfig() +logging.getLogger().setLevel(1) + + +NAME = "org.freedesktop.DBus.TestSuitePythonService" +IFACE = "org.freedesktop.DBus.TestSuiteInterface" +OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject" + +class TestDBusBindings(unittest.TestCase): + # This test case relies on the test service already having been activated. + + def get_conn_and_unique(self): + # since I haven't implemented servers yet, I'll use the bus daemon + # as our peer - note that there's no name tracking because we're not + # using dbus.bus.BusConnection! + conn = dbus.connection.Connection( + os.environ['DBUS_SESSION_BUS_ADDRESS']) + unique = conn.call_blocking('org.freedesktop.DBus', + '/org/freedesktop/DBus', + 'org.freedesktop.DBus', 'Hello', + '', (), utf8_strings=True) + self.assert_(unique.__class__ == dbus.UTF8String, repr(unique)) + self.assert_(unique.startswith(':'), unique) + conn.set_unique_name(unique) + return conn, unique + + def testCall(self): + conn, unique = self.get_conn_and_unique() + ret = conn.call_blocking(NAME, OBJECT, IFACE, 'Echo', 'v', ('V',)) + self.assertEquals(ret, 'V') + + def testCallThroughProxy(self): + conn, unique = self.get_conn_and_unique() + proxy = conn.get_object(NAME, OBJECT) + iface = dbus.Interface(proxy, IFACE) + ret = iface.Echo('V') + self.assertEquals(ret, 'V') + + def testSetUniqueName(self): + conn, unique = self.get_conn_and_unique() + ret = conn.call_blocking(NAME, OBJECT, IFACE, + 'MethodExtraInfoKeywords', '', (), + utf8_strings=True) + self.assertEquals(ret, (unique, OBJECT, NAME, + 'dbus.lowlevel.MethodCallMessage')) + + +if __name__ == '__main__': + gobject.threads_init() + dbus.glib.init_threads() + + unittest.main() diff --git a/test/test-service.py b/test/test-service.py new file mode 100755 index 0000000..51865eb --- /dev/null +++ b/test/test-service.py @@ -0,0 +1,344 @@ +#!/usr/bin/env python + +# Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/> +# Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import sys +import os +import logging +from time import sleep + +builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"]) +pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"]) + +import dbus + +if not dbus.__file__.startswith(pydir): + raise Exception("DBus modules are not being picked up from the package") + +import dbus.service +import dbus.glib +import gobject +import random + +from dbus.gobject_service import ExportedGObject + + +logging.basicConfig(filename=builddir + '/test/test-service.log', filemode='w') +logging.getLogger().setLevel(1) +logger = logging.getLogger('test-service') + + +NAME = "org.freedesktop.DBus.TestSuitePythonService" +IFACE = "org.freedesktop.DBus.TestSuiteInterface" +OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject" + +class RemovableObject(dbus.service.Object): + # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457 + @dbus.service.method(IFACE, in_signature='', out_signature='b') + def IsThere(self): + return True + + @dbus.service.method(IFACE, in_signature='', out_signature='b') + def RemoveSelf(self): + self.remove_from_connection() + return True + +class TestGObject(ExportedGObject): + def __init__(self, bus_name, object_path=OBJECT + '/GObject'): + super(TestGObject, self).__init__(bus_name, object_path) + + @dbus.service.method(IFACE) + def Echo(self, arg): + return arg + +class TestInterface(dbus.service.Interface): + @dbus.service.method(IFACE, in_signature='', out_signature='b') + def CheckInheritance(self): + return False + +class Fallback(dbus.service.FallbackObject): + def __init__(self, conn, object_path=OBJECT + '/Fallback'): + super(Fallback, self).__init__(conn, object_path) + self.add_to_connection(conn, object_path + '/Nested') + + @dbus.service.method(IFACE, in_signature='', out_signature='oos', + path_keyword='path', rel_path_keyword='rel', + connection_keyword='conn') + def TestPathAndConnKeywords(self, path=None, conn=None, rel=None): + return path, rel, conn.get_unique_name() + + @dbus.service.signal(IFACE, signature='s', rel_path_keyword='rel_path') + def SignalOneString(self, test, rel_path=None): + logger.info('SignalOneString(%r) @ %r', test, rel_path) + + # Deprecated usage + @dbus.service.signal(IFACE, signature='ss', path_keyword='path') + def SignalTwoStrings(self, test, test2, path=None): + logger.info('SignalTwoStrings(%r, %r) @ %r', test, test2, path) + + @dbus.service.method(IFACE, in_signature='su', out_signature='', + path_keyword='path') + def EmitSignal(self, signal, value, path=None): + sig = getattr(self, str(signal), None) + assert sig is not None + + assert path.startswith(OBJECT + '/Fallback') + rel_path = path[len(OBJECT + '/Fallback'):] + if rel_path == '': + rel_path = '/' + + if signal == 'SignalOneString': + logger.info('Emitting %s from rel %r', signal, rel_path) + sig('I am a fallback', rel_path=rel_path) + else: + val = ('I am', 'a fallback') + logger.info('Emitting %s from abs %r', signal, path) + sig('I am', 'a deprecated fallback', path=path) + +class MultiPathObject(dbus.service.Object): + SUPPORTS_MULTIPLE_OBJECT_PATHS = True + +class TestObject(dbus.service.Object, TestInterface): + def __init__(self, bus_name, object_path=OBJECT): + dbus.service.Object.__init__(self, bus_name, object_path) + self._removable = RemovableObject() + self._multi = MultiPathObject(bus_name, object_path + '/Multi1') + self._multi.add_to_connection(bus_name.get_bus(), + object_path + '/Multi2') + self._multi.add_to_connection(bus_name.get_bus(), + object_path + '/Multi2/3') + + """ Echo whatever is sent + """ + @dbus.service.method(IFACE) + def Echo(self, arg): + return arg + + @dbus.service.method(IFACE, in_signature='s', out_signature='s') + def AcceptUnicodeString(self, foo): + assert isinstance(foo, unicode), (foo, foo.__class__.__mro__) + return foo + + @dbus.service.method(IFACE, in_signature='s', out_signature='s', utf8_strings=True) + def AcceptUTF8String(self, foo): + assert isinstance(foo, str), (foo, foo.__class__.__mro__) + return foo + + @dbus.service.method(IFACE, in_signature='', out_signature='soss', + sender_keyword='sender', path_keyword='path', + destination_keyword='dest', message_keyword='msg') + def MethodExtraInfoKeywords(self, sender=None, path=None, dest=None, + msg=None): + return (sender, path, dest, + msg.__class__.__module__ + '.' + msg.__class__.__name__) + + @dbus.service.method(IFACE, in_signature='ay', out_signature='ay') + def AcceptListOfByte(self, foo): + assert isinstance(foo, list), (foo, foo.__class__.__mro__) + return foo + + @dbus.service.method(IFACE, in_signature='ay', out_signature='ay', byte_arrays=True) + def AcceptByteArray(self, foo): + assert isinstance(foo, str), (foo, foo.__class__.__mro__) + return foo + + @dbus.service.method(IFACE) + def GetComplexArray(self): + ret = [] + for i in range(0,100): + ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100)))) + + return dbus.Array(ret, signature="(uus)") + + def returnValue(self, test): + if test == 0: + return "" + elif test == 1: + return "","" + elif test == 2: + return "","","" + elif test == 3: + return [] + elif test == 4: + return {} + elif test == 5: + return ["",""] + elif test == 6: + return ["","",""] + + @dbus.service.method(IFACE, in_signature='u', out_signature='s') + def ReturnOneString(self, test): + return self.returnValue(test) + + @dbus.service.method(IFACE, in_signature='u', out_signature='ss') + def ReturnTwoStrings(self, test): + return self.returnValue(test) + + @dbus.service.method(IFACE, in_signature='u', out_signature='(ss)') + def ReturnStruct(self, test): + logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test)) + return self.returnValue(test) + + @dbus.service.method(IFACE, in_signature='u', out_signature='as') + def ReturnArray(self, test): + return self.returnValue(test) + + @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}') + def ReturnDict(self, test): + return self.returnValue(test) + + @dbus.service.signal(IFACE, signature='s') + def SignalOneString(self, test): + logger.info('SignalOneString(%r)', test) + + @dbus.service.signal(IFACE, signature='ss') + def SignalTwoStrings(self, test, test2): + logger.info('SignalTwoStrings(%r, %r)', test, test2) + + @dbus.service.signal(IFACE, signature='(ss)') + def SignalStruct(self, test): + logger.info('SignalStruct(%r)', test) + + @dbus.service.signal(IFACE, signature='as') + def SignalArray(self, test): + pass + + @dbus.service.signal(IFACE, signature='a{ss}') + def SignalDict(self, test): + pass + + @dbus.service.method(IFACE, in_signature='su', out_signature='') + def EmitSignal(self, signal, value): + sig = getattr(self, str(signal), None) + assert(sig != None) + + val = self.returnValue(value) + # make two string case work by passing arguments in by tuple + if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)): + val = tuple(val) + else: + val = tuple([val]) + + logger.info('Emitting %s with %r', signal, val) + sig(*val) + + def CheckInheritance(self): + return True + + @dbus.service.method(IFACE, in_signature='', out_signature='b') + def TestListExportedChildObjects(self): + objs_root = session_bus.list_exported_child_objects('/') + assert objs_root == ['org'], objs_root + objs_org = session_bus.list_exported_child_objects('/org') + assert objs_org == ['freedesktop'], objs_org + return True + + @dbus.service.method(IFACE, in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb')) + def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb): + try: + if async: + gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb) + return + else: + if fail: + raise RuntimeError + else: + return_cb(variant) + + return False # do not run again + except Exception, e: + error_cb(e) + + @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender') + def WhoAmI(self, sender): + return sender + + @dbus.service.method(IFACE, in_signature='', out_signature='b') + def AddRemovableObject(self): + # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457 + # Keep the removable object reffed, since that's the use case for this + self._removable.add_to_connection(self._connection, + OBJECT + '/RemovableObject') + return True + + @dbus.service.method(IFACE, in_signature='', out_signature='b') + def HasRemovableObject(self): + # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457 + objs = session_bus.list_exported_child_objects(OBJECT) + return ('RemovableObject' in objs) + + @dbus.service.method(IFACE) + def MultipleReturnWithoutSignature(self): + # https://bugs.freedesktop.org/show_bug.cgi?id=10174 + return dbus.String('abc'), dbus.Int32(123) + + @dbus.service.method(IFACE, in_signature='', out_signature='') + def BlockFor500ms(self): + sleep(0.5) + + @dbus.service.method(IFACE, in_signature='', out_signature='', + async_callbacks=('return_cb', 'raise_cb')) + def AsyncWait500ms(self, return_cb, raise_cb): + def return_from_async_wait(): + return_cb() + return False + gobject.timeout_add(500, return_from_async_wait) + + @dbus.service.method(IFACE, in_signature='', out_signature='') + def RaiseValueError(self): + raise ValueError('Wrong!') + + @dbus.service.method(IFACE, in_signature='', out_signature='') + def RaiseDBusExceptionNoTraceback(self): + class ServerError(dbus.DBusException): + """Exception representing a normal "environmental" error""" + include_traceback = False + _dbus_error_name = 'com.example.Networking.ServerError' + + raise ServerError('Server not responding') + + @dbus.service.method(IFACE, in_signature='', out_signature='') + def RaiseDBusExceptionWithTraceback(self): + class RealityFailure(dbus.DBusException): + """Exception representing a programming error""" + include_traceback = True + _dbus_error_name = 'com.example.Misc.RealityFailure' + + raise RealityFailure('Botched invariant') + + @dbus.service.method(IFACE, in_signature='', out_signature='', + async_callbacks=('return_cb', 'raise_cb')) + def AsyncRaise(self, return_cb, raise_cb): + class Fdo12403Error(dbus.DBusException): + _dbus_error_name = 'org.freedesktop.bugzilla.bug12403' + + raise_cb(Fdo12403Error()) + +session_bus = dbus.SessionBus() +global_name = dbus.service.BusName(NAME, bus=session_bus) +object = TestObject(global_name) +g_object = TestGObject(global_name) +fallback_object = Fallback(session_bus) +loop = gobject.MainLoop() +loop.run() diff --git a/test/test-signals.py b/test/test-signals.py new file mode 100644 index 0000000..04b1e53 --- /dev/null +++ b/test/test-signals.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python + +# Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/> +# Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import sys +import os +import unittest +import time +import logging + +builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"]) +pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"]) + +import dbus +import _dbus_bindings +import gobject +import dbus.glib +import dbus.service + + +logging.basicConfig() +logging.getLogger().setLevel(1) +logger = logging.getLogger('test-signals') + + +pkg = dbus.__file__ +if not pkg.startswith(pydir): + raise Exception("DBus modules (%s) are not being picked up from the package"%pkg) + +if not _dbus_bindings.__file__.startswith(builddir): + raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__) + + +NAME = "org.freedesktop.DBus.TestSuitePythonService" +IFACE = "org.freedesktop.DBus.TestSuiteInterface" +OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject" + + +class TestSignals(unittest.TestCase): + def setUp(self): + logger.info('setUp()') + self.bus = dbus.SessionBus() + self.remote_object = self.bus.get_object(NAME, OBJECT) + self.remote_object_fallback_trivial = self.bus.get_object(NAME, + OBJECT + '/Fallback') + self.remote_object_fallback = self.bus.get_object(NAME, + OBJECT + '/Fallback/Badger') + self.remote_object_follow = self.bus.get_object(NAME, OBJECT, + follow_name_owner_changes=True) + self.iface = dbus.Interface(self.remote_object, IFACE) + self.iface_follow = dbus.Interface(self.remote_object_follow, IFACE) + self.fallback_iface = dbus.Interface(self.remote_object_fallback, IFACE) + self.fallback_trivial_iface = dbus.Interface( + self.remote_object_fallback_trivial, IFACE) + self.in_test = None + + def signal_test_impl(self, iface, name, test_removal=False): + self.in_test = name + # using append rather than assignment here to avoid scoping issues + result = [] + + def _timeout_handler(): + logger.debug('_timeout_handler for %s: current state %s', name, self.in_test) + if self.in_test == name: + main_loop.quit() + def _signal_handler(s, sender, path): + logger.debug('_signal_handler for %s: current state %s', name, self.in_test) + if self.in_test not in (name, name + '+removed'): + return + logger.info('Received signal from %s:%s, argument is %r', + sender, path, s) + result.append('received') + main_loop.quit() + def _rm_timeout_handler(): + logger.debug('_timeout_handler for %s: current state %s', name, self.in_test) + if self.in_test == name + '+removed': + main_loop.quit() + + logger.info('Testing %s', name) + match = iface.connect_to_signal('SignalOneString', _signal_handler, + sender_keyword='sender', + path_keyword='path') + logger.info('Waiting for signal...') + iface.EmitSignal('SignalOneString', 0) + source_id = gobject.timeout_add(1000, _timeout_handler) + main_loop.run() + if not result: + raise AssertionError('Signal did not arrive within 1 second') + logger.debug('Removing match') + match.remove() + gobject.source_remove(source_id) + if test_removal: + self.in_test = name + '+removed' + logger.info('Testing %s', name) + result = [] + iface.EmitSignal('SignalOneString', 0) + source_id = gobject.timeout_add(1000, _rm_timeout_handler) + main_loop.run() + if result: + raise AssertionError('Signal should not have arrived, but did') + gobject.source_remove(source_id) + + def testFallback(self): + self.signal_test_impl(self.fallback_iface, 'Fallback') + + def testFallbackTrivial(self): + self.signal_test_impl(self.fallback_trivial_iface, 'FallbackTrivial') + + def testSignal(self): + self.signal_test_impl(self.iface, 'Signal') + + def testRemoval(self): + self.signal_test_impl(self.iface, 'Removal', True) + + def testSignalAgain(self): + self.signal_test_impl(self.iface, 'SignalAgain') + + def testRemovalAgain(self): + self.signal_test_impl(self.iface, 'RemovalAgain', True) + + def testSignalF(self): + self.signal_test_impl(self.iface_follow, 'Signal') + + def testRemovalF(self): + self.signal_test_impl(self.iface_follow, 'Removal', True) + + def testSignalAgainF(self): + self.signal_test_impl(self.iface_follow, 'SignalAgain') + + def testRemovalAgainF(self): + self.signal_test_impl(self.iface_follow, 'RemovalAgain', True) + +if __name__ == '__main__': + main_loop = gobject.MainLoop() + gobject.threads_init() + dbus.glib.init_threads() + + logger.info('Starting unit test') + unittest.main() diff --git a/test/test-standalone.py b/test/test-standalone.py new file mode 100755 index 0000000..929fc37 --- /dev/null +++ b/test/test-standalone.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python + +"""Tests that don't need an active D-Bus connection to run, but can be +run in isolation. +""" + +# Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import sys +import os +import unittest +import time +from traceback import print_exc + +builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"]) +pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"]) + +import _dbus_bindings +import dbus +import dbus.types as types + +# Check that we're using the right versions +if not dbus.__file__.startswith(pydir): + raise Exception("DBus modules (%s) are not being picked up from the package"%dbus.__file__) +if not _dbus_bindings.__file__.startswith(builddir): + raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__) +assert (_dbus_bindings._python_version & 0xffff0000 + == sys.hexversion & 0xffff0000), \ + '_dbus_bindings was compiled for Python %x but this is Python %x, '\ + 'a different major version'\ + % (_dbus_bindings._python_version, sys.hexversion) +assert _dbus_bindings.__version__ == os.environ['DBUS_PYTHON_VERSION'], \ + '_dbus_bindings was compiled as version %s but Automake says '\ + 'we should be version %s' \ + % (_dbus_bindings.__version__, os.environ['DBUS_PYTHON_VERSION']) + +class TestTypes(unittest.TestCase): + + def test_Dictionary(self): + self.assertEquals(types.Dictionary({'foo':'bar'}), {'foo':'bar'}) + self.assertEquals(types.Dictionary({}, variant_level=2), {}) + self.assertEquals(types.Dictionary({}, variant_level=2).variant_level, 2) + + def test_Array(self): + self.assertEquals(types.Array(['foo','bar']), ['foo','bar']) + self.assertEquals(types.Array([], variant_level=2), []) + self.assertEquals(types.Array([], variant_level=2).variant_level, 2) + + def test_Double(self): + self.assertEquals(types.Double(0.0), 0.0) + self.assertEquals(types.Double(0.125, variant_level=2), 0.125) + self.assertEquals(types.Double(0.125, variant_level=2).variant_level, 2) + + def test_Struct(self): + x = types.Struct(('',)) + self.assertEquals(x.variant_level, 0) + self.assertEquals(x, ('',)) + x = types.Struct('abc', variant_level=42) + self.assertEquals(x.variant_level, 42) + self.assertEquals(x, ('a','b','c')) + + def test_Byte(self): + self.assertEquals(types.Byte('x', variant_level=2), types.Byte(ord('x'))) + + def test_object_path_attr(self): + class MyObject(object): + __dbus_object_path__ = '/foo' + from _dbus_bindings import SignalMessage + self.assertEquals(SignalMessage.guess_signature(MyObject()), 'o') + + def test_integers(self): + # This is an API guarantee. Note that exactly which of these types + # are ints and which of them are longs is *not* guaranteed. + for cls in (types.Int16, types.UInt16, types.Int32, types.UInt32, + types.Int64, types.UInt64): + self.assert_(issubclass(cls, (int, long))) + self.assert_(isinstance(cls(0), (int, long))) + self.assertEquals(cls(0), 0) + self.assertEquals(cls(23, variant_level=1), 23) + self.assertEquals(cls(23, variant_level=1).variant_level, 1) + + def test_integer_limits_16(self): + self.assertEquals(types.Int16(0x7fff), 0x7fff) + self.assertEquals(types.Int16(-0x8000), -0x8000) + self.assertEquals(types.UInt16(0xffff), 0xffff) + self.assertRaises(Exception, types.Int16, 0x8000) + self.assertRaises(Exception, types.Int16, -0x8001) + self.assertRaises(Exception, types.UInt16, 0x10000) + + def test_integer_limits_32(self): + self.assertEquals(types.Int32(0x7fffffff), 0x7fffffff) + self.assertEquals(types.Int32(-0x80000000L), -0x80000000L) + self.assertEquals(types.UInt32(0xffffffffL), 0xffffffffL) + self.assertRaises(Exception, types.Int32, 0x80000000L) + self.assertRaises(Exception, types.Int32, -0x80000001L) + self.assertRaises(Exception, types.UInt32, 0x100000000L) + + def test_integer_limits_64(self): + self.assertEquals(types.Int64(0x7fffffffffffffffL), 0x7fffffffffffffffL) + self.assertEquals(types.Int64(-0x8000000000000000L), -0x8000000000000000L) + self.assertEquals(types.UInt64(0xffffffffffffffffL), 0xffffffffffffffffL) + self.assertRaises(Exception, types.Int16, 0x8000000000000000L) + self.assertRaises(Exception, types.Int16, -0x8000000000000001L) + self.assertRaises(Exception, types.UInt16, 0x10000000000000000L) + + def test_Signature(self): + self.assertRaises(Exception, types.Signature, 'a') + self.assertEquals(types.Signature('ab', variant_level=23), 'ab') + self.assert_(isinstance(types.Signature('ab'), str)) + self.assertEquals(tuple(types.Signature('ab(xt)a{sv}')), + ('ab', '(xt)', 'a{sv}')) + self.assert_(isinstance(tuple(types.Signature('ab'))[0], + types.Signature)) + + +class TestMessageMarshalling(unittest.TestCase): + + def test_count(self): + from _dbus_bindings import SignalMessage + s = SignalMessage('/', 'foo.bar', 'baz') + try: + s.append('a', signature='ss') + except TypeError: + pass + else: + raise AssertionError('Appending too few things in a message ' + 'should fail') + s = SignalMessage('/', 'foo.bar', 'baz') + try: + s.append('a','b','c', signature='ss') + except TypeError: + pass + else: + raise AssertionError('Appending too many things in a message ' + 'should fail') + + def test_append(self): + aeq = self.assertEquals + from _dbus_bindings import SignalMessage + s = SignalMessage('/', 'foo.bar', 'baz') + s.append([types.Byte(1)], signature='ay') + aeq(s.get_signature(), 'ay') + aeq(s.get_args_list(), [[types.Byte(1)]]) + + s = SignalMessage('/', 'foo.bar', 'baz') + s.append([], signature='ay') + aeq(s.get_args_list(), [[]]) + + def test_append_ByteArray(self): + aeq = self.assertEquals + from _dbus_bindings import SignalMessage + s = SignalMessage('/', 'foo.bar', 'baz') + s.append(types.ByteArray('ab'), signature='ay') + aeq(s.get_args_list(), [[types.Byte('a'), types.Byte('b')]]) + s = SignalMessage('/', 'foo.bar', 'baz') + s.append(types.ByteArray('ab'), signature='av') + aeq(s.get_args_list(), [[types.Byte('a'), types.Byte('b')]]) + + def test_append_Variant(self): + a = self.assert_ + aeq = self.assertEquals + from _dbus_bindings import SignalMessage + s = SignalMessage('/', 'foo.bar', 'baz') + s.append(types.Int32(1, variant_level=0), + types.String('a', variant_level=42), + types.Array([types.Byte('a', variant_level=1), + types.UInt32(123, variant_level=1)], + signature='v'), + signature='vvv') + aeq(s.get_signature(), 'vvv') + args = s.get_args_list() + aeq(args[0].__class__, types.Int32) + aeq(args[0].variant_level, 1) + aeq(args[1].__class__, types.String) + aeq(args[1].variant_level, 42) + aeq(args[2].__class__, types.Array) + aeq(args[2].variant_level, 1) + aeq(args[2].signature, 'v') + + def test_guess_signature(self): + aeq = self.assertEquals + from _dbus_bindings import Message + aeq(Message.guess_signature(('a','b')), '(ss)') + aeq(Message.guess_signature('a','b'), 'ss') + aeq(Message.guess_signature(['a','b']), 'as') + aeq(Message.guess_signature(('a',)), '(s)') + aeq(Message.guess_signature('abc'), 's') + aeq(Message.guess_signature(types.Int32(123)), 'i') + aeq(Message.guess_signature(types.ByteArray('abc')), 'ay') + aeq(Message.guess_signature(('a',)), '(s)') + aeq(Message.guess_signature(['a']), 'as') + aeq(Message.guess_signature({'a':'b'}), 'a{ss}') + + def test_guess_signature_dbus_types(self): + aeq = self.assertEquals + from _dbus_bindings import Message + gs = Message.guess_signature + aeq(gs(types.Dictionary({'a':'b'})), 'a{ss}') + aeq(gs(types.Dictionary({'a':'b'}, signature='sv')), 'a{sv}') + aeq(gs(types.Dictionary({}, signature='iu')), 'a{iu}') + aeq(gs(types.Array([types.Int32(1)])), 'ai') + aeq(gs(types.Array([types.Int32(1)], signature='u')), 'au') + + def test_get_args_options(self): + aeq = self.assertEquals + s = _dbus_bindings.SignalMessage('/', 'foo.bar', 'baz') + s.append('b', 'bytes', -1, 1, 'str', 'var', signature='yayiusv') + aeq(s.get_args_list(), [ord('b'), + [ord('b'),ord('y'),ord('t'),ord('e'), ord('s')], + -1, 1, u'str', u'var']) + byte, bytes, int32, uint32, string, variant = s.get_args_list() + aeq(byte.__class__, types.Byte) + aeq(bytes.__class__, types.Array) + aeq(bytes[0].__class__, types.Byte) + aeq(int32.__class__, types.Int32) + aeq(uint32.__class__, types.UInt32) + aeq(string.__class__, types.String) + aeq(string.variant_level, 0) + aeq(variant.__class__, types.String) + aeq(variant.variant_level, 1) + + byte, bytes, int32, uint32, string, variant = s.get_args_list( + byte_arrays=True) + aeq(byte.__class__, types.Byte) + aeq(bytes.__class__, types.ByteArray) + aeq(bytes, 'bytes') + aeq(bytes[0].__class__, str) + aeq(int32.__class__, types.Int32) + aeq(uint32.__class__, types.UInt32) + aeq(string.__class__, types.String) + aeq(variant.__class__, types.String) + aeq(variant.variant_level, 1) + + byte, bytes, int32, uint32, string, variant = s.get_args_list( + utf8_strings=True) + aeq(byte.__class__, types.Byte) + aeq(bytes.__class__, types.Array) + aeq(bytes[0].__class__, types.Byte) + aeq(int32.__class__, types.Int32) + aeq(uint32.__class__, types.UInt32) + aeq(string.__class__, types.UTF8String) + aeq(string, 'str') + aeq(variant.__class__, types.UTF8String) + aeq(variant.variant_level, 1) + aeq(variant, 'var') + + def test_object_path_attr(self): + from _dbus_bindings import SignalMessage + class MyObject(object): + __dbus_object_path__ = '/foo' + s = SignalMessage('/', 'foo.bar', 'baz') + s.append(MyObject(), signature='o') + s.append(MyObject()) + self.assertEquals(s.get_args_list(), ['/foo', '/foo']) + + def test_struct(self): + from _dbus_bindings import SignalMessage + s = SignalMessage('/', 'foo.bar', 'baz') + try: + s.append(('a',), signature='(ss)') + except TypeError: + pass + else: + raise AssertionError('Appending too few things in a struct ' + 'should fail') + s = SignalMessage('/', 'foo.bar', 'baz') + try: + s.append(('a','b','c'), signature='(ss)') + except TypeError: + pass + else: + raise AssertionError('Appending too many things in a struct ' + 'should fail') + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test-unusable-main-loop.py b/test/test-unusable-main-loop.py new file mode 100644 index 0000000..cee8723 --- /dev/null +++ b/test/test-unusable-main-loop.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python + +# Copyright (C) 2007 Collabora Ltd. <http://www.collabora.co.uk/> +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, copy, +# modify, merge, publish, distribute, sublicense, and/or sell copies +# of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +# DEALINGS IN THE SOFTWARE. + +import dbus + +from dbus_py_test import UnusableMainLoop + +def main(): + UnusableMainLoop(set_as_default=True) + try: + bus = dbus.SessionBus() + except ValueError, e: + print "Correctly got ValueError from UnusableMainLoop" + else: + raise AssertionError("Expected ValueError from UnusableMainLoop") + +if __name__ == '__main__': + main() diff --git a/test/tmp-session-bus.conf.in b/test/tmp-session-bus.conf.in new file mode 100644 index 0000000..9d172ff --- /dev/null +++ b/test/tmp-session-bus.conf.in @@ -0,0 +1,21 @@ +<!-- @configure_input@ --> +<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-BUS Bus Configuration 1.0//EN" + "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd"> +<busconfig> + <!-- Our well-known bus type, don't change this --> + <type>session</type> + <listen>unix:tmpdir=@abs_top_builddir@/test</listen> + + <servicedir>@abs_top_builddir@/test</servicedir> + + <policy context="default"> + <!-- Allow everything to be sent --> + <allow send_destination="*"/> + <!-- Allow everything to be received --> + <allow eavesdrop="true"/> + <!-- Allow anyone to own anything --> + <allow own="*"/> + </policy> + +</busconfig> +<!-- vim:set ft=xml sts=2 sw=2 et: --> |