python - 将 echo 服务器作为守护进程运行以在 python 中进行测试
问题描述
我想自动测试仪器并编写了一个小服务器程序来模拟仪器,它将发送回命令,除非它收到特殊命令“*IDN?”。当我直接在自己的脚本中运行回显服务器,然后单独运行客户端脚本时,一切正常,我得到了预期的结果。现在我想直接从测试脚本运行服务器。所以我想我会使用多处理来启动它。但问题似乎是当服务器套接字到达 s.accept() 行时,它只是在那里等待并且永远不会返回。那么,如果我不能在与测试函数相同的代码中运行这个服务器,我该如何完成自动化测试呢?
import socket
import multiprocessing as mp
import time,sys
HOST = '127.0.0.1' # Standard loopback interface address (localhost),
PORT = 65432 # Port to listen on (non-privileged ports are > 1023),
FTP_PORT = 63217 # Port for ftp testing, change to 21 for device
def handle_connection(conn,addr):
with conn:
conn.send('Connected by', addr)
print("Got connection")
data = conn.recv(1024)
if not data:
return 'Nodata'
elif (data == b'*IDN?\n'):
print('SONY/TEK,AWG520,0,SCPI:95.0 OS:3.0 USR:4.0\n')
conn.sendall(b'SONY/TEK,AWG520,0,SCPI:95.0 OS:3.0 USR:4.0\n')
return 'IDN'
else:
conn.sendall(data)
return 'Data'
def echo_server(c_conn,host=HOST,port=PORT):
# this server simulates the AWG command protocol, simply echoing the command back except for IDN?
p = mp.current_process()
print('Starting echo server:', p.name, p.pid)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind((host, port))
s.listen()
try:
while True:
print("Waiting for connection...")
c_conn.send('waiting for connection...')
conn, addr = s.accept()
handle_connection(conn,addr)
c_conn.send('serving client...')
finally:
conn.close()
c_conn.send('done')
time.sleep(2)
print('Exiting echo server:', p.name, p.pid)
sys.stdout.flush()
def test_echo_server():
print("entering client part")
with socket.socket(socket.AF_INET,socket.SOCK_STREAM) as mysock:
mysock.connect((HOST,PORT))
mysock.sendall('test\n'.encode())
data = mysock.recv(1024)
print('received:',repr(data))
if __name__ == '__main__':
parent_conn, child_conn = mp.Pipe()
echo_demon = mp.Process(name='echo', target=echo_server(child_conn, ))
echo_demon.daemon = True
echo_demon.start()
time.sleep(1)
echo_demon.join(1)
test_echo_server()
if parent_conn.poll(1):
print(parent_conn.recv())
else:
print('Waiting for echo server')
解决方案
我设法使用我在 Romano、Baka 和 Phillips 的“Python 入门”一书中找到的一些代码片段解决了我自己的问题。这是服务器的代码:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((HOST, PORT))
s.listen()
try:
while True:
print("Waiting for connection...")
client, addr = s.accept()
with client:
data = client.recv(1024)
if not data:
break
elif (data == b'*IDN?\n'):
client.sendall(b'SONY/TEK,AWG520,0,SCPI:95.0 OS:3.0 USR:4.0\n')
else:
client.sendall(data)
finally:
time.sleep(1)
print('Exiting echo server:')
这是在单独进程中运行此服务器的测试文件的代码,以及几个简单的测试:
@pytest.fixture(scope="session")
def awgserver():
print("loading server")
p = subprocess.Popen(["python3", "server.py"])
time.sleep(1)
yield p
p.terminate()
@pytest.fixture
def clientsocket(request):
print("entering client part")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as mysock:
mysock.connect((HOST, PORT))
yield mysock
mysock.close()
@pytest.mark.run_this
def test_echo(awgserver, clientsocket):
clientsocket.send(b"*IDN?\n")
#assert clientsocket.recv(1024) == b"SONY/TEK,AWG520,0,SCPI:95.0 OS:3.0 USR:4.0\n"
assert clientsocket.recv(10) == b"TEK" # deliberately make test fail
@pytest.mark.run_this
def test_echo2(awgserver, clientsocket):
clientsocket.send(b"def")
assert clientsocket.recv(3) == b"def"
将 HOST 设置为环回 IP,并将 PORT 设置为 > 1024
推荐阅读
- algorithm - 使用相似度评分算法重新排序字符串
- spring-boot - 为什么 Spring Data JPA 为共享的 EntityManager 调用创建新的 EntityManager?
- html - 只允许身体滚动 x 像素 css
- c++ - Scala 可变数量的参数是 F 有界类型的子类
- python - Twilio 录音调查
- directory - [exchangewebservices] 使用“0”参数调用“更新”的异常:“操作将更改对象类型,这是不允许的。”
- javascript - 如何使用 sapui5 在数据库中创建新表?
- c# - c# iTextSharp:从数据库中替换 pdf 字段(DB 行到列表
) - progress-4gl - 使用 mbpro 运行 QAD mfg/pro 程序,是否需要 mf.p 才能在脚本中运行?
- mariadb - 将列的默认值更改为表达式