May-13-2020, 11:20 AM
Hi everyone,
I am writing a script (Python 3.8) to test an embedded TCP server.
To do this I am creating a script that opens a socket to the server and sends random data and checks its response.
I'm going crazy with this error for two days: "Instance of 'socket' has no 'error' member".
A typical case of exception that I want to handle is the timeout. Unfortunately, however, when I have a timeout, the exception is not handled because I receive this error.
I made three versions of the script, one with functions that wrap the base socket calls, one that handles the socket handle globally and a base with nested tries.
I don't know where to bang my head anymore.
Can anyone help me?
This is my code:
Federico
I am writing a script (Python 3.8) to test an embedded TCP server.
To do this I am creating a script that opens a socket to the server and sends random data and checks its response.
I'm going crazy with this error for two days: "Instance of 'socket' has no 'error' member".
Output:Traceback (most recent call last):
File "D:/Progetti/Python/TCP Client/venv/Scripts/TcpClient.py", line 127, in <module>
s.connect((hostIp, hostPort))
ConnectionRefusedError: [WinError 10061] Impossibile stabilire la connessione. Rifiuto persistente del computer di destinazione
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "D:/Progetti/Python/TCP Client/venv/Scripts/TcpClient.py", line 171, in <module>
except s.timeout:
TypeError: catching classes that do not inherit from BaseException is not allowed
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "C:\Program Files\JetBrains\PyCharm Community Edition 2020.1.1\plugins\python-ce\helpers\pydev\pydevd.py", line 1438, in _exec
pydev_imports.execfile(file, globals, locals) # execute the script
File "C:\Program Files\JetBrains\PyCharm Community Edition 2020.1.1\plugins\python-ce\helpers\pydev\_pydev_imps\_pydev_execfile.py", line 18, in execfile
exec(compile(contents+"\n", file, 'exec'), glob, loc)
File "D:/Progetti/Python/TCP Client/venv/Scripts/TcpClient.py", line 175, in <module>
except s.error as e:
AttributeError: 'socket' object has no attribute 'error'This error occurs when an operation with sockets goes in exception otherwise the code works without problems.A typical case of exception that I want to handle is the timeout. Unfortunately, however, when I have a timeout, the exception is not handled because I receive this error.
I made three versions of the script, one with functions that wrap the base socket calls, one that handles the socket handle globally and a base with nested tries.
I don't know where to bang my head anymore.
Can anyone help me?
This is my code:
import socket
import sys
import os
import time
import logging
import random
import string
import atexit
from distutils.util import strtobool
errorCodes = {
0: 'No Error',
1: 'Socket Timeout',
2: 'Socket Error',
}
errorMessagesToCodes = dict(
(v, k) for k, v in errorCodes.items()
)
######################### SYSTEM FUNCTION #########################
# Disable Print
def DisablePrint():
sys.stdout = open(os.devnull, 'w')
# Restore Print
def EnablePrint():
sys.stdout = sys.__stdout__
def QueryYesNo(question, default='no'):
if default is None:
prompt = " [y/n] "
elif default == 'yes':
prompt = " [Y/n] "
elif default == 'no':
prompt = " [y/N] "
else:
raise ValueError(f"Unknown setting '{default}' for default.")
while True:
try:
resp = input(question + prompt).strip().lower()
if default is None and resp == '':
return True
elif default == 'no' and resp == '':
return False
elif default == 'yes' and resp == '':
return True
else:
return bool(strtobool(resp))
except ValueError:
print("Please respond with 'yes' or 'no' (or 'y' or 'n').\n")
def GenerateRandomString(stringMinLen=4, stringMaxLength=24):
stringLength = random.randint(stringMinLen, stringMaxLength)
letters = string.ascii_lowercase
return ''.join(random.choice(letters) for i in range(stringLength))
######################### MAIN PROGRAM #########################
if __name__ == '__main__':
#### Tests Parameters
##### Echo Test
echoTestDelay = 0.1
echoTestMessage = ''
echoTestechoRepetitions = 20
##### Open Close Test
openCloseDelay = 4
openCloseRepetitions = 100
### Define logging level
logging.basicConfig(level=logging.DEBUG, format='[%(levelname)s][%(funcName)s] %(message)s')
#### Active Tests
testsList = []
random.seed(1)
######################### TEST SETUP #########################
if QueryYesNo("Use default Server IP: 192.168.100.66", "yes") == True:
hostIp = '192.168.100.66'
else:
hostIp = 'localhost'
if QueryYesNo("Use default Server Port: 2000", "yes") == True:
hostPort = 2000
else:
hostIp = 2000
if QueryYesNo("Abilitare \"Echo\" Test") == True:
testsList.append('Echo')
if QueryYesNo("Abilitare \"Open-Close\" Test") == True:
testsList.append('OpenClose')
# Close pending socket on Exit or Crash
# atexit.register( ExitProgram, socketHandle, 1 )
######################### START TESTS #########################
### Echo Test
if 'Echo' in testsList:
# Test Parameters
echoRepetitions = 1000
# Define Test Variables
receivedBufferErrorCnt = 0
socketSendErrorCnt = 0
socketSendTimeoutCnt = 0
socketReceiveErrorCnt = 0
socketReceiveTimeoutCnt = 0
logging.info("Start Echo Test")
try:
logging.debug("Socket Create")
# Create a socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Set Socket Timeout
s.settimeout(5)
logging.debug("Socket Connect")
try:
# Connect to the server
s.connect((hostIp, hostPort))
# Execute Echo Test
for idx in range(echoRepetitions):
# Preparare Buffer
sendBuffer = GenerateRandomString()
logging.debug(f"===== Send Data ({str(idx + 1)}/{str(echoRepetitions)}) =====")
logging.debug("Send Buffer: " + sendBuffer)
# Send Buffer
tic = time.perf_counter()
try:
s.send(sendBuffer.encode('UTF-8'))
# Receive Buffer
try:
receivedBuffer = s.recv(1024)
toc = time.perf_counter()
logging.debug(f"Received Data: {receivedBuffer.decode('UTF-8')}")
# Check received buffer
if( receivedBuffer.decode('utf-8') != sendBuffer ):
logging.warning("!!! Error: Recevived a different message")
receivedBufferErrorCnt = receivedBufferErrorCnt + 1
logging.debug(f"========= ({toc - tic:0.4f}s) =========\n\r")
except s.error as e:
logging.error ("!!! Socket Receive FAILED: %s" % e)
socketReceiveErrorCnt = socketReceiveErrorCnt + 1
except s.timeout:
logging.error("Socket Receive TIMEOUT")
socketReceiveTimeoutCnt = socketReceiveTimeoutCnt + 1
except s.error as e:
logging.error("!!! Socket Send FAILED: %s" % e)
socketSendErrorCnt = socketSendErrorCnt + 1
except s.timeout as e:
logging.error("Socket Connect TIMEOUT")
socketReceiveTimeoutCnt = socketReceiveTimeoutCnt + 1
# Shutdown Socket
try:
s.shutdown(1)
# Socket Close
try:
s.close()
except s.error as e:
logging.error("!!! Socket Close FAILED: %s" % e)
except s.error as e:
logging.error("!!! Socket Shutdown FAILED: %s" % e)
except s.timeout:
logging.error("!!! Socket Connect TIMEOUT: %s" % e)
except s.error as e:
logging.error("!!! Socket Connect FAILED: %s" % e)
except s.error as e:
logging.error("Socket Create FAILED: %s" % e)
if (receivedBufferErrorCnt == 0 and socketSendTimeoutCnt == 0 and socketReceiveTimeoutCnt == 0):
print(f"\n\r=== Echo Test PASSED ({str(echoRepetitions - receivedBufferErrorCnt)}/{str(echoRepetitions)}) =====")
else:
print("\n\r=== Echo Test FAILED - Results ===")
print(f" * Received Buffer Error Count: {str(receivedBufferErrorCnt)}/{str(echoRepetitions)}")
print(f" * Socket Send Timeout Count: {str(socketSendTimeoutCnt)}/{str(echoRepetitions)}")
print(f" * Socket Receive Timeout Count: {str(socketReceiveTimeoutCnt)}/{str(echoRepetitions)}")
print("==============================")
print("=== Test Completed ===")Many thanks in advance.Federico
