tests/check/pipelines/ipcpipeline.c

/* GStreamer
 *
 * tests for the ipcpipelinesrc/ipcpipelinesink elements
 *
 * Copyright (C) 2015-2017 YouView TV Ltd
 *   Author: Vincent Penquerc'h <vincent.penquerch@collabora.co.uk>
 *   Author: George Kiagiadakis <george.kiagiadakis@collabora.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#define _GNU_SOURCE             /* See feature_test_macros(7) */
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>
#include <sys/wait.h>
#include <sys/file.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <gst/check/gstcheck.h>
#include <string.h>

#ifndef HAVE_PIPE2
static int
pipe2 (int pipedes[2], int flags)
{
  int ret = pipe (pipedes);
  if (ret < 0)
    return ret;
  if (flags != 0) {
    ret = fcntl (pipedes[0], F_SETFL, flags);
    if (ret < 0)
      return ret;
    ret = fcntl (pipedes[1], F_SETFL, flags);
    if (ret < 0)
      return ret;
  }
  return 0;
}
#endif

/* This enum contains flags that are used to configure the setup that
 * test_base() will do internally */
typedef enum
{
  /* Features related to the multi-process setup */
  TEST_FEATURE_SPLIT_SINKS = 0x1,       /* separate audio and video sink processes */
  TEST_FEATURE_RECOVERY_SLAVE_PROCESS = 0x2,
  TEST_FEATURE_RECOVERY_MASTER_PROCESS = 0x4,

  TEST_FEATURE_HAS_VIDEO = 0x10,
  TEST_FEATURE_LIVE = 0x20,     /* sets is-live=true in {audio,video}testsrc */
  TEST_FEATURE_ASYNC_SINK = 0x40,       /* sets sync=false in fakesink */
  TEST_FEATURE_ERROR_SINK = 0x80,       /* generates error message in the slave */
  TEST_FEATURE_LONG_DURATION = 0x100,   /* bigger num-buffers in {audio,video}testsrc */
  TEST_FEATURE_FILTER_SINK_CAPS = 0x200,        /* plugs capsfilter before fakesink */

  /* Source selection; Use only one of those, do not combine! */
  TEST_FEATURE_TEST_SOURCE = 0x400,
  TEST_FEATURE_WAV_SOURCE = 0x800,
  TEST_FEATURE_MPEGTS_SOURCE = 0x1000 | TEST_FEATURE_HAS_VIDEO,
  TEST_FEATURE_LIVE_A_SOURCE =
      TEST_FEATURE_TEST_SOURCE | TEST_FEATURE_LIVE | TEST_FEATURE_ASYNC_SINK,
  TEST_FEATURE_LIVE_AV_SOURCE =
      TEST_FEATURE_LIVE_A_SOURCE | TEST_FEATURE_HAS_VIDEO,
} TestFeatures;

/* This is the data structure that each function of the each test receives
 * in user_data. It contains pointers to stack-allocated, test-specific
 * structures that contain the test parameters (input data), the runtime
 * data of the master (source) process (master data) and the runtime data
 * of the slave (sink) process (slave data) */
typedef struct
{
  gpointer id;                  /* input data struct */
  gpointer md;                  /* master data struct */
  gpointer sd;                  /* slave data struct */

  TestFeatures features;        /* the features that this test is running with */

  /* whether there is both an audio and a video stream
   * in this process'es pipeline */
  gboolean two_streams;

  /* the pipeline of this process; could be either master or slave */
  GstElement *p;

  /* this callback will be called in the master process when
   * the master gets STATE_CHANGED with the new state being state_target */
  void (*state_changed_cb) (gpointer);
  GstState state_target;

  /* used by EXCLUSIVE_CALL() */
  gint exclusive_call_counter;
} test_data;

/* All pipelines do not start buffers at exactly zero, so we consider
   timestamps within a small tolerance to be zero */
#define CLOSE_ENOUGH_TO_ZERO (GST_SECOND / 5)

/* milliseconds */
#define STEP_AT  100
#define PAUSE_AT 500
#define SEEK_AT  700
#define QUERY_AT 600
#define MESSAGE_AT 600
#define CRASH_AT 600
#define STOP_AT  600

/* Rough duration of the sample files we use */
#define MPEGTS_SAMPLE_ROUGH_DURATION (GST_SECOND * 64 / 10)
#define WAV_SAMPLE_ROUGH_DURATION (GST_SECOND * 65 / 10)

enum
{
  MSG_ACK = 0,
  MSG_START = 1
};

static GMainLoop *loop;
static gboolean child_dead;
static int pipesfa[2], pipesba[2], pipesfv[2], pipesbv[2];
static int ctlsock[2];
static int recovery_pid = 0;
static int check_fd = -1;
static GList *weak_refs = NULL;

/* lock helpers */

#define FAIL_IF(x) do { lock_check (); fail_if(x); unlock_check (); } while(0)
#define FAIL_UNLESS(x) do { lock_check (); fail_unless(x); unlock_check (); } while(0)
#define FAIL_UNLESS_EQUALS_INT(x,y) do { lock_check (); fail_unless_equals_int(x,y); unlock_check (); } while(0)
#define FAIL() do { lock_check (); fail(); unlock_check (); } while(0)

static void
lock_check (void)
{
  flock (check_fd, LOCK_EX);
}

static void
unlock_check (void)
{
  flock (check_fd, LOCK_UN);
}

static void
setup_lock (void)
{
  gchar *name = NULL;
  check_fd = g_file_open_tmp (NULL, &name, NULL);
  unlink (name);
  g_free (name);
}

/* tracking for ipcpipeline elements; this is used mainly to detect leaks,
 * but also to provide a method for calling "disconnect" on all of them
 * in the tests that require it */

static void
remove_weak_ref (GstElement * element)
{
  weak_refs = g_list_remove (weak_refs, element);
}

static void
add_weak_ref (GstElement * element)
{
  weak_refs = g_list_append (weak_refs, element);
  g_object_weak_ref (G_OBJECT (element), (GWeakNotify) remove_weak_ref,
      element);
}

static void
disconnect_ipcpipeline_elements (void)
{
  GList *l;

  for (l = weak_refs; l; l = l->next) {
    g_signal_emit_by_name (G_OBJECT (l->data), "disconnect", NULL);
  }
}

/* helper functions */

#define EXCLUSIVE_CALL(td,func) \
  G_STMT_START { \
    if (!td->two_streams || \
        g_atomic_int_add (&td->exclusive_call_counter, 1) == 1) { \
      func; \
    } \
  } G_STMT_END

static void
cleanup_bus (GstElement * pipeline)
{
  gst_bus_remove_watch (GST_ELEMENT_BUS (pipeline));
  gst_bus_set_flushing (GST_ELEMENT_BUS (pipeline), TRUE);
}

static void
setup_log (const char *logfile, int append)
{
  FILE *f;

  f = fopen (logfile, append ? "a+" : "w");
  gst_debug_add_log_function (gst_debug_log_default, f, NULL);
}

static GstElement *
create_pipeline (const char *type)
{
  GstElement *pipeline;

  pipeline = gst_element_factory_make (type, NULL);
  FAIL_UNLESS (pipeline);

  return pipeline;
}

static GQuark
to_be_removed_quark (void)
{
  static GQuark q = 0;
  if (!q)
    q = g_quark_from_static_string ("to_be_removed");
  return q;
}

static gboolean
are_caps_audio (const GstCaps * caps)
{
  GstStructure *structure;
  const char *name;

  structure = gst_caps_get_structure (caps, 0);
  name = gst_structure_get_name (structure);
  return g_str_has_prefix (name, "audio/");
}

static gboolean
are_caps_video (const GstCaps * caps)
{
  GstStructure *structure;
  const char *name;

  structure = gst_caps_get_structure (caps, 0);
  name = gst_structure_get_name (structure);
  return (g_str_has_prefix (name, "video/")
      && strcmp (name, "video/x-dvd-subpicture"));
}

static int
caps2idx (GstCaps * caps, gboolean two_streams)
{
  int idx;

  if (!two_streams)
    return 0;

  if (are_caps_audio (caps)) {
    idx = 0;
  } else if (are_caps_video (caps)) {
    idx = 1;
  } else {
    FAIL_IF (1);
    idx = 0;
  }
  return idx;
}

static int
pad2idx (GstPad * pad, gboolean two_streams)
{
  GstCaps *caps;
  int idx;

  if (!two_streams)
    return 0;

  caps = gst_pad_get_current_caps (pad);
  if (!caps)
    caps = gst_pad_get_pad_template_caps (pad);
  FAIL_UNLESS (caps);

  idx = caps2idx (caps, two_streams);

  gst_caps_unref (caps);
  return idx;
}

static gboolean
stop_pipeline (gpointer user_data)
{
  GstElement *pipeline = user_data;
  GstStateChangeReturn ret;

  ret = gst_element_set_state (pipeline, GST_STATE_NULL);
  FAIL_IF (ret == GST_STATE_CHANGE_FAILURE);
  gst_object_unref (pipeline);
  g_main_loop_quit (loop);
  return FALSE;
}

static void
hook_peer_probe_types (const GValue * sinkv, GstPadProbeCallback probe,
    unsigned int types, gpointer user_data)
{
  GstElement *sink;
  GstPad *pad, *peer;

  sink = g_value_get_object (sinkv);
  FAIL_UNLESS (sink);
  pad = gst_element_get_static_pad (sink, "sink");
  FAIL_UNLESS (pad);
  peer = gst_pad_get_peer (pad);
  FAIL_UNLESS (peer);
  gst_pad_add_probe (peer, types, probe, user_data, NULL);
  gst_object_unref (peer);
  gst_object_unref (pad);
}

static void
hook_probe_types (const GValue * sinkv, GstPadProbeCallback probe,
    unsigned int types, gpointer user_data)
{
  GstElement *sink;
  GstPad *pad;

  sink = g_value_get_object (sinkv);
  FAIL_UNLESS (sink);
  pad = gst_element_get_static_pad (sink, "sink");
  FAIL_UNLESS (pad);
  gst_pad_add_probe (pad, types, probe, user_data, NULL);
  gst_object_unref (pad);
}

static void
hook_probe (const GValue * sinkv, GstPadProbeCallback probe, gpointer user_data)
{
  hook_probe_types (sinkv, probe,
      GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM | GST_PAD_PROBE_TYPE_EVENT_FLUSH |
      GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM, user_data);
}

/* the master process'es async GstBus callback */
static gboolean
master_bus_msg (GstBus * bus, GstMessage * message, gpointer user_data)
{
  test_data *td = user_data;

  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_ERROR:{
      GError *err;
      gchar *dbg;

      /* elements we are removing might error out as they are taken out
         of the pipeline, and fail to push. We don't care about those. */
      if (g_object_get_qdata (G_OBJECT (GST_MESSAGE_SRC (message)),
              to_be_removed_quark ()))
        break;

      gst_message_parse_error (message, &err, &dbg);
      g_printerr ("ERROR: %s\n", err->message);
      if (dbg != NULL)
        g_printerr ("ERROR debug information: %s\n", dbg);
      g_error_free (err);
      g_free (dbg);
      g_assert_not_reached ();
      break;
    }
    case GST_MESSAGE_WARNING:{
      GError *err;
      gchar *dbg;

      gst_message_parse_warning (message, &err, &dbg);
      g_printerr ("WARNING: %s\n", err->message);
      if (dbg != NULL)
        g_printerr ("WARNING debug information: %s\n", dbg);
      g_error_free (err);
      g_free (dbg);
      g_assert_not_reached ();
      break;
    }
    case GST_MESSAGE_EOS:
      g_main_loop_quit (loop);
      break;
    case GST_MESSAGE_STATE_CHANGED:
      if (GST_MESSAGE_SRC (message) == GST_OBJECT_CAST (td->p)
          && td->state_changed_cb) {
        GstState state;
        gst_message_parse_state_changed (message, NULL, &state, NULL);
        if (state == td->state_target)
          td->state_changed_cb (td);
      }
      break;
    default:
      break;
  }
  return TRUE;
}

/* source construction functions */

static GstElement *
create_wavparse_source_loc (const char *loc, int fdina, int fdouta)
{
  GstElement *sbin, *pipeline, *filesrc, *ipcpipelinesink;
  GError *e = NULL;

  pipeline = create_pipeline ("pipeline");
  sbin =
      gst_parse_bin_from_description ("pushfilesrc name=filesrc ! wavparse",
      TRUE, &e);
  FAIL_IF (e || !sbin);
  gst_element_set_name (sbin, "source");
  filesrc = gst_bin_get_by_name (GST_BIN (sbin), "filesrc");
  FAIL_UNLESS (filesrc);
  g_object_set (filesrc, "location", loc, NULL);
  gst_object_unref (filesrc);
  ipcpipelinesink =
      gst_element_factory_make ("ipcpipelinesink", "ipcpipelinesink");
  add_weak_ref (ipcpipelinesink);
  g_object_set (ipcpipelinesink, "fdin", fdina, "fdout", fdouta, NULL);
  gst_bin_add_many (GST_BIN (pipeline), sbin, ipcpipelinesink, NULL);
  FAIL_UNLESS (gst_element_link_many (sbin, ipcpipelinesink, NULL));

  return pipeline;
}

static void
on_pad_added (GstElement * element, GstPad * pad, gpointer data)
{
  GstCaps *caps;
  GstElement *next;
  GstBin *pipeline = data;
  GstPad *sink_pad;

  caps = gst_pad_get_current_caps (pad);
  if (!caps)
    caps = gst_pad_get_pad_template_caps (pad);

  if (are_caps_video (caps)) {
    next = gst_bin_get_by_name (GST_BIN (pipeline), "vqueue");
  } else if (are_caps_audio (caps)) {
    next = gst_bin_get_by_name (GST_BIN (pipeline), "aqueue");
  } else {
    gst_caps_unref (caps);
    return;
  }
  gst_caps_unref (caps);

  FAIL_UNLESS (next);
  sink_pad = gst_element_get_static_pad (next, "sink");
  FAIL_UNLESS (sink_pad);
  FAIL_UNLESS (gst_pad_link (pad, sink_pad) == GST_PAD_LINK_OK);
  gst_object_unref (sink_pad);

  gst_object_unref (next);
}

static GstElement *
create_mpegts_source_loc (const char *loc, int fdina, int fdouta, int fdinv,
    int fdoutv)
{
  GstElement *pipeline, *filesrc, *tsdemux, *aqueue, *vqueue, *aipcpipelinesink,
      *vipcpipelinesink;

  pipeline = create_pipeline ("pipeline");
  filesrc = gst_element_factory_make ("filesrc", NULL);
  g_object_set (filesrc, "location", loc, NULL);
  tsdemux = gst_element_factory_make ("tsdemux", NULL);
  g_signal_connect (tsdemux, "pad-added", G_CALLBACK (on_pad_added), pipeline);
  aqueue = gst_element_factory_make ("queue", "aqueue");
  aipcpipelinesink = gst_element_factory_make ("ipcpipelinesink", NULL);
  add_weak_ref (aipcpipelinesink);
  g_object_set (aipcpipelinesink, "fdin", fdina, "fdout", fdouta, NULL);
  vqueue = gst_element_factory_make ("queue", "vqueue");
  vipcpipelinesink = gst_element_factory_make ("ipcpipelinesink", NULL);
  add_weak_ref (vipcpipelinesink);
  g_object_set (vipcpipelinesink, "fdin", fdinv, "fdout", fdoutv, NULL);
  gst_bin_add_many (GST_BIN (pipeline), filesrc, tsdemux, aqueue,
      aipcpipelinesink, vqueue, vipcpipelinesink, NULL);
  FAIL_UNLESS (gst_element_link_many (filesrc, tsdemux, NULL));
  FAIL_UNLESS (gst_element_link_many (aqueue, aipcpipelinesink, NULL));
  FAIL_UNLESS (gst_element_link_many (vqueue, vipcpipelinesink, NULL));

  return pipeline;
}

static GstElement *
create_test_source (gboolean live, int fdina, int fdouta, int fdinv, int fdoutv,
    gboolean audio, gboolean video, gboolean Long)
{
  GstElement *pipeline, *audiotestsrc, *aipcpipelinesink;
  GstElement *videotestsrc, *vipcpipelinesink;
  int L = Long ? 2 : 1;

  pipeline = create_pipeline ("pipeline");

  if (audio) {
    audiotestsrc = gst_element_factory_make ("audiotestsrc", "audiotestsrc");
    g_object_set (audiotestsrc, "is-live", live, "num-buffers",
        live ? 270 * L : 600, NULL);
    aipcpipelinesink = gst_element_factory_make ("ipcpipelinesink",
        "aipcpipelinesink");
    add_weak_ref (aipcpipelinesink);
    g_object_set (aipcpipelinesink, "fdin", fdina, "fdout", fdouta, NULL);
    gst_bin_add_many (GST_BIN (pipeline), audiotestsrc, aipcpipelinesink, NULL);
    FAIL_UNLESS (gst_element_link_many (audiotestsrc, aipcpipelinesink, NULL));
  }

  if (video) {
    videotestsrc = gst_element_factory_make ("videotestsrc", "videotestsrc");
    g_object_set (videotestsrc, "is-live", live, "num-buffers",
        live ? 190 * L : 600, NULL);
    vipcpipelinesink =
        gst_element_factory_make ("ipcpipelinesink", "vipcpipelinesink");
    add_weak_ref (vipcpipelinesink);
    g_object_set (vipcpipelinesink, "fdin", fdinv, "fdout", fdoutv, NULL);
    gst_bin_add_many (GST_BIN (pipeline), videotestsrc, vipcpipelinesink, NULL);
    FAIL_UNLESS (gst_element_link_many (videotestsrc, vipcpipelinesink, NULL));
  }

  return pipeline;
}

static GstElement *
create_source (TestFeatures features, int fdina, int fdouta, int fdinv,
    int fdoutv, test_data * td)
{
  GstElement *pipeline = NULL;
  gboolean live = ! !(features & TEST_FEATURE_LIVE);
  gboolean longdur = ! !(features & TEST_FEATURE_LONG_DURATION);
  gboolean has_video = ! !(features & TEST_FEATURE_HAS_VIDEO);

  if (features & TEST_FEATURE_TEST_SOURCE) {

    pipeline = create_test_source (live, fdina, fdouta, fdinv, fdoutv, TRUE,
        has_video, longdur);
  } else if (features & TEST_FEATURE_WAV_SOURCE) {
    pipeline = create_wavparse_source_loc ("../../tests/files/sine.wav", fdina,
        fdouta);
  } else if (features & TEST_FEATURE_MPEGTS_SOURCE) {
    pipeline = create_mpegts_source_loc ("../../tests/files/test.ts", fdina,
        fdouta, fdinv, fdoutv);
  } else {
    g_assert_not_reached ();
  }

  td->two_streams = has_video;
  td->p = pipeline;

  if (pipeline)
    gst_bus_add_watch (GST_ELEMENT_BUS (pipeline), master_bus_msg, td);

  return pipeline;
}

/* sink construction */

static GstElement *
create_sink (TestFeatures features, GstElement ** slave_pipeline,
    int fdin, int fdout, const char *filter_caps)
{
  GstElement *ipcpipelinesrc, *fakesink, *identity, *capsfilter, *endpoint;
  GstCaps *caps;

  if (!*slave_pipeline)
    *slave_pipeline = create_pipeline ("ipcslavepipeline");
  else
    gst_object_ref (*slave_pipeline);
  ipcpipelinesrc = gst_element_factory_make ("ipcpipelinesrc", NULL);
  add_weak_ref (ipcpipelinesrc);
  g_object_set (ipcpipelinesrc, "fdin", fdin, "fdout", fdout, NULL);
  fakesink = gst_element_factory_make ("fakesink", NULL);
  g_object_set (fakesink, "sync", !(features & TEST_FEATURE_ASYNC_SINK), NULL);
  gst_bin_add_many (GST_BIN (*slave_pipeline), ipcpipelinesrc, fakesink, NULL);
  endpoint = ipcpipelinesrc;

  if (features & TEST_FEATURE_ERROR_SINK &&
      !g_strcmp0 (filter_caps, "audio/x-raw")) {
    identity = gst_element_factory_make ("identity", "error-element");
    g_object_set (identity, "error-after", 5, NULL);
    gst_bin_add (GST_BIN (*slave_pipeline), identity);
    FAIL_UNLESS (gst_element_link_many (endpoint, identity, NULL));
    endpoint = identity;
  }

  if ((features & TEST_FEATURE_FILTER_SINK_CAPS) && filter_caps) {
    capsfilter = gst_element_factory_make ("capsfilter", NULL);
    caps = gst_caps_from_string (filter_caps);
    FAIL_UNLESS (caps);
    g_object_set (capsfilter, "caps", caps, NULL);
    gst_caps_unref (caps);
    gst_bin_add (GST_BIN (*slave_pipeline), capsfilter);
    FAIL_UNLESS (gst_element_link_many (endpoint, capsfilter, NULL));
    endpoint = capsfilter;
  }
  FAIL_UNLESS (gst_element_link_many (endpoint, fakesink, NULL));

  return *slave_pipeline;
}

static void
ensure_sink_setup (GstElement * sink, void (*setup_sink) (GstElement *, void *),
    gpointer user_data)
{
  static GQuark setup_done = 0;
  test_data *td = user_data;

  if (!setup_done)
    setup_done = g_quark_from_static_string ("setup_done");

  if (sink)
    td->p = sink;

  if (sink && setup_sink && !g_object_get_qdata (G_OBJECT (sink), setup_done)) {
    g_object_set_qdata (G_OBJECT (sink), setup_done, GINT_TO_POINTER (1));
    setup_sink (sink, user_data);
  }
}

/* GstCheck multi-process setup helpers */

static void
on_child_exit (int signal)
{
  int status = 0;
  if (waitpid (-1, &status, 0) > 0 && status) {
    FAIL ();
    exit (status);
  } else {
    child_dead = TRUE;
  }
}

static void
die_on_child_death (void)
{
  struct sigaction sa;

  memset (&sa, 0, sizeof (sa));
  sa.sa_handler = on_child_exit;
  sigaction (SIGCHLD, &sa, NULL);
}

static void
wait_for_recovery (void)
{
  int value;

  FAIL_UNLESS (ctlsock[1]);
  FAIL_UNLESS (read (ctlsock[1], &value, sizeof (int)) == sizeof (int));
  FAIL_UNLESS (value == MSG_START);
}

static void
ack_recovery (void)
{
  int value = MSG_ACK;
  FAIL_UNLESS (ctlsock[1]);
  FAIL_UNLESS (write (ctlsock[1], &value, sizeof (int)) == sizeof (int));
}

static void
recreate_crashed_slave_process (void)
{
  int value = MSG_START;
  /* We don't recreate, because there seems to be some subtle issues
     with forking after gst has started running. So we create a new
     recovery process at start, and wake it up after the current
     slave dies, so it can take its place. It's a bit hacky, but it
     works. The spare process waits for SIGUSR2 to setup a replacement
     pipeline and connect to the master. */
  FAIL_UNLESS (recovery_pid);
  FAIL_UNLESS (ctlsock[0]);
  FAIL_UNLESS (write (ctlsock[0], &value, sizeof (int)) == sizeof (int));
  FAIL_UNLESS (read (ctlsock[0], &value, sizeof (int)) == sizeof (int));
  FAIL_UNLESS (value == MSG_ACK);
}

static gboolean
crash (gpointer user_data)
{
  _exit (0);
}

static gboolean
unwind (gpointer user_data)
{
  g_main_loop_quit (loop);
  return FALSE;
}

static void
on_unwind (int signal)
{
  g_idle_add (unwind, NULL);
}

static void
listen_for_unwind (void)
{
  struct sigaction sa;

  memset (&sa, 0, sizeof (sa));
  sa.sa_handler = on_unwind;
  sigaction (SIGUSR1, &sa, NULL);
}

static void
stop_listening_for_unwind (void)
{
  struct sigaction sa;

  memset (&sa, 0, sizeof (sa));
  sa.sa_handler = SIG_DFL;
  sigaction (SIGUSR1, &sa, NULL);
}

#define TEST_BASE(...) test_base(__FUNCTION__,##__VA_ARGS__)

/*
 * This is the main function driving the tests. All tests configure it
 * by way of all the function pointers it takes as arguments, which have
 * self-explanatory names.
 * Most tests are run over a number of different pipelines with the same
 * configuration (eg, a wavparse based pipeline, a live pipeline with
 * test audio/video, etc). Those pipelines that have more than one sink
 * (eg, MPEG-TS source demuxing audio and video) have a version with a
 * single slave pipeline and process, and a version with the audio and
 * video sinks in two different processes, each with its slave pipeline.
 * The master and slave crash tests are also run via this function, and
 * have specific code (grep for recovery).
 * There is a fair amount of hairy stuff to do with letting the main
 * check process when a subprocess has failed. Best not to look at it
 * and let it do its thing.
 * To add new tests, duplicate a set of tests, eg the *_end_of_stream
 * ones, and s/_end_of_stream/new_test_name/g. Then do the same for
 * the functions they pass as parameters to test_base. Typically, the
 * source creation sets a message hook to catch things like async-done
 * messages. Sink creation typically adds a probe to check that events,
 * buffers, etc, come through as expected. The two success functions
 * check all went well for the source and sink. Note that since all of
 * these functions take the same user data structure, and the process
 * will fork, writing something from one process will not be reflected
 * in the other, so there is usually a subset of data relevant to the
 * source, and another to the sink. But some have data relevant to both,
 * it depends on the test and what you are doing.
 * New tests do not have to use this framework, it just avoids spending
 * more time and effort on multi process handling.
 */
static void
test_base (const char *name, TestFeatures features,
    void (*run_source) (GstElement *, void *),
    void (*setup_sink) (GstElement *, void *),
    void (*check_success_source) (void *),
    void (*check_success_sink) (void *),
    gpointer input_data, gpointer master_data, gpointer slave_data)
{
  GstElement *source = NULL, *asink = NULL, *vsink = NULL;
  GstElement *slave_pipeline = NULL;
  GstStateChangeReturn ret;
  gboolean c_src, c_sink;
  pid_t pid = 0;
  unsigned char x;
  int master_recovery_pid_comm[2] = { -1, -1 };
  test_data td = { input_data, master_data, slave_data, features, FALSE, NULL,
    NULL, GST_STATE_NULL, 0
  };

  g_print ("Testing: %s\n", name);

  weak_refs = NULL;

  FAIL_IF (pipe2 (pipesfa, O_NONBLOCK) < 0);
  FAIL_IF (pipe2 (pipesba, O_NONBLOCK) < 0);
  FAIL_IF (pipe2 (pipesfv, O_NONBLOCK) < 0);
  FAIL_IF (pipe2 (pipesbv, O_NONBLOCK) < 0);
  FAIL_IF (socketpair (PF_UNIX, SOCK_STREAM, 0, ctlsock) < 0);

  FAIL_IF (pipesfa[0] < 0);
  FAIL_IF (pipesfa[1] < 0);
  FAIL_IF (pipesba[0] < 0);
  FAIL_IF (pipesba[1] < 0);
  FAIL_IF (pipesfv[0] < 0);
  FAIL_IF (pipesfv[1] < 0);
  FAIL_IF (pipesbv[0] < 0);
  FAIL_IF (pipesbv[1] < 0);

  gst_debug_remove_log_function (gst_debug_log_default);

  listen_for_unwind ();
  child_dead = FALSE;

  if (features & TEST_FEATURE_RECOVERY_MASTER_PROCESS) {
    /* the other master will let us know its child's PID so we can unwind
       it when we're finished */
    FAIL_IF (pipe2 (master_recovery_pid_comm, O_NONBLOCK) < 0);

    recovery_pid = fork ();
    if (recovery_pid > 0) {
      /* we're the main process that libcheck waits for */
      die_on_child_death ();
      while (!child_dead)
        g_usleep (1000);
      /* leave some time for the slave to timeout (1 second), record error, etc */
      g_usleep (1500 * 1000);

      /* Discard anything that was sent to the previous process when it died */
      while (read (pipesba[0], &x, 1) == 1);

      FAIL_UNLESS (read (master_recovery_pid_comm[0], &pid,
              sizeof (pid)) == sizeof (pid));

      setup_log ("gstsrc.log", TRUE);
      source = create_source (features, pipesba[0], pipesfa[1], pipesbv[0],
          pipesfv[1], &td);
      FAIL_UNLESS (source);
      if (run_source)
        run_source (source, &td);
      goto setup_done;
    }
  }

  if (features & TEST_FEATURE_RECOVERY_SLAVE_PROCESS) {
    recovery_pid = fork ();
    if (!recovery_pid) {
      wait_for_recovery ();

      /* Discard anything that was sent to the previous process when it died */
      while (read (pipesfa[0], &x, 1) == 1);

      setup_log ("gstasink.log", TRUE);
      asink = create_sink (features, &slave_pipeline, pipesfa[0], pipesba[1],
          "audio/x-raw");
      FAIL_UNLESS (asink);
      ensure_sink_setup (asink, setup_sink, &td);
      ack_recovery ();
      goto setup_done;
    }
  }

  pid = fork ();
  FAIL_IF (pid < 0);
  if (pid) {
    if (features & TEST_FEATURE_RECOVERY_MASTER_PROCESS) {
      FAIL_UNLESS (write (master_recovery_pid_comm[1], &pid,
              sizeof (pid)) == sizeof (pid));
    }
    die_on_child_death ();
    if (features & TEST_FEATURE_SPLIT_SINKS) {
      pid = fork ();
      FAIL_IF (pid < 0);
      if (pid) {
        die_on_child_death ();
      }
      c_src = ! !pid;
      c_sink = !pid;
    } else {
      c_src = TRUE;
      c_sink = FALSE;
    }
    if (c_src) {
      setup_log ("gstsrc.log", FALSE);
      source = create_source (features, pipesba[0], pipesfa[1], pipesbv[0],
          pipesfv[1], &td);
      FAIL_UNLESS (source);
      run_source (source, &td);
    }
    if (c_sink) {
      setup_log ("gstasink.log", FALSE);
      asink = create_sink (features, &slave_pipeline, pipesfa[0], pipesba[1],
          "audio/x-raw");
      FAIL_UNLESS (asink);
    }
  } else {
    td.two_streams = (features & TEST_FEATURE_HAS_VIDEO) &&
        !(features & TEST_FEATURE_SPLIT_SINKS);

    if (features & TEST_FEATURE_HAS_VIDEO) {
      setup_log ("gstvsink.log", FALSE);
      vsink = create_sink (features, &slave_pipeline, pipesfv[0], pipesbv[1],
          "video/x-raw");
      FAIL_UNLESS (vsink);
    }
    if (!(features & TEST_FEATURE_SPLIT_SINKS)) {
      setup_log ("gstasink.log", FALSE);
      asink = create_sink (features, &slave_pipeline, pipesfa[0], pipesba[1],
          "audio/x-raw");
      FAIL_UNLESS (asink);
    }
  }

setup_done:
  ensure_sink_setup (asink, setup_sink, &td);
  ensure_sink_setup (vsink, setup_sink, &td);

  loop = g_main_loop_new (NULL, FALSE);
  g_main_loop_run (loop);

  /* tell the child process to unwind too */
  stop_listening_for_unwind ();

  if (source) {
    ret = gst_element_set_state (source, GST_STATE_NULL);
    FAIL_UNLESS (ret == GST_STATE_CHANGE_SUCCESS
        || ret == GST_STATE_CHANGE_ASYNC);
  }

  if (pid)
    kill (pid, SIGUSR1);

  g_main_loop_unref (loop);

  if (source) {
    cleanup_bus (source);
    if (check_success_source)
      check_success_source (&td);
  } else {
    if (asink)
      cleanup_bus (asink);
    if (vsink)
      cleanup_bus (vsink);
    if (check_success_sink)
      check_success_sink (&td);
  }

  disconnect_ipcpipeline_elements ();

  close (pipesfa[0]);
  close (pipesfa[1]);
  close (pipesba[0]);
  close (pipesba[1]);
  close (pipesfv[0]);
  close (pipesfv[1]);
  close (pipesbv[0]);
  close (pipesbv[1]);

  /* If we have a child, we must now wait for it to be finished.
     We can't just waitpid, because this child might be still doing
     its shutdown, and might assert, and the die_on_child_death
     function will exit with the right exit code if so. So we wait
     for the child_dead boolean to be set, which die_on_child_death
     sets if the child dies normally. */
  if (pid) {
    while (!child_dead)
      g_usleep (1000);
  }

  if (source) {
    FAIL_UNLESS_EQUALS_INT (GST_OBJECT_REFCOUNT_VALUE (source), 1);
    gst_object_unref (source);
  }
  /* asink and vsink may be the same object, so refcount is not sure to be 1 */
  if (asink)
    gst_object_unref (asink);
  if (vsink)
    gst_object_unref (vsink);

  /* cleanup tasks a bit earlier to make sure all weak refs are gone */
  gst_task_cleanup_all ();

  /* all ipcpipeline elements we created should now be destroyed */
  if (weak_refs) {
#if 1
    /* to make it easier to see what leaks */
    GList *l;
    for (l = weak_refs; l; l = l->next) {
      g_print ("%s has %u refs\n", GST_ELEMENT_NAME (l->data),
          GST_OBJECT_REFCOUNT_VALUE (l->data));
    }
#endif
    FAIL_UNLESS (0);
  }
}

/**** play-pause test ****/

typedef struct
{
  gboolean got_state_changed_to_playing[2];
  gboolean got_state_changed_to_paused;
} play_pause_master_data;

#define PLAY_PAUSE_MASTER_DATA_INIT { { FALSE, FALSE }, FALSE }

typedef struct
{
  gboolean got_caps[2];
  gboolean got_segment[2];
  gboolean got_buffer[2];
} play_pause_slave_data;

#define PLAY_PAUSE_SLAVE_DATA_INIT \
  { { FALSE, FALSE }, { FALSE, FALSE }, { FALSE, FALSE } }

static gboolean
idlenull (gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  ret = gst_element_set_state (td->p, GST_STATE_NULL);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_SUCCESS);
  gst_object_unref (td->p);
  g_main_loop_quit (loop);
  return G_SOURCE_REMOVE;
}

static gboolean idleplay (gpointer user_data);
static gboolean
idlepause (gpointer user_data)
{
  test_data *td = user_data;
  play_pause_master_data *d = td->md;
  GstStateChangeReturn ret;

  ret = gst_element_set_state (td->p, GST_STATE_PAUSED);
  FAIL_IF (ret == GST_STATE_CHANGE_FAILURE);
  if (ret == GST_STATE_CHANGE_SUCCESS || ret == GST_STATE_CHANGE_NO_PREROLL) {
    /* if the state change is not async, we won't get an aync-done, but
       this is expected, so set the flag here */
    d->got_state_changed_to_paused = TRUE;
    td->state_target = GST_STATE_PLAYING;
    g_timeout_add (STEP_AT, idleplay, user_data);
    return G_SOURCE_REMOVE;
  }
  gst_object_unref (td->p);
  return G_SOURCE_REMOVE;
}

static gboolean
idleplay (gpointer user_data)
{
  test_data *td = user_data;
  play_pause_master_data *d = td->md;
  GstStateChangeReturn ret;

  ret = gst_element_set_state (td->p, GST_STATE_PLAYING);
  FAIL_IF (ret == GST_STATE_CHANGE_FAILURE);
  if (ret == GST_STATE_CHANGE_SUCCESS || ret == GST_STATE_CHANGE_NO_PREROLL) {
    /* if the state change is not async, we won't get an aync-done, but
       this is expected, so set the flag here */
    d->got_state_changed_to_playing[1] = TRUE;
    td->state_target = GST_STATE_NULL;
    g_timeout_add (STEP_AT, idlenull, user_data);
    return G_SOURCE_REMOVE;
  }
  gst_object_unref (td->p);
  return G_SOURCE_REMOVE;
}

static void
play_pause_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  play_pause_master_data *d = td->md;
  GstStateChangeReturn ret;

  if (d->got_state_changed_to_paused) {
    d->got_state_changed_to_playing[1] = TRUE;
    td->state_target = GST_STATE_NULL;
    ret = gst_element_set_state (td->p, GST_STATE_NULL);
    FAIL_UNLESS (ret == GST_STATE_CHANGE_SUCCESS);
    g_main_loop_quit (loop);
  } else if (d->got_state_changed_to_playing[0]) {
    d->got_state_changed_to_paused = TRUE;
    td->state_target = GST_STATE_PLAYING;
    gst_object_ref (td->p);
    g_timeout_add (STEP_AT, (GSourceFunc) idleplay, td);
  } else {
    d->got_state_changed_to_playing[0] = TRUE;
    td->state_target = GST_STATE_PAUSED;
    gst_object_ref (td->p);
    g_timeout_add (STEP_AT, (GSourceFunc) idlepause, td);
  }
}

static void
play_pause_source (GstElement * source, void *user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = play_pause_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static GstPadProbeReturn
play_pause_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  play_pause_slave_data *d = td->sd;
  GstCaps *caps;

  if (GST_IS_BUFFER (info->data)) {
    d->got_buffer[pad2idx (pad, td->two_streams)] = TRUE;
  } else if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_CAPS) {
      gst_event_parse_caps (info->data, &caps);
      d->got_caps[caps2idx (caps, td->two_streams)] = TRUE;
    } else if (GST_EVENT_TYPE (info->data) == GST_EVENT_SEGMENT) {
      d->got_segment[pad2idx (pad, td->two_streams)] = TRUE;
    }
  }

  return GST_PAD_PROBE_OK;
}

static void
hook_play_pause_probe (const GValue * v, gpointer user_data)
{
  hook_probe (v, play_pause_probe, user_data);
}

static void
setup_sink_play_pause (GstElement * sink, void *user_data)
{
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (sink));
  while (gst_iterator_foreach (it, hook_play_pause_probe, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);
}

static void
check_success_source_play_pause (void *user_data)
{
  test_data *td = user_data;
  play_pause_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing[0]);
  FAIL_UNLESS (d->got_state_changed_to_playing[1]);
  FAIL_UNLESS (d->got_state_changed_to_paused);
}

static void
check_success_sink_play_pause (void *user_data)
{
  test_data *td = user_data;
  play_pause_slave_data *d = td->sd;
  int idx;

  for (idx = 0; idx < (td->two_streams ? 2 : 1); idx++) {
    FAIL_UNLESS (d->got_caps[idx]);
    FAIL_UNLESS (d->got_segment[idx]);
    FAIL_UNLESS (d->got_buffer[idx]);
  }
}

GST_START_TEST (test_empty_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_TEST_SOURCE, play_pause_source, setup_sink_play_pause,
      check_success_source_play_pause, check_success_sink_play_pause, NULL, &md,
      &sd);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_WAV_SOURCE, play_pause_source, setup_sink_play_pause,
      check_success_source_play_pause, check_success_sink_play_pause, NULL, &md,
      &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE, play_pause_source,
      setup_sink_play_pause, check_success_source_play_pause,
      check_success_sink_play_pause, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      play_pause_source, setup_sink_play_pause, check_success_source_play_pause,
      check_success_sink_play_pause, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_a_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE, play_pause_source,
      setup_sink_play_pause, check_success_source_play_pause,
      check_success_sink_play_pause, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE, play_pause_source,
      setup_sink_play_pause, check_success_source_play_pause,
      check_success_sink_play_pause, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_play_pause)
{
  play_pause_master_data md = PLAY_PAUSE_MASTER_DATA_INIT;
  play_pause_slave_data sd = PLAY_PAUSE_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      play_pause_source, setup_sink_play_pause, check_success_source_play_pause,
      check_success_sink_play_pause, NULL, &md, &sd);
}

GST_END_TEST;

/**** flushing seek test ****/

typedef struct
{
  gboolean segment_seek;
  gboolean pause;
} flushing_seek_input_data;

#define FLUSHING_SEEK_INPUT_DATA_INIT { FALSE, FALSE }
#define FLUSHING_SEEK_INPUT_DATA_INIT_PAUSED { FALSE, TRUE }
#define FLUSHING_SEEK_INPUT_DATA_INIT_SEGMENT_SEEK { TRUE, FALSE }

typedef struct
{
  gboolean got_state_changed_to_playing;
  gboolean got_segment_done;
  gboolean seek_sent;
} flushing_seek_master_data;

#define FLUSHING_SEEK_MASTER_DATA_INIT { FALSE, FALSE, FALSE }

typedef struct
{
  GstClockTime first_ts[2];
  gboolean got_caps[2];
  gboolean got_buffer_before_seek[2];
  gboolean got_buffer_after_seek[2];
  gboolean first_buffer_after_seek_has_timestamp_0[2];
  gboolean got_segment_after_seek[2];
  gboolean got_flush_start[2];
  gboolean got_flush_stop[2];
} flushing_seek_slave_data;

#define FLUSHING_SEEK_SLAVE_DATA_INIT { { 0, 0 }, }

static gboolean
send_flushing_seek (gpointer user_data)
{
  test_data *td = user_data;
  const flushing_seek_input_data *i = td->id;
  flushing_seek_master_data *d = td->md;
  GstEvent *seek_event;

  if (i->segment_seek) {
    GST_INFO_OBJECT (td->p, "Sending segment seek");
    seek_event =
        gst_event_new_seek (1.0, GST_FORMAT_TIME,
        GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_FLUSH, GST_SEEK_TYPE_SET, 0,
        GST_SEEK_TYPE_SET, 1 * GST_SECOND);
    FAIL_UNLESS (gst_element_send_event (td->p, seek_event));
  } else {
    GST_INFO_OBJECT (td->p, "Sending flushing seek");
    gst_element_seek_simple (td->p, GST_FORMAT_TIME, GST_SEEK_FLAG_FLUSH, 0);
    g_timeout_add (STEP_AT, (GSourceFunc) stop_pipeline,
        gst_object_ref (td->p));
  }
  d->seek_sent = TRUE;
  return G_SOURCE_REMOVE;
}

static gboolean
pause_before_seek (gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  ret = gst_element_set_state (td->p, GST_STATE_PAUSED);
  FAIL_IF (ret == GST_STATE_CHANGE_FAILURE);

  return G_SOURCE_REMOVE;
}

static gboolean
flushing_seek_bus_msg (GstBus * bus, GstMessage * message, gpointer user_data)
{
  test_data *td = user_data;
  flushing_seek_master_data *d = td->md;

  if (GST_IS_PIPELINE (GST_MESSAGE_SRC (message))) {
    if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_SEGMENT_DONE) {
      d->got_segment_done = TRUE;
      g_timeout_add (STEP_AT, (GSourceFunc) stop_pipeline,
          gst_object_ref (td->p));
    }
  }
  return master_bus_msg (bus, message, user_data);
}

static void
flushing_seek_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  const flushing_seek_input_data *i = td->id;
  flushing_seek_master_data *d = td->md;

  if (!d->got_state_changed_to_playing) {
    d->got_state_changed_to_playing = TRUE;
    if (i->pause)
      g_timeout_add (PAUSE_AT, (GSourceFunc) pause_before_seek, td);
    g_timeout_add (SEEK_AT, (GSourceFunc) send_flushing_seek, td);
  }
}

static void
flushing_seek_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  /* we're on the source, there's already the basic master_bus_msg watch,
     and gst doesn't want more than one watch, so we remove the watch and
     call it directly when done in the new watch */
  gst_bus_remove_watch (GST_ELEMENT_BUS (source));
  gst_bus_add_watch (GST_ELEMENT_BUS (source), flushing_seek_bus_msg,
      user_data);
  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = flushing_seek_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static GstPadProbeReturn
flushing_seek_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  flushing_seek_slave_data *d = td->sd;
  GstClockTime ts;
  int idx;
  GstCaps *caps;

  if (GST_IS_BUFFER (info->data)) {
    idx = pad2idx (pad, td->two_streams);
    if (d->got_flush_stop[idx]) {
      if (!d->got_buffer_after_seek[idx]) {
        ts = GST_BUFFER_TIMESTAMP (info->data);
        d->first_buffer_after_seek_has_timestamp_0[idx] =
            (ts < d->first_ts[idx] + 10 * GST_MSECOND);
        d->got_buffer_after_seek[idx] = TRUE;
      }
    } else if (!d->got_buffer_before_seek[idx]) {
      d->got_buffer_before_seek[idx] = TRUE;
      d->first_ts[idx] = GST_BUFFER_TIMESTAMP (info->data);
    }
  } else if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_CAPS) {
      gst_event_parse_caps (info->data, &caps);
      if (are_caps_audio (caps) || are_caps_video (caps)) {
        idx = caps2idx (caps, td->two_streams);
        d->got_caps[idx] = TRUE;
      }
    } else if (GST_EVENT_TYPE (info->data) == GST_EVENT_SEGMENT) {
      /* from the sink pipeline, we don't know whether the master issued a seek,
         as the seek_sent memory location isn't directly accesible to us, so we
         look for a segment after a buffer to mean a seek was sent */
      idx = pad2idx (pad, td->two_streams);
      if (d->got_buffer_before_seek[idx])
        d->got_segment_after_seek[idx] = TRUE;
    } else if (GST_EVENT_TYPE (info->data) == GST_EVENT_FLUSH_START) {
      idx = pad2idx (pad, td->two_streams);
      d->got_flush_start[idx] = TRUE;
    } else if (GST_EVENT_TYPE (info->data) == GST_EVENT_FLUSH_STOP) {
      idx = pad2idx (pad, td->two_streams);
      if (d->got_buffer_before_seek[idx])
        d->got_flush_stop[idx] = TRUE;
    }
  }

  return GST_PAD_PROBE_OK;
}

static void
hook_flushing_seek_probe (const GValue * v, gpointer user_data)
{
  hook_probe (v, flushing_seek_probe, user_data);
}

static void
setup_sink_flushing_seek (GstElement * sink, gpointer user_data)
{
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (sink));
  while (gst_iterator_foreach (it, hook_flushing_seek_probe, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);
}

static void
check_success_source_flushing_seek (gpointer user_data)
{
  test_data *td = user_data;
  const flushing_seek_input_data *i = td->id;
  flushing_seek_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing);
  FAIL_UNLESS (d->seek_sent);
  FAIL_UNLESS (d->got_segment_done == i->segment_seek);
}

static void
check_success_sink_flushing_seek (gpointer user_data)
{
  test_data *td = user_data;
  flushing_seek_slave_data *d = td->sd;
  gint idx;

  for (idx = 0; idx < (td->two_streams ? 2 : 1); idx++) {
    FAIL_UNLESS (d->got_caps[idx]);
    FAIL_UNLESS (d->got_buffer_before_seek[idx]);
    FAIL_UNLESS (d->got_buffer_after_seek[idx]);
    FAIL_UNLESS (d->got_segment_after_seek[idx]);
    FAIL_UNLESS (d->got_flush_start[idx]);
    FAIL_UNLESS (d->got_flush_stop[idx]);
    FAIL_UNLESS (d->first_buffer_after_seek_has_timestamp_0[idx]);
  }
}

GST_START_TEST (test_empty_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_TEST_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_WAV_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      flushing_seek_source, setup_sink_flushing_seek,
      check_success_source_flushing_seek, check_success_sink_flushing_seek, &id,
      &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_a_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_flushing_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      flushing_seek_source, setup_sink_flushing_seek,
      check_success_source_flushing_seek, check_success_sink_flushing_seek, &id,
      &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_empty_flushing_seek_in_pause)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_PAUSED;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_TEST_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_flushing_seek_in_pause)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_PAUSED;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_WAV_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_flushing_seek_in_pause)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_PAUSED;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_flushing_seek_in_pause)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_PAUSED;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      flushing_seek_source, setup_sink_flushing_seek,
      check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_empty_segment_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_SEGMENT_SEEK;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_TEST_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_segment_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_SEGMENT_SEEK;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_WAV_SOURCE, flushing_seek_source,
      setup_sink_flushing_seek, check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_a_segment_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_SEGMENT_SEEK;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE,
      flushing_seek_source, setup_sink_flushing_seek,
      check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_segment_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_SEGMENT_SEEK;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE,
      flushing_seek_source, setup_sink_flushing_seek,
      check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_segment_seek)
{
  flushing_seek_input_data id = FLUSHING_SEEK_INPUT_DATA_INIT_SEGMENT_SEEK;
  flushing_seek_master_data md = FLUSHING_SEEK_MASTER_DATA_INIT;
  flushing_seek_slave_data sd = FLUSHING_SEEK_SLAVE_DATA_INIT;

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      flushing_seek_source, setup_sink_flushing_seek,
      check_success_source_flushing_seek,
      check_success_sink_flushing_seek, &id, &md, &sd);
}

GST_END_TEST;

/**** seek stress test ****/

typedef struct
{
  gint n_flushing_seeks;
  gint n_paused_seeks;
  gint n_segment_seeks;
} seek_stress_input_data;

typedef struct
{
  gboolean got_state_changed_to_playing;
  gboolean got_eos;
  gboolean seek_sent;
  guint64 t0;
} seek_stress_master_data;

static gboolean
send_seek_stress (gpointer user_data)
{
  test_data *td = user_data;
  seek_stress_input_data *i = td->id;
  seek_stress_master_data *d = td->md;
  GstEvent *seek_event;
  unsigned int available, seekidx;
  GstClockTime t, base;

  /* Live streams don't like to be seeked too far away from the
     "current" time, since they're live, so always seek near the
     "real" time, so we still exercise seeking to another position
     but still land somewhere close enough to "live" position. */
  t = (g_get_monotonic_time () - d->t0) * 1000;
  base = t > GST_SECOND / 2 ? t - GST_SECOND / 2 : 0;
  t = base + g_random_int_range (0, GST_SECOND);

  /* pick a random seek type among the ones we have left */
  available = i->n_flushing_seeks + i->n_paused_seeks + i->n_segment_seeks;
  if (available == 0) {
    GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS (GST_BIN (td->p),
        GST_DEBUG_GRAPH_SHOW_ALL, "inter.test.toplaying");
    FAIL_UNLESS (gst_element_set_state (td->p,
            GST_STATE_PLAYING) != GST_STATE_CHANGE_FAILURE);
    g_timeout_add (STEP_AT, (GSourceFunc) stop_pipeline,
        gst_object_ref (td->p));
    gst_object_unref (td->p);
    return G_SOURCE_REMOVE;
  }

  seekidx = rand () % available;
  if (seekidx < i->n_flushing_seeks) {
    GST_INFO_OBJECT (td->p, "Sending flushing seek to %" GST_TIME_FORMAT,
        GST_TIME_ARGS (t));
    FAIL_UNLESS (gst_element_set_state (td->p,
            GST_STATE_PLAYING) != GST_STATE_CHANGE_FAILURE);
    FAIL_UNLESS (gst_element_seek_simple (td->p, GST_FORMAT_TIME,
            GST_SEEK_FLAG_FLUSH, t));
    --i->n_flushing_seeks;
    return G_SOURCE_CONTINUE;
  }
  seekidx -= i->n_flushing_seeks;

  if (seekidx < i->n_paused_seeks) {
    GST_INFO_OBJECT (td->p,
        "Sending flushing seek in paused to %" GST_TIME_FORMAT,
        GST_TIME_ARGS (t));
    FAIL_UNLESS (gst_element_set_state (td->p,
            GST_STATE_PAUSED) != GST_STATE_CHANGE_FAILURE);
    FAIL_UNLESS (gst_element_seek_simple (td->p, GST_FORMAT_TIME,
            GST_SEEK_FLAG_FLUSH, t));
    --i->n_paused_seeks;
    return G_SOURCE_CONTINUE;
  }
  seekidx -= i->n_paused_seeks;

  GST_INFO_OBJECT (td->p, "Sending segment seek to %" GST_TIME_FORMAT,
      GST_TIME_ARGS (t));
  seek_event =
      gst_event_new_seek (1.0, GST_FORMAT_TIME,
      GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_FLUSH, GST_SEEK_TYPE_SET, t,
      GST_SEEK_TYPE_SET, t + 5 * GST_SECOND);
  FAIL_UNLESS (gst_element_send_event (td->p, seek_event));
  --i->n_segment_seeks;
  return G_SOURCE_CONTINUE;
}

static gboolean
seek_stress_bus_msg (GstBus * bus, GstMessage * message, gpointer user_data)
{
  test_data *td = user_data;
  seek_stress_master_data *d = td->md;

  if (GST_IS_PIPELINE (GST_MESSAGE_SRC (message))) {
    if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_EOS ||
        GST_MESSAGE_TYPE (message) == GST_MESSAGE_SEGMENT_DONE) {
      d->got_eos = TRUE;
    }
  }
  return master_bus_msg (bus, message, user_data);
}

static void
seek_stress_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  seek_stress_master_data *d = td->md;

  if (!d->got_state_changed_to_playing) {
    d->got_state_changed_to_playing = TRUE;
    d->t0 = g_get_monotonic_time ();
    gst_object_ref (td->p);
    g_timeout_add (10, (GSourceFunc) send_seek_stress, td);
  }
}

static void
seek_stress_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  /* we're on the source, there's already the basic master_bus_msg watch,
     and gst doesn't want more than one watch, so we remove the watch and
     call it directly when done in the new watch */
  gst_bus_remove_watch (GST_ELEMENT_BUS (source));
  gst_bus_add_watch (GST_ELEMENT_BUS (source), seek_stress_bus_msg, user_data);
  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = seek_stress_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static void
check_success_source_seek_stress (gpointer user_data)
{
  test_data *td = user_data;
  seek_stress_input_data *i = td->id;
  seek_stress_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing);
  FAIL_UNLESS_EQUALS_INT (i->n_flushing_seeks, 0);
  FAIL_UNLESS_EQUALS_INT (i->n_paused_seeks, 0);
  FAIL_UNLESS_EQUALS_INT (i->n_segment_seeks, 0);
  FAIL_IF (d->got_eos);
}

GST_START_TEST (test_empty_seek_stress)
{
  seek_stress_input_data id = { 100, 100, 100 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_TEST_SOURCE, seek_stress_source, NULL,
      check_success_source_seek_stress, NULL, &id, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_seek_stress)
{
  seek_stress_input_data id = { 100, 100, 100 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_WAV_SOURCE, seek_stress_source, NULL,
      check_success_source_seek_stress, NULL, &id, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_seek_stress)
{
  seek_stress_input_data id = { 100, 100, 0 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE, seek_stress_source, NULL,
      check_success_source_seek_stress, NULL, &id, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_seek_stress)
{
  seek_stress_input_data id = { 100, 100, 0 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      seek_stress_source, NULL, check_success_source_seek_stress, NULL, &id,
      &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_live_a_seek_stress)
{
  seek_stress_input_data id = { 100, 0, 100 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE | TEST_FEATURE_LONG_DURATION,
      seek_stress_source, NULL, check_success_source_seek_stress, NULL, &id,
      &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_live_av_seek_stress)
{
  seek_stress_input_data id = { 100, 0, 100 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_LONG_DURATION,
      seek_stress_source, NULL, check_success_source_seek_stress, NULL, &id,
      &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_seek_stress)
{
  seek_stress_input_data id = { 100, 0, 100 };
  seek_stress_master_data md = { 0 };

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_LONG_DURATION |
      TEST_FEATURE_SPLIT_SINKS,
      seek_stress_source, NULL, check_success_source_seek_stress, NULL, &id,
      &md, NULL);
}

GST_END_TEST;

/**** upstream query test ****/

typedef struct
{
  GstClockTime expected_duration;

  /* In this test, the source does a position query (in the source pipeline
     process), and must check its return against the last buffer timestamp
     in the sink pipeline process. We open a pipe to let the sink send us
     the timestamps it receives so the source can make the comparison. */
  gint ts_pipes[2];
} upstream_query_input_data;

typedef struct
{
  gboolean got_state_changed_to_playing;
  gboolean got_correct_position;
  gboolean got_correct_duration;
  GstClockTime last_buffer_ts;
} upstream_query_master_data;

typedef struct
{
  gboolean got_caps[2];
  gboolean got_buffer[2];
  GstClockTime last_buffer_ts;
} upstream_query_slave_data;

static gboolean
send_upstream_queries (gpointer user_data)
{
  test_data *td = user_data;
  upstream_query_input_data *i = td->id;
  upstream_query_master_data *d = td->md;
  gint64 pos, dur, last;

  FAIL_UNLESS (gst_element_query_position (td->p, GST_FORMAT_TIME, &pos));

  /* read up the buffer ts sent by the sink process till the last one */
  while (read (i->ts_pipes[0], &last, sizeof (last)) == sizeof (last)) {
    /* timestamps may not be increasing because we are getting ts from
     * both the audio and video streams; the position query will report
     * the higher */
    if (last > d->last_buffer_ts)
      d->last_buffer_ts = last;
  }
  if (ABS ((gint64) (pos - d->last_buffer_ts)) <= CLOSE_ENOUGH_TO_ZERO)
    d->got_correct_position = TRUE;

  FAIL_UNLESS (gst_element_query_duration (td->p, GST_FORMAT_TIME, &dur));
  if (GST_CLOCK_TIME_IS_VALID (i->expected_duration)) {
    GstClockTimeDiff diff = GST_CLOCK_DIFF (dur, i->expected_duration);
    if (diff >= -CLOSE_ENOUGH_TO_ZERO && diff <= CLOSE_ENOUGH_TO_ZERO)
      d->got_correct_duration = TRUE;
  } else {
    if (!GST_CLOCK_TIME_IS_VALID (dur))
      d->got_correct_duration = TRUE;
  }

  g_timeout_add (STEP_AT, (GSourceFunc) stop_pipeline, td->p);
  return FALSE;
}

static void
upstream_query_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  upstream_query_master_data *d = td->md;

  if (!d->got_state_changed_to_playing) {
    d->got_state_changed_to_playing = TRUE;
    gst_object_ref (td->p);
    g_timeout_add (QUERY_AT, (GSourceFunc) send_upstream_queries, td);
  }
}

static void
upstream_query_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  td->state_changed_cb = upstream_query_on_state_changed;
  td->state_target = GST_STATE_PLAYING;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static GstPadProbeReturn
upstream_query_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  upstream_query_input_data *i = td->id;
  upstream_query_slave_data *d = td->sd;
  GstCaps *caps;

  if (GST_IS_BUFFER (info->data)) {
    d->got_buffer[pad2idx (pad, td->two_streams)] = TRUE;
    if (GST_BUFFER_TIMESTAMP_IS_VALID (info->data)) {
      d->last_buffer_ts = GST_BUFFER_TIMESTAMP (info->data);
      FAIL_UNLESS (write (i->ts_pipes[1], &d->last_buffer_ts,
              sizeof (d->last_buffer_ts)) == sizeof (d->last_buffer_ts));
    }
  } else if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_CAPS) {
      gst_event_parse_caps (info->data, &caps);
      d->got_caps[caps2idx (caps, td->two_streams)] = TRUE;
    }
  }

  return GST_PAD_PROBE_OK;
}

static void
hook_upstream_query_probe (const GValue * v, gpointer user_data)
{
  hook_probe (v, upstream_query_probe, user_data);
}

static void
setup_sink_upstream_query (GstElement * sink, gpointer user_data)
{
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (sink));
  while (gst_iterator_foreach (it, hook_upstream_query_probe, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);
}

static void
check_success_source_upstream_query (gpointer user_data)
{
  test_data *td = user_data;
  upstream_query_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing);
  FAIL_UNLESS (d->got_correct_position);
  FAIL_UNLESS (d->got_correct_duration);
}

static void
check_success_sink_upstream_query (gpointer user_data)
{
  test_data *td = user_data;
  upstream_query_slave_data *d = td->sd;
  int idx;

  for (idx = 0; idx < (td->two_streams ? 2 : 1); ++idx) {
    FAIL_UNLESS (d->got_caps[idx]);
    FAIL_UNLESS (d->got_buffer[idx]);
  }
}

GST_START_TEST (test_empty_upstream_query)
{
  upstream_query_input_data id = { GST_CLOCK_TIME_NONE, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_TEST_SOURCE, upstream_query_source,
      setup_sink_upstream_query, check_success_source_upstream_query,
      check_success_sink_upstream_query, &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_upstream_query)
{
  upstream_query_input_data id = { WAV_SAMPLE_ROUGH_DURATION, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_WAV_SOURCE, upstream_query_source,
      setup_sink_upstream_query, check_success_source_upstream_query,
      check_success_sink_upstream_query, &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_upstream_query)
{
  upstream_query_input_data id = { MPEGTS_SAMPLE_ROUGH_DURATION, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE, upstream_query_source,
      setup_sink_upstream_query, check_success_source_upstream_query,
      check_success_sink_upstream_query, &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_upstream_query)
{
  upstream_query_input_data id = { MPEGTS_SAMPLE_ROUGH_DURATION, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      upstream_query_source, setup_sink_upstream_query,
      check_success_source_upstream_query, check_success_sink_upstream_query,
      &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

GST_START_TEST (test_live_a_upstream_query)
{
  upstream_query_input_data id = { GST_CLOCK_TIME_NONE, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE,
      upstream_query_source, setup_sink_upstream_query,
      check_success_source_upstream_query, check_success_sink_upstream_query,
      &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

GST_START_TEST (test_live_av_upstream_query)
{
  upstream_query_input_data id = { GST_CLOCK_TIME_NONE, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE,
      upstream_query_source, setup_sink_upstream_query,
      check_success_source_upstream_query, check_success_sink_upstream_query,
      &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_upstream_query)
{
  upstream_query_input_data id = { GST_CLOCK_TIME_NONE, };
  upstream_query_master_data md = { 0 };
  upstream_query_slave_data sd = { {0}
  };

  FAIL_UNLESS (pipe2 (id.ts_pipes, O_NONBLOCK) == 0);
  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      upstream_query_source, setup_sink_upstream_query,
      check_success_source_upstream_query, check_success_sink_upstream_query,
      &id, &md, &sd);
  close (id.ts_pipes[0]);
  close (id.ts_pipes[1]);
}

GST_END_TEST;

/**** message test ****/

typedef struct
{
  gboolean got_state_changed_to_playing;
  guint8 num_got_message;
  guint8 num_sent_message;
} message_master_data;

static void
send_ipcpipeline_test_message_event (const GValue * v, gpointer user_data)
{
  test_data *td = user_data;
  message_master_data *d = td->md;
  GstElement *element = g_value_get_object (v);
  GstMessage *msg;
  gboolean ret;

  d->num_sent_message++;

  msg = gst_message_new_element (GST_OBJECT (element),
      gst_structure_new_empty ("ipcpipeline-test"));
  ret = gst_element_send_event (element,
      gst_event_new_sink_message ("ipcpipeline-test", msg));
  FAIL_UNLESS (ret);
  gst_message_unref (msg);
}

static gboolean
send_sink_message (gpointer user_data)
{
  test_data *td = user_data;
  GstIterator *it;

  it = gst_bin_iterate_sources (GST_BIN (td->p));
  while (gst_iterator_foreach (it, send_ipcpipeline_test_message_event, td))
    gst_iterator_resync (it);
  gst_iterator_free (it);

  gst_object_unref (td->p);
  return G_SOURCE_REMOVE;
}

static gboolean
message_bus_msg (GstBus * bus, GstMessage * message, gpointer user_data)
{
  test_data *td = user_data;
  message_master_data *d = td->md;
  const GstStructure *structure;

  if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ELEMENT) {
    structure = gst_message_get_structure (message);
    FAIL_UNLESS (structure);
    if (gst_structure_has_name (structure, "ipcpipeline-test")) {
      d->num_got_message++;
      if (d->num_got_message == d->num_sent_message)
        g_main_loop_quit (loop);
    }
  }
  return master_bus_msg (bus, message, user_data);
}

static void
message_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  message_master_data *d = td->md;

  if (!d->got_state_changed_to_playing) {
    d->got_state_changed_to_playing = TRUE;
    gst_object_ref (td->p);
    g_timeout_add (MESSAGE_AT, (GSourceFunc) send_sink_message, td);
  }
}

static void
message_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  /* we're on the source, there's already the basic master_bus_msg watch,
     and gst doesn't want more than one watch, so we remove the watch and
     call it directly when done in the new watch */
  gst_bus_remove_watch (GST_ELEMENT_BUS (source));
  gst_bus_add_watch (GST_ELEMENT_BUS (source), message_bus_msg, user_data);
  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = message_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static void
check_success_source_message (gpointer user_data)
{
  test_data *td = user_data;
  message_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing);
  FAIL_UNLESS_EQUALS_INT (d->num_got_message, d->num_sent_message);
}

GST_START_TEST (test_empty_message)
{
  message_master_data md = { 0 };
  TEST_BASE (TEST_FEATURE_TEST_SOURCE, message_source, NULL,
      check_success_source_message, NULL, NULL, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_message)
{
  message_master_data md = { 0 };
  TEST_BASE (TEST_FEATURE_WAV_SOURCE, message_source, NULL,
      check_success_source_message, NULL, NULL, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_live_a_message)
{
  message_master_data md = { 0 };
  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE, message_source, NULL,
      check_success_source_message, NULL, NULL, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_live_av_message)
{
  message_master_data md = { 0 };
  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE, message_source, NULL,
      check_success_source_message, NULL, NULL, &md, NULL);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_message)
{
  message_master_data md = { 0 };
  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      message_source, NULL, check_success_source_message, NULL, NULL, &md,
      NULL);
}

GST_END_TEST;

/**** end of stream test ****/

typedef struct
{
  gboolean got_state_changed_to_playing;
} end_of_stream_master_data;

typedef struct
{
  gboolean got_buffer[2];
  gboolean got_eos[2];
} end_of_stream_slave_data;

static void
end_of_stream_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  end_of_stream_master_data *d = td->md;

  if (!d->got_state_changed_to_playing)
    d->got_state_changed_to_playing = TRUE;
}

static void
end_of_stream_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  td->state_changed_cb = end_of_stream_on_state_changed;
  td->state_target = GST_STATE_PLAYING;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static GstPadProbeReturn
end_of_stream_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  end_of_stream_slave_data *d = td->sd;

  if (GST_IS_BUFFER (info->data)) {
    d->got_buffer[pad2idx (pad, td->two_streams)] = TRUE;
  } else if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_EOS) {
      d->got_eos[pad2idx (pad, td->two_streams)] = TRUE;
    }
  }

  return GST_PAD_PROBE_OK;
}

static void
hook_end_of_stream_probe (const GValue * v, gpointer user_data)
{
  hook_probe (v, end_of_stream_probe, user_data);
}

static void
setup_sink_end_of_stream (GstElement * sink, gpointer user_data)
{
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (sink));
  while (gst_iterator_foreach (it, hook_end_of_stream_probe, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);
}

static void
check_success_source_end_of_stream (gpointer user_data)
{
  test_data *td = user_data;
  end_of_stream_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing);
}

static void
check_success_sink_end_of_stream (gpointer user_data)
{
  test_data *td = user_data;
  end_of_stream_slave_data *d = td->sd;
  int idx;

  for (idx = 0; idx < (td->two_streams ? 2 : 1); idx++) {
    FAIL_UNLESS (d->got_buffer[idx]);
    FAIL_UNLESS (d->got_eos[idx]);
  }
}

GST_START_TEST (test_empty_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_TEST_SOURCE | TEST_FEATURE_ASYNC_SINK,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_WAV_SOURCE | TEST_FEATURE_ASYNC_SINK,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_ASYNC_SINK,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS |
      TEST_FEATURE_ASYNC_SINK,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_a_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_end_of_stream)
{
  end_of_stream_master_data md = { 0 };
  end_of_stream_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      end_of_stream_source, setup_sink_end_of_stream,
      check_success_source_end_of_stream, check_success_sink_end_of_stream,
      NULL, &md, &sd);
}

GST_END_TEST;

/**** reverse playback test ****/

typedef struct
{
  gboolean got_state_changed_to_playing;
  gboolean seek_sent;
} reverse_playback_master_data;

typedef struct
{
  gboolean got_segment_with_negative_rate;
  gboolean got_buffer_after_segment_with_negative_rate;
  GstClockTime first_backward_buffer_timestamp;
  gboolean got_buffer_one_second_early;
} reverse_playback_slave_data;

static gboolean
play_backwards (gpointer user_data)
{
  test_data *td = user_data;
  reverse_playback_master_data *d = td->md;
  gint64 pos;
  gboolean ret;

  FAIL_UNLESS (gst_element_query_position (td->p, GST_FORMAT_TIME, &pos));

  ret =
      gst_element_seek (td->p, -0.5, GST_FORMAT_TIME, 0, GST_SEEK_TYPE_SET, 0,
      GST_SEEK_TYPE_SET, pos);
  FAIL_UNLESS (ret);
  d->seek_sent = TRUE;

  gst_object_unref (td->p);
  return G_SOURCE_REMOVE;
}

static void
reverse_playback_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  reverse_playback_master_data *d = td->md;

  if (!d->got_state_changed_to_playing) {
    d->got_state_changed_to_playing = TRUE;
    gst_object_ref (td->p);
    g_timeout_add (2000, (GSourceFunc) play_backwards, td);
  }
}

static void
reverse_playback_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = reverse_playback_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static GstPadProbeReturn
reverse_playback_probe (GstPad * pad, GstPadProbeInfo * info,
    gpointer user_data)
{
  test_data *td = user_data;
  reverse_playback_slave_data *d = td->sd;

  if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_SEGMENT) {
      const GstSegment *s;
      gst_event_parse_segment (GST_EVENT (info->data), &s);
      if (s->rate < 0)
        d->got_segment_with_negative_rate = TRUE;
    }
  } else if (GST_IS_BUFFER (info->data)) {
    GstClockTime ts = GST_BUFFER_TIMESTAMP (info->data);
    if (GST_CLOCK_TIME_IS_VALID (ts)) {
      if (d->got_segment_with_negative_rate) {
        if (d->got_buffer_after_segment_with_negative_rate) {
          /* We test for 1 second, not just earlier, to make sure we don't
             just see B frames, or whatever else */
          if (ts < d->first_backward_buffer_timestamp - GST_SECOND) {
            d->got_buffer_one_second_early = TRUE;
          }
        } else {
          d->got_buffer_after_segment_with_negative_rate = TRUE;
          d->first_backward_buffer_timestamp = ts;
        }
      }
    }
  }

  return GST_PAD_PROBE_OK;
}

static void
hook_reverse_playback_probe (const GValue * v, gpointer user_data)
{
  hook_probe (v, reverse_playback_probe, user_data);
}

static void
setup_sink_reverse_playback (GstElement * sink, gpointer user_data)
{
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (sink));
  while (gst_iterator_foreach (it, hook_reverse_playback_probe, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);
}

static void
check_success_source_reverse_playback (gpointer user_data)
{
  test_data *td = user_data;
  reverse_playback_master_data *d = td->md;

  FAIL_UNLESS (d->got_state_changed_to_playing);
  FAIL_UNLESS (d->seek_sent);
}

static void
check_success_sink_reverse_playback (gpointer user_data)
{
  test_data *td = user_data;
  reverse_playback_slave_data *d = td->sd;

  FAIL_UNLESS (d->got_segment_with_negative_rate);
  FAIL_UNLESS (d->got_buffer_after_segment_with_negative_rate);
  FAIL_UNLESS (GST_CLOCK_TIME_IS_VALID (d->first_backward_buffer_timestamp));
  FAIL_UNLESS (d->first_backward_buffer_timestamp >= GST_SECOND);
  FAIL_UNLESS (d->got_buffer_one_second_early);
}

GST_START_TEST (test_a_reverse_playback)
{
  reverse_playback_master_data md = { 0 };
  reverse_playback_slave_data sd = { 0 };

  TEST_BASE (TEST_FEATURE_TEST_SOURCE,
      reverse_playback_source, setup_sink_reverse_playback,
      check_success_source_reverse_playback,
      check_success_sink_reverse_playback, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_av_reverse_playback)
{
  reverse_playback_master_data md = { 0 };
  reverse_playback_slave_data sd = { 0 };

  TEST_BASE (TEST_FEATURE_TEST_SOURCE | TEST_FEATURE_HAS_VIDEO,
      reverse_playback_source, setup_sink_reverse_playback,
      check_success_source_reverse_playback,
      check_success_sink_reverse_playback, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_av_2_reverse_playback)
{
  reverse_playback_master_data md = { 0 };
  reverse_playback_slave_data sd = { 0 };

  TEST_BASE (TEST_FEATURE_TEST_SOURCE | TEST_FEATURE_HAS_VIDEO |
      TEST_FEATURE_SPLIT_SINKS,
      reverse_playback_source, setup_sink_reverse_playback,
      check_success_source_reverse_playback,
      check_success_sink_reverse_playback, NULL, &md, &sd);
}

GST_END_TEST;

/**** tags test ****/

enum
{
  TEST_TAG_EMPTY,
  TEST_TAG_TWO_TAGS,
  N_TEST_TAGS
};

typedef struct
{
  gboolean got_state_changed_to_playing;
  gboolean tags_sent[2][N_TEST_TAGS];
} tags_master_data;

typedef struct
{
  gboolean tags_received[N_TEST_TAGS];
} tags_slave_data;

static void
send_tags_on_pad (GstPad * pad, gpointer user_data)
{
  test_data *td = user_data;
  tags_master_data *d = td->md;
  GstEvent *e;
  gint idx;

  idx = pad2idx (pad, td->two_streams);

  e = gst_event_new_tag (gst_tag_list_new_empty ());
  FAIL_UNLESS (gst_pad_send_event (pad, e));
  d->tags_sent[idx][TEST_TAG_EMPTY] = TRUE;

  e = gst_event_new_tag (gst_tag_list_new (GST_TAG_TITLE, "title",
          GST_TAG_BITRATE, 56000, NULL));
  FAIL_UNLESS (gst_pad_send_event (pad, e));
  d->tags_sent[idx][TEST_TAG_TWO_TAGS] = TRUE;
}

static GstPadProbeReturn
tags_probe_source (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  tags_master_data *d = td->md;
  GstClockTime ts;

  if (GST_IS_BUFFER (info->data)) {
    ts = GST_BUFFER_TIMESTAMP (info->data);
    if (GST_CLOCK_TIME_IS_VALID (ts) && ts > STEP_AT * GST_MSECOND) {
      gint idx = pad2idx (pad, td->two_streams);
      if (!d->tags_sent[idx][0]) {
        GstPad *peer = gst_pad_get_peer (pad);
        FAIL_UNLESS (peer);
        send_tags_on_pad (peer, td);
        gst_object_unref (peer);
        EXCLUSIVE_CALL (td, g_timeout_add (STEP_AT, (GSourceFunc) stop_pipeline,
                gst_object_ref (td->p)));
      }
    }
  }

  return GST_PAD_PROBE_OK;
}

static void
hook_tags_probe_source (const GValue * v, gpointer user_data)
{
  hook_peer_probe_types (v, tags_probe_source,
      GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM, user_data);
}

static void
tags_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  tags_master_data *d = td->md;
  GstIterator *it;

  if (!d->got_state_changed_to_playing) {
    d->got_state_changed_to_playing = TRUE;

    it = gst_bin_iterate_sinks (GST_BIN (td->p));
    while (gst_iterator_foreach (it, hook_tags_probe_source, user_data))
      gst_iterator_resync (it);
    gst_iterator_free (it);
  }
}

static void
tags_source (GstElement * source, gpointer user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;

  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = tags_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static GstPadProbeReturn
tags_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  tags_slave_data *d = td->sd;
  guint funsigned;
  gchar *fstring = NULL;

  if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_TAG) {
      GstTagList *taglist = NULL;
      gst_event_parse_tag (GST_EVENT (info->data), &taglist);
      FAIL_UNLESS (taglist);
      if (gst_tag_list_is_empty (taglist)) {
        d->tags_received[TEST_TAG_EMPTY] = TRUE;
      } else if (gst_tag_list_get_string (taglist, GST_TAG_TITLE, &fstring)
          && !strcmp (fstring, "title")
          && gst_tag_list_get_uint (taglist, GST_TAG_BITRATE, &funsigned)
          && funsigned == 56000) {
        d->tags_received[TEST_TAG_TWO_TAGS] = TRUE;
      }
    }
  }
  g_free (fstring);

  return GST_PAD_PROBE_OK;
}

static void
hook_tags_probe (const GValue * v, gpointer user_data)
{
  hook_probe (v, tags_probe, user_data);
}

static void
setup_sink_tags (GstElement * sink, gpointer user_data)
{
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (sink));
  while (gst_iterator_foreach (it, hook_tags_probe, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);
}

static void
check_success_source_tags (gpointer user_data)
{
  test_data *td = user_data;
  tags_master_data *d = td->md;
  gint n;

  FAIL_UNLESS (d->got_state_changed_to_playing);
  for (n = 0; n < N_TEST_TAGS; ++n) {
    FAIL_UNLESS (d->tags_sent[n]);
  }
}

static void
check_success_sink_tags (gpointer user_data)
{
  test_data *td = user_data;
  tags_slave_data *d = td->sd;
  gint n;

  for (n = 0; n < N_TEST_TAGS; ++n) {
    FAIL_UNLESS (d->tags_received[n]);
  }
}

GST_START_TEST (test_empty_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_TEST_SOURCE, tags_source, setup_sink_tags,
      check_success_source_tags, check_success_sink_tags, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_wavparse_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_WAV_SOURCE, tags_source, setup_sink_tags,
      check_success_source_tags, check_success_sink_tags, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE, tags_source, setup_sink_tags,
      check_success_source_tags, check_success_sink_tags, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_mpegts_2_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_MPEGTS_SOURCE | TEST_FEATURE_SPLIT_SINKS, tags_source,
      setup_sink_tags, check_success_source_tags, check_success_sink_tags, NULL,
      &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_a_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_LIVE_A_SOURCE, tags_source, setup_sink_tags,
      check_success_source_tags, check_success_sink_tags, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE, tags_source, setup_sink_tags,
      check_success_source_tags, check_success_sink_tags, NULL, &md, &sd);
}

GST_END_TEST;

GST_START_TEST (test_live_av_2_tags)
{
  tags_master_data md = { 0 };
  tags_slave_data sd = { {0}
  };

  TEST_BASE (TEST_FEATURE_LIVE_AV_SOURCE | TEST_FEATURE_SPLIT_SINKS,
      tags_source, setup_sink_tags, check_success_source_tags,
      check_success_sink_tags, NULL, &md, &sd);
}

GST_END_TEST;

/**** nagivation test ****/

enum
{
  TEST_NAV_MOUSE_MOVE,
  TEST_NAV_KEY_PRESS,
  N_NAVIGATION_EVENTS
};

typedef struct
{
  gboolean got_state_changed_to_playing;
  gboolean navigation_received[N_NAVIGATION_EVENTS];
} navigation_master_data;

typedef struct
{
  gboolean started;
  gboolean navigation_sent[N_NAVIGATION_EVENTS];
  gint step;
} navigation_slave_data;

static GstPadProbeReturn
navigation_probe_source (GstPad * pad, GstPadProbeInfo * info,
    gpointer user_data)
{
  test_data *td = user_data;
  navigation_master_data *d = td->md;
  const GstStructure *s;
  const gchar *string, *key;
  double x, y;

  if (GST_IS_EVENT (info->data)) {
    if (GST_EVENT_TYPE (info->data) == GST_EVENT_NAVIGATION) {
      s = gst_event_get_structure (info->data);
      FAIL_UNLESS (s);

      /* mouse-move */
      string = gst_structure_get_string (s, "event");
      if (string && !strcmp (string, "mouse-move")) {
        if (gst_structure_get_double (s, "pointer_x", &x) && x == 4.7) {
          if (gst_structure_get_double (s, "pointer_y", &y) && y == 0.1) {
            d->navigation_received[TEST_NAV_MOUSE_MOVE] = TRUE;
          }
        }
      }

      /* key-press */
      string = gst_structure_get_string (s, "event");
      if (string && !strcmp (string, "key-press")) {
        key = gst_structure_get_string (s, "key");
        if (key && !strcmp (key, "Left")) {
          d->navigation_received[TEST_NAV_KEY_PRESS] = TRUE;
        }
      }

      /* drop at this point to imply successful handling; the upstream filesrc
       * does not know how to handle navigation events and returns FALSE,
       * which makes the test fail */
      return GST_PAD_PROBE_DROP;
    }
  }
  return GST_PAD_PROBE_OK;
}

static void
hook_navigation_probe_source (const GValue * v, gpointer user_data)
{
  hook_probe_types (v, navigation_probe_source,
      GST_PAD_PROBE_TYPE_EVENT_UPSTREAM, user_data);
}

static void
navigation_on_state_changed (gpointer user_data)
{
  test_data *td = user_data;
  navigation_master_data *d = td->md;

  if (!d->got_state_changed_to_playing)
    d->got_state_changed_to_playing = TRUE;
}

static void
navigation_source (GstElement * source, void *user_data)
{
  test_data *td = user_data;
  GstStateChangeReturn ret;
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (source));
  while (gst_iterator_foreach (it, hook_navigation_probe_source, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);

  td->state_target = GST_STATE_PLAYING;
  td->state_changed_cb = navigation_on_state_changed;
  ret = gst_element_set_state (source, GST_STATE_PLAYING);
  FAIL_UNLESS (ret == GST_STATE_CHANGE_ASYNC);
}

static void
send_navigation_event (const GValue * v, gpointer user_data)
{
  test_data *td = user_data;
  navigation_slave_data *d = td->sd;
  GstElement *sink;
  GstPad *pad, *peer;
  GstStructure *s;
  GstEvent *e = NULL;

  sink = g_value_get_object (v);
  FAIL_UNLESS (sink);
  pad = gst_element_get_static_pad (sink, "sink");
  FAIL_UNLESS (pad);
  peer = gst_pad_get_peer (pad);
  FAIL_UNLESS (peer);
  gst_object_unref (pad);

  switch (d->step) {
    case TEST_NAV_MOUSE_MOVE:
      s = gst_structure_new ("application/x-gst-navigation", "event",
          G_TYPE_STRING, "mouse-move", "button", G_TYPE_INT, 0, "pointer_x",
          G_TYPE_DOUBLE, 4.7, "pointer_y", G_TYPE_DOUBLE, 0.1, NULL);
      e = gst_event_new_navigation (s);
      break;
    case TEST_NAV_KEY_PRESS:
      s = gst_structure_new ("application/x-gst-navigation", "event",
          G_TYPE_STRING, "key-press", "key", G_TYPE_STRING, "Left", NULL);
      e = gst_event_new_navigation (s);
      break;
  }

  FAIL_UNLESS (e);
  FAIL_UNLESS (gst_pad_send_event (peer, e));
  d->navigation_sent[d->step] = TRUE;

  gst_object_unref (peer);
}

static gboolean
step_navigation (gpointer user_data)
{
  test_data *td = user_data;
  navigation_slave_data *d = td->sd;
  GstIterator *it;

  it = gst_bin_iterate_sinks (GST_BIN (td->p));
  while (gst_iterator_foreach (it, send_navigation_event, user_data))
    gst_iterator_resync (it);
  gst_iterator_free (it);

  if (++d->step < N_NAVIGATION_EVENTS)
    return G_SOURCE_CONTINUE;

  /* we are in the slave; send EOS to force the master to stop the pipeline */
  gst_element_post_message (GST_ELEMENT (td->p),
      gst_message_new_eos (GST_OBJECT (td->p)));

  gst_object_unref (td->p);
  return G_SOURCE_REMOVE;
}

static GstPadProbeReturn
navigation_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
  test_data *td = user_data;
  navigation_slave_data *d = td->sd;
  GstClockTime ts;

  if (GST_IS_BUFFER (info->data)) {
    ts = GST_BUFFER_TIMESTAMP (info->data);
    if (GST_CLOCK_TIME_IS_VALID (ts) && ts > STEP_AT