Рассмотрим этот код:
Код: Выделить всё
import socket
def send_recv(xml_message):
try:
address = ('127.0.0.1', 12000)
xml_bytes = xml_message.encode(encoding="utf-8")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:
client_socket.settimeout(10)
client_socket.connect(address)
client_socket.sendall(xml_bytes)
reply = ''
while True:
data = client_socket.recv(1024).decode()
print(data)
reply += data
if "" in reply or not data:
break
client_socket.shutdown(socket.SHUT_WR)
return reply
except ConnectionError as e:
error_description = "Error connecting to TCP/IP: {}".format(e)
print(error_description)
return None
except socket.timeout:
error_description = "Error receiving message. No reply received or message termination not found."
print(error_description)
return None
Код: Выделить всё
import unittest
from unittest.mock import patch, MagicMock
from socket_example import only_recv, send_recv
class TestSocketExample(unittest.TestCase):
@patch('socket_example.socket.socket')
def test_send_recv(self, mock_socket):
mock_socket = mock_socket.return_value
mock_socket.recv.return_value = b''
#TODO: I have to insert an assert to check the result
- Если запустить тест, то распечатка полученных данных
Код: Выделить всё
print(data)
Код: Выделить всё
- В тестируемом коде я вставил следующий код
Код: Выделить всё
client_socket.settimeout(10)
Подробнее здесь: https://stackoverflow.com/questions/792 ... settimeout
Мобильная версия