udpecho_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 
00018 import re
00019 import sys
00020 import uuid
00021 from sys import stdout
00022 from socket import socket, AF_INET, SOCK_DGRAM
00023 
00024 class UDPEchoServerTest():
00025     ECHO_SERVER_ADDRESS = ""
00026     ECHO_PORT = 0
00027     s = None # Socket
00028 
00029     PATTERN_SERVER_IP = "Server IP Address is (\d+).(\d+).(\d+).(\d+):(\d+)"
00030     re_detect_server_ip = re.compile(PATTERN_SERVER_IP)
00031 
00032     def test(self, selftest):
00033         result = True
00034         serial_ip_msg = selftest.mbed.serial_readline()
00035         if serial_ip_msg is None:
00036             return selftest.RESULT_IO_SERIAL
00037         selftest.notify(serial_ip_msg)
00038         # Searching for IP address and port prompted by server
00039         m = self.re_detect_server_ip.search(serial_ip_msg)
00040         if m and len(m.groups()):
00041             self.ECHO_SERVER_ADDRESS = ".".join(m.groups()[:4])
00042             self.ECHO_PORT = int(m.groups()[4]) # must be integer for socket.connect method
00043             selftest.notify("HOST: UDP Server found at: " + self.ECHO_SERVER_ADDRESS + ":" + str(self.ECHO_PORT))
00044 
00045             # We assume this test fails so can't send 'error' message to server
00046             try:
00047                 self.s = socket(AF_INET, SOCK_DGRAM)
00048             except Exception, e:
00049                 self.s = None
00050                 selftest.notify("HOST: Socket error: %s"% e)
00051                 return selftest.RESULT_ERROR
00052 
00053             for i in range(0, 100):
00054                 TEST_STRING = str(uuid.uuid4())
00055                 self.s.sendto(TEST_STRING, (self.ECHO_SERVER_ADDRESS, self.ECHO_PORT))
00056                 data = self.s.recv(len(TEST_STRING))
00057                 received_str = repr(data)[1:-1]
00058                 if TEST_STRING != received_str:
00059                     result = False
00060                     break
00061                 sys.stdout.write('.')
00062                 stdout.flush()
00063         else:
00064             result = False
00065 
00066         if self.s is not None:
00067             self.s.close()
00068         return selftest.RESULT_SUCCESS if result else selftest.RESULT_FAILURE