Added a Error Msg Type to underlying RPC infra; RPC service now returns error for add/modify/delete stream if transmit is running; added prints for start/stop transmit/capture NOP cases; added a rpctest.py script

binding/rpc.py

@@ -74,6 +74,7 @@ class OstinatoRpcChannel(RpcChannel):
         MSG_TYPE_REQUEST = 1
         MSG_TYPE_RESPONSE = 2
         MSG_TYPE_BLOB = 3
+        MSG_TYPE_ERROR = 4
 
         error = ''
         try:
@@ -118,6 +119,8 @@ class OstinatoRpcChannel(RpcChannel):
                     self.log.debug('parsed response %s', response)
             elif msg_type == MSG_TYPE_BLOB:
                 response = resp
+            elif msg_type == MSG_TYPE_ERROR:
+                raise RpcError(unicode(resp, 'utf-8'))
             else:
                 raise RpcError('unknown RPC msg type %d' % msg_type)
 
@@ -152,7 +155,7 @@ class OstinatoRpcChannel(RpcChannel):
             self.log.exception(error)
             raise
         except RpcError as e:
-            error = 'ERROR: Unknown reply received for RPC %s() (%s) ' % (
+            error = 'ERROR: error received for RPC %s() (%s) ' % (
                     method.name, e)
             self.log.exception(error)
             raise

rpc/pbrpcchannel.cpp

@@ -22,6 +22,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 
 #include <qendian.h>
 
+static uchar msgBuf[4096];
+
 PbRpcChannel::PbRpcChannel(QHostAddress ip, quint16 port)
 {
     isPending = false;
@@ -94,8 +96,7 @@ void PbRpcChannel::CallMethod(
     ::google::protobuf::Message *response,
     ::google::protobuf::Closure* done)
 {
-    char    msgBuf[PB_HDR_SIZE];
+    char* const msg = (char*) &msgBuf[0];
-    char* const msg = &msgBuf[0];
     int     len;
     bool    ret;
   
@@ -161,8 +162,7 @@ void PbRpcChannel::CallMethod(
 
 void PbRpcChannel::on_mpSocket_readyRead()
 {
-    uchar   msg[PB_HDR_SIZE];
+    uchar   *msg = (uchar*) &msgBuf;
-    uchar   *p = (uchar*) &msg;
     int        msgLen;
     static bool parsing = false;
     static quint16    type, method;
@@ -183,9 +183,9 @@ void PbRpcChannel::on_mpSocket_readyRead()
 
         Q_ASSERT(msgLen == PB_HDR_SIZE);
 
-        type = qFromBigEndian<quint16>(p+0);
+        type = qFromBigEndian<quint16>(msg+0);
-        method = qFromBigEndian<quint16>(p+2);
+        method = qFromBigEndian<quint16>(msg+2);
-        len = qFromBigEndian<quint32>(p+4);
+        len = qFromBigEndian<quint32>(msg+4);
 
         //BUFDUMP(msg, PB_HDR_SIZE);
         //qDebug("type = %hu, method = %hu, len = %u", type, method, len);
@@ -207,8 +207,8 @@ void PbRpcChannel::on_mpSocket_readyRead()
             {
                 int l;
 
-                l = mpSocket->read((char*)msg, sizeof(msg));
+                l = mpSocket->read((char*)msgBuf, sizeof(msgBuf));
-                blob->write((char*)msg, l);
+                blob->write((char*)msgBuf, l);
                 cumLen += l;
             }
 
@@ -221,13 +221,13 @@ void PbRpcChannel::on_mpSocket_readyRead()
 
             if (!isPending)
             {
-                qDebug("not waiting for response");
+                qWarning("not waiting for response");
                 goto _error_exit2;
             }
 
             if (pendingMethodId != method)
             {
-                qDebug("invalid method id %d (expected = %d)", method, 
+                qWarning("invalid method id %d (expected = %d)", method, 
                     pendingMethodId);
                 goto _error_exit2;
             }
@@ -241,13 +241,13 @@ void PbRpcChannel::on_mpSocket_readyRead()
 
             if (!isPending)
             {
-                qDebug("not waiting for response");
+                qWarning("not waiting for response");
                 goto _error_exit;
             }
 
             if (pendingMethodId != method)
             {
-                qDebug("invalid method id %d (expected = %d)", method, 
+                qWarning("invalid method id %d (expected = %d)", method, 
                     pendingMethodId);
                 goto _error_exit;
             }
@@ -274,6 +274,47 @@ void PbRpcChannel::on_mpSocket_readyRead()
             }
             break;
 
+        case PB_MSG_TYPE_ERROR:
+        {
+            static quint32 cumLen = 0;
+            static QByteArray error;
+
+            while ((cumLen < len) && mpSocket->bytesAvailable())
+            {
+                int l;
+
+                l = mpSocket->read((char*)msgBuf, sizeof(msgBuf));
+                error.append(QByteArray((char*)msgBuf,l));
+                cumLen += l;
+            }
+
+            qDebug("%s: error rcvd %d/%d", __PRETTY_FUNCTION__, cumLen, len);
+
+            if (cumLen < len)
+                return;
+
+            static_cast<PbRpcController*>(controller)->SetFailed(
+                    QString::fromUtf8(error, len));
+
+            cumLen = 0;
+            error.resize(0);
+
+            if (!isPending)
+            {
+                qWarning("not waiting for response");
+                goto _error_exit2;
+            }
+
+            if (pendingMethodId != method)
+            {
+                qWarning("invalid method id %d (expected = %d)", method, 
+                    pendingMethodId);
+                goto _error_exit2;
+            }
+
+            break;
+        }
+
         default:
             qFatal("%s: unexpected type %d", __PRETTY_FUNCTION__, type);
             goto _error_exit;

rpc/pbrpccommon.h

@@ -35,5 +35,6 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 #define PB_MSG_TYPE_REQUEST        1
 #define PB_MSG_TYPE_RESPONSE    2
 #define PB_MSG_TYPE_BINBLOB        3
+#define PB_MSG_TYPE_ERROR          4
 
 #endif

rpc/pbrpccontroller.h

@@ -44,14 +44,17 @@ public:
     ::google::protobuf::Message* response() { return response_; }
 
     // Client Side Methods
-    void Reset() { failed = false; blob = NULL; }
+    void Reset() { failed = false; blob = NULL; errStr = ""; }
     bool Failed() const { return failed; }
     void StartCancel() { /*! \todo (MED) */}
-    std::string ErrorText() const { return errStr; }
+    std::string ErrorText() const { return errStr.toStdString(); }
 
     // Server Side Methods
+    void SetFailed(const QString &reason) 
+        { failed = true; errStr = reason; qWarning(qPrintable(errStr)); }
     void SetFailed(const std::string &reason) 
-        { failed = true; errStr = reason; }
+        { SetFailed(QString::fromStdString(reason)); }
+    QString ErrorString() const { return errStr; }
     bool IsCanceled() const { return false; };
     void NotifyOnCancel(::google::protobuf::Closure* /* callback */) {
         /*! \todo (MED) */ 
@@ -64,7 +67,7 @@ public:
 private:
     bool failed;
     QIODevice *blob;
-    std::string errStr;
+    QString errStr;
     ::google::protobuf::Message *request_;
     ::google::protobuf::Message *response_;
 

rpc/rpcconn.cpp

@@ -101,6 +101,14 @@ void RpcConnection::start()
         this, SLOT(on_clientSock_error(QAbstractSocket::SocketError)));
 }
 
+void RpcConnection::writeHeader(char* header, quint16 type, quint16 method, 
+                                quint32 length)
+{
+    *((quint16*)(header+0)) = qToBigEndian(type);
+    *((quint16*)(header+2)) = qToBigEndian(method);
+    *((quint32*)(header+4)) = qToBigEndian(length);
+}
+
 void RpcConnection::sendRpcReply(PbRpcController *controller)
 {
     google::protobuf::Message *response = controller->response();
@@ -111,7 +119,14 @@ void RpcConnection::sendRpcReply(PbRpcController *controller)
 
     if (controller->Failed())
     {
-        qDebug("rpc failed");
+        QByteArray err = controller->ErrorString().toUtf8();
+
+        qWarning("rpc failed (%s)", qPrintable(controller->ErrorString()));
+        len = err.size();
+        writeHeader(msg, PB_MSG_TYPE_ERROR, pendingMethodId, len);
+        clientSock->write(msg, PB_HDR_SIZE);
+        clientSock->write(err.constData(), len);
+
         goto _exit;
     }
 
@@ -121,10 +136,7 @@ void RpcConnection::sendRpcReply(PbRpcController *controller)
         len = blob->size();
         qDebug("is binary blob of len %d", len);
 
-        *((quint16*)(msg+0)) = qToBigEndian(quint16(PB_MSG_TYPE_BINBLOB)); // type
+        writeHeader(msg, PB_MSG_TYPE_BINBLOB, pendingMethodId, len);
-        *((quint16*)(msg+2)) = qToBigEndian(quint16(pendingMethodId)); // method
-        (*(quint32*)(msg+4)) = qToBigEndian(quint32(len)); // len
-
         clientSock->write(msg, PB_HDR_SIZE);
 
         blob->seek(0);
@@ -152,10 +164,7 @@ void RpcConnection::sendRpcReply(PbRpcController *controller)
     }
 
     len = response->ByteSize();
-
+    writeHeader(msg, PB_MSG_TYPE_RESPONSE, pendingMethodId, len);
-    *((quint16*)(msg+0)) = qToBigEndian(quint16(PB_MSG_TYPE_RESPONSE)); // type
-    *((quint16*)(msg+2)) = qToBigEndian(quint16(pendingMethodId)); // method
-    *((quint32*)(msg+4)) = qToBigEndian(quint32(len)); // len
 
     // Avoid printing stats since it happens once every couple of seconds
     if (pendingMethodId != 13)

rpc/rpcconn.h

@@ -45,6 +45,8 @@ public:
     static void connIdMsgHandler(QtMsgType type, const char* msg);
 
 private:
+    void writeHeader(char* header, quint16 type, quint16 method, 
+                     quint32 length);
     void sendRpcReply(PbRpcController *controller);
 
 signals:

server/myservice.cpp

@@ -208,6 +208,9 @@ void MyService::addStream(::google::protobuf::RpcController* controller,
     if ((portId < 0) || (portId >= portInfo.size()))
         goto _invalid_port;
 
+    if (portInfo[portId]->isTransmitOn())
+        goto _port_busy;
+
     portLock[portId]->lockForWrite();
     for (int i = 0; i < request->stream_id_size(); i++)
     {
@@ -232,8 +235,13 @@ void MyService::addStream(::google::protobuf::RpcController* controller,
     done->Run();
     return;
 
+_port_busy:
+    controller->SetFailed("Port Busy");
+    goto _exit;
+
 _invalid_port:
     controller->SetFailed("invalid portid");
+_exit:
     done->Run();
 }
 
@@ -250,6 +258,9 @@ void MyService::deleteStream(::google::protobuf::RpcController* controller,
     if ((portId < 0) || (portId >= portInfo.size()))
         goto _invalid_port;
 
+    if (portInfo[portId]->isTransmitOn())
+        goto _port_busy;
+
     portLock[portId]->lockForWrite();
     for (int i = 0; i < request->stream_id_size(); i++)
         portInfo[portId]->deleteStream(request->stream_id(i).id());
@@ -260,8 +271,12 @@ void MyService::deleteStream(::google::protobuf::RpcController* controller,
     done->Run();
     return;
 
+_port_busy:
+    controller->SetFailed("Port Busy");
+    goto _exit;
 _invalid_port:
     controller->SetFailed("invalid portid");
+_exit:
     done->Run();
 }
 
@@ -278,6 +293,9 @@ void MyService::modifyStream(::google::protobuf::RpcController* controller,
     if ((portId < 0) || (portId >= portInfo.size()))
         goto _invalid_port;
 
+    if (portInfo[portId]->isTransmitOn())
+        goto _port_busy;
+
     portLock[portId]->lockForWrite();
     for (int i = 0; i < request->stream_size(); i++)
     {
@@ -300,8 +318,12 @@ void MyService::modifyStream(::google::protobuf::RpcController* controller,
     done->Run();
     return;
 
+_port_busy:
+    controller->SetFailed("Port Busy");
+    goto _exit;
 _invalid_port:
     controller->SetFailed("invalid portid");
+_exit:
     done->Run();
 }
 

server/pcapport.cpp

@@ -597,8 +597,10 @@ _exit:
 void PcapPort::PortTransmitter::start()
 {
     // FIXME: return error
-    if (state_ == kRunning)
+    if (state_ == kRunning) {
+        qWarning("Transmit start requested but is already running!");
         return;
+    }
 
     state_ = kNotStarted;
     QThread::start();
@@ -614,6 +616,16 @@ void PcapPort::PortTransmitter::stop()
         while (state_ == kRunning)
             QThread::msleep(10);
     }
+    else {
+        // FIXME: return error
+        qWarning("Transmit stop requested but is not running!");
+        return;
+    }
+}
+
+bool PcapPort::PortTransmitter::isRunning()
+{
+    return (state_ == kRunning);
 }
 
 int PcapPort::PortTransmitter::sendQueueTransmit(pcap_t *p,
@@ -808,8 +820,10 @@ _exit:
 void PcapPort::PortCapturer::start()
 {
     // FIXME: return error
-    if (state_ == kRunning)
+    if (state_ == kRunning) {
+        qWarning("Capture start requested but is already running!");
         return;
+    }
 
     state_ = kNotStarted;
     QThread::start();
@@ -825,6 +839,16 @@ void PcapPort::PortCapturer::stop()
         while (state_ == kRunning)
             QThread::msleep(10);
     }
+    else {
+        // FIXME: return error
+        qWarning("Capture stop requested but is not running!");
+        return;
+    }
+}
+
+bool PcapPort::PortCapturer::isRunning()
+{
+    return (state_ == kRunning);
 }
 
 QFile* PcapPort::PortCapturer::captureFile()

server/pcapport.h

@@ -116,6 +116,7 @@ protected:
         void run();
         void start();
         void stop();
+        bool isRunning();
     private:
         enum State 
         {
@@ -201,6 +202,7 @@ protected:
         void run();
         void start();
         void stop();
+        bool isRunning();
         QFile* captureFile();
 
     private:

test/rpctest.py

@@ -0,0 +1,412 @@
+#! /usr/bin/env python
+
+# standard modules
+import logging
+import os
+import subprocess
+import sys
+import time
+
+sys.path.append('../binding')
+from core import ost_pb, DroneProxy
+from rpc import RpcError
+from protocols.mac_pb2 import mac
+from protocols.ip4_pb2 import ip4, Ip4
+
+class Test:
+    pass
+
+class TestSuite:
+    def __init__(self):
+        self.results = []
+        self.total = 0
+        self.passed = 0
+        self.completed = False
+
+    def test_begin(self, name):
+        test = Test()
+        test.name = name
+        test.passed = False
+        self.running = test
+        print('-----------------------------------------------------------')
+        print('@@TEST: %s' % name)
+        print('-----------------------------------------------------------')
+
+    def test_end(self, result):
+        if self.running:
+            self.running.passed = result
+            self.results.append(self.running)
+            self.total = self.total + 1
+            if result:
+                self.passed = self.passed + 1
+            self.running = None
+            print('@@RESULT: %s' % ('PASS' if result else 'FAIL'))
+        else:
+            raise Exception('Test end without a test begin')
+
+    def report(self):
+        print('===========================================================')
+        print('TEST REPORT')
+        print('===========================================================')
+        for test in self.results:
+            print('%s: %d' % (test.name, test.passed))
+        print('Passed: %d/%d' % (self.passed, self.total))
+        print('Completed: %d' % (self.completed))
+
+    def complete(self):
+        self.completed = True
+
+    def passed(self):
+        return passed == total and self.completed
+
+# initialize defaults
+host_name = '127.0.0.1'
+tx_port_number = -1
+rx_port_number = -1 
+
+if sys.platform == 'win32':
+    tshark = r'C:\Program Files\Wireshark\tshark.exe'
+else:
+    tshark = 'tshark'
+    
+
+# setup logging
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+print('')
+print('This test uses the following topology -')
+print('')
+print(' +-------+           ')
+print(' |       |Tx--->----+')
+print(' | Drone |          |')
+print(' |       |Rx---<----+')
+print(' +-------+           ')
+print('')
+print('A loopback port is used as both the Tx and Rx ports')
+print('')
+
+suite = TestSuite()
+drone = DroneProxy(host_name)
+
+try:
+    # ----------------------------------------------------------------- #
+    # Baseline Configuration
+    # ----------------------------------------------------------------- #
+
+    # connect to drone
+    log.info('connecting to drone(%s:%d)' 
+            % (drone.hostName(), drone.portNumber()))
+    drone.connect()
+
+    # retreive port id list
+    log.info('retreiving port list')
+    port_id_list = drone.getPortIdList()
+
+    # retreive port config list
+    log.info('retreiving port config for all ports')
+    port_config_list = drone.getPortConfig(port_id_list)
+
+    if len(port_config_list.port) == 0:
+        log.warning('drone has no ports!')
+        sys.exit(1)
+
+    # iterate port list to find a loopback port to use as the tx/rx port id 
+    print('Port List')
+    print('---------')
+    for port in port_config_list.port:
+        print('%d.%s (%s)' % (port.port_id.id, port.name, port.description))
+        # use a loopback port as default tx/rx port 
+        if ('lo' in port.name or 'loopback' in port.description.lower()):
+            tx_port_number = port.port_id.id
+            rx_port_number = port.port_id.id
+
+    if tx_port_number < 0 or rx_port_number < 0:
+        log.warning('loopback port not found')
+        sys.exit(1)
+
+    print('Using port %d as tx/rx port(s)')
+
+    tx_port = ost_pb.PortIdList()
+    tx_port.port_id.add().id = tx_port_number;
+
+    rx_port = ost_pb.PortIdList()
+    rx_port.port_id.add().id = rx_port_number;
+
+    # add a stream
+    stream_id = ost_pb.StreamIdList()
+    stream_id.port_id.CopyFrom(tx_port.port_id[0])
+    stream_id.stream_id.add().id = 1
+    log.info('adding tx_stream %d' % stream_id.stream_id[0].id)
+    drone.addStream(stream_id)
+
+    # configure the stream
+    stream_cfg = ost_pb.StreamConfigList()
+    stream_cfg.port_id.CopyFrom(tx_port.port_id[0])
+    s = stream_cfg.stream.add()
+    s.stream_id.id = stream_id.stream_id[0].id
+    s.core.is_enabled = True
+    s.control.num_packets = 10
+
+    # setup stream protocols as mac:eth2:ip4:udp:payload
+    p = s.protocol.add()
+    p.protocol_id.id = ost_pb.Protocol.kMacFieldNumber
+    p.Extensions[mac].dst_mac = 0x001122334455
+    p.Extensions[mac].src_mac = 0x00aabbccddee
+
+    p = s.protocol.add()
+    p.protocol_id.id = ost_pb.Protocol.kEth2FieldNumber
+
+    p = s.protocol.add()
+    p.protocol_id.id = ost_pb.Protocol.kIp4FieldNumber
+    # reduce typing by creating a shorter reference to p.Extensions[ip4]
+    ip = p.Extensions[ip4]
+    ip.src_ip = 0x01020304
+    ip.dst_ip = 0x05060708
+    ip.dst_ip_mode = Ip4.e_im_inc_host
+
+    s.protocol.add().protocol_id.id = ost_pb.Protocol.kUdpFieldNumber
+    s.protocol.add().protocol_id.id = ost_pb.Protocol.kPayloadFieldNumber
+
+    log.info('configuring tx_stream %d' % stream_id.stream_id[0].id)
+    drone.modifyStream(stream_cfg)
+
+    # clear tx/rx stats
+    log.info('clearing tx/rx stats')
+    drone.clearStats(tx_port)
+    drone.clearStats(rx_port)
+
+    # ----------------------------------------------------------------- #
+    # TESTCASE: Verify invoking addStream() during transmit fails
+    # TESTCASE: Verify invoking modifyStream() during transmit fails
+    # TESTCASE: Verify invoking deleteStream() during transmit fails
+    # ----------------------------------------------------------------- #
+    sid = ost_pb.StreamIdList()
+    sid.port_id.CopyFrom(tx_port.port_id[0])
+    sid.stream_id.add().id = 2
+
+    passed = False
+    suite.test_begin('addStreamDuringTransmitFails')
+    drone.startTx(tx_port)
+    try:
+        log.info('adding tx_stream %d' % sid.stream_id[0].id)
+        drone.addStream(sid)
+    except RpcError as e:
+        if ('Port Busy' in str(e)):
+            passed = True
+        else:
+            raise
+    finally:
+        drone.stopTx(tx_port)
+        suite.test_end(passed)
+
+    passed = False
+    suite.test_begin('modifyStreamDuringTransmitFails')
+    scfg = ost_pb.StreamConfigList()
+    scfg.port_id.CopyFrom(tx_port.port_id[0])
+    s = scfg.stream.add()
+    s.stream_id.id = sid.stream_id[0].id
+    s.protocol.add().protocol_id.id = ost_pb.Protocol.kMacFieldNumber
+    s.protocol.add().protocol_id.id = ost_pb.Protocol.kArpFieldNumber
+    s.protocol.add().protocol_id.id = ost_pb.Protocol.kPayloadFieldNumber
+    drone.startTx(tx_port)
+    try:
+        log.info('configuring tx_stream %d' % sid.stream_id[0].id)
+        drone.modifyStream(scfg)
+    except RpcError as e:
+        if ('Port Busy' in str(e)):
+            passed = True
+        else:
+            raise
+    finally:
+        drone.stopTx(tx_port)
+        suite.test_end(passed)
+
+    passed = False
+    suite.test_begin('deleteStreamDuringTransmitFails')
+    drone.startTx(tx_port)
+    try:
+        log.info('deleting tx_stream %d' % sid.stream_id[0].id)
+        drone.deleteStream(sid)
+    except RpcError as e:
+        if ('Port Busy' in str(e)):
+            passed = True
+        else:
+            raise
+    finally:
+        drone.stopTx(tx_port)
+        suite.test_end(passed)
+
+
+    # ----------------------------------------------------------------- #
+    # TESTCASE: Verify invoking startTx() during transmit is a NOP, 
+    #           not a restart
+    # ----------------------------------------------------------------- #
+    passed = False
+    suite.test_begin('startTxDuringTransmitIsNopNotRestart')
+    drone.startCapture(rx_port)
+    drone.startTx(tx_port)
+    try:
+        log.info('sleeping for 4s ...')
+        time.sleep(4)
+        log.info('starting transmit multiple times')
+        drone.startTx(tx_port)
+        time.sleep(1)
+        drone.startTx(tx_port)
+        time.sleep(1)
+        drone.startTx(tx_port)
+        time.sleep(1)
+        log.info('waiting for transmit to finish ...')
+        time.sleep(5)
+        drone.stopTx(tx_port)
+        drone.stopCapture(rx_port)
+
+        buff = drone.getCaptureBuffer(rx_port.port_id[0])
+        drone.saveCaptureBuffer(buff, 'capture.pcap')
+        log.info('dumping Rx capture buffer')
+        cap_pkts = subprocess.check_output([tshark, '-r', 'capture.pcap'])
+        print(cap_pkts)
+        if '5.6.7.8' in cap_pkts:
+            passed = True
+        os.remove('capture.pcap')
+    except RpcError as e:
+            raise
+    finally:
+        drone.stopTx(tx_port)
+        suite.test_end(passed)
+
+
+    # ----------------------------------------------------------------- #
+    # TESTCASE: Verify invoking startCapture() during capture is a NOP, 
+    #           not a restart
+    # ----------------------------------------------------------------- #
+    passed = False
+    suite.test_begin('startCaptureDuringTransmitIsNopNotRestart')
+    try:
+        drone.startCapture(rx_port)
+        drone.startTx(tx_port)
+        log.info('sleeping for 4s ...')
+        time.sleep(4)
+        log.info('starting capture multiple times')
+        drone.startCapture(rx_port)
+        time.sleep(1)
+        drone.startCapture(rx_port)
+        time.sleep(1)
+        drone.startCapture(rx_port)
+        time.sleep(1)
+        log.info('waiting for transmit to finish ...')
+        time.sleep(5)
+        drone.stopTx(tx_port)
+        drone.stopCapture(rx_port)
+
+        buff = drone.getCaptureBuffer(rx_port.port_id[0])
+        drone.saveCaptureBuffer(buff, 'capture.pcap')
+        log.info('dumping Rx capture buffer')
+        cap_pkts = subprocess.check_output([tshark, '-r', 'capture.pcap'])
+        print(cap_pkts)
+        if '5.6.7.8' in cap_pkts:
+            passed = True
+        os.remove('capture.pcap')
+    except RpcError as e:
+            raise
+    finally:
+        drone.stopTx(tx_port)
+        suite.test_end(passed)
+
+    # ----------------------------------------------------------------- #
+    # TESTCASE: Verify invoking stopTx() when transmit is not running
+    #           is a NOP 
+    # ----------------------------------------------------------------- #
+    passed = False
+    suite.test_begin('stopTxWhenTransmitNotRunningIsNop')
+    try:
+        tx_stats = drone.getStats(tx_port)
+        log.info('--> (tx_stats)' + tx_stats.__str__())
+        if tx_stats.port_stats[0].state.is_transmit_on:
+            raise Exception('Unexpected transmit ON state')
+        log.info('stopping transmit multiple times')
+        drone.stopTx(tx_port)
+        time.sleep(1)
+        drone.stopTx(tx_port)
+        time.sleep(1)
+        drone.stopTx(tx_port)
+
+        # if we reached here, that means there was no exception
+        passed = True
+    except RpcError as e:
+            raise
+    finally:
+        suite.test_end(passed)
+
+    # ----------------------------------------------------------------- #
+    # TESTCASE: Verify invoking stopCapture() when capture is not running 
+    #           is a NOP 
+    # ----------------------------------------------------------------- #
+    passed = False
+    suite.test_begin('stopCaptureWhenCaptureNotRunningIsNop')
+    try:
+        rx_stats = drone.getStats(rx_port)
+        log.info('--> (rx_stats)' + rx_stats.__str__())
+        if rx_stats.port_stats[0].state.is_capture_on:
+            raise Exception('Unexpected capture ON state')
+        log.info('stopping capture multiple times')
+        drone.stopCapture(rx_port)
+        time.sleep(1)
+        drone.stopCapture(rx_port)
+        time.sleep(1)
+        drone.stopCapture(rx_port)
+
+        # if we reached here, that means there was no exception
+        passed = True
+    except RpcError as e:
+            raise
+    finally:
+        suite.test_end(passed)
+
+    # ----------------------------------------------------------------- #
+    # TESTCASE: Verify startCapture(), startTx() sequence captures the
+    #           first packet
+    # TESTCASE: Verify stopTx(), stopCapture() sequence captures the
+    #           last packet
+    # ----------------------------------------------------------------- #
+    passed = False
+    suite.test_begin('startStopTransmitCaptureOrderCapturesAllPackets')
+    try:
+        drone.startCapture(rx_port)
+        drone.startTx(tx_port)
+        log.info('waiting for transmit to finish ...')
+        time.sleep(12)
+        drone.stopTx(tx_port)
+        drone.stopCapture(rx_port)
+
+        log.info('getting Rx capture buffer')
+        buff = drone.getCaptureBuffer(rx_port.port_id[0])
+        drone.saveCaptureBuffer(buff, 'capture.pcap')
+        log.info('dumping Rx capture buffer')
+        cap_pkts = subprocess.check_output([tshark, '-r', 'capture.pcap'])
+        print(cap_pkts)
+        if '5.6.7.8' in cap_pkts and '5.6.7.17' in cap_pkts:
+            passed = True
+        os.remove('capture.pcap')
+    except RpcError as e:
+            raise
+    finally:
+        drone.stopTx(tx_port)
+        suite.test_end(passed)
+
+    suite.complete()
+
+    # delete streams
+    log.info('deleting tx_stream %d' % stream_id.stream_id[0].id)
+    drone.deleteStream(stream_id)
+
+    # bye for now
+    drone.disconnect()
+
+except Exception as ex:
+    log.exception(ex)
+
+finally:
+    suite.report()
+    if not suite.passed:
+        sys.exit(2);