4

切断後にサーバーに再接続するコードをテストしようとしています。これはテストの外では完全に正常に機能しますが、テストの実行時にソケットが切断されたことを確認できません。

Gevent Stream Serverを使用して、実際のリスニングサーバーをモックしています。

import gevent.server
from gevent import queue


class TestServer(gevent.server.StreamServer):

    def __init__(self, *args, **kwargs):
        super(TestServer, self).__init__(*args, **kwargs)
        self.sockets = {}

    def handle(self, socket, address):
        self.sockets[address] = (socket, queue.Queue())
        socket.sendall('testing the connection\r\n')
        gevent.spawn(self.recv, address)

    def recv(self, address):
        socket = self.sockets[address][0]
        queue = self.sockets[address][1]
        print 'Connection accepted %s:%d' % address
        try:
            for data in socket.recv(1024):
                queue.put(data)
        except:
            pass

    def murder(self):
        self.stop()
        for sock in self.sockets.iteritems():
            print sock
            sock[1][0].shutdown(socket.SHUT_RDWR)
            sock[1][0].close()
        self.sockets = {}


def run_server():
    test_server = TestServer(('127.0.0.1', 10666))
    test_server.start()
    return test_server

そして私のテストは次のようになります：

def test_can_reconnect(self):
    test_server = run_server()
    client_config = {'host': '127.0.0.1', 'port': 10666}
    client = Connection('test client', client_config, get_config())
    client.connect()
    assert client.socket_connected
    test_server.murder()
    #time.sleep(4) #tried sleeping. no dice.
    assert not client.socket_connected
    assert client.server_disconnect
    test_server = run_server()
    client.reconnect()
    assert client.socket_connected

で失敗しassert not client.socket_connectedます。

recv中に「データではない」を検出します。Noneの場合、他のコードが再接続するかどうかを決定できるようにいくつかの変数を設定します（user_disconnectの場合は再接続しないでくださいなど）。この動作は機能し、過去には常に機能してきました。これまでテストを試みたことがありません。ソケット接続やローカル関数スコープなどに何か奇妙なことがありますか？サーバーを停止した後も接続が存在しているようです。

私がテストしようとしているコードは開いています：https ：//github.com/kyleterry/tenyks.git

テストを実行すると、修正しようとしているテストが失敗することがわかります。

4

4 に答える 4

2

実際のソケットを使用して単体テストを実行しようとするのは難しいことです。サーバーポートが使用されるため、一度に実行できるテストのセットは1つだけであるため、注意が必要です。また、ソケットがセットアップされて破棄されると、処理が遅くなります。これが本当にユニットテストである場合は、ソケットをテストするのではなく、ソケットを使用しているコードだけをテストする必要があります。

ソケット呼び出しをモックする場合は、モックされたコードから例外をスローし、ソケットを使用するコードが正しいことを実行することを確認できます。テスト対象のクラスが正しいことを実行することを保証するために実際のソケットは必要ありません。ソケット呼び出しをオブジェクトにラップできる場合は、それを偽造することができます。クラスを構築するときにソケットオブジェクトへの参照を渡すと、準備が整います。

私の提案は、sendall、recv、およびソケットで呼び出すすべてのメソッドをサポートするクラスでソケット呼び出しをラップすることです。次に、実際のSocketクラスをTestReconnectSocket（またはその他）と交換して、テストを実行できます。

Pythonのモックフレームワークであるmoxを見てください。

于 2012-10-25T18:13:05.340 に答える
0

このようなソケットをテストするときに覚えておくべきことの1つは、オペレーティングシステムは、使用後すぐにソケットを再度開くことを好まないということです。ソケットオプションを設定して、先に進んでとにかく再利用するように指示できます。ソケットを作成した直後に、ソケットのオプションを設定します。

mysocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

うまくいけば、これで問題が解決するでしょう。どちらが問題を引き起こしているかによっては、サーバー側とクライアント側の両方でそれを行う必要がある場合があります。

于 2012-10-25T18:00:39.087 に答える
0

あいまいな応答ですが、私の即時の反応は、recv（）呼び出しがブロックされ、ソケットを存続させているということです-ソケットを非ブロックにして、代わりに閉じるときにエラーをキャッチしようとしましたか？

于 2012-10-25T09:43:12.127 に答える
0
  • あなたが呼んでいるので、これはrecvブロッキングshutdown(socket.SHUT_RDWR)の問題のようには見えません。
  • ただし、を使用しているため、バージョンgevent.socket.socket.recvを確認してください。基になるファイル記述子が閉じている場合（バージョン< ）にブロックされる問題があります。geventrecv()v0.13.0
  • gevent.sleep()それでも、協調的な譲歩を行い、クライアントに通話を終了する機会を与える必要がある場合がありrecv()ます。
于 2012-10-25T21:33:48.250 に答える