PyGILState_STATE state;
PyObject *callback, *args;
PyObject *ret;
+ PyObject *py_user_data;
gboolean res;
+ gint len, i;
g_return_val_if_fail(user_data != NULL, FALSE);
state = pyg_gil_state_ensure();
- callback = PyTuple_GetItem((PyObject *)user_data, 0);
- args = Py_BuildValue("(NNO)",
+ py_user_data = (PyObject *) user_data;
+
+ callback = PyTuple_GetItem(py_user_data, 0);
+ args = Py_BuildValue("(NN)",
pyg_boxed_new(GST_TYPE_PROBE, probe, TRUE, TRUE),
- pyg_boxed_new(GST_TYPE_DATA, *data, TRUE, TRUE),
- PyTuple_GetItem((PyObject *)user_data, 1));
+ pyg_boxed_new(GST_TYPE_DATA, *data, TRUE, TRUE));
+
+ len = PyTuple_Size(py_user_data);
+ for (i = 1; i < len; ++i) {
+ PyObject *tuple = args;
+ args = PySequence_Concat(tuple, PyTuple_GetItem(py_user_data, i));
+ Py_DECREF(tuple);
+ }
ret = PyObject_CallObject(callback, args);
if (!ret) {
self.assertRaises(TypeError, probe.perform, None)
self.assertRaises(TypeError, probe.perform, "nodata")
- def testPerform(self):
+ def testPerformNoArg(self):
+ probe = gst.Probe(True, self._probe_callback_no_arg)
+ buffer = gst.Buffer()
+ probe.perform(buffer)
+ self.assertEqual(self._no_arg, None)
+
+ def _probe_callback_no_arg(self, probe, data):
+ self._no_arg = None
+
+ def testPerformOneArg(self):
probe = gst.Probe(True, self._probe_callback, "yeeha")
buffer = gst.Buffer()
probe.perform(buffer)
def _probe_callback(self, probe, data, result):
self._probe_result = result
return True
-
+
+ def testPerformTwoArgs(self):
+ probe = gst.Probe(True, self._probe_callback_two, "yeeha", "works")
+ buffer = gst.Buffer()
+ probe.perform(buffer)
+ self.assertEqual(self._probe_result1, "yeeha")
+ self.assertEqual(self._probe_result2, "works")
+
+ def _probe_callback_two(self, probe, data, result1, result2):
+ self._probe_result1 = result1
+ self._probe_result2 = result2
+ return True
+
if __name__ == "__main__":
unittest.main()