add stricter test to detect invalid webcam frames

git-svn-id: https://xpra.org/svn/Xpra/trunk@23198 3bb7dfac-3a0b-4e04-842a-767bc560f471

src/unittests/unit/server/source_mixins_test.py

@@ -72,17 +72,26 @@ def send(*args):
             assert wm.start_virtual_webcam(device_id, w, h)
             assert wm.get_info().get("webcam", {}).get("active-devices", 0)==1
             assert len(packets)==1    #ack sent
+            assert packets[0][0]=="webcam-ack"
             frame_no = 0
-            encoding = "png"
-            buf = BytesIO()
             from PIL import Image
             image = Image.new('RGB', size=(w, h), color=(155, 0, 0))
-            image.save(buf, 'jpeg')
+            buf = BytesIO()
+            image.save(buf, "png")
             data = buf.getvalue()
             buf.close()
-            wm.process_webcam_frame(device_id, frame_no, encoding, w, h, data)
+            assert wm.process_webcam_frame(device_id, frame_no, "png", w, h, data)
             assert len(packets)==2    #ack sent
-            wm.stop_virtual_webcam(device_id)
+            assert packets[1][0]=="webcam-ack"
+            #now send a jpeg as png,
+            #which should fail and stop:
+            buf = BytesIO()
+            image.save(buf, "jpeg")
+            data = buf.getvalue()
+            buf.close()
+            assert not wm.process_webcam_frame(device_id, frame_no, "png", w, h, data)
+            assert len(packets)==3
+            assert packets[2][0]=="webcam-stop"
         finally:
             wm.cleanup()
 

src/xpra/server/source/webcam_mixin.py

@@ -160,7 +160,7 @@ def process_webcam_frame(self, device_id, frame_no, encoding, w, h, data):
         if not webcam:
             log.error("Error: webcam forwarding is not active, dropping frame")
             self.send_webcam_stop(device_id, "not started")
-            return
+            return False
         try:
             assert encoding in self.webcam_encodings, "invalid encoding specified: %s (must be one of %s)" % (encoding, self.webcam_encodings)
             rgb_pixel_format = "BGRX"       #BGRX
@@ -171,7 +171,7 @@ def process_webcam_frame(self, device_id, frame_no, encoding, w, h, data):
             src_format = webcam.get_src_format()
             if not src_format:
                 #closed / closing
-                return
+                return False
             #one of those two should be present
             try:
                 csc_mod = "csc_libyuv"
@@ -182,7 +182,7 @@ def process_webcam_frame(self, device_id, frame_no, encoding, w, h, data):
                     )
             except ImportError:
                 self.send_webcam_stop(device_id, "no csc module")
-                return
+                return False
             try:
                 assert rgb_pixel_format in get_input_colorspaces(), "unsupported RGB pixel format %s" % rgb_pixel_format
                 assert src_format in get_output_colorspaces(rgb_pixel_format), "unsupported output colourspace format %s" % src_format
@@ -191,7 +191,7 @@ def process_webcam_frame(self, device_id, frame_no, encoding, w, h, data):
                 log.error(" input-colorspaces: %s", csv(get_input_colorspaces()))
                 log.error(" output-colorspaces: %s", csv(get_output_colorspaces(rgb_pixel_format)))
                 self.send_webcam_stop(device_id, "csc format error")
-                return
+                return False
             tw = webcam.get_width()
             th = webcam.get_height()
             csc = ColorspaceConverter()
@@ -200,6 +200,7 @@ def process_webcam_frame(self, device_id, frame_no, encoding, w, h, data):
             webcam.push_image(image)
             #tell the client all is good:
             self.send_webcam_ack(device_id, frame_no)
+            return True
         except Exception as e:
             log("error on %ix%i frame %i using encoding %s", w, h, frame_no, encoding, exc_info=True)
             log.error("Error processing webcam frame:")
@@ -210,3 +211,4 @@ def process_webcam_frame(self, device_id, frame_no, encoding, w, h, data):
             log.error(" %s", msg)
             self.send_webcam_stop(device_id, msg)
             self.stop_virtual_webcam(device_id)
+            return False