Имитация длинного сетевого запроса для Python тестирования

Question

Мне нужно проверить функцию обновления устройства. Функция открывает сокет на хосте и отправляет блок текста.

Обновление может занять до 120 секунд. Возвращает код для успеха / неудачи. Чтобы продолжить работу программы, обновление запускается в потоке.

Я не могу контролировать реакцию устройства. Симуляция должна быть в состоянии удерживать открытое соединение не менее 120 секунд.

Он не должен быть безопасным или масштабируемым, поскольку он будет использоваться только для интеграционного теста. Самое простое решение является предпочтительным. Чистый python лучше, но docker также приемлем.

Answer 1

Я написал это на основе указателя rdas.

import json
import logging
import socket
import socketserver
import threading
import time

log = logging.getLogger(__name__)
log.setLevel(logging.INFO)

class LongRequestHandler(socketserver.BaseRequestHandler):

    def handle(self):
        # Echo the back to the client
        data = json.loads(self.request.recv(1024).decode())
        t = 0
        while t < data['delay']:
            time.sleep(1)
            print(".", end='')
            t += 1
            if t % 80 == 0:
                print("\n")
        print("\n")
        self.request.send(b"ok")

class Server():

    def __init__(self, host='localhost', port=0):
        self.host = host
        self.port = port
        self.ip = None
        self.server = None

    def run(self):
        address = (self.host, self.port)  # let the kernel assign port if port=0
        self.server = socketserver.TCPServer(address, LongRequestHandler)
        self.ip, self.port = self.server.server_address  # what port was assigned?

        t = threading.Thread(target=self.server.serve_forever)
        t.setDaemon(True)  # don't hang on exit
        t.start()
        return True

    def send_request(self, data: dict ):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((self.ip, self.port))
        message = json.dumps(data).encode()
        s.send(message)
        response = s.recv(1024)
        s.close()
        return response

    def __exit__(self):
        self.server.shutdown()
        self.server.socket.close()



if __name__ == '__main__':
    # For simple testing and config example...
    server = Server()
    server.run()
    # Send the data
    d = dict(delay=5)  # set delay here to desired
    out = server.send_request(d)
    print('Received: {!r}'.format(out))