gst-plugins-rs/mux/fmp4/tests/tests.rs

// Copyright (C) 2021 Sebastian Dröge <sebastian@centricular.com>
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
// If a copy of the MPL was not distributed with this file, You can obtain one at
// <https://mozilla.org/MPL/2.0/>.
//
// SPDX-License-Identifier: MPL-2.0
//

use gst::prelude::*;

fn init() {
    use std::sync::Once;
    static INIT: Once = Once::new();

    INIT.call_once(|| {
        gst::init().unwrap();
        gstfmp4::plugin_register_static().unwrap();
    });
}

fn to_completion(pipeline: &gst::Pipeline) {
    pipeline
        .set_state(gst::State::Playing)
        .expect("Unable to set the pipeline to the `Playing` state");

    for msg in pipeline.bus().unwrap().iter_timed(gst::ClockTime::NONE) {
        use gst::MessageView;

        match msg.view() {
            MessageView::Eos(..) => break,
            MessageView::Error(err) => {
                panic!(
                    "Error from {:?}: {} ({:?})",
                    err.src().map(|s| s.path_string()),
                    err.error(),
                    err.debug()
                );
            }
            _ => (),
        }
    }

    pipeline
        .set_state(gst::State::Null)
        .expect("Unable to set the pipeline to the `Null` state");
}

fn test_buffer_flags_single_stream(cmaf: bool, set_dts: bool, caps: gst::Caps) {
    let mut h = if cmaf {
        gst_check::Harness::new("cmafmux")
    } else {
        gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"))
    };

    // 5s fragment duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h.set_src_caps(caps);
    h.play();

    let output_offset = if cmaf {
        gst::ClockTime::ZERO
    } else {
        (60 * 60 * 1000).seconds()
    };

    // Push 7 buffers of 1s each, 1st and 6 buffer without DELTA_UNIT flag
    for i in 0..7 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            if set_dts {
                buffer.set_dts(i.seconds());
            }
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 5 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    if set_dts {
        assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));
    }

    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    if set_dts {
        assert_eq!(
            fragment_header.dts(),
            Some(gst::ClockTime::ZERO + output_offset)
        );
    }
    assert_eq!(fragment_header.duration(), Some(5.seconds()));

    for i in 0..5 {
        let buffer = h.pull().unwrap();
        if i == 4 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
        if set_dts {
            assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
        }
        assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
    }

    h.push_event(gst::event::Eos::new());

    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(5.seconds() + output_offset));
    if set_dts {
        assert_eq!(fragment_header.dts(), Some(5.seconds() + output_offset));
    }
    assert_eq!(fragment_header.duration(), Some(2.seconds()));

    for i in 5..7 {
        let buffer = h.pull().unwrap();
        if i == 6 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
        if set_dts {
            assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
        }
        assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
    }

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_buffer_flags_single_h264_stream_cmaf() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    test_buffer_flags_single_stream(true, true, caps);
}

#[test]
fn test_buffer_flags_single_h264_stream_iso() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    test_buffer_flags_single_stream(false, true, caps);
}

#[test]
fn test_buffer_flags_single_vp9_stream_iso() {
    init();

    let caps = gst::Caps::builder("video/x-vp9")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("profile", "0")
        .field("chroma-format", "4:2:0")
        .field("bit-depth-luma", 8u32)
        .field("bit-depth-chroma", 8u32)
        .field("colorimetry", "bt709")
        .build();

    test_buffer_flags_single_stream(false, false, caps);
}

#[test]
fn test_buffer_flags_single_av1_stream_cmaf() {
    init();

    let caps = gst::Caps::builder("video/x-av1")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("profile", "main")
        .field("tier", "main")
        .field("level", "4.1")
        .field("chroma-format", "4:2:0")
        .field("bit-depth-luma", 8u32)
        .field("bit-depth-chroma", 8u32)
        .field("colorimetry", "bt709")
        .build();

    test_buffer_flags_single_stream(true, false, caps);
}

#[test]
fn test_buffer_flags_multi_stream() {
    init();

    let mut h1 = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));
    let mut h2 = gst_check::Harness::with_element(&h1.element().unwrap(), Some("sink_1"), None);

    // 5s fragment duration
    h1.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h1.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h1.play();

    h2.set_src_caps(
        gst::Caps::builder("audio/mpeg")
            .field("mpegversion", 4i32)
            .field("channels", 1i32)
            .field("rate", 44100i32)
            .field("stream-format", "raw")
            .field("base-profile", "lc")
            .field("profile", "lc")
            .field("level", "2")
            .field(
                "codec_data",
                gst::Buffer::from_slice([0x12, 0x08, 0x56, 0xe5, 0x00]),
            )
            .build(),
    );
    h2.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 7 buffers of 1s each, 1st and last buffer without DELTA_UNIT flag
    for i in 0..7 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 5 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h1.push(buffer), Ok(gst::FlowSuccess::Ok));

        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
        }
        assert_eq!(h2.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h1.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );

            let ev = loop {
                let ev = h2.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h1.crank_single_clock_wait().unwrap();

    let header = h1.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(
        fragment_header.dts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(fragment_header.duration(), Some(5.seconds()));

    for i in 0..5 {
        for j in 0..2 {
            let buffer = h1.pull().unwrap();
            if i == 4 && j == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }

            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));

            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    h1.push_event(gst::event::Eos::new());
    h2.push_event(gst::event::Eos::new());

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(5.seconds() + output_offset));
    assert_eq!(fragment_header.dts(), Some(5.seconds() + output_offset));
    assert_eq!(fragment_header.duration(), Some(2.seconds()));

    for i in 5..7 {
        for j in 0..2 {
            let buffer = h1.pull().unwrap();
            if i == 6 && j == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_live_timeout() {
    init();

    let mut h1 = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));
    let mut h2 = gst_check::Harness::with_element(&h1.element().unwrap(), Some("sink_1"), None);

    h1.use_testclock();

    // 5s fragment duration
    h1.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h1.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h1.play();

    h2.set_src_caps(
        gst::Caps::builder("audio/mpeg")
            .field("mpegversion", 4i32)
            .field("channels", 1i32)
            .field("rate", 44100i32)
            .field("stream-format", "raw")
            .field("base-profile", "lc")
            .field("profile", "lc")
            .field("level", "2")
            .field(
                "codec_data",
                gst::Buffer::from_slice([0x12, 0x08, 0x56, 0xe5, 0x00]),
            )
            .build(),
    );
    h2.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 7 buffers of 1s each, 1st and last buffer without DELTA_UNIT flag
    for i in 0..7 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 5 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h1.push(buffer), Ok(gst::FlowSuccess::Ok));

        // Skip buffer 4th and 6th buffer (end of fragment / stream)
        if i == 4 || i == 6 {
            continue;
        } else {
            let mut buffer = gst::Buffer::with_size(1).unwrap();
            {
                let buffer = buffer.get_mut().unwrap();
                buffer.set_pts(i.seconds());
                buffer.set_dts(i.seconds());
                buffer.set_duration(gst::ClockTime::SECOND);
            }
            assert_eq!(h2.push(buffer), Ok(gst::FlowSuccess::Ok));
        }

        if i == 2 {
            let ev = loop {
                let ev = h1.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );

            let ev = loop {
                let ev = h2.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h1.crank_single_clock_wait().unwrap();

    let header = h1.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(
        fragment_header.dts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(fragment_header.duration(), Some(5.seconds()));

    for i in 0..5 {
        for j in 0..2 {
            // Skip gap events that don't result in buffers
            if j == 1 && i == 4 {
                // Advance time and crank the clock another time. This brings us at the end of the
                // EOS.
                h1.crank_single_clock_wait().unwrap();
                continue;
            }

            let buffer = h1.pull().unwrap();
            if i == 4 && j == 0 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else if i == 5 && j == 0 {
                assert_eq!(buffer.flags(), gst::BufferFlags::HEADER);
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }

            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));

            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    h1.push_event(gst::event::Eos::new());
    h2.push_event(gst::event::Eos::new());

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(5.seconds() + output_offset));
    assert_eq!(fragment_header.dts(), Some(5.seconds() + output_offset));
    assert_eq!(fragment_header.duration(), Some(2.seconds()));

    for i in 5..7 {
        for j in 0..2 {
            // Skip gap events that don't result in buffers
            if j == 1 && i == 6 {
                continue;
            }

            let buffer = h1.pull().unwrap();
            if i == 6 && j == 0 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_gap_events() {
    init();

    let mut h1 = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));
    let mut h2 = gst_check::Harness::with_element(&h1.element().unwrap(), Some("sink_1"), None);

    h1.use_testclock();

    // 5s fragment duration
    h1.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h1.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h1.play();

    h2.set_src_caps(
        gst::Caps::builder("audio/mpeg")
            .field("mpegversion", 4i32)
            .field("channels", 1i32)
            .field("rate", 44100i32)
            .field("stream-format", "raw")
            .field("base-profile", "lc")
            .field("profile", "lc")
            .field("level", "2")
            .field(
                "codec_data",
                gst::Buffer::from_slice([0x12, 0x08, 0x56, 0xe5, 0x00]),
            )
            .build(),
    );
    h2.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 7 buffers of 1s each, 1st and last buffer without DELTA_UNIT flag
    for i in 0..7 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 5 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h1.push(buffer), Ok(gst::FlowSuccess::Ok));

        // Replace buffer 3 and 6 with a gap event
        if i == 3 || i == 6 {
            let ev = gst::event::Gap::builder(i.seconds())
                .duration(gst::ClockTime::SECOND)
                .build();
            assert!(h2.push_event(ev));
        } else {
            let mut buffer = gst::Buffer::with_size(1).unwrap();
            {
                let buffer = buffer.get_mut().unwrap();
                buffer.set_pts(i.seconds());
                buffer.set_dts(i.seconds());
                buffer.set_duration(gst::ClockTime::SECOND);
            }
            assert_eq!(h2.push(buffer), Ok(gst::FlowSuccess::Ok));
        }

        if i == 2 {
            let ev = loop {
                let ev = h1.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );

            let ev = loop {
                let ev = h2.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h1.crank_single_clock_wait().unwrap();

    let header = h1.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(
        fragment_header.dts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(fragment_header.duration(), Some(5.seconds()));

    for i in 0..5 {
        for j in 0..2 {
            // Skip gap events that don't result in buffers
            if j == 1 && i == 3 {
                continue;
            }

            let buffer = h1.pull().unwrap();
            if i == 4 && j == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }

            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));

            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    h1.push_event(gst::event::Eos::new());
    h2.push_event(gst::event::Eos::new());

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(5.seconds() + output_offset));
    assert_eq!(fragment_header.dts(), Some(5.seconds() + output_offset));
    assert_eq!(fragment_header.duration(), Some(2.seconds()));

    for i in 5..7 {
        for j in 0..2 {
            // Skip gap events that don't result in buffers
            if j == 1 && i == 6 {
                continue;
            }

            let buffer = h1.pull().unwrap();
            if i == 6 && j == 0 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_single_stream_short_gops() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    // 5s fragment duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 8 buffers of 1s each, 1st, 4th and 7th buffer without DELTA_UNIT flag
    for i in 0..8 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 3 && i != 6 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 || i == 7 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            let fku_time = if i == 2 { 5.seconds() } else { 8.seconds() };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(fku_time),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(
        fragment_header.dts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(fragment_header.duration(), Some(3.seconds()));

    for i in 0..3 {
        let buffer = h.pull().unwrap();
        if i == 2 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
    }

    h.push_event(gst::event::Eos::new());

    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(3.seconds() + output_offset));
    assert_eq!(fragment_header.dts(), Some(3.seconds() + output_offset));
    assert_eq!(fragment_header.duration(), Some(5.seconds()));

    for i in 3..8 {
        let buffer = h.pull().unwrap();
        if i == 7 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
    }

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_single_stream_long_gops() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    // 5s fragment duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 10 buffers of 1s each, 1st and 7th buffer without DELTA_UNIT flag
    for i in 0..10 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 6 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 || i == 7 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            let fku_time = if i == 2 { 5.seconds() } else { 11.seconds() };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(fku_time),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(
        fragment_header.dts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(fragment_header.duration(), Some(6.seconds()));

    for i in 0..6 {
        let buffer = h.pull().unwrap();
        if i == 5 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
    }

    h.push_event(gst::event::Eos::new());

    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(6.seconds() + output_offset));
    assert_eq!(fragment_header.dts(), Some(6.seconds() + output_offset));
    assert_eq!(fragment_header.duration(), Some(4.seconds()));

    for i in 6..10 {
        let buffer = h.pull().unwrap();
        if i == 9 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
        assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
    }

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_buffer_multi_stream_short_gops() {
    init();

    let mut h1 = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));
    let mut h2 = gst_check::Harness::with_element(&h1.element().unwrap(), Some("sink_1"), None);

    // 5s fragment duration
    h1.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());

    h1.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h1.play();

    h2.set_src_caps(
        gst::Caps::builder("audio/mpeg")
            .field("mpegversion", 4i32)
            .field("channels", 1i32)
            .field("rate", 44100i32)
            .field("stream-format", "raw")
            .field("base-profile", "lc")
            .field("profile", "lc")
            .field("level", "2")
            .field(
                "codec_data",
                gst::Buffer::from_slice([0x12, 0x08, 0x56, 0xe5, 0x00]),
            )
            .build(),
    );
    h2.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 9 buffers of 1s each, 1st, 4th and 7th buffer without DELTA_UNIT flag
    for i in 0..9 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
            if i != 0 && i != 3 && i != 6 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h1.push(buffer), Ok(gst::FlowSuccess::Ok));

        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i.seconds());
            buffer.set_dts(i.seconds());
            buffer.set_duration(gst::ClockTime::SECOND);
        }
        assert_eq!(h2.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 || i == 8 {
            let ev = loop {
                let ev = h1.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            let fku_time = if i == 2 { 5.seconds() } else { 8.seconds() };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(fku_time),
                    all_headers: true,
                    count: 0
                }
            );

            let ev = loop {
                let ev = h2.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(fku_time),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    let header = h1.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(
        fragment_header.pts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(
        fragment_header.dts(),
        Some(gst::ClockTime::ZERO + output_offset)
    );
    assert_eq!(fragment_header.duration(), Some(3.seconds()));

    for i in 0..3 {
        for j in 0..2 {
            let buffer = h1.pull().unwrap();
            if i == 2 && j == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }

            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));

            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    h1.push_event(gst::event::Eos::new());
    h2.push_event(gst::event::Eos::new());

    let fragment_header = h1.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(3.seconds() + output_offset));
    assert_eq!(fragment_header.dts(), Some(3.seconds() + output_offset));
    assert_eq!(fragment_header.duration(), Some(6.seconds()));

    for i in 3..9 {
        for j in 0..2 {
            let buffer = h1.pull().unwrap();
            if i == 8 && j == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(i.seconds() + output_offset));
            if j == 0 {
                assert_eq!(buffer.dts(), Some(i.seconds() + output_offset));
            } else {
                assert!(buffer.dts().is_none());
            }
            assert_eq!(buffer.duration(), Some(gst::ClockTime::SECOND));
        }
    }

    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_single_stream_manual_fragment() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    let mut h = gst_check::Harness::new("cmafmux");

    // fragment duration long enough to be ignored, 1s chunk duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.hours());

    h.set_src_caps(caps);
    h.play();

    // request fragment at 4 seconds, should be created at 11th buffer
    h.element()
        .unwrap()
        .emit_by_name::<()>("split-at-running-time", &[&4.seconds()]);

    // Push 15 buffers of 0.5s each, 1st, 11th and 16th buffer without DELTA_UNIT flag
    for i in 0..20 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
            if i != 0 && i != 10 && i != 15 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(4.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO));

    // first fragment
    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(gst::ClockTime::ZERO));
    assert_eq!(fragment_header.dts(), Some(gst::ClockTime::ZERO));
    assert_eq!(fragment_header.duration(), Some(5.seconds()));

    for buffer_idx in 0..10 {
        let buffer = h.pull().unwrap();
        if buffer_idx == 9 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(buffer.pts(), Some(buffer_idx * 500.mseconds()));
        assert_eq!(buffer.dts(), Some(buffer_idx * 500.mseconds()));
        assert_eq!(buffer.duration(), Some(500.mseconds()));
    }

    // second manual fragment
    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(5.seconds()));
    assert_eq!(fragment_header.dts(), Some(5.seconds()));
    assert_eq!(fragment_header.duration(), Some(2500.mseconds()));

    for buffer_idx in 0..5 {
        let buffer = h.pull().unwrap();
        if buffer_idx == 4 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(
            buffer.pts(),
            Some(5.seconds() + buffer_idx * 500.mseconds())
        );
        assert_eq!(
            buffer.dts(),
            Some(5.seconds() + buffer_idx * 500.mseconds())
        );
        assert_eq!(buffer.duration(), Some(500.mseconds()));
    }

    h.push_event(gst::event::Eos::new());

    // There should be the second fragment now
    let fragment_header = h.pull().unwrap();
    assert_eq!(fragment_header.flags(), gst::BufferFlags::HEADER);
    assert_eq!(fragment_header.pts(), Some(7500.mseconds()));
    assert_eq!(fragment_header.dts(), Some(7500.mseconds()));
    assert_eq!(fragment_header.duration(), Some(2500.mseconds()));

    for buffer_idx in 0..5 {
        let buffer = h.pull().unwrap();
        if buffer_idx == 4 {
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
        } else {
            assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(
            buffer.pts(),
            Some(7500.mseconds() + buffer_idx * 500.mseconds())
        );
        assert_eq!(
            buffer.dts(),
            Some(7500.mseconds() + buffer_idx * 500.mseconds())
        );
        assert_eq!(buffer.duration(), Some(500.mseconds()));
    }

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_chunking_single_stream_manual_fragment() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    let mut h = gst_check::Harness::new("cmafmux");

    // fragment duration long enough to be ignored, 1s chunk duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.hours());
    h.element()
        .unwrap()
        .set_property("chunk-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    // request fragment at 4 seconds, should be created at 11th buffer
    h.element()
        .unwrap()
        .emit_by_name::<()>("split-at-running-time", &[&4.seconds()]);

    // Push 15 buffers of 0.5s each, 1st and 11th buffer without DELTA_UNIT flag
    for i in 0..15 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
            if i != 0 && i != 10 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(4.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO));

    // There should be 7 chunks now, and the 1st and 6th are starting a fragment.
    // Each chunk should have two buffers.
    for chunk in 0..7 {
        let chunk_header = h.pull().unwrap();
        if chunk == 0 || chunk == 5 {
            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
        } else {
            assert_eq!(
                chunk_header.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
            );
        }
        assert_eq!(chunk_header.pts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.dts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.duration(), Some(1.seconds()));

        for buffer_idx in 0..2 {
            let buffer = h.pull().unwrap();
            if buffer_idx == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(
                buffer.pts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(
                buffer.dts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(buffer.duration(), Some(500.mseconds()));
        }
    }

    h.push_event(gst::event::Eos::new());

    // There should be the remaining chunk now, containing one 500ms buffer.
    for chunk in 7..8 {
        let chunk_header = h.pull().unwrap();
        assert_eq!(
            chunk_header.flags(),
            gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
        );
        assert_eq!(chunk_header.pts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.dts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.duration(), Some(500.mseconds()));

        for buffer_idx in 0..1 {
            let buffer = h.pull().unwrap();
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
            assert_eq!(
                buffer.pts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(
                buffer.dts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(buffer.duration(), Some(500.mseconds()));
        }
    }

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_chunking_single_stream() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    let mut h = gst_check::Harness::new("cmafmux");

    // 5s fragment duration, 1s chunk duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());
    h.element()
        .unwrap()
        .set_property("chunk-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    // Push 15 buffers of 0.5s each, 1st and 11th buffer without DELTA_UNIT flag
    for i in 0..15 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
            if i != 0 && i != 10 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO));

    // There should be 7 chunks now, and the 1st and 6th are starting a fragment.
    // Each chunk should have two buffers.
    for chunk in 0..7 {
        let chunk_header = h.pull().unwrap();
        if chunk == 0 || chunk == 5 {
            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
        } else {
            assert_eq!(
                chunk_header.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
            );
        }
        assert_eq!(chunk_header.pts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.dts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.duration(), Some(1.seconds()));

        for buffer_idx in 0..2 {
            let buffer = h.pull().unwrap();
            if buffer_idx == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(
                buffer.pts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(
                buffer.dts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(buffer.duration(), Some(500.mseconds()));
        }
    }

    h.push_event(gst::event::Eos::new());

    // There should be the remaining chunk now, containing one 500ms buffer.
    for chunk in 7..8 {
        let chunk_header = h.pull().unwrap();
        assert_eq!(
            chunk_header.flags(),
            gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
        );
        assert_eq!(chunk_header.pts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.dts(), Some(chunk * 1.seconds()));
        assert_eq!(chunk_header.duration(), Some(500.mseconds()));

        for buffer_idx in 0..1 {
            let buffer = h.pull().unwrap();
            assert_eq!(
                buffer.flags(),
                gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
            );
            assert_eq!(
                buffer.pts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(
                buffer.dts(),
                Some((chunk * 2 + buffer_idx) * 500.mseconds())
            );
            assert_eq!(buffer.duration(), Some(500.mseconds()));
        }
    }

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_chunking_multi_stream() {
    init();

    let mut h1 = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));
    let mut h2 = gst_check::Harness::with_element(&h1.element().unwrap(), Some("sink_1"), None);

    // 5s fragment duration, 1s chunk duration
    h1.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());
    h1.element()
        .unwrap()
        .set_property("chunk-duration", 1.seconds());

    h1.set_src_caps(
        gst::Caps::builder("video/x-h264")
            .field("width", 1920i32)
            .field("height", 1080i32)
            .field("framerate", gst::Fraction::new(30, 1))
            .field("stream-format", "avc")
            .field("alignment", "au")
            .field("codec_data", gst::Buffer::with_size(1).unwrap())
            .build(),
    );
    h1.play();

    h2.set_src_caps(
        gst::Caps::builder("audio/mpeg")
            .field("mpegversion", 4i32)
            .field("channels", 1i32)
            .field("rate", 44100i32)
            .field("stream-format", "raw")
            .field("base-profile", "lc")
            .field("profile", "lc")
            .field("level", "2")
            .field(
                "codec_data",
                gst::Buffer::from_slice([0x12, 0x08, 0x56, 0xe5, 0x00]),
            )
            .build(),
    );
    h2.play();

    let output_offset = (60 * 60 * 1000).seconds();

    // Push 15 buffers of 0.5s each, 1st and 11th buffer without DELTA_UNIT flag
    for i in 0..15 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
            if i != 0 && i != 10 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h1.push(buffer), Ok(gst::FlowSuccess::Ok));

        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
        }
        assert_eq!(h2.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h1.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );

            let ev = loop {
                let ev = h2.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h1.crank_single_clock_wait().unwrap();

    let header = h1.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(gst::ClockTime::ZERO + output_offset));
    assert_eq!(header.dts(), Some(gst::ClockTime::ZERO + output_offset));

    // There should be 7 chunks now, and the 1st and 6th are starting a fragment.
    // Each chunk should have two buffers.
    for chunk in 0..7 {
        let chunk_header = h1.pull().unwrap();
        if chunk == 0 || chunk == 5 {
            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
        } else {
            assert_eq!(
                chunk_header.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
            );
        }
        assert_eq!(
            chunk_header.pts(),
            Some(chunk * 1.seconds() + output_offset)
        );
        assert_eq!(
            chunk_header.dts(),
            Some(chunk * 1.seconds() + output_offset)
        );
        assert_eq!(chunk_header.duration(), Some(1.seconds()));

        for buffer_idx in 0..2 {
            for stream_idx in 0..2 {
                let buffer = h1.pull().unwrap();
                if buffer_idx == 1 && stream_idx == 1 {
                    assert_eq!(
                        buffer.flags(),
                        gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                    );
                } else {
                    assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
                }
                assert_eq!(
                    buffer.pts(),
                    Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
                );

                if stream_idx == 0 {
                    assert_eq!(
                        buffer.dts(),
                        Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
                    );
                } else {
                    assert!(buffer.dts().is_none());
                }
                assert_eq!(buffer.duration(), Some(500.mseconds()));
            }
        }
    }

    h1.push_event(gst::event::Eos::new());
    h2.push_event(gst::event::Eos::new());

    // There should be the remaining chunk now, containing one 500ms buffer.
    for chunk in 7..8 {
        let chunk_header = h1.pull().unwrap();
        assert_eq!(
            chunk_header.flags(),
            gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
        );
        assert_eq!(
            chunk_header.pts(),
            Some(chunk * 1.seconds() + output_offset)
        );
        assert_eq!(
            chunk_header.dts(),
            Some(chunk * 1.seconds() + output_offset)
        );
        assert_eq!(chunk_header.duration(), Some(500.mseconds()));

        for buffer_idx in 0..1 {
            for stream_idx in 0..2 {
                let buffer = h1.pull().unwrap();
                if buffer_idx == 0 && stream_idx == 1 {
                    assert_eq!(
                        buffer.flags(),
                        gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                    );
                } else {
                    assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
                }

                assert_eq!(
                    buffer.pts(),
                    Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
                );
                if stream_idx == 0 {
                    assert_eq!(
                        buffer.dts(),
                        Some((chunk * 2 + buffer_idx) * 500.mseconds() + output_offset)
                    );
                } else {
                    assert!(buffer.dts().is_none());
                }
                assert_eq!(buffer.duration(), Some(500.mseconds()));
            }
        }
    }

    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h1.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_chunking_single_stream_gops_after_fragment_end_before_next_chunk_end() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    let mut h = gst_check::Harness::new("cmafmux");

    // 5s fragment duration, 1s chunk duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());
    h.element()
        .unwrap()
        .set_property("chunk-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    // Push 15 buffers of 0.5s each, 1st and 12th buffer without DELTA_UNIT flag
    for i in 0..15 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
            if i != 0 && i != 11 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let mut expected_ts = gst::ClockTime::ZERO;
    let mut num_buffers = 0;

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(expected_ts));
    assert_eq!(header.dts(), Some(expected_ts));

    // There should be 7 chunks now, and the 1st and 7th are starting a fragment.
    // Each chunk should have two buffers except for the 6th.
    for chunk in 0..7 {
        let chunk_header = h.pull().unwrap();
        if chunk == 0 || chunk == 6 {
            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
        } else {
            assert_eq!(
                chunk_header.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
            );
        }
        assert_eq!(chunk_header.pts(), Some(expected_ts));
        assert_eq!(chunk_header.dts(), Some(expected_ts));
        if chunk == 5 {
            assert_eq!(chunk_header.duration(), Some(500.mseconds()));
        } else {
            assert_eq!(chunk_header.duration(), Some(1.seconds()));
        }

        for buffer_idx in 0..2 {
            let buffer = h.pull().unwrap();
            num_buffers += 1;
            if buffer_idx == 1 || (chunk == 5 && buffer_idx == 0) {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(expected_ts));
            assert_eq!(buffer.dts(), Some(expected_ts));
            assert_eq!(buffer.duration(), Some(500.mseconds()));

            expected_ts += 500.mseconds();

            // Only one buffer in this chunk
            if chunk == 5 && buffer_idx == 0 {
                break;
            }
        }
    }

    h.push_event(gst::event::Eos::new());

    // There should be one remaining chunk now, containing two 500ms buffer.
    for _chunk in 7..8 {
        let chunk_header = h.pull().unwrap();
        assert_eq!(
            chunk_header.flags(),
            gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
        );
        assert_eq!(chunk_header.pts(), Some(expected_ts));
        assert_eq!(chunk_header.dts(), Some(expected_ts));
        assert_eq!(chunk_header.duration(), Some(1.seconds()));

        for buffer_idx in 0..2 {
            let buffer = h.pull().unwrap();
            num_buffers += 1;
            if buffer_idx == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(expected_ts));
            assert_eq!(buffer.dts(), Some(expected_ts));
            assert_eq!(buffer.duration(), Some(500.mseconds()));
            expected_ts += 500.mseconds();
        }
    }

    assert_eq!(num_buffers, 15);

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_chunking_single_stream_gops_after_fragment_end_after_next_chunk_end() {
    init();

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    let mut h = gst_check::Harness::new("cmafmux");

    // 5s fragment duration, 1s chunk duration
    h.element()
        .unwrap()
        .set_property("fragment-duration", 5.seconds());
    h.element()
        .unwrap()
        .set_property("chunk-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    // Push 15 buffers of 0.5s each, 1st and 14th buffer without DELTA_UNIT flag
    for i in 0..15 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 500.mseconds());
            buffer.set_dts(i * 500.mseconds());
            buffer.set_duration(500.mseconds());
            if i != 0 && i != 13 {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));

        if i == 2 {
            let ev = loop {
                let ev = h.pull_upstream_event().unwrap();
                if ev.type_() != gst::EventType::Reconfigure
                    && ev.type_() != gst::EventType::Latency
                {
                    break ev;
                }
            };

            assert_eq!(ev.type_(), gst::EventType::CustomUpstream);
            assert_eq!(
                gst_video::UpstreamForceKeyUnitEvent::parse(&ev).unwrap(),
                gst_video::UpstreamForceKeyUnitEvent {
                    running_time: Some(5.seconds()),
                    all_headers: true,
                    count: 0
                }
            );
        }
    }

    // Crank the clock: this should bring us to the end of the first fragment
    h.crank_single_clock_wait().unwrap();

    let mut expected_ts = gst::ClockTime::ZERO;
    let mut num_buffers = 0;

    let header = h.pull().unwrap();
    assert_eq!(
        header.flags(),
        gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
    );
    assert_eq!(header.pts(), Some(expected_ts));
    assert_eq!(header.dts(), Some(expected_ts));

    // There should be 7 chunks now, and the 1st is starting a fragment.
    // Each chunk should have two buffers except for the 7th.
    for chunk in 0..7 {
        let chunk_header = h.pull().unwrap();
        if chunk == 0 {
            assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
        } else {
            assert_eq!(
                chunk_header.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
            );
        }
        assert_eq!(chunk_header.pts(), Some(expected_ts));
        assert_eq!(chunk_header.dts(), Some(expected_ts));
        if chunk == 6 {
            assert_eq!(chunk_header.duration(), Some(500.mseconds()));
        } else {
            assert_eq!(chunk_header.duration(), Some(1.seconds()));
        }

        for buffer_idx in 0..2 {
            let buffer = h.pull().unwrap();
            num_buffers += 1;
            if buffer_idx == 1 || (chunk == 6 && buffer_idx == 0) {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(expected_ts));
            assert_eq!(buffer.dts(), Some(expected_ts));
            assert_eq!(buffer.duration(), Some(500.mseconds()));

            expected_ts += 500.mseconds();

            // Only one buffer in this chunk
            if chunk == 6 && buffer_idx == 0 {
                break;
            }
        }
    }

    h.push_event(gst::event::Eos::new());

    // There should be two remaining chunks now, containing two 500ms buffers.
    // This should start a new fragment.
    for _chunk in 7..8 {
        let chunk_header = h.pull().unwrap();
        assert_eq!(chunk_header.flags(), gst::BufferFlags::HEADER);
        assert_eq!(chunk_header.pts(), Some(expected_ts));
        assert_eq!(chunk_header.dts(), Some(expected_ts));
        assert_eq!(chunk_header.duration(), Some(1.seconds()));

        for buffer_idx in 0..2 {
            let buffer = h.pull().unwrap();
            num_buffers += 1;
            if buffer_idx == 1 {
                assert_eq!(
                    buffer.flags(),
                    gst::BufferFlags::DELTA_UNIT | gst::BufferFlags::MARKER
                );
            } else {
                assert_eq!(buffer.flags(), gst::BufferFlags::DELTA_UNIT);
            }
            assert_eq!(buffer.pts(), Some(expected_ts));
            assert_eq!(buffer.dts(), Some(expected_ts));
            assert_eq!(buffer.duration(), Some(500.mseconds()));
            expected_ts += 500.mseconds();
        }
    }

    assert_eq!(num_buffers, 15);

    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::StreamStart);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Caps);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Segment);
    let ev = h.pull_event().unwrap();
    assert_eq!(ev.type_(), gst::EventType::Eos);
}

#[test]
fn test_early_eos() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::with_size(1).unwrap())
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    for i in 0..5 {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * 100.mseconds());
            buffer.set_dts(i * 100.mseconds());
            buffer.set_duration(100.mseconds());

            buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));
    }

    h.push_event(gst::event::Eos::new());
    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_roundtrip_vp9_flac() {
    init();

    let pipeline = gst::parse::launch(
        r#"
        videotestsrc num-buffers=99 ! vp9enc ! vp9parse ! mux.
        audiotestsrc num-buffers=149 ! flacenc ! flacparse ! mux.
        isofmp4mux name=mux ! qtdemux name=demux
        demux.audio_0 ! queue ! flacdec ! fakesink
        demux.video_0 ! queue ! vp9dec ! fakesink
        "#,
    )
    .unwrap();
    let pipeline = pipeline.downcast().unwrap();
    to_completion(&pipeline);
}

#[track_caller]
fn test_caps_changed_verify(
    h: &mut gst_check::Harness,
    num_bufs: usize,
    caps_changed: bool,
    chunk: bool,
) {
    for i in 0..num_bufs {
        let b = h.pull().unwrap();
        // FIXME: Rust 1.71 does not detect that the match is exhaustive so a `_` pattern has to be
        // added, but newer Rust warns (correctly) about that pattern being unreachable.
        #[allow(unreachable_patterns)]
        match (caps_changed, i, chunk) {
            (true, 0, _) => assert_eq!(
                b.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DISCONT
            ),
            (false, 0, false) | (true, 1, false) => assert_eq!(b.flags(), gst::BufferFlags::HEADER),
            (false, 0, true) | (true, 1, true) => assert_eq!(
                b.flags(),
                gst::BufferFlags::HEADER | gst::BufferFlags::DELTA_UNIT
            ),
            (false, 1, _) | (_, 2.., _) => {
                if i == num_bufs - 1 {
                    assert_eq!(
                        b.flags(),
                        gst::BufferFlags::MARKER | gst::BufferFlags::DELTA_UNIT
                    );
                } else {
                    assert_eq!(b.flags(), gst::BufferFlags::DELTA_UNIT);
                }
            }
            _ => unreachable!(),
        }
    }
}

#[track_caller]
fn test_caps_changed_buffers(
    h: &mut gst_check::Harness,
    num_bufs: u64,
    gop_size: u64,
    caps_change: u64,
    duration: u64,
    key_frame_on_caps_change: bool,
    drop_first_buffer: bool,
) {
    for i in 0..num_bufs {
        let mut buffer = gst::Buffer::with_size(1).unwrap();
        {
            let buffer = buffer.get_mut().unwrap();
            buffer.set_pts(i * duration.mseconds());
            buffer.set_dts(i * duration.mseconds());
            buffer.set_duration(duration.mseconds());

            if i % gop_size != 0 && (i != caps_change || !key_frame_on_caps_change) {
                buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
            }
        }

        if i == 0 && drop_first_buffer {
            continue;
        }

        if i == caps_change {
            let caps = gst::Caps::builder("video/x-h264")
                .field("width", 1280i32)
                .field("height", 720i32)
                .field("framerate", gst::Fraction::new(30, 1))
                .field("stream-format", "avc")
                .field("alignment", "au")
                .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
                .build();
            h.push_event(gst::event::Caps::new(&caps));
        }
        assert_eq!(h.push(buffer), Ok(gst::FlowSuccess::Ok));
    }
}

#[test]
fn test_caps_change_at_gop_boundary() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 30, 10, 10, 100, true, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    // Full GOP with HEADER and DISCONT due to caps change
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Full GOP with HEADER but no DISCONT because no caps change
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_at_gop_boundary_compatible() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1280i32)
        .field("height", 720i32)
        .field("framerate", gst::Fraction::new(10, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 30, 10, 10, 100, true, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    // Full GOP with HEADER but no DISCONT because compatible caps
    // change
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Full GOP with HEADER but no DISCONT because no caps change
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_at_gop_boundary_not_allowed() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.element()
        .unwrap()
        .set_property_from_str("header-update-mode", "rewrite");

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 30, 10, 10, 100, true, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    // Full GOP with HEADER but no DISCONT because caps change not
    // allowed from header-update-modex
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Full GOP with HEADER but no DISCONT because no caps change
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_within_gop() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 20, 10, 5, 100, true, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 5, true, false);

    h.crank_single_clock_wait().unwrap();
    // Reduced GOP with HEADER and DISCONT due to caps change
    test_caps_changed_verify(&mut h, 1 + 1 + 5, true, false);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Full GOP with HEADER but no DISCONT because no caps change
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_within_gop_start_without_key() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 20, 10, 5, 100, true, true);

    // Same as test_caps_change_within_gop() but without the first
    // fragment since all frames are dropped due to missing key frame

    h.crank_single_clock_wait().unwrap();
    // Reduced GOP with HEADER and DISCONT due to caps change
    test_caps_changed_verify(&mut h, 1 + 1 + 5, true, false);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Full GOP with HEADER but no DISCONT because no caps change
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_within_gop_chunked() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());
    h.element()
        .unwrap()
        .set_property("chunk-duration", 300.mseconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 22, 10, 5, 30, true, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 5, true, false);

    h.crank_single_clock_wait().unwrap();
    // Fragment with HEADER and DISCONT due to caps change
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    // Reduced chunk due to GOP end inbetween
    test_caps_changed_verify(&mut h, 1 + 5, false, true);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Everything left until EOS
    test_caps_changed_verify(&mut h, 1 + 2, false, true);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_within_gop_no_key() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 22, 10, 5, 100, false, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 5, true, false);

    h.crank_single_clock_wait().unwrap();
    // Reduced GOP with HEADER and DISCONT due to caps change
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    h.push_event(gst::event::Eos::new());
    // Everything left until EOS
    test_caps_changed_verify(&mut h, 1 + 2, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}

#[test]
fn test_caps_change_before_first_frame() {
    init();

    let mut h = gst_check::Harness::with_padnames("isofmp4mux", Some("sink_0"), Some("src"));

    let caps = gst::Caps::builder("video/x-h264")
        .field("width", 1920i32)
        .field("height", 1080i32)
        .field("framerate", gst::Fraction::new(30, 1))
        .field("stream-format", "avc")
        .field("alignment", "au")
        .field("codec_data", gst::Buffer::from_slice([1, 2, 3, 4]))
        .build();

    h.element()
        .unwrap()
        .set_property("fragment-duration", 1.seconds());

    h.set_src_caps(caps);
    h.play();

    test_caps_changed_buffers(&mut h, 22, 10, 0, 100, true, false);

    h.crank_single_clock_wait().unwrap();
    // Initial fragment with HEADER and DISCONT
    test_caps_changed_verify(&mut h, 1 + 1 + 10, true, false);

    h.crank_single_clock_wait().unwrap();
    // 2nd fragment with HEADER
    test_caps_changed_verify(&mut h, 1 + 10, false, false);

    assert_eq!(h.buffers_in_queue(), 0);
}