TCPSOCKET_ECHOTEST_BURST_SHORT.py

00001 """
00002 Copyright 2018 ARM Limited
00003 Licensed under the Apache License, Version 2.0 (the "License");
00004 you may not use this file except in compliance with the License.
00005 You may obtain a copy of the License at
00006 
00007     http://www.apache.org/licenses/LICENSE-2.0
00008 
00009 Unless required by applicable law or agreed to in writing, software
00010 distributed under the License is distributed on an "AS IS" BASIS,
00011 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
00012 See the License for the specific language governing permissions and
00013 limitations under the License.
00014 """
00015 import string
00016 
00017 from icetea_lib.Randomize.randomize import Randomize
00018 from icetea_lib.bench import Bench, TestStepFail
00019 
00020 
00021 class Testcase(Bench):
00022     def __init__(self):
00023         Bench.__init__(self,
00024                        name="TCPSOCKET_ECHOTEST_BURST_SHORT",
00025                        title="TCPSOCKET_ECHOTEST_BURST_SHORT",
00026                        purpose="Verify that TCPSocket can send burst of packets to echo server and read incoming packets",
00027                        status="released",
00028                        component=["mbed-os", "netsocket"],
00029                        author="Juha Ylinen <juha.ylinen@arm.com>",
00030                        type="smoke",
00031                        subtype="socket",
00032                        requirements={
00033                            "duts": {
00034                                '*': {  # requirements for all nodes
00035                                    "count": 1,
00036                                    "type": "hardware",
00037                                    "application": {
00038                                        "name": "TEST_APPS-device-socket_app"
00039                                    }
00040                                },
00041                                "1": {"nick": "dut1"},
00042                            }
00043                        }
00044                        )
00045 
00046     def setup(self):
00047         self.command("dut1", "ifup")
00048 
00049     def case(self):
00050         response = self.command("dut1", "socket new TCPSocket")
00051         socket_id = int(response.parsed['socket_id'])
00052 
00053         self.command("dut1", "socket " + str(socket_id) + " open")
00054         self.command("dut1", "socket " + str(socket_id) + " connect echo.mbedcloudtesting.com 7")
00055 
00056         for i in range(2):
00057             sentData = ""
00058             for size in (100, 200, 300, 120, 500):
00059                 packet = Randomize.random_string(max_len=size, min_len=size, chars=string.ascii_uppercase)
00060                 sentData += packet
00061                 response = self.command("dut1", "socket " + str(socket_id) + " send " + str(packet))
00062                 response.verify_trace("TCPSocket::send() returned: " + str(size))
00063 
00064             received = 0
00065             data = ""
00066             totalSize = 1220
00067             while received < totalSize:
00068                 response = self.command("dut1", "socket " + str(socket_id) + " recv " + str(totalSize))
00069                 data += response.parsed['data'].replace(":", "")
00070                 received += int(response.parsed['received_bytes'])
00071 
00072             if data != sentData:
00073                 raise TestStepFail("Received data doesn't match the sent data")
00074 
00075         self.command("dut1", "socket " + str(socket_id) + " delete")
00076 
00077     def teardown(self):
00078         self.command("dut1", "ifdown")