tcpecho_server_auto.py

00001 """
00002 mbed SDK
00003 Copyright (c) 2011-2013 ARM Limited
00004 
00005 Licensed under the Apache License, Version 2.0 (the "License");
00006 you may not use this file except in compliance with the License.
00007 You may obtain a copy of the License at
00008 
00009     http://www.apache.org/licenses/LICENSE-2.0
00010 
00011 Unless required by applicable law or agreed to in writing, software
00012 distributed under the License is distributed on an "AS IS" BASIS,
00013 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
00014 See the License for the specific language governing permissions and
00015 limitations under the License.
00016 """
00017 from __future__ import print_function
00018 
00019 import re
00020 import sys
00021 import uuid
00022 import socket
00023 from sys import stdout
00024 
00025 class TCPEchoServerTest():
00026     ECHO_SERVER_ADDRESS = ""
00027     ECHO_PORT = 0
00028     ECHO_LOOPs = 100
00029     s = None # Socket
00030 
00031     PATTERN_SERVER_IP = "Server IP Address is (\d+).(\d+).(\d+).(\d+):(\d+)"
00032     re_detect_server_ip = re.compile(PATTERN_SERVER_IP)
00033 
00034     def test(self, selftest):
00035         result = False
00036         c = selftest.mbed.serial_readline()
00037         if c is None:
00038             return selftest.RESULT_IO_SERIAL
00039         selftest.notify(c)
00040 
00041         m = self.re_detect_server_ip.search(c)
00042         if m and len(m.groups()):
00043             self.ECHO_SERVER_ADDRESS = ".".join(m.groups()[:4])
00044             self.ECHO_PORT = int(m.groups()[4]) # must be integer for socket.connect method
00045             selftest.notify("HOST: TCP Server found at: " + self.ECHO_SERVER_ADDRESS + ":" + str(self.ECHO_PORT))
00046 
00047             # We assume this test fails so can't send 'error' message to server
00048             try:
00049                 self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
00050                 self.s.connect((self.ECHO_SERVER_ADDRESS, self.ECHO_PORT))
00051             except Exception as e:
00052                 self.s = None
00053                 selftest.notify("HOST: Socket error: %s"% e)
00054                 return selftest.RESULT_ERROR
00055 
00056             print('HOST: Sending %d echo strings...'% self.ECHO_LOOPs,)
00057             for i in range(0, self.ECHO_LOOPs):
00058                 TEST_STRING = str(uuid.uuid4())
00059                 try:
00060                     self.s.sendall(TEST_STRING)
00061                     data = self.s.recv(128)
00062                 except Exception as e:
00063                     self.s = None
00064                     selftest.notify("HOST: Socket error: %s"% e)
00065                     return selftest.RESULT_ERROR
00066 
00067                 received_str = repr(data)[1:-1]
00068                 if TEST_STRING == received_str: # We need to cut not needed single quotes from the string
00069                     sys.stdout.write('.')
00070                     stdout.flush()
00071                     result = True
00072                 else:
00073                     print("Expected: ")
00074                     print("'%s'"% TEST_STRING)
00075                     print("received: ")
00076                     print("'%s'"% received_str)
00077                     result = False
00078                     break
00079 
00080             if self.s is not None:
00081                 self.s.close()
00082         else:
00083             selftest.notify("HOST: TCP Server not found")
00084             result = False
00085         return selftest.RESULT_SUCCESS if result else selftest.RESULT_FAILURE