How to testing/debugging IoT application (TCP, UDP, HTTP, MQTT)

1. Introduction
- When working with IoT application, we often work with protocols such as TCP/UDP, HTTP, MQTT, ... So we need to know some tools/utilities for debugging, testing. Beside the most famous tool for Internet monitor is Wireshark, I often write some simple application for testing, debugging using Python. I choose Python because it is easy to setup, learn, use. Here are some simple Python programs that are useful for our purpose.
1.1 Scan network
Before going to next steps we must ensure that IoT nodes exist on the network.
We can use "ping ip_or_mdns_name_of_node"
or an application called Fing-Network Scanner that is available on Apple store and Google store

Figure: Fing network scanner, alive node has black color

2. Testing TCP
- Create a TCP server and TCP client. Client will send the data to server, server convert to upper case and respond to client.

2.1 TCP Server

import socket

# bind all IP address
HOST = '0.0.0.0' 
# Listen on Port 
PORT = 44444 
#Size of receive buffer   
BUFFER_SIZE = 1024    
# Create a TCP/IP socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Bind the socket to the host and port
s.bind((HOST, PORT))
print('Listen for incoming connections')
s.listen(1)
while True:
    # Accept the incomming connection
    conn, addr = s.accept()
    # Print the info of client
    print ('Client connected with ' , addr)
    # Receive BUFFER_SIZE bytes data
    data = conn.recv(BUFFER_SIZE)
    if data:
        #print received data
        print('Client to Server: ' , data)
        # Convert to upper case and send back to Client
        conn.send(data.upper())
# Close connection
conn.close()

2.2 TCP Client

import socket

# Ip of local host
HOST = '127.0.0.1' 
# Connect to Port 
PORT = 44444 
#Size of send buffer   
BUFFER_SIZE = 1024   
# data to sent to server
data = 'hello, world!'
# Create a TCP/IP socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Connect to server
s.connect((HOST, PORT))
# send data to server
s.send(bytearray(data, 'utf-8'))
# Receive response from server
data = s.recv(BUFFER_SIZE)
# Close connection
s.close()
print ('Server to Client: ' , data)

2.3 Result

Figure: Testing TCP

3. Testing UDP
- Create a UDP server and UDP client. Client will send the data to server, server convert to upper case and respond to client.

3.1 UDP Server

import socket

# bind all IP address
HOST = '0.0.0.0' 
# Listen on Port 
PORT = 44444 
#Size of receive buffer   
BUFFER_SIZE = 1024    
# Create a TCP/IP socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# Bind the socket to the host and port
s.bind((HOST, PORT))
while True:
    # Receive BUFFER_SIZE bytes data
    # data is a list with 2 elements
    # first is data
    #second is client address
    data = s.recvfrom(BUFFER_SIZE)
    if data:
        #print received data
        print('Client to Server: ' , data)
        # Convert to upper case and send back to Client
        s.sendto(data[0].upper(), data[1])
# Close connection
s.close()

3.2 UDP Client

import socket

# Ip of local host
HOST = '127.0.0.1' 
# Connect to Port 
PORT = 44444 
#Size of send buffer   
BUFFER_SIZE = 1024   
# data to sent to server
data = 'hello, world!'
# Create a TCP/IP socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# send data to server
s.sendto(bytearray(data, 'utf-8'), (HOST, PORT))
# Receive response from server
# data is a list with 2 elements
# first is data
#second is client address
data = s.recvfrom(BUFFER_SIZE)
# Close connection
s.close()
print ('Server to Client: ' , data[0])

3.3 Result

Figure: Testing UDP

4. Testing HTTP

- In order to test HTTP we create HTTP GET or POST request and monitor the response (data and status code) from server. We will use the Requests library in Python.

- Install it by using command line: "sudo pip install requests" and to use it just "import requests"

4.1 HTTP GET request

import requests

# we use HTTP GET to get content of web page
r = requests.get('http://iotsharing.com')
# print status code response
print (r.status_code)
# print header response
print(r.headers)
# print content response
print (r.text)

4.2 HTTP POST request
- Using "r = requests.post(http://iotsharing.com/post) "
4.3 Result

Figure: Response for HTTP request testing

5. Testing MQTT Client

- In order to test MQTT Client we will use paho MQTT. You can refer Demo 15: How to build a system to update Price Tag automatically using Arduino ESP32

5.1 Testing MQTT publish

import sys
try:
    import paho.mqtt.publish as publish
except ImportError:
    import os
    import inspect
    cmd_subfolder = os.path.realpath(os.path.abspath(os.path.join(os.path.split(inspect.getfile( inspect.currentframe() ))[0],"../src")))
    if cmd_subfolder not in sys.path:
        sys.path.insert(0, cmd_subfolder)
    import paho.mqtt.publish as publish
mosquitto_ip = "192.168.1.104"
publish.single("test/message", hostname=mosquitto_ip)

5.2 Testing MQTT subscribe

import sys
try:
    import paho.mqtt.subscribe as subscribe
except ImportError:
    # This part is only required to run the example from within the examples
    # directory when the module itself is not installed.
    #
    # If you have the module installed, just use "import paho.mqtt.subscribe"
    import os
    import inspect
    cmd_subfolder = os.path.realpath(os.path.abspath(os.path.join(os.path.split(inspect.getfile( inspect.currentframe() ))[0],"../src")))
    if cmd_subfolder not in sys.path:
        sys.path.insert(0, cmd_subfolder)
    import paho.mqtt.subscribe as subscribe
    import paho.mqtt.client
mosquitto_ip = "192.168.1.104"
#callback function will be invoked when message is available
def print_msg(client, userdata, message):
    print("%s : %s" % (message.topic, message.payload))
#subscribe all topics
subscribe.callback(print_msg, "#", hostname=mosquitto_ip)