:right-sidebar: True TestClock =================================================================== .. currentmodule:: gi.repository.GstCheck .. versionadded:: 1.2 .. class:: TestClock(**properties: ~typing.Any) :no-contents-entry: Superclasses: :class:`~gi.repository.Gst.Clock`, :class:`~gi.repository.Gst.Object`, :class:`~gi.repository.GObject.InitiallyUnowned`, :class:`~gi.repository.GObject.Object` GstTestClock is an implementation of :obj:`~gi.repository.Gst.Clock` which has different behaviour compared to ``GstSystemClock``. Time for ``GstSystemClock`` advances according to the system time, while time for :obj:`~gi.repository.GstCheck.TestClock` changes only when :func:`~gi.repository.GstCheck.TestClock.set_time` or :func:`~gi.repository.GstCheck.TestClock.advance_time` are called. :obj:`~gi.repository.GstCheck.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 -------------------------------------------------------------------------------- .. code-block:: C :dedent: #include #include 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))); ... :obj:`~gi.repository.Gst.Clock` allows for setting up single shot or periodic clock notifications as well as waiting for these notifications synchronously (using :func:`~gi.repository.Gst.Gst.Clock.id_wait`) or asynchronously (using :func:`~gi.repository.Gst.Gst.Clock.id_wait_async` or :func:`~gi.repository.Gst.Gst.Clock.id_wait_async`). This is used by many GStreamer elements, among them ``GstBaseSrc`` and ``GstBaseSink``. :obj:`~gi.repository.GstCheck.TestClock` keeps track of these clock notifications. By calling :func:`~gi.repository.GstCheck.TestClock.wait_for_next_pending_id` or :func:`~gi.repository.GstCheck.TestClock.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 :func:`~gi.repository.GstCheck.TestClock.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 :obj:`~gi.repository.GstCheck.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 :func:`~gi.repository.GstCheck.TestClock.wait_for_next_pending_id` or :func:`~gi.repository.GstCheck.TestClock.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 :obj:`~gi.repository.GstCheck.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 :func:`~gi.repository.Gst.Gst.Clock.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 -------------------------------------------------------------------------------- .. code-block:: C :dedent: #include #include #include 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 :obj:`~gi.repository.GstCheck.TestClock` is only supposed to be used in unit tests it calls :func:`~gi.repository.GLib.assert`, :func:`~gi.repository.GLib.assert_cmpint` or :func:`~gi.repository.GLib.assert_cmpuint` to validate all function arguments. This will highlight any issues with the unit test code itself. Constructors ------------ .. rst-class:: interim-class .. class:: TestClock :no-index: .. classmethod:: new() -> ~gi.repository.Gst.Clock Creates a new test clock with its time set to zero. MT safe. .. versionadded:: 1.2 .. classmethod:: new_with_start_time(start_time: int) -> ~gi.repository.Gst.Clock Creates a new test clock with its time set to the specified time. MT safe. .. versionadded:: 1.2 :param start_time: a :obj:`~gi.repository.Gst.ClockTime` set to the desired start time of the clock. Methods ------- .. rst-class:: interim-class .. class:: TestClock :no-index: .. method:: 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. .. versionadded:: 1.2 :param delta: a positive :obj:`~gi.repository.Gst.ClockTimeDiff` to be added to the time of the clock .. method:: crank() -> bool A "crank" consists of three steps: 1: Wait for a :obj:`~gi.repository.Gst.ClockID` to be registered with the :obj:`~gi.repository.GstCheck.TestClock`\. 2: Advance the :obj:`~gi.repository.GstCheck.TestClock` to the time the :obj:`~gi.repository.Gst.ClockID` is waiting, unless the clock time is already passed the clock id (Since: 1.18). 3: Release the :obj:`~gi.repository.Gst.ClockID` wait. A "crank" can be though of as the notion of manually driving the clock forward to its next logical step. .. versionadded:: 1.8 .. method:: get_next_entry_time() -> int Retrieve the requested time for the next pending clock notification. MT safe. .. versionadded:: 1.2 .. method:: has_id(id: ~typing.Any) -> bool Checks whether ``test_clock`` was requested to provide the clock notification given by ``id``\. MT safe. .. versionadded:: 1.2 :param id: a :obj:`~gi.repository.Gst.ClockID` clock notification .. classmethod:: id_list_get_latest_time() -> int Finds the latest time inside the list. MT safe. .. versionadded:: 1.4 .. method:: peek_id_count() -> int Determine the number of pending clock notifications that have been requested from the ``test_clock``\. MT safe. .. versionadded:: 1.2 .. method:: peek_next_pending_id() -> ~typing.Tuple[bool, ~typing.Any] Determines if the ``pending_id`` is the next clock notification scheduled to be triggered given the current time of the ``test_clock``\. MT safe. .. versionadded:: 1.2 .. method:: process_id(pending_id: ~typing.Any) -> bool Processes and releases the pending ID. MT safe. .. versionadded:: 1.18 :param pending_id: :obj:`~gi.repository.Gst.ClockID` .. method:: process_id_list(pending_list: list[~typing.Any] | None = None) -> int Processes and releases the pending IDs in the list. MT safe. .. versionadded:: 1.4 :param pending_list: List of pending :obj:`~gi.repository.Gst.ClockID` .. method:: process_next_clock_id() -> ~typing.Any | None MT safe. .. versionadded:: 1.2 .. method:: 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 :func:`~gi.repository.Gst.Gst.Clock.get_time` is a programming error. MT safe. .. versionadded:: 1.2 :param new_time: a :obj:`~gi.repository.Gst.ClockTime` later than that returned by :func:`~gi.repository.Gst.Gst.Clock.get_time` .. method:: timed_wait_for_multiple_pending_ids(count: int, timeout_ms: int) -> ~typing.Tuple[bool, list[~typing.Any]] Blocks until at least ``count`` clock notifications have been requested from ``test_clock``\, or the timeout expires. MT safe. .. versionadded:: 1.16 :param count: the number of pending clock notifications to wait for :param timeout_ms: the timeout in milliseconds .. method:: wait_for_multiple_pending_ids(count: int) -> list[~typing.Any] 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 :obj:`~gi.repository.GstCheck.TestClock`\. MT safe. .. versionadded:: 1.4 :param count: the number of pending clock notifications to wait for .. method:: wait_for_next_pending_id() -> ~typing.Any Waits until a clock notification is requested from ``test_clock``\. There is no timeout for this wait, see the main description of :obj:`~gi.repository.GstCheck.TestClock`\. A reference to the pending clock notification is stored in ``pending_id``\. MT safe. .. versionadded:: 1.2 .. method:: 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 :obj:`~gi.repository.GstCheck.TestClock`\. .. versionadded:: 1.2 .. deprecated:: Unknown use :func:`~gi.repository.GstCheck.TestClock.wait_for_multiple_pending_ids` instead. :param count: the number of pending clock notifications to wait for Properties ---------- .. rst-class:: interim-class .. class:: TestClock :no-index: .. attribute:: props.clock_type :type: ~gi.repository.Gst.ClockType The type of the None singleton. .. attribute:: props.start_time :type: int The type of the None singleton. Fields ------ .. rst-class:: interim-class .. class:: TestClock :no-index: .. attribute:: parent .. attribute:: priv