praxis-proxy-filter 0.5.6

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024 Praxis Contributors

//! Body capabilities computation for filter pipelines.
//!
//! Scans every filter's [`BodyAccess`] and [`BodyMode`] declarations to
//! produce a single [`BodyCapabilities`] that the handler layer uses to
//! decide whether to enable body filter hooks. Mode merging picks the
//! most demanding mode (`StreamBuffer` > `SizeLimit` > `Stream`) and
//! keeps the largest buffer limit so every filter gets enough data.
//!
//! Called once at pipeline build time by [`FilterPipeline::from_filters`].
//!
//! [`BodyAccess`]: crate::body::BodyAccess
//! [`BodyMode`]: crate::body::BodyMode
//! [`BodyCapabilities`]: crate::body::BodyCapabilities
//! [`FilterPipeline::from_filters`]: super::FilterPipeline

use praxis_core::config::{ABSOLUTE_MAX_BODY_BYTES, ResponseCondition};

use super::filter::PipelineFilter;
use crate::{
    any_filter::AnyFilter,
    body::{BodyAccess, BodyCapabilities, BodyMode},
};

// -----------------------------------------------------------------------------
// Body Mode Merging
// -----------------------------------------------------------------------------

/// Merge two optional size limits, keeping the largest value.
///
/// `None` represents unbounded buffering and is treated as larger
/// than any finite limit. When both sides are `Some`, the larger
/// value wins so that every filter in the pipeline gets enough
/// buffer to do its job. The pipeline-level body ceiling (applied
/// separately via [`apply_body_limits`]) remains the hard safety cap.
///
/// [`apply_body_limits`]: super::FilterPipeline::apply_body_limits
pub(super) fn merge_optional_limits(a: Option<usize>, b: Option<usize>) -> Option<usize> {
    match (a, b) {
        (Some(x), Some(y)) => Some(x.max(y)),
        (None, _) | (_, None) => None,
        // unreachable, but spelled out for clarity — both None is still None
    }
}

/// Merge a filter's body mode into the current accumulated mode.
///
/// Precedence: `StreamBuffer` > `SizeLimit` > `Stream`.
/// When two `StreamBuffer` modes merge, the **largest** limit wins
/// so that every filter gets enough buffer to do its job. The
/// pipeline-level body ceiling is applied separately and acts as the
/// hard safety cap.
pub(crate) fn merge_body_mode(current: &mut BodyMode, filter_mode: BodyMode) {
    match filter_mode {
        BodyMode::StreamBuffer { max_bytes } => {
            *current = match *current {
                BodyMode::Stream | BodyMode::SizeLimit { .. } => BodyMode::StreamBuffer { max_bytes },
                BodyMode::StreamBuffer { max_bytes: existing } => BodyMode::StreamBuffer {
                    max_bytes: merge_optional_limits(existing, max_bytes),
                },
            };
        },
        BodyMode::SizeLimit { .. } | BodyMode::Stream => {},
    }
}

// -----------------------------------------------------------------------------
// Body Capabilities
// -----------------------------------------------------------------------------

/// Merge all filters' body access declarations into a single capability set.
pub(super) fn compute_body_capabilities(filters: &[PipelineFilter]) -> BodyCapabilities {
    let mut caps = BodyCapabilities::default();
    accumulate_caps(&mut caps, filters);
    caps
}

/// Precompute the pipeline indices of filters that declared body access.
pub(super) fn body_filter_indices(filters: &[PipelineFilter]) -> (Vec<usize>, Vec<usize>) {
    let mut request = Vec::new();
    let mut response = Vec::new();
    for (idx, pf) in filters.iter().enumerate() {
        if let AnyFilter::Http(f) = &pf.filter {
            if f.request_body_access() != BodyAccess::None {
                request.push(idx);
            }
            if f.response_body_access() != BodyAccess::None {
                response.push(idx);
            }
        }
    }
    (request, response)
}

/// Precompute the pipeline indices of filters that declared
/// selected-upstream request-body access.
///
/// Top-level only: branch filters never run body hooks, so a
/// selected-upstream declaration inside a branch is rejected at build
/// time rather than collected here.
pub(super) fn selected_upstream_request_body_indices(filters: &[PipelineFilter]) -> Vec<usize> {
    let mut indices = Vec::new();
    for (idx, pf) in filters.iter().enumerate() {
        if let AnyFilter::Http(f) = &pf.filter
            && f.selected_upstream_request_body_access() != BodyAccess::None
        {
            indices.push(idx);
        }
    }
    indices
}

/// Recursively accumulate body capabilities from a slice of pipeline filters.
pub(super) fn accumulate_caps(caps: &mut BodyCapabilities, filters: &[PipelineFilter]) {
    accumulate_caps_inner(caps, filters, false);
}

/// Recursive worker for [`accumulate_caps`].
///
/// Branch sub-chains only run `on_request`: body hooks never execute for
/// filters inside branches, so their body access declarations must not
/// enable pipeline-wide buffering (`in_branch` skips body accumulation).
/// Request-context needs still accumulate because `on_request` runs.
fn accumulate_caps_inner(caps: &mut BodyCapabilities, filters: &[PipelineFilter], in_branch: bool) {
    for pf in filters {
        let http_filter = match &pf.filter {
            AnyFilter::Http(f) => f.as_ref(),
            AnyFilter::Tcp(_) => continue,
        };

        if !in_branch {
            accumulate_request_body(caps, http_filter);
            accumulate_selected_upstream_request_body(caps, http_filter);
            accumulate_response_body(caps, http_filter, &pf.response_conditions);
            if !caps.any_response_condition_uses_headers {
                caps.any_response_condition_uses_headers = resp_conditions_use_headers(&pf.response_conditions);
            }
        }

        if http_filter.needs_request_context() {
            caps.needs_request_context = true;
        }

        for branch in &pf.branches {
            accumulate_caps_inner(caps, &branch.filters, true);
        }
    }
}

/// Accumulate request body capabilities from a single filter.
fn accumulate_request_body(caps: &mut BodyCapabilities, filter: &dyn crate::filter::HttpFilter) {
    let access = filter.request_body_access();
    if access != BodyAccess::None {
        caps.needs_request_body = true;
        if access == BodyAccess::ReadWrite {
            caps.any_request_body_writer = true;
        }
        merge_body_mode(&mut caps.request_body_mode, filter.request_body_mode());
    }
}

/// Accumulate selected-upstream request body capabilities from a single filter.
///
/// A participating filter contributes to the *global* request-body
/// capabilities (`needs_request_body`, the request-body writer flag, and
/// the effective `request_body_mode`) because the selected-upstream phase
/// operates on the same buffered request body: the pipeline must buffer it
/// (a bounded `StreamBuffer`) for the phase to have anything to run on. It
/// also sets selected-specific flags so downstream layers can skip the
/// phase entirely when no filter participates.
///
/// The contributed mode is promoted defensively: a participant that
/// declared a non-buffering `request_body_mode` (rejected by validation,
/// but not relied on here) still forces a bounded `StreamBuffer` capped at
/// [`ABSOLUTE_MAX_BODY_BYTES`] so the phase never runs against an
/// unbuffered or unbounded body.
///
/// [`ABSOLUTE_MAX_BODY_BYTES`]: praxis_core::config::ABSOLUTE_MAX_BODY_BYTES
fn accumulate_selected_upstream_request_body(caps: &mut BodyCapabilities, filter: &dyn crate::filter::HttpFilter) {
    let access = filter.selected_upstream_request_body_access();
    if access == BodyAccess::None {
        return;
    }
    caps.needs_selected_upstream_request_body = true;
    caps.needs_request_body = true;
    if access == BodyAccess::ReadWrite {
        caps.any_selected_upstream_request_body_writer = true;
        caps.any_request_body_writer = true;
    }
    merge_body_mode(&mut caps.request_body_mode, selected_upstream_body_mode(filter));
}

/// The buffered body mode a selected-upstream participant contributes.
///
/// The phase needs the complete body, so any declaration that is not
/// already a bounded `StreamBuffer` is promoted to one capped at the
/// absolute ceiling. Validation rejects such declarations, but the
/// capability computation stays safe on its own: it never yields an
/// unbuffered or unbounded mode for a selected-upstream participant.
fn selected_upstream_body_mode(filter: &dyn crate::filter::HttpFilter) -> BodyMode {
    match filter.request_body_mode() {
        BodyMode::StreamBuffer { max_bytes: Some(limit) } => BodyMode::StreamBuffer { max_bytes: Some(limit) },
        _ => BodyMode::StreamBuffer {
            max_bytes: Some(ABSOLUTE_MAX_BODY_BYTES),
        },
    }
}

/// Accumulate response body capabilities from a single filter.
fn accumulate_response_body(
    caps: &mut BodyCapabilities,
    filter: &dyn crate::filter::HttpFilter,
    response_conditions: &[ResponseCondition],
) {
    let access = filter.response_body_access();
    if access != BodyAccess::None {
        caps.needs_response_body = true;
        if !response_conditions.is_empty() {
            caps.any_response_body_condition = true;
        }
        if access == BodyAccess::ReadWrite {
            caps.any_response_body_writer = true;
        }
        merge_body_mode(&mut caps.response_body_mode, filter.response_body_mode());
    }
}

/// Check whether any response condition references headers.
fn resp_conditions_use_headers(conditions: &[ResponseCondition]) -> bool {
    conditions.iter().any(|c| {
        let m = match c {
            ResponseCondition::When(m) | ResponseCondition::Unless(m) => m,
        };
        m.headers.is_some()
    })
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    clippy::too_many_lines,
    reason = "tests"
)]
mod tests {
    use std::collections::HashMap;

    use praxis_core::config::{FailureMode, ResponseConditionMatch};

    use super::*;

    #[test]
    fn merge_body_mode_stream_buffer_wins_over_stream() {
        let mut mode = BodyMode::Stream;
        merge_body_mode(&mut mode, BodyMode::StreamBuffer { max_bytes: Some(1024) });
        assert_eq!(
            mode,
            BodyMode::StreamBuffer { max_bytes: Some(1024) },
            "StreamBuffer should replace Stream"
        );
    }

    #[test]
    fn merge_body_mode_stream_buffer_wins_over_size_limit() {
        let mut mode = BodyMode::SizeLimit { max_bytes: 4096 };
        merge_body_mode(&mut mode, BodyMode::StreamBuffer { max_bytes: Some(2048) });
        assert_eq!(
            mode,
            BodyMode::StreamBuffer { max_bytes: Some(2048) },
            "StreamBuffer should replace SizeLimit"
        );
    }

    #[test]
    fn merge_body_mode_size_limit_is_noop() {
        let mut mode = BodyMode::Stream;
        merge_body_mode(&mut mode, BodyMode::SizeLimit { max_bytes: 4096 });
        assert_eq!(
            mode,
            BodyMode::Stream,
            "SizeLimit should not change Stream (treated as noop in merge)"
        );
    }

    #[test]
    fn merge_body_mode_stream_buffer_merges_limits() {
        let mut mode = BodyMode::StreamBuffer { max_bytes: Some(2048) };
        merge_body_mode(&mut mode, BodyMode::StreamBuffer { max_bytes: Some(1024) });
        assert_eq!(
            mode,
            BodyMode::StreamBuffer { max_bytes: Some(2048) },
            "larger StreamBuffer limit should win"
        );
    }

    #[test]
    fn merge_body_mode_stream_buffer_none_with_some() {
        let mut mode = BodyMode::StreamBuffer { max_bytes: None };
        merge_body_mode(&mut mode, BodyMode::StreamBuffer { max_bytes: Some(1024) });
        assert_eq!(
            mode,
            BodyMode::StreamBuffer { max_bytes: None },
            "None (unbounded) should win over Some"
        );
    }

    #[test]
    fn merge_body_mode_stream_is_noop() {
        let mut mode = BodyMode::StreamBuffer { max_bytes: Some(1024) };
        merge_body_mode(&mut mode, BodyMode::Stream);
        assert_eq!(
            mode,
            BodyMode::StreamBuffer { max_bytes: Some(1024) },
            "Stream should not change existing mode"
        );
    }

    #[test]
    fn merge_optional_limits_both_some_picks_larger() {
        assert_eq!(
            merge_optional_limits(Some(100), Some(50)),
            Some(100),
            "should pick larger of two Some values"
        );
    }

    #[test]
    fn merge_optional_limits_one_none() {
        assert_eq!(
            merge_optional_limits(Some(100), None),
            None,
            "None (unbounded) should win over Some (left)"
        );
        assert_eq!(
            merge_optional_limits(None, Some(200)),
            None,
            "None (unbounded) should win over Some (right)"
        );
    }

    #[test]
    fn merge_optional_limits_both_none() {
        assert_eq!(merge_optional_limits(None, None), None, "both None should yield None");
    }

    #[test]
    fn resp_conditions_use_headers_true_when_headers_present() {
        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
            status: None,
            headers: Some(HashMap::from([("x-key".to_owned(), "val".to_owned())])),
        })];
        assert!(
            resp_conditions_use_headers(&conds),
            "should return true when a condition has headers"
        );
    }

    #[test]
    fn resp_conditions_use_headers_false_when_status_only() {
        let conds = vec![ResponseCondition::When(ResponseConditionMatch {
            status: Some(vec![200]),
            headers: None,
        })];
        assert!(
            !resp_conditions_use_headers(&conds),
            "should return false when conditions only use status"
        );
    }

    #[test]
    fn resp_conditions_use_headers_false_when_empty() {
        assert!(
            !resp_conditions_use_headers(&[]),
            "should return false when no conditions"
        );
    }

    #[test]
    fn resp_conditions_use_headers_unless_variant() {
        let conds = vec![ResponseCondition::Unless(ResponseConditionMatch {
            status: None,
            headers: Some(HashMap::from([("x-skip".to_owned(), "yes".to_owned())])),
        })];
        assert!(
            resp_conditions_use_headers(&conds),
            "should return true for Unless variant with headers"
        );
    }

    #[test]
    fn body_caps_marks_response_body_conditions_for_status_only() {
        use crate::{FilterAction, FilterError};

        /// Minimal response-body filter for capability tests.
        struct ResponseBodyFilter;

        #[async_trait::async_trait]
        impl crate::HttpFilter for ResponseBodyFilter {
            fn name(&self) -> &'static str {
                "response_body"
            }

            async fn on_request(&self, _ctx: &mut crate::HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
                Ok(FilterAction::Continue)
            }

            fn response_body_access(&self) -> BodyAccess {
                BodyAccess::ReadOnly
            }
        }

        let conditions = vec![ResponseCondition::When(ResponseConditionMatch {
            status: Some(vec![200]),
            headers: None,
        })];
        let filter = PipelineFilter::new(0, AnyFilter::Http(Box::new(ResponseBodyFilter)), vec![], conditions);
        let caps = compute_body_capabilities(&[filter]);

        assert!(
            caps.any_response_body_condition,
            "status-only response body conditions should require response header snapshots"
        );
        assert!(
            !caps.any_response_condition_uses_headers,
            "status-only conditions should not set the header-specific flag"
        );
    }

    #[test]
    fn body_caps_ignore_branch_body_filters() {
        use std::sync::Arc;

        use async_trait::async_trait;
        use bytes::Bytes;

        use crate::{
            FilterAction, FilterError,
            filter::HttpFilter,
            pipeline::branch::{RejoinTarget, ResolvedBranch},
        };

        struct BranchBodyFilter;

        #[async_trait]
        impl HttpFilter for BranchBodyFilter {
            fn name(&self) -> &'static str {
                "branch_body"
            }

            async fn on_request(&self, _ctx: &mut crate::HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
                Ok(FilterAction::Continue)
            }

            fn request_body_access(&self) -> BodyAccess {
                BodyAccess::ReadWrite
            }

            fn request_body_mode(&self) -> BodyMode {
                BodyMode::StreamBuffer { max_bytes: Some(4096) }
            }

            async fn on_request_body(
                &self,
                _ctx: &mut crate::HttpFilterContext<'_>,
                _body: &mut Option<Bytes>,
                _eos: bool,
            ) -> Result<FilterAction, FilterError> {
                Ok(FilterAction::Continue)
            }
        }

        let branch_filter = PipelineFilter {
            filter_id: 100,
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(BranchBodyFilter)),
            name: None,
            response_conditions: vec![],
        };
        let branch = ResolvedBranch {
            condition: None,
            filters: vec![branch_filter],
            max_iterations: None,
            name: Arc::from("body_branch"),
            rejoin: RejoinTarget::Next,
        };
        let parent = PipelineFilter {
            filter_id: 0,
            is_security: false,
            branches: vec![branch],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NoopHttpFilter)),
            name: None,
            response_conditions: vec![],
        };
        let caps = compute_body_capabilities(&[parent]);
        assert!(
            !caps.needs_request_body,
            "branch filters only run on_request; their body access must not enable buffering"
        );
        assert!(
            !caps.any_request_body_writer,
            "ReadWrite filter in branch must not set writer flag"
        );
        assert_eq!(
            caps.request_body_mode,
            BodyMode::Stream,
            "StreamBuffer mode from branch filter must not propagate"
        );
    }

    #[test]
    fn body_caps_branch_filters_still_accumulate_request_context() {
        use std::sync::Arc;

        use crate::pipeline::branch::{RejoinTarget, ResolvedBranch};

        /// Branch filter that needs the request context on `on_request`.
        struct ContextFilter;

        #[async_trait::async_trait]
        impl crate::filter::HttpFilter for ContextFilter {
            fn name(&self) -> &'static str {
                "context_filter"
            }

            async fn on_request(
                &self,
                _ctx: &mut crate::HttpFilterContext<'_>,
            ) -> Result<crate::FilterAction, crate::FilterError> {
                Ok(crate::FilterAction::Continue)
            }

            fn needs_request_context(&self) -> bool {
                true
            }
        }

        let branch = ResolvedBranch {
            condition: None,
            filters: vec![PipelineFilter::new(
                100,
                AnyFilter::Http(Box::new(ContextFilter)),
                vec![],
                vec![],
            )],
            max_iterations: None,
            name: Arc::from("context_branch"),
            rejoin: RejoinTarget::Next,
        };
        let parent = PipelineFilter {
            filter_id: 0,
            is_security: false,
            branches: vec![branch],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NoopHttpFilter)),
            name: None,
            response_conditions: vec![],
        };
        let caps = compute_body_capabilities(&[parent]);
        assert!(
            caps.needs_request_context,
            "branch filters run on_request, so request-context needs must accumulate"
        );
    }

    #[test]
    fn body_caps_no_branch_body_filters_has_no_effect() {
        use std::sync::Arc;

        use crate::pipeline::branch::{RejoinTarget, ResolvedBranch};

        let branch = ResolvedBranch {
            condition: None,
            filters: vec![PipelineFilter::new(
                100,
                AnyFilter::Http(Box::new(NoopHttpFilter)),
                vec![],
                vec![],
            )],
            max_iterations: None,
            name: Arc::from("noop_branch"),
            rejoin: RejoinTarget::Next,
        };
        let parent = PipelineFilter {
            filter_id: 0,
            is_security: false,
            branches: vec![branch],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NoopHttpFilter)),
            name: None,
            response_conditions: vec![],
        };
        let caps = compute_body_capabilities(&[parent]);
        assert!(
            !caps.needs_request_body,
            "branch without body filters should not enable request body"
        );
    }

    #[test]
    fn body_caps_selected_upstream_read_write_participation() {
        let filter = PipelineFilter::new(
            0,
            AnyFilter::Http(Box::new(SelectedUpstreamCapFilter {
                access: BodyAccess::ReadWrite,
                mode: BodyMode::StreamBuffer { max_bytes: Some(4096) },
            })),
            vec![],
            vec![],
        );
        let caps = compute_body_capabilities(&[filter]);

        assert!(
            caps.needs_selected_upstream_request_body,
            "selected participant should set needs_selected_upstream_request_body"
        );
        assert!(
            caps.needs_request_body,
            "selected participant must contribute to the global request-body need"
        );
        assert!(
            caps.any_selected_upstream_request_body_writer,
            "ReadWrite selected participant should set the selected writer flag"
        );
        assert!(
            caps.any_request_body_writer,
            "ReadWrite selected participant must contribute to the global writer flag"
        );
        assert_eq!(
            caps.request_body_mode,
            BodyMode::StreamBuffer { max_bytes: Some(4096) },
            "selected participant's bounded StreamBuffer should drive the effective mode"
        );
    }

    #[test]
    fn body_caps_selected_upstream_read_only_is_not_writer() {
        let filter = PipelineFilter::new(
            0,
            AnyFilter::Http(Box::new(SelectedUpstreamCapFilter {
                access: BodyAccess::ReadOnly,
                mode: BodyMode::StreamBuffer { max_bytes: Some(2048) },
            })),
            vec![],
            vec![],
        );
        let caps = compute_body_capabilities(&[filter]);

        assert!(
            caps.needs_selected_upstream_request_body,
            "read-only selected participant should still need the phase"
        );
        assert!(
            caps.needs_request_body,
            "read-only selected participant needs request body"
        );
        assert!(
            !caps.any_selected_upstream_request_body_writer,
            "read-only selected participant must not set the selected writer flag"
        );
        assert!(
            !caps.any_request_body_writer,
            "read-only selected participant must not set the global writer flag"
        );
    }

    #[test]
    fn body_caps_selected_upstream_defensive_ceiling() {
        let filter = PipelineFilter::new(
            0,
            AnyFilter::Http(Box::new(SelectedUpstreamCapFilter {
                access: BodyAccess::ReadOnly,
                mode: BodyMode::Stream,
            })),
            vec![],
            vec![],
        );
        let caps = compute_body_capabilities(&[filter]);

        assert_eq!(
            caps.request_body_mode,
            BodyMode::StreamBuffer {
                max_bytes: Some(ABSOLUTE_MAX_BODY_BYTES),
            },
            "a non-bounded selected participant must fall back to the finite absolute ceiling"
        );
    }

    #[test]
    fn body_caps_ignore_branch_selected_upstream_filters() {
        use std::sync::Arc;

        use crate::pipeline::branch::{RejoinTarget, ResolvedBranch};

        let branch = ResolvedBranch {
            condition: None,
            filters: vec![PipelineFilter::new(
                100,
                AnyFilter::Http(Box::new(SelectedUpstreamCapFilter {
                    access: BodyAccess::ReadWrite,
                    mode: BodyMode::StreamBuffer { max_bytes: Some(4096) },
                })),
                vec![],
                vec![],
            )],
            max_iterations: None,
            name: Arc::from("selected_branch"),
            rejoin: RejoinTarget::Next,
        };
        let parent = PipelineFilter {
            filter_id: 0,
            is_security: false,
            branches: vec![branch],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NoopHttpFilter)),
            name: None,
            response_conditions: vec![],
        };
        let caps = compute_body_capabilities(&[parent]);

        assert!(
            !caps.needs_selected_upstream_request_body,
            "selected participant inside a branch must not enable the phase"
        );
        assert!(
            !caps.needs_request_body,
            "branch selected participant must not enable request buffering"
        );
        assert_eq!(
            caps.request_body_mode,
            BodyMode::Stream,
            "StreamBuffer mode from a branch selected participant must not propagate"
        );
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    /// Noop HTTP filter for body capability branch testing.
    struct NoopHttpFilter;

    #[async_trait::async_trait]
    impl crate::filter::HttpFilter for NoopHttpFilter {
        fn name(&self) -> &'static str {
            "noop"
        }

        async fn on_request(
            &self,
            _ctx: &mut crate::HttpFilterContext<'_>,
        ) -> Result<crate::FilterAction, crate::FilterError> {
            Ok(crate::FilterAction::Continue)
        }
    }

    /// Selected-upstream request-body participant for capability tests, with
    /// a configurable declared access and body mode.
    struct SelectedUpstreamCapFilter {
        access: BodyAccess,
        mode: BodyMode,
    }

    #[async_trait::async_trait]
    impl crate::filter::HttpFilter for SelectedUpstreamCapFilter {
        fn name(&self) -> &'static str {
            "selected_upstream_cap"
        }

        async fn on_request(
            &self,
            _ctx: &mut crate::HttpFilterContext<'_>,
        ) -> Result<crate::FilterAction, crate::FilterError> {
            Ok(crate::FilterAction::Continue)
        }

        fn selected_upstream_request_body_access(&self) -> BodyAccess {
            self.access
        }

        fn request_body_mode(&self) -> BodyMode {
            self.mode
        }
    }
}