Report a data update download cut short as an error accessing the service, whatever urllib3 is installed

pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py

@@ -199,6 +199,20 @@ def update_data_file(self, data_file):
             self._log_check_error(message)
             return self._update_failed(data_file, status)
 
+        # A response that ended before the whole file arrived is an error
+        # accessing the service. urllib3 2 raises one, which the request
+        # above catches, but urllib3 1 gives the bytes that arrived, so the
+        # length is checked here as well.
+        short = self._cut_short(response)
+        if short is not None:
+            self._log_check_error(
+                "Error accessing data update service at '" + shown_url +
+                "' for engine '" + engine + "'",
+                "The response ended after " + str(short[0]) + " of the " +
+                str(short[1]) + " bytes it gave as its Content-Length.")
+            return self._update_failed(
+                data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR)
+
         # Verify the MD5 hash if enabled and the answer gives one.
 
         header_md5 = response.headers.get("content-md5")
@@ -431,6 +445,23 @@ def _with_cause(text, cause):
         return (text + ("" if text.endswith(".") else ".") +
                 " Error detail: " + mask_licence_keys_in_text(cause))
 
+    @staticmethod
+    def _cut_short(response):
+        # The bytes received and the Content-Length, when fewer arrived than
+        # it gave, or None. With a Content-Encoding the length counts the
+        # encoded bytes and requests gives the decoded ones, so the two are
+        # not compared.
+        promised = response.headers.get("content-length")
+        encoding = response.headers.get("content-encoding", "identity")
+        if promised is None or encoding.strip().lower() != "identity":
+            return None
+        try:
+            promised = int(promised)
+        except ValueError:
+            return None
+        received = len(response.content)
+        return (received, promised) if received < promised else None
+
     @staticmethod
     def _engine_type(data_file):
         return type(getattr(data_file, "flow_element", data_file)).__name__

pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py

@@ -22,6 +22,7 @@
 
 import datetime
 import gzip
+import http.server
 import os
 import random
 import shutil
@@ -783,5 +784,115 @@ def testFailedMd5CheckOnStartUpIsNotRepeatedAtOnce(self):
             self.statuses)
 
 
+# The bytes a short download sends, and how many it promises.
+SENT = b"x" * 100
+PROMISED = 1000
+
+
+class ShortDownloadHandler(http.server.BaseHTTPRequestHandler):
+    """Answers 200 and closes the connection before the whole file has been
+    sent, with the length given in Content-Length or in chunks."""
+
+    protocol_version = "HTTP/1.1"
+    chunked = False
+
+    def do_GET(self):
+        self.send_response(200)
+        if self.chunked:
+            self.send_header("Transfer-Encoding", "chunked")
+            self.end_headers()
+            # One chunk, then the connection closes with no last chunk.
+            self.wfile.write(b"%x\r\n" % len(SENT) + SENT + b"\r\n")
+        else:
+            self.send_header("Content-Length", str(PROMISED))
+            self.end_headers()
+            self.wfile.write(SENT)
+        self.wfile.flush()
+        self.close_connection = True
+
+    def log_message(self, format, *args):
+        pass
+
+
+class ShortDownloadTests(unittest.TestCase):
+    """A response that ends before the whole file arrived is reported as an
+    error accessing the data update service, and the data in use is kept."""
+
+    def serve(self, chunked):
+        handler = type(
+            "Handler", (ShortDownloadHandler,), {"chunked": chunked})
+        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(server.server_close)
+        self.addCleanup(server.shutdown)
+        return "http://127.0.0.1:" + str(server.server_address[1]) + "/update"
+
+    def check_short_download(self, chunked):
+        pipeline = RecordingPipeline()
+        service = DataFileUpdateService(pipeline)
+        statuses = []
+        service.on_complete(lambda status, data_file: statuses.append(status))
+        data_file = HttpDataFile()
+        data_file.verify_if_modified_since = False
+        data_file.url = self.serve(chunked)
+
+        self.assertFalse(service.update_data_file(data_file))
+
+        self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], statuses)
+        errors = [message for level, message in pipeline.logs
+                  if level == "error"]
+        self.assertTrue(
+            any(message.startswith(
+                CHECK_ERROR + "Error accessing data update service at '" +
+                data_file.url + "' for engine 'FakeEngine'")
+                for message in errors),
+            errors)
+        self.assertEqual(b"old", data_file.data)
+        self.assertEqual([], data_file.refreshed)
+
+    def testAResponseShorterThanItsContentLengthIsAnAccessError(self):
+        self.check_short_download(chunked=False)
+
+    def testTheLengthIsCheckedWhicheverHttpLibraryIsInstalled(self):
+        # urllib3 before version 2 gives requests the bytes that arrived
+        # with no error, so the service compares them with Content-Length
+        # itself. A reply that holds fewer bytes stands in for that here.
+        pipeline = RecordingPipeline()
+        service = DataFileUpdateService(pipeline)
+        statuses = []
+        service.on_complete(lambda status, data_file: statuses.append(status))
+        data_file = HttpDataFile()
+        reply = response(200, SENT)
+        reply.headers = {"content-length": str(PROMISED)}
+        with mock.patch.object(requests, "get", lambda url, **kwargs: reply):
+            self.assertFalse(service.update_data_file(data_file))
+
+        self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], statuses)
+        self.assertIn(
+            ("error",
+             CHECK_ERROR + "Error accessing data update service at '" + URL +
+             "' for engine 'FakeEngine'. Error detail: The response ended "
+             "after 100 of the 1000 bytes it gave as its Content-Length."),
+            pipeline.logs)
+        self.assertEqual(b"old", data_file.data)
+
+    def testAResponseInAContentEncodingIsNotMeasured(self):
+        # Content-Length counts the encoded bytes, and requests gives the
+        # decoded ones, so the two are not compared.
+        pipeline = RecordingPipeline()
+        service = DataFileUpdateService(pipeline)
+        data_file = HttpDataFile()
+        reply = response(200, SENT)
+        reply.headers = {
+            "content-length": str(PROMISED), "content-encoding": "gzip"}
+        with mock.patch.object(requests, "get", lambda url, **kwargs: reply):
+            self.assertTrue(service.update_data_file(data_file))
+        self.assertEqual(SENT, data_file.data)
+
+    def testAChunkedResponseCutShortIsAnAccessError(self):
+        self.check_short_download(chunked=True)
+
+
 if __name__ == "__main__":
     unittest.main()