9:51 a.m.
From: Inaky Perez-Gonzalez <inaky.perez-gonzalez(a)intel.com>
---
test/test-sms-msg-cancel | 173 ++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 173 insertions(+), 0 deletions(-)
create mode 100755 test/test-sms-msg-cancel
diff --git a/test/test-sms-msg-cancel b/test/test-sms-msg-cancel
new file mode 100755
index 0000000..e9bb40f
--- /dev/null
+++ b/test/test-sms-msg-cancel
@@ -0,0 +1,173 @@
+#!/usr/bin/python
+#
+# Sends a message and inmediately cancels it once the state moves to
+# "queued"; fails if the message doesn't move to the CANCELLED state.
+#
+# Arguments are: DEST-PHONE-NUMBER MSG-TEXT BOOLEAN
+#
+# BOOLEAN: request delivery confirmation / no
+#
+# Notes: accessing the globals result and reason using 'globals'
+#
+
+import gobject
+import sys
+import dbus
+import dbus.mainloop.glib
+
+dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+class sms_msg_cancel_test:
+ number = ""
+ text = ""
+ wsr = ""
+ target_state = ""
+
+ result = ""
+ reason = ""
+
+ sms_msg_path = ""
+
+ # FIXME
+ def __init__(self, number, text, wsr):
+ self.number = number
+ self.text = text
+ self.wsr = wsr
+ self.mainloop = gobject.MainLoop()
+
+ self.dbus = dbus.SystemBus()
+ ofono_manager = dbus.Interface(self.dbus.get_object('org.ofono',
'/'),
+ 'org.ofono.Manager')
+ path = ofono_manager.GetProperties()["Modems"][0]
+ self.sms_manager =
dbus.Interface(self.dbus.get_object('org.ofono', path),
+ 'org.ofono.SmsManager')
+ self.sms_manager.SetProperty("UseDeliveryReports",
dbus.Boolean(int(self.wsr)))
+
+ # Connect the signal *before* creating the message, otherwise we miss
+ # the transition - this also means we can't use
sms_msg.connect_to_signal()
+ self.dbus.add_signal_receiver(self.property_changed,
bus_name="org.ofono",
+ signal_name = "PropertyChanged",
+ path_keyword="path",
+ interface_keyword="interface")
+ self.result = "FAILED"
+ self.reason = "didn't see CANCELING state"
+
+ # Cancel the message and update test state
+ def msg_cancel(self):
+ print "I: %s[%s]: SMS message: canceling because " \
+ "target state was reached" \
+ % (self.sms_msg_path, self.target_state)
+ sms_msg = dbus.Interface(self.dbus.get_object('org.ofono',
self.sms_msg_path),
+ 'org.ofono.SmsMessage')
+ try:
+ sms_msg.Cancel()
+ except dbus.DBusException as dbus_ex:
+ if dbus_ex.get_dbus_name() !=
'org.freedesktop.DBus.Error.UnknownMethod':
+ pass
+ # Object has been cancelled already--this is
+ # why we get the D-Bus exception, so we can
+ # consider this a win
+ print "I: %s[%s]: SMS message: ok, D-Bus object already
removed" \
+ % (self.sms_msg_path, self.target_state)
+ self.result = "SUCCESS"
+ self.reason = "object dissapeared from the bus, meaning it
was cancelled"
+ self.mainloop.quit()
+ return
+
+ # Run the actual test FIXME
+ def run(self, target_state):
+ self.target_state = target_state
+
+ self.result = "FAILED"
+ self.reason = "didn't see %s state to allow cancelling" \
+ % target_state
+
+ print "I: [%s]: starting test" % self.target_state
+ self.sms_msg_path = self.sms_manager.SendMessage(number, text)
+ print "I: %s[%s]: SMS message: D-Bus object created" \
+ % (self.sms_msg_path, self.target_state)
+
+ # wait for property changed signals, property_changed() will be called
+ # when there is a transition and the thing cancelled.
+ gobject.timeout_add(20000, self.timeout_fail)
+ # If the target state to cancel on is QUEUED, schedule a
+ # cancellation inmediately -- note we need to schedule it in
+ # the mainloop to ensure property_changed() signals are
+ # properly propagated.
+ if target_state == "queued" \
+ or target_state == "canceling" \
+ or target_state == "cancelled":
+ gobject.idle_add(self.msg_cancel)
+ self.mainloop.run()
+ if self.result == "SUCCESS":
+ letter = "I"
+ result = 0
+ else:
+ letter = "E"
+ result = 1
+ print "%s: %s[%s]: %s: %s" % (letter, self.sms_msg_path,
+ self.target_state,
+ self.result, self.reason)
+ return result
+
+ def timeout_fail(self):
+ self.result = "FAILED"
+ self.reason = "%s (timedout)" % self.reason
+ self.mainloop.quit()
+ # Don't rearm the timeout ... doesn't really matter as we
+ # quit the mainloop.
+ return 0
+
+
+ # When the message's state (the one we created) switches to QUEUED, cancel
+ def property_changed(self, property_name, property_value,
+ path, interface):
+ if interface != "org.ofono.SmsMessage":
+ return
+ if self.sms_msg_path != path:
+ return
+ if property_name != "State":
+ return
+ print "I: %s[%s]: SMS message: transitioning to %s state" \
+ % (self.sms_msg_path, self.target_state, property_value)
+ if property_value == "canceling":
+ print "I: %s[%s]: SMS message: ok, saw canceling state"
\
+ % (self.sms_msg_path, self.target_state)
+ self.result = "FAIL"
+ self.reason = "didn't see the CANCELLED state"
+ if property_value == "cancelled":
+ print "I: %s[%s]: SMS message: ok, saw cancelled state,
quiting" \
+ % (self.sms_msg_path, self.target_state)
+ self.result = "SUCCESS"
+ self.reason = "saw CANCELLED state"
+ self.mainloop.quit()
+ # If we get to the state that is the target state for
+ # the test, cancel the message -- we need to do this
+ # after evaluating propery_value or we might enter
+ # into conflicts (as self.msg_cancel() will change
+ # self.{result,reason} if it turns out the operation
+ # is complete).
+ if self.target_state == property_value:
+ # Cancel the message
+ self.msg_cancel()
+
+
+# Send the message
+if len(sys.argv) == 4:
+ wsr = int(sys.argv[3])
+elif len(sys.argv) == 3:
+ wsr = 0
+else:
+ raise NameError("E: Bad number of arguments (expected NUMBER TEXT
[BOOLEAN])")
+
+number = sys.argv[1]
+text = sys.argv[2]
+global test
+
+test = sms_msg_cancel_test(number, text, wsr)
+
+test.run('queued')
+test.run('wsr')
+test.run('canceling')
+test.run('cancelled')
+test.run('done')
--
1.6.6.1