MinitestのTCPSocketにいくつかのテストを書く方法

Question

ネットワーキングアプリケーションをテストしようとしています。しかし、私は脳に結びついているかもしれません。 私が知っている限り、最小限のテストは並行して走ります。

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を指定すると、ソケットはランダムな未使用ポートにバインドされます。

bound_port = @server_socket.addr[1] 

使用する方法の1つ：

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

あなたはTCPサーバーソケットがバインドされたポートを見つけることができます

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を使用してサーバーインスタンスを作成します：

これらの事実は、このようなサーバーの何かを持っていることです10

server = Server.new(port: 0) 

サーバーへの接続を行う、テストがに接続するポートを見つけるために#bound_portアクセサを使用しています。

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

、その後、通常に運びます。