0

我想测试一个复杂的类，它包装了socket模块的一些方法： connect,sendall和recv. 特别是，我想测试recv这个类的方法。

下面的工作示例代码显示了我如何做到这一点（以基本的底层形式保持简单，testsocket将对应于复杂的包装类）：

import socket

# This is just a socket for testing purposes, binds to the loopback device
sock =  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 1234))
sock.listen(5)

# This is the socket later part of the complex socket wrapper.
# It just contains calls to connect, sendall and recv
testsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
testsocket.connect(("127.0.0.1", 1234))
testsocket.sendall("test_send")

# The testing socket connects to a client
(client, adr) = sock.accept()
print client.recv(1024)

# Now I can do the actual test: Test the receive method of the socket
# wrapped in the complex class
client.sendall("test_recv")
print testsocket.recv(1024)  # <--  This is what I want to test !!

# close everything
testsocket.close()
client.close()
sock.close()

但是为了测试testsocket.recv我需要使用testsocket.sendall之前。

是否可以以简单的方式（没有分叉或线程）修改此代码以便在testsocket.recv不使用该方法的情况下进行测试testsocket.sendall？

4

2 回答 2

1

怎么用socket.socketpair？：

import socket

client, testsocket = socket.socketpair()
client.sendall("test_recv")
print testsocket.recv(1024)
testsocket.close()
client.close()

注意仅在 Unix 中可用。

使用mock

import mock

testsocket = mock.Mock()
testsocket.configure_mock(**{'recv.return_value': 'test_recv'})
print testsocket.recv(1024)
于 2013-10-21T07:41:29.023 回答
0

您不能在同一个进程/线程中运行客户端和服务器套接字，因为这server.recv()是一个阻塞调用

我的日常：

import socket, threading

# Protocols supported
TCP = (0x00)
UDP = (0x01)
UDP_Multicast = (0x02)

# Client/ Server mode
CLIENT = (0x00)
SERVER = (0x01)

# Data to be sent
data = 'Test. Please Ignore'


# Server processing
def simple_processing(data):
    print "messsage : ", data


def start_socket( protocol, client, processing_callback):

    # switch on protocol
    if protocol == TCP:
        sock =  socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    else: 
        return

    # switch on client
    if client == SERVER:

        # Server mode = listening to incoming datas
        sock.bind(("127.0.0.1", 1234))
        sock.listen(5)
        (sock, adr) = sock.accept()
        processing_callback( sock.recv(1024) ) # processing data

    elif client == CLIENT:

        # Client mode : connecting and sending data
        sock.connect(("127.0.0.1", 1234))
        sock.sendall(data)


    else:
        return

    sock.close()


def test():

    # Thread creations
    server = threading.Thread( target = start_socket,
                               args=(   TCP,
                                        SERVER,
                                        simple_processing, )  
                              )

    client = threading.Thread(  target= start_socket, 
                                args=(  TCP,
                                        CLIENT,
                                        None)
                                )

    server.start()
    client.start()


    # Join : wait on every thread to finish
    client.join()
    server.join()

if __name__ == '__main__':
    # Launch the test
    test()
于 2013-10-21T07:58:23.400 回答