add tests for handle exception in radio receiver

mercury/tests/test_radio_receiver.py

@@ -11,16 +11,13 @@ def fake_event(event_uuid):
     """
         Mock a dummy AGEvent model
     """
-    if str(event_uuid) == "d81cac8d-26e1-4983-a942-1922e54a943d":
-        return AGEvent(
-            event_uuid=event_uuid,
-            event_name="fake event",
-            event_description="fake event",
-            event_date=datetime.datetime(2020, 2, 2, 20, 21, 22),
-            event_location="nyu",
-        )
-    else:
-        return None
+    return AGEvent(
+        event_uuid=event_uuid,
+        event_name="fake event",
+        event_description="fake event",
+        event_date=datetime.datetime(2020, 2, 2, 20, 21, 22),
+        event_location="nyu",
+    )
 
 
 def fake_valid(res):
@@ -54,11 +51,35 @@ def post_radio_data(self):
         )
         return response
 
-    @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
+    def post_defect_data(self):
+        response = self.client.post(
+            reverse(self.post_url, args=[self.uuid]),
+            data={
+                "values": {"power": "2", "speed": 1},
+                "date": datetime.datetime(2020, 2, 2, 20, 21, 22),
+            },
+        )
+        return response
+
     def test_Radio_Receiver_GET_No_Related_Event(self):
         response = self.client.get(reverse(self.get_url, args=[self.uuid2]))
         self.assertEqual(400, response.status_code)
 
+    @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_GET_Missing_Enable(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
+                "timeout": 1,
+            },
+        )
+        self.assertEqual(400, response.status_code)
+        self.assertEqual("Missing enable value in url", json.loads(response.content))
+
     @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
     @mock.patch("mercury.views.radioreceiver.serial_ports", fake_invalid_port)
     def test_Radio_Receiver_GET_No_Valid_Port(self):
@@ -69,13 +90,12 @@ def test_Radio_Receiver_GET_No_Valid_Port(self):
                 "baudrate": 9000,
                 "bytesize": 8,
                 "parity": "N",
-                "stop bits": 1,
+                "stopbits": 1,
                 "timeout": 1,
             },
         )
         self.assertEqual(200, response.status_code)
         self.assertEqual("No valid ports on the backend", json.loads(response.content))
-        # self.assertEqual("", response.)
 
     @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
     @mock.patch("mercury.views.radioreceiver.serial_ports", fake_valid_port)
@@ -87,7 +107,24 @@ def test_Radio_Receiver_GET_Success(self):
                 "baudrate": 9000,
                 "bytesize": 8,
                 "parity": "N",
-                "stop bits": 1,
+                "stopbits": 1,
+                "timeout": 1,
+            },
+        )
+        self.assertEqual(200, response.status_code)
+
+    @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
+    @mock.patch("mercury.views.radioreceiver.serial_ports", fake_valid_port)
+    @mock.patch("mercury.views.radioreceiver.check_port", fake_valid)
+    def test_Radio_Receiver_GET_Close_Port_Success(self):
+        response = self.client.get(
+            reverse(self.get_url, args=[self.uuid]),
+            data={
+                "enable": 0,
+                "baudrate": 9000,
+                "bytesize": 8,
+                "parity": "N",
+                "stopbits": 1,
                 "timeout": 1,
             },
         )
@@ -102,17 +139,28 @@ def test_Radio_Receiver_Fake_GET_Success(self):
                 "baudrate": 9000,
                 "bytesize": 8,
                 "parity": "N",
-                "stop bits": 1,
+                "stopbits": 1,
                 "timeout": 1,
                 "fake": 1,
             },
         )
         self.assertEqual(200, response.status_code)
 
-    @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
     def test_Radio_Receiver_POST_Event_Not_Exist(self):
         response = self.client.post(reverse(self.get_url, args=[self.uuid2]))
         self.assertEqual(400, response.status_code)
+        self.assertEqual("Wrong uuid in url", json.loads(response.content))
+
+    @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_POST_Missing_Params(self):
+        response = self.post_defect_data()
+        self.assertEqual(400, response.status_code)
+        self.assertTrue("Missing required params" in json.loads(response.content))
+
+    @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
+    def test_Radio_Receiver_POST_Fail_to_Save(self):
+        response = self.post_radio_data()
+        self.assertEqual(400, response.status_code)
 
     @mock.patch("mercury.models.AGEvent.objects.get", fake_event)
     @mock.patch("mercury.serializers.AGMeasurementSerializer.is_valid", fake_valid)

mercury/views/radioreceiver.py

@@ -25,11 +25,11 @@ def serial_ports():
     """
     if sys.platform.startswith("win"):
         ports = ["COM%s" % (i + 1) for i in range(256)]
+    elif sys.platform.startswith("darwin"):
+        ports = glob.glob("/dev/tty.*")
     elif sys.platform.startswith("linux") or sys.platform.startswith("cygwin"):
         # this excludes your current terminal "/dev/tty"
         ports = glob.glob("/dev/tty[A-Za-z]*")
-    elif sys.platform.startswith("darwin"):
-        ports = glob.glob("/dev/tty.*")
     else:
         raise EnvironmentError("Unsupported platform")
 
@@ -43,7 +43,7 @@ def serial_ports():
             pass
 
     if "TRAVIS" in os.environ:
-        return ["dev/tty.USB"]
+        result = ["dev/tty.USB"]
     return result
 
 
@@ -73,6 +73,14 @@ def call_script(uuid, port, fake):
     subprocess.call(command, shell=True)
 
 
+def check_port(ser):
+    """
+        Package ser.is_open for test purpose
+        TODO: Add other checks on Serial port
+    """
+    return ser.is_open
+
+
 class RadioReceiverView(APIView):
     """
     This is a Django REST API supporting user to send GET request fetching the RADIO
@@ -99,8 +107,8 @@ def get(self, request, event_uuid=None):
         try:
             event = AGEvent.objects.get(event_uuid=event_uuid)
         except AGEvent.DoesNotExist:
-            event = None
-        if event is None:
+            event = False
+        if event is False:
             return Response("Wrong uuid in url", status=status.HTTP_400_BAD_REQUEST)
 
         # Check Serial port parameters
@@ -129,7 +137,7 @@ def get(self, request, event_uuid=None):
         if params.get("parity"):
             ser.parity = params.get("parity")
         if params.get("stopbits"):
-            ser.stopbits = params.get("stopbits")
+            ser.stopbits = int(params.get("stopbits"))
         if params.get("timeout"):
             timeout = int(params.get("timeout"))
             ser.timeout = timeout
@@ -140,13 +148,13 @@ def get(self, request, event_uuid=None):
         elif enable:
             try:
                 ser.open()
-                if ser.is_open:
+                if check_port(ser):
                     # Call Script
                     call_script(event_uuid, ser.port, fake)
             except serial.serialutil.SerialException:
                 pass
         else:
-            if ser.is_open:
+            if check_port(ser):
                 ser.close()
 
         # Response data
@@ -174,7 +182,10 @@ def post(self, request, event_uuid=None):
         }
         """
         # First check event_uuid exists
-        event = AGEvent.objects.get(event_uuid=event_uuid)
+        try:
+            event = AGEvent.objects.get(event_uuid=event_uuid)
+        except AGEvent.DoesNotExist:
+            event = False
         if event is False:
             return Response("Wrong uuid in url", status=status.HTTP_400_BAD_REQUEST)
 
@@ -191,7 +202,8 @@ def post(self, request, event_uuid=None):
         for d in dic:
             if json_data.get(dic[d]) is None:
                 return Response(
-                    "Missing required params", status=status.HTTP_400_BAD_REQUEST
+                    "Missing required params " + dic[d],
+                    status=status.HTTP_400_BAD_REQUEST,
                 )
             res[d] = json_data[dic[d]]
 

scripts/radioport.py

@@ -42,6 +42,7 @@ def post_fake_request(self, data):
         print("Body: " + str(r.content))
 
     def listen_port(self):
+        print("Start listening port")
         while self.serial_port.is_open:
             data = self.serial_port.readline()
             self.post_request(data=data)
@@ -51,7 +52,7 @@ def listen_port(self):
     print("Call radioport.py script")
     parser = argparse.ArgumentParser(description="Process some integers.")
     parser.add_argument("--uuid", required=True, help="Event_uuid for AGEvent")
-    parser.add_argument("--port", required=True, help="Port name for serial")
+    parser.add_argument("--port", help="Port name for serial")
     parser.add_argument(
         "--fake",
         default=False,
@@ -65,17 +66,19 @@ def listen_port(self):
     )
     args = parser.parse_args()
 
-    print(args.data)
-    try:
-        ser = serial.Serial(args.port)
-        radio_port = RadioPort(args.uuid, ser)
-        if args.fake:
-            print("Send fake data")
-            radio_port.post_fake_request(args.data)
-        elif ser.is_open:
-            print("Start sending data")
-            radio_port.listen_port()
-        else:
+    if args.fake:
+        radio_port = RadioPort(args.uuid, None)
+        print("Send fake data")
+        print(args.data)
+        radio_port.post_fake_request(args.data)
+    else:
+        try:
+            ser = serial.Serial(args.port)
+            radio_port = RadioPort(args.uuid, ser)
+            if ser.is_open:
+                print("Start sending data")
+                radio_port.listen_port()
+            else:
+                print("Serial is invalid")
+        except serial.serialutil.SerialException:
             print("Serial is invalid")
-    except serial.serialutil.SerialException:
-        print("Serial is invalid")