TestClock

Added in version 1.2.

class TestClock(**properties: Any)

Superclasses: Clock, Object, InitiallyUnowned, Object

GstTestClock is an implementation of Clock which has different behaviour compared to GstSystemClock. Time for GstSystemClock advances according to the system time, while time for TestClock changes only when set_time() or advance_time() are called. TestClock provides unit tests with the possibility to precisely advance the time in a deterministic manner, independent of the system time or any other external factors.

Advancing the time of a TestClock

``include`` <gst/gst.h>
``include`` <gst/check/gsttestclock.h>

GstClock *clock;
GstTestClock *test_clock;

clock = gst_test_clock_new ();
test_clock = GST_TEST_CLOCK (clock);
GST_INFO ("Time: %" GST_TIME_FORMAT, GST_TIME_ARGS (gst_clock_get_time (clock)));
gst_test_clock_advance_time ( test_clock, 1 * GST_SECOND);
GST_INFO ("Time: %" GST_TIME_FORMAT, GST_TIME_ARGS (gst_clock_get_time (clock)));
g_usleep (10 * G_USEC_PER_SEC);
GST_INFO ("Time: %" GST_TIME_FORMAT, GST_TIME_ARGS (gst_clock_get_time (clock)));
gst_test_clock_set_time (test_clock, 42 * GST_SECOND);
GST_INFO ("Time: %" GST_TIME_FORMAT, GST_TIME_ARGS (gst_clock_get_time (clock)));
...

Clock allows for setting up single shot or periodic clock notifications as well as waiting for these notifications synchronously (using id_wait()) or asynchronously (using id_wait_async() or id_wait_async()). This is used by many GStreamer elements, among them GstBaseSrc and GstBaseSink.

TestClock keeps track of these clock notifications. By calling wait_for_next_pending_id() or wait_for_multiple_pending_ids() a unit tests may wait for the next one or several clock notifications to be requested. Additionally unit tests may release blocked waits in a controlled fashion by calling process_next_clock_id(). This way a unit test can control the inaccuracy (jitter) of clock notifications, since the test can decide to release blocked waits when the clock time has advanced exactly to, or past, the requested clock notification time.

There are also interfaces for determining if a notification belongs to a TestClock or not, as well as getting the number of requested clock notifications so far.

N.B.: When a unit test waits for a certain amount of clock notifications to be requested in wait_for_next_pending_id() or wait_for_multiple_pending_ids() then these functions may block for a long time. If they block forever then the expected clock notifications were never requested from TestClock, and so the assumptions in the code of the unit test are wrong. The unit test case runner in gstcheck is expected to catch these cases either by the default test case timeout or the one set for the unit test by calling tcase_set_timeout().

The sample code below assumes that the element under test will delay a buffer pushed on the source pad by some latency until it arrives on the sink pad. Moreover it is assumed that the element will at some point call id_wait() to synchronously wait for a specific time. The first buffer sent will arrive exactly on time only delayed by the latency. The second buffer will arrive a little late (7ms) due to simulated jitter in the clock notification.

Demonstration of how to work with clock notifications and TestClock

``include`` <gst/gst.h>
``include`` <gst/check/gstcheck.h>
``include`` <gst/check/gsttestclock.h>

GstClockTime latency;
GstElement *element;
GstPad *srcpad;
GstClock *clock;
GstTestClock *test_clock;
GstBuffer buf;
GstClockID pending_id;
GstClockID processed_id;

latency = 42 * GST_MSECOND;
element = create_element (latency, ...);
srcpad = get_source_pad (element);

clock = gst_test_clock_new ();
test_clock = GST_TEST_CLOCK (clock);
gst_element_set_clock (element, clock);

GST_INFO ("Set time, create and push the first buffer\n");
gst_test_clock_set_time (test_clock, 0);
buf = create_test_buffer (gst_clock_get_time (clock), ...);
gst_assert_cmpint (gst_pad_push (srcpad, buf), ==, GST_FLOW_OK);

GST_INFO ("Block until element is waiting for a clock notification\n");
gst_test_clock_wait_for_next_pending_id (test_clock, &pending_id);
GST_INFO ("Advance to the requested time of the clock notification\n");
gst_test_clock_advance_time (test_clock, latency);
GST_INFO ("Release the next blocking wait and make sure it is the one from element\n");
processed_id = gst_test_clock_process_next_clock_id (test_clock);
g_assert (processed_id == pending_id);
g_assert_cmpint (GST_CLOCK_ENTRY_STATUS (processed_id), ==, GST_CLOCK_OK);
gst_clock_id_unref (pending_id);
gst_clock_id_unref (processed_id);

GST_INFO ("Validate that element produced an output buffer and check its timestamp\n");
g_assert_cmpint (get_number_of_output_buffer (...), ==, 1);
buf = get_buffer_pushed_by_element (element, ...);
g_assert_cmpint (GST_BUFFER_TIMESTAMP (buf), ==, latency);
gst_buffer_unref (buf);
GST_INFO ("Check that element does not wait for any clock notification\n");
g_assert (!gst_test_clock_peek_next_pending_id (test_clock, NULL));

GST_INFO ("Set time, create and push the second buffer\n");
gst_test_clock_advance_time (test_clock, 10 * GST_SECOND);
buf = create_test_buffer (gst_clock_get_time (clock), ...);
gst_assert_cmpint (gst_pad_push (srcpad, buf), ==, GST_FLOW_OK);

GST_INFO ("Block until element is waiting for a new clock notification\n");
(gst_test_clock_wait_for_next_pending_id (test_clock, &pending_id);
GST_INFO ("Advance past 7ms beyond the requested time of the clock notification\n");
gst_test_clock_advance_time (test_clock, latency + 7 * GST_MSECOND);
GST_INFO ("Release the next blocking wait and make sure it is the one from element\n");
processed_id = gst_test_clock_process_next_clock_id (test_clock);
g_assert (processed_id == pending_id);
g_assert_cmpint (GST_CLOCK_ENTRY_STATUS (processed_id), ==, GST_CLOCK_OK);
gst_clock_id_unref (pending_id);
gst_clock_id_unref (processed_id);

GST_INFO ("Validate that element produced an output buffer and check its timestamp\n");
g_assert_cmpint (get_number_of_output_buffer (...), ==, 1);
buf = get_buffer_pushed_by_element (element, ...);
g_assert_cmpint (GST_BUFFER_TIMESTAMP (buf), ==,
    10 * GST_SECOND + latency + 7 * GST_MSECOND);
gst_buffer_unref (buf);
GST_INFO ("Check that element does not wait for any clock notification\n");
g_assert (!gst_test_clock_peek_next_pending_id (test_clock, NULL));
...

Since TestClock is only supposed to be used in unit tests it calls assert(), assert_cmpint() or assert_cmpuint() to validate all function arguments. This will highlight any issues with the unit test code itself.

Constructors

class TestClock

classmethod new() → Clock

Creates a new test clock with its time set to zero.

MT safe.

Added in version 1.2.

classmethod new_with_start_time(start_time: int) → Clock

Creates a new test clock with its time set to the specified time.

MT safe.

Added in version 1.2.

Parameters:

start_time – a ClockTime set to the desired start time of the clock.

Methods

class TestClock

advance_time(delta: int) → None

Advances the time of the test_clock by the amount given by delta. The time of test_clock is monotonically increasing, therefore providing a delta which is negative or zero is a programming error.

MT safe.

Added in version 1.2.

Parameters:

delta – a positive ClockTimeDiff to be added to the time of the clock

crank() → bool

A “crank” consists of three steps: 1: Wait for a ClockID to be registered with the TestClock. 2: Advance the TestClock to the time the ClockID is waiting, unless

the clock time is already passed the clock id (Since: 1.18).

3: Release the ClockID wait. A “crank” can be though of as the notion of manually driving the clock forward to its next logical step.

Added in version 1.8.

get_next_entry_time() → int

Retrieve the requested time for the next pending clock notification.

MT safe.

Added in version 1.2.

has_id(id: None) → bool

Checks whether test_clock was requested to provide the clock notification given by id.

MT safe.

Added in version 1.2.

Parameters:

id – a ClockID clock notification

id_list_get_latest_time(pending_list: list[None] | None = None) → int

Finds the latest time inside the list.

MT safe.

Added in version 1.4.

Parameters:

pending_list – List of of pending ClockID

peek_id_count() → int

Determine the number of pending clock notifications that have been requested from the test_clock.

MT safe.

Added in version 1.2.

peek_next_pending_id() → tuple[bool, None]

Determines if the pending_id is the next clock notification scheduled to be triggered given the current time of the test_clock.

MT safe.

Added in version 1.2.

process_id(pending_id: None) → bool

Processes and releases the pending ID.

MT safe.

Added in version 1.18.

Parameters:

pending_id – ClockID

process_id_list(pending_list: list[None] | None = None) → int

Processes and releases the pending IDs in the list.

MT safe.

Added in version 1.4.

Parameters:

pending_list – List of pending ClockID

process_next_clock_id() → None

MT safe.

Added in version 1.2.

set_time(new_time: int) → None

Sets the time of test_clock to the time given by new_time. The time of test_clock is monotonically increasing, therefore providing a new_time which is earlier or equal to the time of the clock as given by get_time() is a programming error.

MT safe.

Added in version 1.2.

Parameters:

new_time – a ClockTime later than that returned by get_time()

timed_wait_for_multiple_pending_ids(count: int, timeout_ms: int) → tuple[bool, list[None]]

Blocks until at least count clock notifications have been requested from test_clock, or the timeout expires.

MT safe.

Added in version 1.16.

Parameters:

• count – the number of pending clock notifications to wait for

• timeout_ms – the timeout in milliseconds

wait_for_multiple_pending_ids(count: int) → list[None]

Blocks until at least count clock notifications have been requested from test_clock. There is no timeout for this wait, see the main description of TestClock.

MT safe.

Added in version 1.4.

Parameters:

count – the number of pending clock notifications to wait for

wait_for_next_pending_id() → None

Waits until a clock notification is requested from test_clock. There is no timeout for this wait, see the main description of TestClock. A reference to the pending clock notification is stored in pending_id.

MT safe.

Added in version 1.2.

wait_for_pending_id_count(count: int) → None

Blocks until at least count clock notifications have been requested from test_clock. There is no timeout for this wait, see the main description of TestClock.

Added in version 1.2.

Deprecated since version Unknown: use wait_for_multiple_pending_ids() instead.

Parameters:

count – the number of pending clock notifications to wait for

Properties

class TestClock

props.clock_type: ClockType

props.start_time: int

When a TestClock is constructed it will have a certain start time set. If the clock was created using new_with_start_time() then this property contains the value of the start_time argument. If new() was called the clock started at time zero, and thus this property contains the value 0.

Fields

class TestClock

parent

priv