#[test]
fn runtime_pump_output_turn_drains_bounded_sources_without_vec_staging() {
    let owner = fmp_owner(86);
    let open_key = 75;
    let seal_key = 76;
    let path = live_path(8600);
    let mut driver = DataplaneTurnDriver::new(AdmissionConfig::new(4, 8));
    driver.register_owner(owner, OwnerConfig::new(3, 8).with_next_send_counter(700));
    driver
        .owner_mut(owner)
        .unwrap()
        .set_crypto_keys(OwnerCryptoKeys::new(test_key(open_key), test_key(seal_key)));

    let mut raw_source = VecDeque::from([
        DataplaneRawIngress::from_received(
            PacketProtocol::Fmp,
            path.clone(),
            ReceivedPacket::with_timestamp(
                TransportId::new(6),
                TransportAddr::from_string("198.51.100.10:9000"),
                PacketBuffer::new(fmp_encrypted_wire(86, 1300, 0, b"raw-a", open_key)),
                1,
            ),
        ),
        DataplaneRawIngress::from_received(
            PacketProtocol::Fmp,
            path.clone(),
            ReceivedPacket::with_timestamp(
                TransportId::new(6),
                TransportAddr::from_string("198.51.100.10:9000"),
                PacketBuffer::new(fmp_encrypted_wire(86, 1301, 0, b"raw-b", open_key)),
                2,
            ),
        ),
    ]);

    let mut outbound_source = VecDeque::from([
        OutboundPacket::fmp(owner, 3, PacketClass::Bulk, 860, 0, PacketBuffer::new(b"out-a".to_vec())),
        OutboundPacket::fmp(owner, 3, PacketClass::Bulk, 860, 0, PacketBuffer::new(b"out-b".to_vec())),
    ]);

    let mut router = SimpleIngressRouter {
        owner,
        generation: 3,
        class: PacketClass::Liveness,
        output: OutputTarget::Transport,
    };
    let mut sink = BatchRecordingOutputSink::default();
    let mut completions = VecDeque::<CryptoCompletion>::new();

    let first = pump_aead_output_completion_turn(
        &mut driver,
        AeadOutputCompletionTurn {
            completions: &mut completions,
            completion_limit: 0,
            raw_ingress: &mut raw_source,
            router: &mut router,
            raw_ingress_limit: 1,
            outbound: &mut outbound_source,
            outbound_limit: 1,
            sink: &mut sink,
            crypto_limit: 8,
        },
    );
    assert_eq!(first.summary().raw_ingress_dropped(), 0);
    assert_eq!(first.summary().inbound_admitted(), 1);
    assert_eq!(first.summary().outbound_admitted(), 1);
    assert_eq!(first.summary().dispatched(), 2);
    assert_eq!(first.summary().outputs(), 2);
    assert_eq!(first.summary().outputs_sent(), 2);
    assert!(first.outputs().is_empty());
    assert!(first.output_drops().is_empty());
    assert_eq!(raw_source.len(), 1);
    assert_eq!(outbound_source.len(), 1);
    assert_eq!(sink.batch_calls, 1);

    let second = pump_aead_output_completion_turn(
        &mut driver,
        AeadOutputCompletionTurn {
            completions: &mut completions,
            completion_limit: 0,
            raw_ingress: &mut raw_source,
            router: &mut router,
            raw_ingress_limit: 1,
            outbound: &mut outbound_source,
            outbound_limit: 1,
            sink: &mut sink,
            crypto_limit: 8,
        },
    );
    assert_eq!(second.summary().inbound_admitted(), 1);
    assert_eq!(second.summary().outbound_admitted(), 1);
    assert_eq!(second.summary().outputs_sent(), 2);
    assert!(second.outputs().is_empty());
    assert!(second.output_drops().is_empty());
    assert_eq!(raw_source.len(), 0);
    assert_eq!(outbound_source.len(), 0);
    assert_eq!(sink.batch_calls, 2);
    assert_eq!(
        sink.outputs
            .iter()
            .map(PacketOutput::counter)
            .collect::<Vec<_>>(),
        vec![1300, 700, 1301, 701]
    );
    assert_eq!(
        sink.outputs
            .iter()
            .map(PacketOutput::target)
            .collect::<Vec<_>>(),
        vec![
            OutputTarget::Transport,
            OutputTarget::Transport,
            OutputTarget::Transport,
            OutputTarget::Transport,
        ]
    );
    assert_eq!(
        sink.outputs
            .iter()
            .map(|output| output.path.clone())
            .collect::<Vec<_>>(),
        vec![None, Some(path.clone()), None, Some(path)]
    );
    assert_eq!(open_sealed_output(&sink.outputs[1], seal_key), b"out-a");
    assert_eq!(open_sealed_output(&sink.outputs[3], seal_key), b"out-b");
}
