Further improve ext_proc code coverage

Signed-off-by: Ethan Truong <ethantruong@google.com>

Author: etruong42

source/extensions/filters/http/ext_proc/ext_proc.h

@@ -490,6 +490,7 @@ class FilterConfigPerRoute : public Router::RouteSpecificFilterConfig {
 class Filter : public Logger::Loggable<Logger::Id::ext_proc>,
                public Http::PassThroughFilter,
                public ExternalProcessorCallbacks {
+  friend class FilterAccessor;
   // The result of an attempt to open the stream
   enum class StreamOpenState {
     // The stream was opened successfully

test/extensions/filters/http/ext_proc/filter_local_reply_streaming_test.cc

@@ -1329,6 +1329,56 @@ TEST_F(StreamingLocalReplyTest, SaveStreamingLocalResponse) {
   filter_->onDestroy();
 }
 
+TEST_F(StreamingLocalReplyTest, StartLocalResponseInvalidHeaders) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, false));
+
+  auto response = std::make_unique<ProcessingResponse>();
+  auto* headers_response =
+      response->mutable_streamed_immediate_response()->mutable_headers_response();
+  auto* h = headers_response->mutable_headers()->add_headers();
+  h->set_key(":status");
+  h->set_raw_value(std::string(1, '\0')); // Invalid header value
+
+  stream_callbacks_->onReceiveMessage(std::move(response));
+}
+
+TEST_F(StreamingLocalReplyTest, ProcessLocalTrailersInvalid) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, true));
+
+  // Start local response
+  auto response1 = std::make_unique<ProcessingResponse>();
+  auto* headers_response =
+      response1->mutable_streamed_immediate_response()->mutable_headers_response();
+  auto* h1 = headers_response->mutable_headers()->add_headers();
+  h1->set_key(":status");
+  h1->set_raw_value("200");
+  headers_response->set_end_of_stream(false);
+
+  stream_callbacks_->onReceiveMessage(std::move(response1));
+
+  // Send invalid trailers
+  auto response2 = std::make_unique<ProcessingResponse>();
+  auto* trailers_response =
+      response2->mutable_streamed_immediate_response()->mutable_trailers_response();
+  auto* h2 = trailers_response->add_headers();
+  h2->set_key("invalid name");
+  h2->set_raw_value("value");
+
+  stream_callbacks_->onReceiveMessage(std::move(response2));
+}
+
 } // namespace
 } // namespace ExternalProcessing
 } // namespace HttpFilters

test/extensions/filters/http/ext_proc/filter_test.cc

@@ -46,8 +46,17 @@ namespace Envoy {
 namespace Extensions {
 namespace HttpFilters {
 namespace ExternalProcessing {
+// Accessor for private members of Filter to improve coverage
+class FilterAccessor {
+public:
+  static DecodingProcessorState& decodingState(Filter& filter) { return filter.decoding_state_; }
+  static EncodingProcessorState& encodingState(Filter& filter) { return filter.encoding_state_; }
+};
+
 namespace {
 
+using ::testing::ReturnRef;
+
 using ::envoy::extensions::filters::http::ext_proc::v3::ExtProcPerRoute;
 using ::envoy::extensions::filters::http::ext_proc::v3::ProcessingMode;
 using ::envoy::service::ext_proc::v3::BodyResponse;
@@ -106,9 +115,15 @@ class HttpFilterTest : public testing::Test {
         {{"envoy.reloadable_features.ext_proc_inject_data_with_state_update", "true"}});
     scoped_runtime_.mergeValues(
         {{"envoy.reloadable_features.ext_proc_return_stop_iteration", "true"}});
-    client_ = std::make_unique<MockClient>();
+    if (!client_) {
+      client_ = std::make_unique<MockClient>();
+    }
+    client_ptr_ = client_.get();
     route_ = std::make_shared<NiceMock<Router::MockRoute>>();
-    EXPECT_CALL(*client_, start(_, _, _, _)).WillOnce(Invoke(this, &HttpFilterTest::doStart));
+    EXPECT_CALL(*client_ptr_, cancel()).Times(AnyNumber());
+    EXPECT_CALL(*client_ptr_, start(_, _, _, _))
+        .Times(AnyNumber())
+        .WillRepeatedly(Invoke(this, &HttpFilterTest::doStart));
     EXPECT_CALL(encoder_callbacks_, dispatcher()).WillRepeatedly(ReturnRef(dispatcher_));
     EXPECT_CALL(decoder_callbacks_, dispatcher()).WillRepeatedly(ReturnRef(dispatcher_));
     EXPECT_CALL(decoder_callbacks_, route())
@@ -680,6 +695,7 @@ class HttpFilterTest : public testing::Test {
   absl::optional<envoy::config::core::v3::GrpcService> final_expected_grpc_service_;
   Grpc::GrpcServiceConfigWithHashKey config_with_hash_key_;
   std::unique_ptr<MockClient> client_;
+  MockClient* client_ptr_ = nullptr;
   ExternalProcessorCallbacks* stream_callbacks_ = nullptr;
   ProcessingRequest last_request_;
   bool server_closed_stream_ = false;
@@ -6168,6 +6184,242 @@ TEST_F(HttpFilterTest, StreamingSendDataRandomGrpcLatencyReturnContinue) {
   expectNoGrpcCall(envoy::config::core::v3::TrafficDirection::OUTBOUND);
 }
 
+TEST_F(HttpFilterTest, ResponseCaseToStringCoverage) {
+  EXPECT_EQ("request headers",
+            responseCaseToString(ProcessingResponse::ResponseCase::kRequestHeaders));
+  EXPECT_EQ("response headers",
+            responseCaseToString(ProcessingResponse::ResponseCase::kResponseHeaders));
+  EXPECT_EQ("request body", responseCaseToString(ProcessingResponse::ResponseCase::kRequestBody));
+  EXPECT_EQ("response body", responseCaseToString(ProcessingResponse::ResponseCase::kResponseBody));
+  EXPECT_EQ("request trailers",
+            responseCaseToString(ProcessingResponse::ResponseCase::kRequestTrailers));
+  EXPECT_EQ("response trailers",
+            responseCaseToString(ProcessingResponse::ResponseCase::kResponseTrailers));
+  EXPECT_EQ("immediate response",
+            responseCaseToString(ProcessingResponse::ResponseCase::kImmediateResponse));
+  EXPECT_EQ("streamed immediate response",
+            responseCaseToString(ProcessingResponse::ResponseCase::kStreamedImmediateResponse));
+  EXPECT_EQ("unknown", responseCaseToString(ProcessingResponse::ResponseCase::RESPONSE_NOT_SET));
+}
+
+TEST_F(HttpFilterTest, OnErrorAfterProcessingComplete) {
+  initialize("");
+
+  filter_->onDestroy(); // Sets processing_complete_ = true
+
+  // This should hit the if (processing_complete_) block and return early.
+  filter_->onError();
+  EXPECT_EQ(1, config_->stats().http_not_ok_resp_received_.value());
+}
+
+TEST_F(HttpFilterTest, OnTrailersOpenStreamFail) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  processing_mode:
+    request_header_mode: "SEND"
+    request_body_mode: "BUFFERED"
+  )EOF");
+
+  // Setting the client to return null stream will cause openStream to return IgnoreError
+  // We MUST do this after initialize() because initialize() might set its own expectation.
+  EXPECT_CALL(*client_ptr_, start(_, _, _, _))
+      .WillRepeatedly(
+          Invoke([](ExternalProcessorCallbacks&, const Grpc::GrpcServiceConfigWithHashKey&,
+                    const Envoy::Http::AsyncClient::StreamOptions&,
+                    Envoy::Http::StreamFilterSidestreamWatermarkCallbacks&) { return nullptr; }));
+
+  // decodeHeaders calls onHeaders which calls openStream which returns IgnoreError.
+  filter_->decodeHeaders(request_headers_, false);
+
+  // decodeTrailers calls onTrailers, which hits the BUFFERED mode check and calls openStream again.
+  filter_->decodeTrailers(request_trailers_);
+}
+
+TEST_F(HttpFilterTest, CloseGrpcStreamLastRespImmediate) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+
+  auto response = std::make_unique<ProcessingResponse>();
+  response->mutable_immediate_response();
+
+  // Start stream
+  filter_->decodeHeaders(request_headers_, false);
+
+  filter_->onReceiveMessage(std::move(response));
+}
+
+TEST_F(HttpFilterTest, HandleStreamingImmediateResponseDefault) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+
+  auto response = std::make_unique<ProcessingResponse>();
+  response->mutable_streamed_immediate_response(); // response_case is NOT_SET by default inside it
+
+  // Start stream
+  filter_->decodeHeaders(request_headers_, false);
+
+  filter_->onReceiveMessage(std::move(response));
+}
+
+TEST_F(HttpFilterTest, DeferredCloseAlreadyClosed) {
+  initialize("");
+
+  auto stream = std::make_unique<NiceMock<MockStream>>();
+  DeferredDeletableStream wrapper(std::move(stream), config_->threadLocalStreamManager(),
+                                  config_->stats(), std::chrono::milliseconds(100));
+
+  // Null out the stream
+  wrapper.stream_ = nullptr;
+  wrapper.closeStreamOnTimer();
+}
+
+TEST_F(HttpFilterTest, BufferedPartialBodyMutationAndLeftover) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  processing_mode:
+    request_header_mode: "SKIP"
+    request_body_mode: "BUFFERED_PARTIAL"
+  )EOF");
+
+  EXPECT_CALL(decoder_callbacks_, bufferLimit()).WillRepeatedly(Return(10));
+  EXPECT_EQ(FilterHeadersStatus::Continue, filter_->decodeHeaders(request_headers_, false));
+
+  Buffer::OwnedImpl data1("this is more than 10 bytes");
+  EXPECT_EQ(FilterDataStatus::StopIterationNoBuffer, filter_->decodeData(data1, false));
+
+  Buffer::OwnedImpl data2("more data");
+  EXPECT_EQ(FilterDataStatus::StopIterationNoBuffer, filter_->decodeData(data2, false));
+
+  EXPECT_CALL(decoder_callbacks_, injectDecodedDataToFilterChain(_, _)).Times(2);
+
+  processRequestBody([](const envoy::service::ext_proc::v3::HttpBody&, ProcessingResponse&,
+                        BodyResponse& body_resp) {
+    auto* common = body_resp.mutable_response();
+    auto* mut = common->mutable_header_mutation();
+    auto* h = mut->add_set_headers();
+    h->mutable_header()->set_key("x-new-header");
+    h->mutable_header()->set_value("foo");
+  });
+
+  filter_->onDestroy();
+}
+
+TEST_F(HttpFilterTest, ValidateContentLengthSimpleAtoiFail) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  processing_mode:
+    request_header_mode: "SEND"
+    request_body_mode: "BUFFERED"
+  )EOF");
+
+  request_headers_.addCopy(LowerCaseString("content-length"), "invalid");
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, false));
+  processRequestHeaders(true, absl::nullopt);
+
+  Buffer::OwnedImpl req_data("body");
+  Buffer::OwnedImpl buffered_data;
+  setUpDecodingBuffering(buffered_data, true);
+
+  // This will call handleDataBufferedMode
+  EXPECT_EQ(FilterDataStatus::StopIterationNoBuffer, filter_->decodeData(req_data, true));
+
+  EXPECT_TRUE(last_request_.has_request_body());
+
+  processRequestBody([](const HttpBody&, ProcessingResponse&, BodyResponse& body_resp) {
+    body_resp.mutable_response()->mutable_body_mutation()->set_body("new body");
+  });
+
+  filter_->onDestroy();
+}
+
+TEST_F(HttpFilterTest, EncodingWatermark) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  processing_mode:
+    request_header_mode: "SKIP"
+    response_header_mode: "SKIP"
+    response_body_mode: "STREAMED"
+  )EOF");
+
+  EXPECT_EQ(FilterHeadersStatus::Continue, filter_->decodeHeaders(request_headers_, true));
+  EXPECT_EQ(FilterHeadersStatus::Continue, filter_->encodeHeaders(response_headers_, false));
+
+  EXPECT_CALL(encoder_callbacks_, bufferLimit()).WillRepeatedly(Return(10));
+
+  Buffer::OwnedImpl data("this is a long body that exceeds 10 bytes");
+  EXPECT_CALL(encoder_callbacks_, onEncoderFilterAboveWriteBufferHighWatermark());
+  EXPECT_EQ(FilterDataStatus::StopIterationNoBuffer, filter_->encodeData(data, false));
+
+  EXPECT_CALL(encoder_callbacks_, onEncoderFilterBelowWriteBufferLowWatermark());
+  processResponseBody(
+      [](const envoy::service::ext_proc::v3::HttpBody&, ProcessingResponse&, BodyResponse&) {},
+      /*should_continue=*/false);
+
+  filter_->onDestroy();
+}
+
+TEST_F(HttpFilterTest, TrafficDirectionDebugStrOutbound) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, false));
+  EXPECT_EQ("OUTBOUND", FilterAccessor::encodingState(*filter_).trafficDirectionDebugStr());
+  filter_->onDestroy();
+}
+
+TEST_F(HttpFilterTest, DecodingAddTrailers) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, false));
+  EXPECT_CALL(decoder_callbacks_, addDecodedTrailers()).WillOnce(ReturnRef(request_trailers_));
+  auto* trailers = FilterAccessor::decodingState(*filter_).addTrailers();
+  EXPECT_EQ(trailers, &request_trailers_);
+  filter_->onDestroy();
+}
+
+TEST_F(HttpFilterTest, LocalResponseStarted) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, false));
+  EXPECT_FALSE(FilterAccessor::decodingState(*filter_).localResponseStarted());
+  filter_->onDestroy();
+}
+
+TEST_F(HttpFilterTest, EncodingAddTrailers) {
+  initialize(R"EOF(
+  grpc_service:
+    envoy_grpc:
+      cluster_name: "ext_proc_server"
+  )EOF");
+  EXPECT_EQ(FilterHeadersStatus::StopIteration, filter_->decodeHeaders(request_headers_, false));
+  EXPECT_CALL(encoder_callbacks_, addEncodedTrailers()).WillOnce(ReturnRef(response_trailers_));
+  auto* trailers = FilterAccessor::encodingState(*filter_).addTrailers();
+  EXPECT_EQ(trailers, &response_trailers_);
+  filter_->onDestroy();
+}
+
 } // namespace
 } // namespace ExternalProcessing
 } // namespace HttpFilters

test/extensions/filters/http/ext_proc/logging_info_test.cc

@@ -108,6 +108,16 @@ TEST_F(ExtProcLoggingInfoTest, GetField) {
   EXPECT_TRUE(absl::holds_alternative<absl::monostate>(logging_info_.getField("non_existent")));
 }
 
+TEST_F(ExtProcLoggingInfoTest, ProcessingEffectsConst) {
+  const auto& const_logging_info = logging_info_;
+
+  // Exercise both directions
+  EXPECT_EQ(Effect::None,
+            const_logging_info.processingEffects(TrafficDirection::INBOUND).header_effect_);
+  EXPECT_EQ(Effect::None,
+            const_logging_info.processingEffects(TrafficDirection::OUTBOUND).header_effect_);
+}
+
 } // namespace
 } // namespace ExternalProcessing
 } // namespace HttpFilters