Improved test coverage.

Author: samuel-williams-shopify

fixtures/async/container/supervisor/a_server.rb

@@ -37,10 +37,10 @@ def remove(connection)
 				let(:ipc_path) {File.join(@root, "supervisor.ipc")}
 				let(:endpoint) {Async::Container::Supervisor.endpoint(ipc_path)}
 
-				def around(&block)
+				def around
 					Dir.mktmpdir do |directory|
 						@root = directory
-						super(&block)
+						super
 					end
 				end
 

gems.rb

@@ -30,6 +30,7 @@
 	gem "rubocop-socketry"
 
 	gem "sus-fixtures-async"
+	gem "sus-fixtures-console"
 
 	gem "bake-test"
 	gem "bake-test-external"

lib/async/container/supervisor/server.rb

@@ -27,11 +27,9 @@ def do_register(call)
 					call.connection.state.merge!(call.message[:state])
 
 					@monitors.each do |monitor|
-						begin
-							monitor.register(call.connection)
-						rescue => error
-							Console.error(self, "Error while registering process!", monitor: monitor, exception: error)
-						end
+						monitor.register(call.connection)
+					rescue => error
+						Console.error(self, "Error while registering process!", monitor: monitor, exception: error)
 					end
 				ensure
 					call.finish
@@ -59,22 +57,18 @@ def do_status(call)
 
 				def remove(connection)
 					@monitors.each do |monitor|
-						begin
-							monitor.remove(connection)
-						rescue => error
-							Console.error(self, "Error while removing process!", monitor: monitor, exception: error)
-						end
+						monitor.remove(connection)
+					rescue => error
+						Console.error(self, "Error while removing process!", monitor: monitor, exception: error)
 					end
 				end
 
 				def run(parent: Task.current)
 					parent.async do |task|
 						@monitors.each do |monitor|
-							begin
-								monitor.run
-							rescue => error
-								Console.error(self, "Error while starting monitor!", monitor: monitor, exception: error)
-							end
+							monitor.run
+						rescue => error
+							Console.error(self, "Error while starting monitor!", monitor: monitor, exception: error)
 						end
 
 						@endpoint.accept do |peer|

test/async/container/connection.rb

@@ -0,0 +1,142 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "async/container/supervisor/connection"
+require "stringio"
+
+describe Async::Container::Supervisor::Connection do
+	let(:stream) {StringIO.new}
+	let(:connection) {Async::Container::Supervisor::Connection.new(stream)}
+	
+	with subject::Call do
+		let(:test_call) {Async::Container::Supervisor::Connection::Call.new(connection, 1, {do: :test, data: "value"})}
+		
+		it "can serialize call to JSON" do
+			json = test_call.to_json
+			parsed = JSON.parse(json)
+			
+			expect(parsed).to have_keys(
+				"do" => be == "test",
+				"data" => be == "value"
+			)
+		end
+		
+		it "can get call message via as_json" do
+			expect(test_call.as_json).to have_keys(
+				do: be == :test,
+				data: be == "value"
+			)
+		end
+		
+		it "can iterate over call responses with each" do
+			# Push some responses
+			test_call.push(status: "working")
+			test_call.push(status: "done")
+			test_call.close
+			
+			responses = []
+			test_call.each do |response|
+				responses << response
+			end
+			
+			expect(responses.size).to be == 2
+			expect(responses[0]).to have_keys(status: be == "working")
+			expect(responses[1]).to have_keys(status: be == "done")
+		end
+		
+		it "can fail a call" do
+			test_call.fail(error: "Something went wrong")
+			
+			response = test_call.pop
+			expect(response).to have_keys(
+				id: be == 1,
+				finished: be == true,
+				failed: be == true,
+				error: be == "Something went wrong"
+			)
+			
+			expect(test_call.closed?).to be == true
+		end
+		
+		it "can access message via []" do
+			expect(test_call[:do]).to be == :test
+			expect(test_call[:data]).to be == "value"
+		end
+		
+		it "reports closed? correctly" do
+			expect(test_call.closed?).to be == false
+			
+			test_call.finish
+			
+			expect(test_call.closed?).to be == true
+		end
+	end
+	
+	it "writes JSON with newline" do
+		connection.write(id: 1, do: :test)
+		
+		stream.rewind
+		output = stream.read
+		
+		# Check it's valid JSON with a newline
+		expect(output[-1]).to be == "\n"
+		
+		parsed = JSON.parse(output.chomp)
+		expect(parsed).to have_keys(
+			"id" => be == 1,
+			"do" => be == "test"
+		)
+	end
+	
+	it "parses JSON lines" do
+		stream.string = JSON.dump({id: 1, do: "test"}) << "\n"
+		stream.rewind
+		
+		message = connection.read
+		
+		# Connection.read uses symbolize_names: true (keys are symbols, values are as-is)
+		expect(message).to have_keys(
+			id: be == 1,
+			do: be == "test"
+		)
+	end
+	
+	it "returns nil when stream is closed" do
+		stream.string = ""
+		stream.rewind
+		
+		message = connection.read
+		
+		expect(message).to be_nil
+	end
+	
+	it "increments id by 2" do
+		first_id = connection.next_id
+		second_id = connection.next_id
+		
+		expect(second_id).to be == first_id + 2
+	end
+	
+	with "#close" do
+		it "closes all pending calls" do
+			call1 = Async::Container::Supervisor::Connection::Call.new(connection, 1, {do: :test})
+			call2 = Async::Container::Supervisor::Connection::Call.new(connection, 2, {do: :test})
+			
+			connection.calls[1] = call1
+			connection.calls[2] = call2
+			
+			connection.close
+			
+			expect(call1.closed?).to be == true
+			expect(call2.closed?).to be == true
+			expect(connection.calls).to be(:empty?)
+		end
+		
+		it "closes the stream" do
+			connection.close
+			expect(stream).to be(:closed?)
+		end
+	end
+end

test/async/container/server.rb

@@ -4,11 +4,11 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "async/container/supervisor/a_server"
-require "sus/fixtures/console/null_logger"
+require "sus/fixtures/console/captured_logger"
 
 describe Async::Container::Supervisor::Server do
-	include Sus::Fixtures::Console::NullLogger
 	include Async::Container::Supervisor::AServer
+	include Sus::Fixtures::Console::CapturedLogger
 
 	it "can handle unexpected failures" do
 		# First, send invalid JSON to trigger the error:
@@ -36,6 +36,10 @@
 		)
 
 		stream.close
+		
+		# Verify error was logged about the parsing failure
+		error_logs = console_capture.select {|log| log[:severity] == :warn}
+		expect(error_logs).not.to be(:empty?)
 	end
 
 	with "failing monitor" do
@@ -45,9 +49,11 @@ def run
 				end
 
 				def register(connection)
+					raise "Monitor failed to register!"
 				end
 
 				def remove(connection)
+					raise "Monitor failed to remove!"
 				end
 
 				def status(call)
@@ -58,6 +64,30 @@ def status(call)
 
 		let(:monitors) {[failing_monitor]}
 
+		it "can handle monitor registration failures" do
+			# Send a register message:
+			stream = endpoint.connect
+			
+			message = {id: 1, do: :register, state: {process_id: ::Process.pid}}
+			stream.puts(JSON.dump(message))
+			stream.flush
+			
+			# Read the response:
+			response = JSON.parse(stream.gets, symbolize_names: true)
+			
+			# The server should still finish despite the monitor error:
+			expect(response).to have_keys(
+				id: be == 1,
+				finished: be == true
+			)
+			
+			# Verify error was logged about the monitor failure:
+			error_log = console_capture.find {|log| log[:severity] == :error && log[:message] =~ /Error while registering process/}
+			expect(error_log).to be_truthy
+			
+			stream.close
+		end
+		
 		it "can handle monitor status failures" do
 			# Send a status request:
 			stream = endpoint.connect
@@ -71,14 +101,41 @@ def status(call)
 
 			# The server should still respond with a finished message despite the monitor error:
 			expect(response).to have_keys(
-					id: be == 1,
-					finished: be == true,
-					error: have_keys(
-						class: be == "RuntimeError",
-						message: be == "Monitor failed to get status!",
-						backtrace: be_a(Array)
-					)
+				id: be == 1,
+				finished: be == true,
+				error: have_keys(
+					class: be == "RuntimeError",
+					message: be == "Monitor failed to get status!",
+					backtrace: be_a(Array)
 				)
+			)
+			
+			stream.close
+		end
+		
+		it "can handle monitor removal failures" do
+			# Connect then disconnect to trigger removal:
+			stream = endpoint.connect
+			stream.close
+			
+			# Give time for removal to process
+			reactor.sleep(0.01)
+			
+			# Verify error was logged about the monitor removal failure:
+			error_log = console_capture.find {|log| log[:severity] == :error && log[:message] =~ /Error while removing process/}
+			expect(error_log).to be_truthy
+			
+			# Verify server is still working by sending a new request:
+			stream = endpoint.connect
+			message = {id: 1, do: :status}
+			stream.puts(JSON.dump(message))
+			stream.flush
+			
+			response = JSON.parse(stream.gets, symbolize_names: true)
+			expect(response).to have_keys(
+				id: be == 1,
+				finished: be == true
+			)
 
 			stream.close
 		end
@@ -98,6 +155,13 @@ def status(call)
 		stream.puts(JSON.dump(message))
 		stream.flush
 
+		# Wait for the message to be processed
+		reactor.sleep(0.01)
+		
+		# Verify a debug warning was logged about ignoring the message:
+		debug_log = console_capture.find {|log| log[:severity] == :debug && log[:message] =~ /Ignoring message/}
+		expect(debug_log).to be_truthy
+		
 		# Send a valid message to confirm the server is still working:
 		valid_message = {id: 3, do: :register, state: {process_id: ::Process.pid}}
 		stream.puts(JSON.dump(valid_message))
@@ -145,4 +209,4 @@ def status(call)
 
 		stream.close
 	end
-end
+end