ruby - 如何在 Minitest 中为 TCPSocket 编写多个测试

Question

我尝试测试网络应用程序。但是我的脑子里可能有个结。 据我所知，minitest 中的测试是并行运行的。基于这个假设，我认为很明显，当我在 setup() 中分配一个端口时，它会在运行多个测试时失败:

RuntimeError: Address already in use - bind(2) for nil port 2000 TCPServer new failed

那么在监听端口的服务器上进行多项测试的最佳做法是什么？

   class ServerTest < Minitest::Test

      def setup
        # -- set env variable
        ENV["conf"] = "./tc_gpio.config" 
        Thread::abort_on_exception = true
        @my_thr = Thread.start() do 
          @server = Server.new       
          @server.start
          assert @server.hi() != nil
        end
      end


      def teardown
        Thread.kill(@my_thr) # sends exit() to thr
      end

      def test_connect_and_disconnect
        sleep 1
        hostname = 'localhost'
        port = 2000
        s = TCPSocket.open(hostname, port)
        my_s = s.recvmsg()
        s.sendmsg(:set.to_s, 0) # Failes since a serialized object is expected
        my_s2 = s.recvmsg()

        assert_equal(  "READY" , my_s[0] )
        assert_equal( "eeFIN" , my_s2[0])
      end

      def test_send_command

        # fill command
        com = Command.new
        com.type = :set
        com.device_name = 'pump0'
        com.device_address = 'resource0'
        com.value = 2

        serial_com = YAML::dump(com)

        sleep 1
        hostname = 'localhost'
        port = 2000
        s = TCPSocket.open(hostname, port)
        my_s = s.recvmsg()
        s.sendmsg(serial_com, 0)
        my_s2 = s.recvmsg()


        assert_equal(  "READY" , my_s[0] )
        assert_equal( "FIN" , my_s2[0])
      end
    end

Answer 1

并行测试 TCP 服务器时，服务器的每个实例都应使用不同的端口启动。这可以通过在创建套接字时指定端口号 0 来完成。当给定端口号 0 时，套接字将绑定(bind)到一个随机未使用的端口:

interface = "0.0.0.0"
port = 0
tcp_server = TCPServer.new(interface, port)

您可以找出 TCP 服务器套接字绑定(bind)到哪个端口:

bound_port = @server_socket.addr[1]

使用这些事实的一种方法是拥有这样的服务器:

class Server

  # Create a server instance.  If the port is unspecified, 
  # or 0, then a random ephemeral port is bound to.
  def initialize(interface: "127.0.0.1", port: 0)
    @server_socket = TCPServer.new(interface, port)
    ...
  end

  # Return the port the server socket is bound to.
  def  bound_port
    @server_socket.addr[1]
  end

  ...

end

测试然后使用端口 0 创建服务器实例:

server = Server.new(port: 0)

当连接到服务器时，测试使用#bound_port 访问器找出要连接到的端口:

client = TCPSocket.open("localhost", server.bound_port)

然后正常进行。