mirror of https://github.com/ARMmbed/mbed-os.git
Host test plugins: Ethernet tests (NET_*) refactored to new host test copy paradigm
parent
7646c51d7e
commit
54c06c8b84
|
@ -251,6 +251,7 @@ class HostTestResults:
|
|||
self.RESULT_IO_SERIAL = 'ioerr_serial'
|
||||
self.RESULT_NO_IMAGE = 'no_image'
|
||||
self.RESULT_IOERR_COPY = "ioerr_copy"
|
||||
self.RESULT_PASSIVE = "passive"
|
||||
|
||||
|
||||
class Test(HostTestResults):
|
||||
|
@ -284,7 +285,10 @@ class Test(HostTestResults):
|
|||
# Run test
|
||||
try:
|
||||
result = self.test()
|
||||
self.print_result(result)
|
||||
if result is not None:
|
||||
self.print_result(result)
|
||||
else:
|
||||
self.notify("HOST: Passive mode...")
|
||||
except Exception, e:
|
||||
print str(e)
|
||||
self.print_result(self.RESULT_ERROR)
|
||||
|
@ -324,7 +328,7 @@ class Simple(DefaultTest):
|
|||
output from MUT, no supervision over test running in MUT is executed.
|
||||
"""
|
||||
def test(self):
|
||||
result = True
|
||||
result = self.RESULT_SUCCESS
|
||||
try:
|
||||
while True:
|
||||
c = self.mbed.serial_read(512)
|
||||
|
@ -334,7 +338,7 @@ class Simple(DefaultTest):
|
|||
stdout.flush()
|
||||
except KeyboardInterrupt, _:
|
||||
self.notify("\r\n[CTRL+C] exit")
|
||||
result = False
|
||||
result = self.RESULT_ERROR
|
||||
return result
|
||||
|
||||
|
||||
|
|
|
@ -18,7 +18,7 @@ limitations under the License.
|
|||
import sys
|
||||
import socket
|
||||
from sys import stdout
|
||||
from host_test import Test
|
||||
from host_test import HostTestResults, Test
|
||||
from SocketServer import BaseRequestHandler, TCPServer
|
||||
|
||||
|
||||
|
@ -28,15 +28,12 @@ SERVER_PORT = 7
|
|||
|
||||
class TCPEchoClientTest(Test):
|
||||
def __init__(self):
|
||||
HostTestResults.__init__(self)
|
||||
Test.__init__(self)
|
||||
self.mbed.init_serial()
|
||||
|
||||
def send_server_ip_port(self, ip_address, port_no):
|
||||
""" Set up network host. Reset target and and send server IP via serial to Mbed
|
||||
"""
|
||||
self.notify("HOST: Resetting target...")
|
||||
self.mbed.reset()
|
||||
|
||||
c = self.mbed.serial_readline() # 'TCPCllient waiting for server IP and port...'
|
||||
if c is None:
|
||||
self.print_result(self.RESULT_IO_SERIAL)
|
||||
|
@ -57,10 +54,15 @@ class TCPEchoClientTest(Test):
|
|||
return
|
||||
self.notify(c.strip())
|
||||
|
||||
def test(self):
|
||||
# Returning none will suppress host test from printing success code
|
||||
return None
|
||||
|
||||
|
||||
class TCPEchoClient_Handler(BaseRequestHandler):
|
||||
def handle(self):
|
||||
""" One handle per connection """
|
||||
""" One handle per connection
|
||||
"""
|
||||
print "HOST: Connection received...",
|
||||
count = 1;
|
||||
while True:
|
||||
|
@ -78,9 +80,10 @@ class TCPEchoClient_Handler(BaseRequestHandler):
|
|||
|
||||
|
||||
server = TCPServer((SERVER_IP, SERVER_PORT), TCPEchoClient_Handler)
|
||||
print "HOST: Listening for connections: " + SERVER_IP + ":" + str(SERVER_PORT)
|
||||
print "HOST: Listening for TCP connections: " + SERVER_IP + ":" + str(SERVER_PORT)
|
||||
|
||||
mbed_test = TCPEchoClientTest();
|
||||
mbed_test.run()
|
||||
mbed_test.send_server_ip_port(SERVER_IP, SERVER_PORT)
|
||||
|
||||
server.serve_forever()
|
||||
|
|
|
@ -18,7 +18,7 @@ limitations under the License.
|
|||
import sys
|
||||
import socket
|
||||
from sys import stdout
|
||||
from host_test import Test
|
||||
from host_test import HostTestResults, Test
|
||||
from SocketServer import BaseRequestHandler, UDPServer
|
||||
|
||||
|
||||
|
@ -28,20 +28,17 @@ SERVER_PORT = 7
|
|||
|
||||
class UDPEchoClientTest(Test):
|
||||
def __init__(self):
|
||||
HostTestResults.__init__(self)
|
||||
Test.__init__(self)
|
||||
self.mbed.init_serial()
|
||||
|
||||
def send_server_ip_port(self, ip_address, port_no):
|
||||
print "HOST: Resetting target..."
|
||||
self.mbed.reset()
|
||||
|
||||
c = self.mbed.serial_readline() # 'UDPCllient waiting for server IP and port...'
|
||||
if c is None:
|
||||
self.print_result(self.RESULT_IO_SERIAL)
|
||||
return
|
||||
self.notify(c.strip())
|
||||
|
||||
print "HOST: Sending server IP Address to target..."
|
||||
self.notify("HOST: Sending server IP Address to target...")
|
||||
connection_str = ip_address + ":" + str(port_no) + "\n"
|
||||
self.mbed.serial_write(connection_str)
|
||||
|
||||
|
@ -50,6 +47,11 @@ class UDPEchoClientTest(Test):
|
|||
self.print_result(self.RESULT_IO_SERIAL)
|
||||
return
|
||||
self.notify(c.strip())
|
||||
return self.RESULT_PASSIVE
|
||||
|
||||
def test(self):
|
||||
# Returning none will suppress host test from printing success code
|
||||
return None
|
||||
|
||||
|
||||
class UDPEchoClient_Handler(BaseRequestHandler):
|
||||
|
@ -67,9 +69,10 @@ class UDPEchoClient_Handler(BaseRequestHandler):
|
|||
|
||||
|
||||
server = UDPServer((SERVER_IP, SERVER_PORT), UDPEchoClient_Handler)
|
||||
print "HOST: Listening for connections..."
|
||||
print "HOST: Listening for UDP connections..."
|
||||
|
||||
mbed_test = UDPEchoClientTest();
|
||||
mbed_test.run()
|
||||
mbed_test.send_server_ip_port(SERVER_IP, SERVER_PORT)
|
||||
|
||||
server.serve_forever()
|
||||
|
|
|
@ -657,7 +657,7 @@ TESTS = [
|
|||
"peripherals": ["ethernet"],
|
||||
},
|
||||
{
|
||||
"id": "NET_2", "description": "UDP client hello world",
|
||||
"id": "NET_2", "description": "NIST Internet Time Service",
|
||||
"source_dir": join(TEST_DIR, "net", "helloworld", "udpclient"),
|
||||
"dependencies": [MBED_LIBRARIES, RTOS_LIBRARIES, ETH_LIBRARY, TEST_MBED_LIB],
|
||||
"duration": 15,
|
||||
|
@ -697,7 +697,7 @@ TESTS = [
|
|||
"peripherals": ["ethernet"],
|
||||
},
|
||||
{
|
||||
"id": "NET_7", "description": "HTTP client",
|
||||
"id": "NET_7", "description": "HTTP client hello world",
|
||||
"source_dir": join(TEST_DIR, "net", "protocols", "HTTPClient_HelloWorld"),
|
||||
"dependencies": [MBED_LIBRARIES, RTOS_LIBRARIES, ETH_LIBRARY, TEST_MBED_LIB],
|
||||
"automated": True,
|
||||
|
|
Loading…
Reference in New Issue