Achieve 100% test coverage

lib/protocol/grpc/body/readable.rb

@@ -7,6 +7,9 @@
 require "protocol/http/body/wrapper"
 require "zlib"
 
+require_relative "../error"
+require_relative "../status"
+
 module Protocol
 	module GRPC
 		# @namespace
@@ -47,8 +50,6 @@ def initialize(body, message_class: nil, encoding: nil)
 				# Overrides Wrapper#read to transform raw HTTP body chunks into decoded gRPC messages.
 				# @returns [Object | String | Nil] Decoded message, raw binary, or `Nil` if stream ended
 				def read
-					return nil if @body.nil? || @body.empty?
-
 					# Read 5-byte prefix: 1 byte compression flag + 4 bytes length
 					prefix = read_exactly(5)
 					return nil unless prefix
@@ -58,7 +59,9 @@ def read
 
 					# Read the message body:
 					data = read_exactly(length)
-					return nil unless data
+					unless data
+						raise Error.new(Status::INTERNAL, "Truncated gRPC frame: expected #{length} bytes, received 0")
+					end
 
 					# Decompress if needed:
 					data = decompress(data) if compressed
@@ -76,19 +79,25 @@ def read
 			private
 
 				# Read exactly n bytes from the underlying body.
-				# @parameter n [Integer] The number of bytes to read
-				# @returns [String | Nil] The data read, or `Nil` if the stream ended
+				# @parameter n [Integer] The number of bytes to read.
+				# @returns [String | Nil] The data read, or `Nil` if the stream ended before reading any bytes.
+				# @raises [Error] If the stream ends after reading a partial value.
 				def read_exactly(n)
 					# Fill buffer until we have enough data:
 					while @buffer.bytesize < n
-						return nil if @body.nil? || @body.empty?
+						if @body.nil? || @body.empty?
+							return nil if @buffer.empty?
+
+							raise Error.new(Status::INTERNAL, "Truncated gRPC frame: expected #{n} bytes, received #{@buffer.bytesize}")
+						end
 
 						# Read chunk from underlying body:
 						chunk = @body.read
 
 						if chunk.nil?
-							# End of stream:
-							return nil
+							return nil if @buffer.empty?
+
+							raise Error.new(Status::INTERNAL, "Truncated gRPC frame: expected #{n} bytes, received #{@buffer.bytesize}")
 						end
 
 						# Append to buffer:
@@ -122,8 +131,10 @@ def decompress(data)
 						inflater.close
 						result
 					else
-						data
+						raise Error.new(Status::UNIMPLEMENTED, "Unsupported compression encoding: #{@encoding.inspect}")
 					end
+				rescue Error
+					raise
 				rescue StandardError => error
 					raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}")
 				end

lib/protocol/grpc/body/writable.rb

@@ -8,6 +8,9 @@
 require "zlib"
 require "stringio"
 
+require_relative "../error"
+require_relative "../status"
+
 module Protocol
 	module GRPC
 		# @namespace
@@ -97,7 +100,7 @@ def compress(data)
 						# This matches HTTP's "deflate" content-encoding
 						Zlib::Deflate.deflate(data, @level)
 					else
-						data # No compression or identity
+						raise ArgumentError, "Unsupported compression encoding: #{@encoding.inspect}"
 					end
 				rescue StandardError => error
 					raise Error.new(Status::INTERNAL, "Failed to compress message: #{error.message}")

lib/protocol/grpc/metadata.rb

@@ -3,7 +3,6 @@
 # Released under the MIT License.
 # Copyright, 2025-2026, by Samuel Williams.
 
-require "uri"
 require_relative "header"
 require_relative "status"
 
@@ -12,7 +11,6 @@ module GRPC
 		# @namespace
 		module Metadata
 			# Extract gRPC status from headers.
-			# Convenience method that handles both Header::Status instances and raw values.
 			# Returns Status::UNKNOWN if status is not present.
 			#
 			# Note: In Protocol::HTTP::Headers, trailers are merged into the headers
@@ -29,31 +27,10 @@ def self.extract_status(headers)
 				status = headers["grpc-status"]
 				return Status::UNKNOWN unless status
 
-				if status.is_a?(Header::Status)
-					status.to_i
-				else
-					# Fallback for when header policy isn't used
-					# Handle Array case (may occur with external clients)
-					status_value = if status.is_a?(Array)
-						# Flatten and take first non-nil value, recursively handle nested arrays
-						flattened = status.flatten.compact.first
-						# If still an array, take first element
-						flattened.is_a?(Array) ? flattened.first : flattened
-					else
-						status
-					end
-
-					# Convert to string then integer to handle various types
-					# Handle case where status_value might still be an array somehow
-					if status_value.is_a?(Array)
-						status_value = status_value.first
-					end
-					status_value.to_s.to_i
-				end
+				return status.to_i
 			end
 
 			# Extract gRPC status message from headers.
-			# Convenience method that handles both Header::Message instances and raw values.
 			# Returns `Nil` if message is not present.
 			#
 			# @parameter headers [Protocol::HTTP::Headers]
@@ -66,13 +43,7 @@ def self.extract_message(headers)
 				message = headers["grpc-message"]
 				return nil unless message
 
-				if message.is_a?(Header::Message)
-					message.decode
-				else
-					# Fallback for when header policy isn't used
-					message_value = message.is_a?(Array) ? message.first : message.to_s
-					URI.decode_www_form_component(message_value)
-				end
+				return message.decode
 			end
 
 			# Assign gRPC status, message, and optional backtrace to headers.

test/protocol/grpc/body/readable.rb

@@ -7,6 +7,8 @@
 require "protocol/http/body/buffered"
 require_relative "../../../../fixtures/protocol/grpc/test_message"
 
+require "zlib"
+
 describe Protocol::GRPC::Body::Readable do
 	let(:message_class) {Protocol::GRPC::Fixtures::TestMessage}
 	let(:source_body) {Protocol::HTTP::Body::Buffered.new}
@@ -19,6 +21,28 @@ def write_message(message, compressed: false)
 		source_body.write(prefix + data)
 	end
 
+	def write_data(data, compressed: false)
+		compression_flag = compressed ? 1 : 0
+		prefix = [compression_flag].pack("C") + [data.bytesize].pack("N")
+		source_body.write(prefix + data)
+	end
+
+	with ".wrap" do
+		it "wraps a message body" do
+			message = Struct.new(:body).new(source_body)
+			wrapped_body = subject.wrap(message, message_class: message_class)
+
+			expect(wrapped_body).to be_a(subject)
+			expect(message.body).to be_equal(wrapped_body)
+		end
+
+		it "returns nil when the message has no body" do
+			message = Struct.new(:body).new(nil)
+
+			expect(subject.wrap(message)).to be_nil
+		end
+	end
+
 	it "has body attribute" do
 		expect(body.body).to be == source_body
 	end
@@ -70,6 +94,62 @@ def write_message(message, compressed: false)
 			read_message = body.read
 			expect(read_message).to be == message
 		end
+
+		it "returns nil when the underlying body reports clean EOF" do
+			source_body = Object.new
+			def source_body.empty?
+				false
+			end
+
+			def source_body.read
+				nil
+			end
+
+			body = subject.new(source_body)
+			expect(body.read).to be_nil
+		end
+
+		it "raises an error for a truncated prefix" do
+			source_body.write("\x00\x00".b)
+
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+				expect(error.message).to be =~ /expected 5 bytes, received 2/
+			end
+		end
+
+		it "raises an error when a partial prefix is followed by nil" do
+			chunks = ["\x00".b, nil]
+			source_body = Object.new
+			source_body.define_singleton_method(:empty?){false}
+			source_body.define_singleton_method(:read){chunks.shift}
+			body = subject.new(source_body)
+
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+				expect(error.message).to be =~ /expected 5 bytes, received 1/
+			end
+		end
+
+		it "raises an error for a truncated payload" do
+			write_data("ab")
+			framed_data = source_body.read
+			source_body.write(framed_data.byteslice(0...5) + "a")
+
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+				expect(error.message).to be =~ /expected 2 bytes, received 1/
+			end
+		end
+
+		it "raises an error when the payload is missing" do
+			source_body.write("\x00".b + [2].pack("N"))
+
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+				expect(error.message).to be =~ /expected 2 bytes, received 0/
+			end
+		end
 	end
 
 	with "#each" do
@@ -127,6 +207,34 @@ def write_message(message, compressed: false)
 			read_message = body.read
 			expect(read_message).to be == message
 		end
+
+		it "decompresses deflate messages" do
+			body = subject.new(source_body, message_class: message_class, encoding: "deflate")
+			message = message_class.new(value: "Hello")
+			write_data(Zlib::Deflate.deflate(message.to_proto), compressed: true)
+
+			expect(body.read).to be == message
+		end
+
+		it "rejects unsupported encodings" do
+			body = subject.new(source_body, message_class: message_class, encoding: "custom")
+			message = message_class.new(value: "Hello")
+			write_message(message, compressed: true)
+
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::UNIMPLEMENTED
+				expect(error.message).to be =~ /Unsupported compression encoding: "custom"/
+			end
+		end
+
+		it "raises a gRPC error for invalid compressed data" do
+			body = subject.new(source_body, encoding: "gzip")
+			write_data("invalid", compressed: true)
+
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+			end
+		end
 	end
 
 	with "empty stream" do

test/protocol/grpc/body/writable.rb

@@ -212,6 +212,35 @@ def message.encode
 			compressed = prefix[0].unpack1("C")
 			expect(compressed).to be == 0
 		end
+
+		it "compresses messages using deflate" do
+			body = subject.new(encoding: "deflate")
+			message = message_class.new(value: "Hello")
+			body.write(message)
+			body.close_write
+
+			framed_data = body.join
+			expect(framed_data.getbyte(0)).to be == 1
+			expect(Zlib::Inflate.inflate(framed_data.byteslice(5..))).to be == message.to_proto
+		end
+
+		it "rejects unsupported encodings" do
+			body = subject.new(encoding: "custom")
+			message = message_class.new(value: "Hello")
+
+			expect{body.write(message)}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+				expect(error.message).to be =~ /Unsupported compression encoding: "custom"/
+			end
+		end
+
+		it "raises a gRPC error when compression fails" do
+			body = subject.new(encoding: "deflate", level: 100)
+
+			expect{body.write("Hello")}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+			end
+		end
 	end
 
 	with "message framing" do

test/protocol/grpc/error.rb

@@ -86,4 +86,43 @@
 			expect(error.status_code).to be == Protocol::GRPC::Status::UNAUTHENTICATED
 		end
 	end
+
+	with ".error_class_for_status" do
+		it "maps status codes to specialized error classes" do
+			expected_classes = {
+				Protocol::GRPC::Status::CANCELLED => Protocol::GRPC::Cancelled,
+				Protocol::GRPC::Status::INVALID_ARGUMENT => Protocol::GRPC::InvalidArgument,
+				Protocol::GRPC::Status::DEADLINE_EXCEEDED => Protocol::GRPC::DeadlineExceeded,
+				Protocol::GRPC::Status::NOT_FOUND => Protocol::GRPC::NotFound,
+				Protocol::GRPC::Status::INTERNAL => Protocol::GRPC::Internal,
+				Protocol::GRPC::Status::UNAVAILABLE => Protocol::GRPC::Unavailable,
+				Protocol::GRPC::Status::UNAUTHENTICATED => Protocol::GRPC::Unauthenticated,
+			}
+
+			expected_classes.each do |status_code, error_class|
+				expect(subject.error_class_for_status(status_code)).to be == error_class
+			end
+		end
+
+		it "uses the base error class for other status codes" do
+			expect(subject.error_class_for_status(Protocol::GRPC::Status::UNKNOWN)).to be == subject
+		end
+	end
+
+	with ".for" do
+		it "creates a specialized error" do
+			error = subject.for(Protocol::GRPC::Status::NOT_FOUND, "Missing", metadata: {"key" => "value"})
+
+			expect(error).to be_a(Protocol::GRPC::NotFound)
+			expect(error.message).to be == "Missing"
+			expect(error.metadata).to be == {"key" => "value"}
+		end
+
+		it "creates a base error for an unmapped status" do
+			error = subject.for(Protocol::GRPC::Status::UNKNOWN)
+
+			expect(error.class).to be == subject
+			expect(error.status_code).to be == Protocol::GRPC::Status::UNKNOWN
+		end
+	end
 end