Piik/native/capture/linux/main.test.c

// build.sh --check uses synthetic GStreamer elements; no capture or codec is used.
#define main capture_main
#include "main.c"
#undef main

typedef struct { GstElement parent; guint bitrate; } FixtureEncoder;
typedef struct { GstElementClass parent; } FixtureEncoderClass;
G_DEFINE_TYPE(FixtureEncoder, fixture_encoder, GST_TYPE_ELEMENT)

static void fixture_set_property(GObject *object, guint id,
                                 const GValue *value, GParamSpec *property) {
  (void)id;
  (void)property;
  ((FixtureEncoder *)object)->bitrate = g_value_get_uint(value);
}

static void fixture_get_property(GObject *object, guint id,
                                 GValue *value, GParamSpec *property) {
  (void)id;
  (void)property;
  g_value_set_uint(value, ((FixtureEncoder *)object)->bitrate);
}

static void fixture_encoder_init(FixtureEncoder *encoder) {
  encoder->bitrate = 3000000;
}

static void fixture_encoder_class_init(FixtureEncoderClass *klass) {
  GObjectClass *object_class = G_OBJECT_CLASS(klass);
  object_class->set_property = fixture_set_property;
  object_class->get_property = fixture_get_property;
  g_object_class_install_property(object_class, 1,
      g_param_spec_uint("bitrate", "Bitrate", "Bitrate in bps", 1000, 12000000,
                        3000000, G_PARAM_READWRITE | GST_PARAM_MUTABLE_PLAYING));
}

static void check_encoder_admission_and_control(void) {
  GstElement *encoder = g_object_new(fixture_encoder_get_type(), NULL);
  GParamSpec *property = g_object_class_find_property(G_OBJECT_GET_CLASS(encoder), "bitrate");
  // Only this synthetic class is varied, before any pipeline owns an instance.
  const GParamFlags flags = property->flags;
  property->flags = G_PARAM_READWRITE;
  g_assert_cmpuint(bitrate_divisor(encoder), ==, 0);
  property->flags = G_PARAM_READABLE | GST_PARAM_MUTABLE_PLAYING;
  g_assert_cmpuint(bitrate_divisor(encoder), ==, 0);
  property->flags = flags;
  g_assert_cmpuint(bitrate_divisor(encoder), ==, 1);

  EncoderInfo info = {.bit_rate_divisor = bitrate_divisor(encoder)};
  CaptureRun run = {.output_count = 1, .encoder_info = &info};
  g_mutex_init(&run.lock);
  run.outputs[0] = (VideoOutput){.run = &run, .encoder = encoder,
      .profile = {.bit_rate = 3000000, .frame_rate = 30}};
  g_assert_true(configure_encoder(&run.outputs[0]));
  g_assert_true(apply_control(&run, "B 0 3000000"));
  g_assert_true(apply_control(&run, "B 0 1000000"));
  guint applied = 0;
  g_object_get(encoder, "bitrate", &applied, NULL);
  g_assert_cmpuint(applied, ==, 1000000);
  g_assert_null(run.failure);
  g_mutex_clear(&run.lock);
  gst_object_unref(encoder);
}

static void check_bus_error_retains_primary_cause(void) {
  CaptureRun run = {.loop = g_main_loop_new(NULL, FALSE), .output_count = 1};
  g_mutex_init(&run.lock);
  run.outputs[0].run = &run;
  run.outputs[0].encoder = gst_element_factory_make("identity", NULL);
  g_assert_nonnull(run.outputs[0].encoder);
  GError *error = g_error_new_literal(GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_WRITE,
                                     "device returned fixture failure 42");
  GstMessage *message = gst_message_new_error(GST_OBJECT(run.outputs[0].encoder), error,
                                             "fixture encoder write");
  char *frame_path = NULL;
  int frames = g_file_open_tmp("piik-native-test-XXXXXX", &frame_path, NULL);
  g_assert_cmpint(frames, >=, 0);
  int original_stdout = dup(STDOUT_FILENO);
  g_assert_cmpint(original_stdout, >=, 0);
  g_assert_cmpint(dup2(frames, STDOUT_FILENO), >=, 0);
  g_assert_true(bus_message(NULL, message, &run));
  g_assert_cmpint(dup2(original_stdout, STDOUT_FILENO), >=, 0);
  close(original_stdout);
  close(frames);
  unlink(frame_path);
  g_free(frame_path);
  g_assert_cmpstr(run.failure, ==, error->message);
  gst_message_unref(message);
  g_error_free(error);
  gst_object_unref(run.outputs[0].encoder);
  g_free(run.failure);
  g_main_loop_unref(run.loop);
  g_mutex_clear(&run.lock);
}

static void check_slot_profiles(void) {
  char *arguments[23 + 7 * 5] = {
      "capture", "--capture-video", "picker", "1", "0", "0",
      "--adapter-index", "0", "--mft-index", "0", "--width", "1280",
      "--height", "720", "--fps", "30", "--bitrate", "3000000",
      "--preference", "balanced", "--codec", "h264", "--protocol-v7"};
  for (guint index = 0; index < 7; ++index) {
    guint offset = 23 + index * 5;
    arguments[offset] = "--output";
    arguments[offset + 1] = index == 1 ? "1280" : "854";
    arguments[offset + 2] = index == 1 ? "720" : "480";
    arguments[offset + 3] = "30";
    arguments[offset + 4] = index == 1 ? "3000000" : "2000000";
  }
  CaptureRun run = {0};
  guint encoder = 0;
  g_assert_true(parse_profile(53, arguments, &run, &encoder));
  g_assert_cmpuint(run.output_count, ==, 6);
  g_assert_cmpuint(run.original_output, ==, 1);
  for (guint index = 0; index < run.output_count; ++index) {
    g_assert_cmpint(run.outputs[index].enabled, ==, index < 2);
  }
  g_assert_false(parse_profile(58, arguments, &run, &encoder));
  arguments[23 + 5 * 5 + 1] = "1920";
  g_assert_false(parse_profile(53, arguments, &run, &encoder));
  arguments[23 + 5 * 5 + 1] = "854";
  arguments[23 + 5 + 1] = "854";
  g_assert_false(parse_profile(53, arguments, &run, &encoder));
  arguments[23 + 5 + 1] = "1280";

  char *encoded[11 + 6 * 5] = {
      "capture", "--encoded-video", "--codec", "h264", "--adapter-index", "0",
      "--mft-index", "0", "--preference", "balanced", "--protocol-v7"};
  memcpy(encoded + 11, arguments + 23, 6 * 5 * sizeof(char *));
  run = (CaptureRun){0};
  g_assert_true(parse_profile(G_N_ELEMENTS(encoded), encoded, &run, &encoder));
  g_assert_cmpuint(run.profile.width, ==, 1280);
  g_assert_cmpuint(run.profile.height, ==, 720);
  g_assert_cmpuint(run.original_output, ==, 0);
  for (guint index = 0; index < run.output_count; ++index) {
    g_assert_false(run.outputs[index].enabled);
  }
}

static GstPadProbeReturn record_key_request(GstPad *pad, GstPadProbeInfo *info,
                                           gpointer data) {
  (void)pad;
  GstEvent *event = GST_PAD_PROBE_INFO_EVENT(info);
  if (gst_video_event_is_force_key_unit(event)) {
    GstEvent **request = data;
    gst_event_replace(request, event);
  }
  return GST_PAD_PROBE_OK;
}

static void check_encoder_timestamp_segment(GstClockTime offset) {
  // NVENC shifts both its PTS and segment by 1000 hours. Emit the source
  // running time, including the segment base, rather than that private PTS.
  const guint8 h264[] = {0, 0, 1, 0x67, 0x42, 0xc0, 0x1f,
                        0, 0, 1, 0x68, 0,
                        0, 0, 1, 0x65, 0};
  GError *error = NULL;
  GstElement *pipeline = gst_parse_launch(
      "appsrc name=source format=time handle-segment-change=true ! "
      "appsink name=output sync=false async=false", &error);
  g_assert_no_error(error);
  GstElement *source = gst_bin_get_by_name(GST_BIN(pipeline), "source");
  GstElement *sink = gst_bin_get_by_name(GST_BIN(pipeline), "output");
  CaptureRun run = {.active = TRUE, .last_input_timestamp = GST_CLOCK_TIME_NONE,
                    .profile = {.frame_rate = 30}};
  g_mutex_init(&run.lock);
  VideoOutput output = {.run = &run, .enabled = TRUE,
      .activation_timestamp = GST_CLOCK_TIME_NONE, .key_timestamp = GST_CLOCK_TIME_NONE,
      .key_requested = 7,
      .profile = {.width = 1280, .height = 720, .frame_rate = 30}};
  GstPad *source_pad = gst_element_get_static_pad(source, "src");
  GstPad *sink_pad = gst_element_get_static_pad(sink, "sink");
  GstEvent *request = NULL;
  gst_pad_add_probe(source_pad, GST_PAD_PROBE_TYPE_BUFFER, begin_input_frame, &run, NULL);
  gst_pad_add_probe(source_pad, GST_PAD_PROBE_TYPE_BUFFER, output_input_frame, &output, NULL);
  gst_pad_add_probe(sink_pad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM, record_key_request, &request, NULL);
  GstCaps *caps = gst_caps_from_string("video/x-h264,stream-format=byte-stream,alignment=au");
  GstBuffer *buffer = gst_buffer_new_allocate(NULL, sizeof(h264), NULL);
  gst_buffer_fill(buffer, 0, h264, sizeof(h264));
  GST_BUFFER_PTS(buffer) = offset + GST_SECOND;
  GstSegment segment;
  gst_segment_init(&segment, GST_FORMAT_TIME);
  segment.start = offset;
  segment.base = 2 * GST_SECOND;
  GstSample *sample = gst_sample_new(buffer, caps, &segment, NULL);
  char *frame_path = NULL;
  int frames = g_file_open_tmp("piik-timestamp-test-XXXXXX", &frame_path, NULL);
  int original_stdout = dup(STDOUT_FILENO);
  g_assert_cmpint(frames, >=, 0);
  g_assert_cmpint(original_stdout, >=, 0);
  g_assert_cmpint(dup2(frames, STDOUT_FILENO), >=, 0);
  g_assert_cmpint(gst_element_set_state(pipeline, GST_STATE_PLAYING), !=,
                 GST_STATE_CHANGE_FAILURE);
  g_assert_cmpint(gst_app_src_push_sample(GST_APP_SRC(source), sample), ==, GST_FLOW_OK);
  gst_sample_unref(sample);
  gst_caps_unref(caps);
  gst_buffer_unref(buffer);

  g_assert_cmpint(video_sample(GST_APP_SINK(sink), &output), ==, GST_FLOW_OK);
  g_assert_cmpint(dup2(original_stdout, STDOUT_FILENO), >=, 0);
  close(original_stdout);
  close(frames);
  char *bytes = NULL;
  gsize size = 0;
  g_assert_true(g_file_get_contents(frame_path, &bytes, &size, NULL));
  g_assert_cmpuint(size, ==, 64 + sizeof(h264));
  g_assert_cmpint(bytes[5], ==, 5);
  g_assert_cmpuint(read_be((guint8 *)bytes + 8, 8), ==, 3 * GST_SECOND / 100);
  g_assert_cmpint(bytes[37], ==, 2);
  g_assert_cmpint(bytes[38], ==, 1);
  g_assert_cmpuint(read_be((guint8 *)bytes + 40, 8), ==, 3 * GST_SECOND / 100);
  g_assert_cmpmem(bytes + 64, sizeof(h264), h264, sizeof(h264));
  g_assert_cmpuint(output.activation_timestamp, ==, 3 * GST_SECOND);
  g_assert_nonnull(request);
  GstClockTime pts, running_time;
  guint count;
  g_assert_true(gst_video_event_parse_downstream_force_key_unit(
      request, &pts, NULL, &running_time, NULL, &count));
  g_assert_cmpuint(pts, ==, offset + GST_SECOND);
  g_assert_cmpuint(running_time, ==, 3 * GST_SECOND);
  g_assert_cmpuint(count, ==, 7);
  g_free(bytes);
  unlink(frame_path);
  g_free(frame_path);
  gst_element_set_state(pipeline, GST_STATE_NULL);
  gst_event_unref(request);
  gst_object_unref(source_pad);
  gst_object_unref(sink_pad);
  gst_object_unref(source);
  gst_object_unref(sink);
  gst_object_unref(pipeline);
  g_mutex_clear(&run.lock);
}

static GstState branch_state(VideoOutput *output) {
  GstState state = GST_STATE_VOID_PENDING;
  g_assert_cmpint(gst_element_get_state(output->branch, &state, NULL, GST_SECOND),
                 !=, GST_STATE_CHANGE_FAILURE);
  return state;
}

static void check_slot_retirement(void) {
  CaptureRun run = {.output_count = 6};
  g_mutex_init(&run.lock);
  GString *description = g_string_new("fakesrc is-live=true num-buffers=8 ! tee name=frames ");
  for (guint index = 0; index < run.output_count; ++index) {
    g_string_append_printf(description,
        "frames. ! valve name=gate%u drop=true ! ( identity name=encoder%u ! "
        "fakesink sync=false async=false ) ", index, index);
  }
  GError *error = NULL;
  run.pipeline = gst_parse_launch(description->str, &error);
  g_string_free(description, TRUE);
  g_assert_no_error(error);
  g_assert_nonnull(run.pipeline);
  for (guint index = 0; index < run.output_count; ++index) {
    VideoOutput *output = &run.outputs[index];
    output->run = &run;
    output->profile.bit_rate = 2000000;
    char *name = g_strdup_printf("encoder%u", index);
    output->encoder = gst_bin_get_by_name(GST_BIN(run.pipeline), name);
    g_free(name);
    g_assert_nonnull(output->encoder);
    output->branch = GST_ELEMENT(gst_object_get_parent(GST_OBJECT(output->encoder)));
    g_assert_true(output->branch != run.pipeline);
    name = g_strdup_printf("gate%u", index);
    output->gate = gst_bin_get_by_name(GST_BIN(run.pipeline), name);
    g_free(name);
    g_assert_nonnull(output->gate);
    gst_element_set_locked_state(output->branch, TRUE);
  }
  g_assert_cmpint(gst_element_set_state(run.pipeline, GST_STATE_PLAYING), !=, GST_STATE_CHANGE_FAILURE);
  g_assert_true(apply_control(&run, "A 1 1"));
  g_assert_true(apply_control(&run, "A 5 1"));
  g_assert_cmpint(branch_state(&run.outputs[1]), ==, GST_STATE_PLAYING);
  g_assert_cmpint(branch_state(&run.outputs[5]), ==, GST_STATE_PLAYING);
  g_assert_cmpint(branch_state(&run.outputs[2]), ==, GST_STATE_NULL);
  g_assert_true(apply_control(&run, "A 1 0"));
  g_assert_cmpint(branch_state(&run.outputs[1]), ==, GST_STATE_NULL);
  g_assert_cmpint(branch_state(&run.outputs[5]), ==, GST_STATE_PLAYING);
  g_assert_true(apply_control(&run, "A 1 1"));
  g_assert_cmpint(branch_state(&run.outputs[1]), ==, GST_STATE_PLAYING);
  g_assert_true(apply_control(&run, "K 5"));
  g_assert_false(apply_control(&run, "A 2"));
  g_assert_false(apply_control(&run, "A 6 1"));
  g_assert_false(apply_control(&run, "A 0 2"));
  g_assert_false(apply_control(&run, "B 5 999"));
  gst_element_set_state(run.pipeline, GST_STATE_NULL);
  for (guint index = 0; index < run.output_count; ++index) {
    gst_object_unref(run.outputs[index].encoder);
    gst_object_unref(run.outputs[index].branch);
    gst_object_unref(run.outputs[index].gate);
  }
  gst_object_unref(run.pipeline);
  g_mutex_clear(&run.lock);
}

typedef struct {
  CaptureRun run;
  gint frames;
  gint frames_at_failure;
  guint checks;
} FailureFixture;

static void count_healthy_frame(GstElement *sink, GstBuffer *buffer,
                                GstPad *pad, gpointer data) {
  (void)sink;
  (void)buffer;
  (void)pad;
  g_atomic_int_inc(&((FailureFixture *)data)->frames);
}

static GstFlowReturn fail_without_bus(GstPad *pad, GstObject *parent, GstBuffer *buffer) {
  (void)pad;
  (void)parent;
  gst_buffer_unref(buffer);
  return GST_FLOW_ERROR;
}

static gboolean verify_sibling_survives(gpointer data) {
  FailureFixture *fixture = data;
  CaptureRun *run = &fixture->run;
  g_assert_cmpuint(++fixture->checks, <, 200);
  g_assert_null(run->failure);
  if (!run->outputs[0].failed) return G_SOURCE_CONTINUE;
  g_assert_false(run->outputs[0].enabled);
  g_assert_cmpint(branch_state(&run->outputs[0]), ==, GST_STATE_NULL);
  gint frames = g_atomic_int_get(&fixture->frames);
  if (fixture->frames_at_failure < 0) fixture->frames_at_failure = frames;
  if (frames - fixture->frames_at_failure < 10) return G_SOURCE_CONTINUE;
  // A failed slot cannot be revived by a later demand or bitrate update.
  g_assert_true(apply_control(run, "A 0 1"));
  g_assert_true(apply_control(run, "B 0 1000000"));
  g_assert_false(run->outputs[0].enabled);
  g_assert_cmpint(branch_state(&run->outputs[1]), ==, GST_STATE_PLAYING);
  GstElement *source = gst_bin_get_by_name(GST_BIN(run->pipeline), "source");
  GError *error = g_error_new_literal(GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_READ,
                                     "shared source fixture failure");
  gst_element_post_message(source, gst_message_new_error(GST_OBJECT(source), error, NULL));
  g_error_free(error);
  gst_object_unref(source);
  return G_SOURCE_REMOVE;
}

static void check_output_failure_isolation(gboolean post_error) {
  FailureFixture fixture = {.run = {.output_count = 2}, .frames_at_failure = -1};
  CaptureRun *run = &fixture.run;
  g_mutex_init(&run->lock);
  run->loop = g_main_loop_new(NULL, FALSE);
  char *description = g_strdup_printf(
      "fakesrc name=source is-live=true sizetype=fixed sizemax=4 ! "
      "identity sleep-time=1000 ! tee name=frames "
      "frames. ! valve name=gate0 drop=true ! ( queue max-size-buffers=1 "
      "max-size-bytes=0 max-size-time=0 leaky=downstream ! "
      "identity name=encoder0 error-after=%d ! fakesink name=bad sync=false async=false ) "
      "frames. ! valve name=gate1 drop=true ! ( queue max-size-buffers=1 "
      "max-size-bytes=0 max-size-time=0 leaky=downstream ! "
      "identity name=encoder1 ! fakesink name=good signal-handoffs=true sync=false async=false )",
      post_error ? 3 : -1);
  GError *error = NULL;
  run->pipeline = gst_parse_launch(description, &error);
  g_free(description);
  g_assert_no_error(error);
  for (guint index = 0; index < run->output_count; ++index) {
    VideoOutput *output = &run->outputs[index];
    output->run = run;
    output->layer = index;
    output->profile.bit_rate = 2000000;
    char *name = g_strdup_printf("encoder%u", index);
    output->encoder = gst_bin_get_by_name(GST_BIN(run->pipeline), name);
    g_free(name);
    output->branch = GST_ELEMENT(gst_object_get_parent(GST_OBJECT(output->encoder)));
    name = g_strdup_printf("gate%u", index);
    output->gate = gst_bin_get_by_name(GST_BIN(run->pipeline), name);
    g_free(name);
    g_assert_true(isolate_output(output));
  }
  GstElement *sink = gst_bin_get_by_name(GST_BIN(run->pipeline), post_error ? "good" : "bad");
  if (!post_error) {
    GstPad *pad = gst_element_get_static_pad(sink, "sink");
    gst_pad_set_chain_function(pad, fail_without_bus);
    gst_object_unref(pad);
    gst_object_unref(sink);
    sink = gst_bin_get_by_name(GST_BIN(run->pipeline), "good");
  }
  g_signal_connect(sink, "handoff", G_CALLBACK(count_healthy_frame), &fixture);
  gst_object_unref(sink);
  GstBus *bus = gst_element_get_bus(run->pipeline);
  guint watch = gst_bus_add_watch(bus, bus_message, run);
  gst_object_unref(bus);
  char *frame_path = NULL;
  int frames = g_file_open_tmp("piik-output-test-XXXXXX", &frame_path, NULL);
  int original_stdout = dup(STDOUT_FILENO);
  g_assert_cmpint(frames, >=, 0);
  g_assert_cmpint(original_stdout, >=, 0);
  g_assert_cmpint(dup2(frames, STDOUT_FILENO), >=, 0);
  g_assert_cmpint(gst_element_set_state(run->pipeline, GST_STATE_PLAYING), !=, GST_STATE_CHANGE_FAILURE);
  g_assert_true(set_output_active(&run->outputs[0], TRUE));
  g_assert_true(set_output_active(&run->outputs[1], TRUE));
  g_timeout_add(10, verify_sibling_survives, &fixture);
  g_main_loop_run(run->loop);
  g_assert_cmpstr(run->failure, ==, "shared source fixture failure");
  for (guint index = 0; index < run->output_count; ++index) {
    g_assert_true(set_output_active(&run->outputs[index], FALSE));
  }
  gst_element_set_state(run->pipeline, GST_STATE_NULL);
  g_source_remove(watch);
  g_assert_cmpint(dup2(original_stdout, STDOUT_FILENO), >=, 0);
  close(original_stdout);
  close(frames);
  char *bytes = NULL;
  gsize size = 0;
  g_assert_true(g_file_get_contents(frame_path, &bytes, &size, NULL));
  g_assert_cmpuint(size, >, 32);
  g_assert_cmpint(bytes[5], ==, 6);
  g_assert_cmpint(bytes[7], ==, 0);
  g_assert_cmpuint(size, ==, 32 + read_be((guint8 *)bytes + 28, 4));
  if (post_error) g_assert_nonnull(strstr(bytes + 32, "Failed after iterations as requested"));
  g_free(bytes);
  unlink(frame_path);
  g_free(frame_path);
  for (guint index = 0; index < run->output_count; ++index) {
    gst_object_unref(run->outputs[index].encoder);
    gst_object_unref(run->outputs[index].branch);
    gst_object_unref(run->outputs[index].gate);
  }
  gst_object_unref(run->pipeline);
  g_free(run->failure);
  g_main_loop_unref(run->loop);
  g_mutex_clear(&run->lock);
}

int main(int argc, char **argv) {
  gst_init(&argc, &argv);
  check_encoder_admission_and_control();
  check_bus_error_retains_primary_cause();
  check_slot_profiles();
  check_encoder_timestamp_segment(1000ULL * 60 * 60 * GST_SECOND);
  check_encoder_timestamp_segment(0);
  check_slot_retirement();
  check_output_failure_isolation(TRUE);
  check_output_failure_isolation(FALSE);
  return 0;
}