From df11c1c0c505b2be8bbb18ed1c1b7a54d17d9e25 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 16:13:33 +0000 Subject: [PATCH 01/30] Align the data update service to the specification Send If-Modified-Since as an HTTP date, wait for the data update expected time before the first poll, add update_from_memory, and log the actions with the messages and levels in the Logging section of the data updates specification. --- .../datafile_update_service.py | 255 ++++++++++++-- .../tests/test_datafile_update_http.py | 328 ++++++++++++++++++ .../tests/test_datafile_update_service.py | 6 +- 3 files changed, 551 insertions(+), 38 deletions(-) create mode 100644 pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index c4c6b047..f318688a 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -20,6 +20,9 @@ # such notice(s) shall fulfill the requirements of that article. # ********************************************************************* +import calendar +import datetime +from email.utils import formatdate from enum import Enum import gzip import shutil @@ -99,44 +102,68 @@ def update_data_file(self, data_file): if data_file.updating: pass + self._log("info", "Checking for update") + # Get the update url url = data_file.get_update_url() + engine = self._engine_type(data_file) headers = {} status = UpdateStatus.AUTO_UPDATE_IN_PROGRESS # Add if modified since header if enabled and available + since = None if data_file.verify_if_modified_since: - try: - headers["If-Modified-Since"] = data_file.get_date_published().strftime('%Y-%m-%dT%H:%M:%SZ') - except: - pass - - response = requests.get(url, headers=headers) + since = self._if_modified_since(data_file) + if since: + headers["If-Modified-Since"] = since + + self._log( + "info", + "Checking for update from '" + url + "' for Engine '" + + engine + "'") + + retry = self._retry_text(data_file) + + try: + response = requests.get(url, headers=headers) + except requests.RequestException as e: + self._log( + "warning", + "An error occurred when connecting to " + url + + " in order to check for data file updates for " + engine + + "." + retry + " Error detail: " + str(e)) + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR, True) + + if response.status_code == 304: + self._log( + "info", + "No data newer than " + (since or "the data in use") + + " found at '" + url + "' for Engine '" + engine + "'") + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_NOT_NEEDED, True) if response.status_code != 200: # Handle errors if response.status_code == 429: - self.pipeline.log("error", "Too many requests to " + url + "for engine " + data_file.flow_element.datakey) status = UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS - elif response.status_code == 304: - self.pipeline.log("warn", "No data update available from " + url + "for engine " + data_file.flow_element.datakey) - status = UpdateStatus.AUTO_UPDATE_NOT_NEEDED elif response.status_code == 403: - self.pipeline.log("error", "Access denied from " + url + "for engine " + data_file.flow_element.datakey) status = UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN else: - self.pipeline.log("error", str(response.status_code) + url + "for engine " + data_file.flow_element.datakey) status = UpdateStatus.AUTO_UPDATE_HTTPS_ERR - data_file.updating = False - self.__emit(status, data_file) - self.check_next_update(data_file, run_after_error = True) - return False + self._log( + "warning", + "An error occurred while downloading a data file update " + "for " + engine + " from " + url + "." + retry + + " Error detail: HTTP status code " + + str(response.status_code)) + return self._update_failed(data_file, status, True) # Verify MD5 hash if enabled @@ -148,15 +175,23 @@ def update_data_file(self, data_file): for data in response.iter_content(8192): hash_md5.update(data) - - if header_md5 != hash_md5.hexdigest(): - self.pipeline.log("error", "MD5 doesn't match from " + url + "for engine " + data_file.flow_element.datakey) - status = UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED - data_file.updating = False - self.__emit(status, data_file) - self.check_next_update(data_file) - return False + if header_md5 != hash_md5.hexdigest(): + self._log( + "warning", + "An error occurred during the integrity check of new " + "data file for " + engine + "." + retry + + " Error detail: The MD5 hash of the data does not " + "match the content-md5 header.") + return self._update_failed( + data_file, + UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED, + False) + + self._log( + "info", + "Downloaded new data from '" + url + "' for Engine '" + + engine + "'") data = response.content @@ -181,14 +216,33 @@ def update_data_file(self, data_file): filename = data_file.path - with open(filename, 'wb') as f: - f.write(data) + try: + with open(filename, 'wb') as f: + f.write(data) + except OSError as e: + self._log( + "error", + "An error occurred when writing to " + filename + + ". Error detail: " + str(e)) + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM, + True) if data_file.create_temp_copy: # Move to real file - shutil.move(filename, data_file.path) + try: + shutil.move(filename, data_file.path) + except (OSError, shutil.Error): + self._log( + "error", + "An error occurred when copying a data file to " + "replace the existing one at '" + data_file.path + + "'.") + return self._update_failed( + data_file, + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, True) else: @@ -201,19 +255,143 @@ def update_data_file(self, data_file): # Done. Refresh the file. - try: - data_file.refresh() - status = UpdateStatus.AUTO_UPDATE_SUCCESS - except: - status = UpdateStatus.AUTO_UPDATE_REFRESH_FAILED + status = self._refresh(data_file) data_file.updating = False - + self.__emit(status, data_file) self.check_next_update(data_file) return True + def update_from_memory(self, data_file, data): + + """! + Refresh a data file that is kept in memory from the bytes given, + without downloading anything. + @type data_file: DataFile + @param data_file: Datafile to update, created with download set to + False + @type data: bytes + @param data: The new content of the data file + @rtype: UpdateStatus + @return: AUTO_UPDATE_SUCCESS, AUTO_UPDATE_REFRESH_FAILED if the + engine could not use the data (the data it had is kept), or + AUTO_UPDATE_NOT_NEEDED if the data file's engine has been closed + """ + + if data_file.download: + raise ValueError( + "update_from_memory needs a data file that is kept in " + "memory, created with download set to False.") + + if not self.__is_registered(data_file): + return UpdateStatus.AUTO_UPDATE_NOT_NEEDED + + previous = data_file.data + data_file.data = data + data_file.updating = True + try: + status = self._refresh(data_file) + finally: + data_file.updating = False + + if status != UpdateStatus.AUTO_UPDATE_SUCCESS: + data_file.data = previous + + self.__emit(status, data_file) + + return status + + def _refresh(self, data_file): + # Refresh the engine with the data file's new content. + engine = self._engine_type(data_file) + self._log( + "info", + "Attempting to refresh Engine '" + engine + "' with new data") + try: + data_file.refresh() + except Exception as e: + self._log( + "error", + "An error occurred while applying a data file update to " + + engine + ". Error detail: " + str(e)) + return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED + return UpdateStatus.AUTO_UPDATE_SUCCESS + + def _update_failed(self, data_file, status, run_after_error): + # An update that ends without new data. With run_after_error False, a + # check on start-up that has not yet succeeded is made again at once. + data_file.updating = False + self.__emit(status, data_file) + self.check_next_update(data_file, run_after_error=run_after_error) + return False + + def _log(self, level, message): + pipeline = getattr(self, "pipeline", None) + if pipeline is not None: + pipeline.log(level, message) + + @staticmethod + def _engine_type(data_file): + return type(getattr(data_file, "flow_element", data_file)).__name__ + + @staticmethod + def _retry_text(data_file): + # Only an automatic update is tried again. + if getattr(data_file, "auto_update", False): + return " Update will be attempted again later." + return "" + + @staticmethod + def _http_date(moment): + # An HTTP date, as RFC 9110 section 5.6.7 requires. A time with no + # zone is taken as UTC. + if isinstance(moment, datetime.datetime): + fields = moment.utctimetuple() + else: + fields = moment.timetuple() + return formatdate(calendar.timegm(fields), usegmt=True) + + def _if_modified_since(self, data_file): + # The publication date of the data file, or the time the file was + # last modified when it has no publication date. None when neither + # is known. + try: + published = data_file.get_date_published() + except Exception: + published = None + if published is None: + path = getattr(data_file, "path", None) + try: + published = datetime.datetime.fromtimestamp( + os.path.getmtime(path), datetime.timezone.utc) + except (OSError, TypeError): + return None + return self._http_date(published) + + @staticmethod + def _seconds_until_update_expected(data_file): + # Seconds from now until the data file says new data is expected, or + # 0 when it does not say or the time has passed. A date with no time + # is the default of a data file that does not know. + try: + expected = data_file.get_next_update() + except Exception: + return 0 + utc = datetime.timezone.utc + if isinstance(expected, datetime.datetime): + if expected.tzinfo is None: + expected = expected.replace(tzinfo=utc) + elif isinstance(expected, datetime.date): + if expected <= datetime.date.today(): + return 0 + expected = datetime.datetime( + expected.year, expected.month, expected.day, tzinfo=utc) + else: + return 0 + return max(0, (expected - datetime.datetime.now(utc)).total_seconds()) + def async_update_data_file(self, data_file): """! @@ -229,7 +407,9 @@ def async_update_data_file(self, data_file): def check_next_update(self, data_file, run_after_error = False): """! Function to periodically check if a datafile needs updating - also updates the datafile on start if it is set with the update_on_start option + also updates the datafile on start if it is set with the update_on_start option. + Polling does not start until after the time the data file expects + new data, if it gives one. @type data_file: DataFile @param data_file: Datafile to check updates for """ @@ -238,13 +418,17 @@ def check_next_update(self, data_file, run_after_error = False): return if data_file.update_on_start and not data_file.updated_on_start and not run_after_error: + self._log("info", "Updating on start-up") self.update_data_file(data_file) elif data_file.auto_update: - + random_offset = random.randint(0, data_file.update_time_maximum_randomisation) interval = random_offset + (data_file.polling_interval * 60) + until_expected = self._seconds_until_update_expected(data_file) + if until_expected > 0: + interval = until_expected + random_offset t = threading.Timer(interval, self.async_update_data_file, args=[data_file]) t.daemon = True t.name = "AutoUpdate - Scheduled" @@ -270,6 +454,7 @@ def register_data_file(self, data_file): # file_system_watcher is enabled on data_file if data_file.file_system_watcher: + self._log("info", "Creating file system watcher") data_file.last_changed_time = os.stat(data_file.path).st_mtime self.track_local_data_file_changes(data_file) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py new file mode 100644 index 00000000..2aa00668 --- /dev/null +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -0,0 +1,328 @@ + +import datetime +import os +import random +import shutil +import tempfile +import threading +import unittest +from email.utils import formatdate +from unittest import mock + +import requests + +from fiftyone_pipeline_engines.datafile_update_service import DataFileUpdateService +from fiftyone_pipeline_engines.datafile_update_service import UpdateStatus + +from .test_datafile_update_service import RecordingPipeline +from .test_datafile_update_service import ScheduledCheck + +URL = "https://example.com/update" + + +class FakeEngine: + """Stands in for the flow element of a data file.""" + datakey = "fake" + + +class HttpDataFile: + """Stands in for a data file that is kept in memory and updated over + HTTP.""" + + def __init__(self, path=None): + self.flow_element = FakeEngine() + self.identifier = "fake-data" + self.update_on_start = False + self.auto_update = False + self.file_system_watcher = False + self.updating = False + self.verify_if_modified_since = True + self.verify_md5 = False + self.decompress = False + self.download = False + self.create_temp_copy = False + self.polling_interval = 30 + self.update_time_maximum_randomisation = 0 + self.data = b"old" + self.refreshed = [] + self.published = datetime.datetime(2026, 11, 3, 10, 30, 0) + self.next_update = datetime.datetime(2026, 11, 3, 10, 30, 0) + self.refresh_error = None + if path is not None: + self.path = path + + def get_update_url(self): + return URL + + def get_date_published(self): + if isinstance(self.published, Exception): + raise self.published + return self.published + + def get_next_update(self): + return self.next_update + + def refresh(self): + if self.refresh_error: + raise self.refresh_error + self.refreshed.append(self.data) + + +def response(status, content=b"new"): + reply = mock.Mock() + reply.status_code = status + reply.content = content + reply.headers = {} + return reply + + +class UpdateServiceTestCase(unittest.TestCase): + + def setUp(self): + self.pipeline = RecordingPipeline() + self.service = DataFileUpdateService(self.pipeline) + self.statuses = [] + self.service.on_complete( + lambda status, data_file: self.statuses.append(status)) + self.data_file = HttpDataFile() + self.pending = [] + + def get(self, reply): + """Replaces the HTTP request and keeps the headers it was sent.""" + self.sent = [] + + def fake_get(url, headers=None, **kwargs): + self.sent.append(headers) + if isinstance(reply, Exception): + raise reply + return reply + + patch = mock.patch.object(requests, "get", fake_get) + patch.start() + self.addCleanup(patch.stop) + + def hold_timers(self): + """Replaces the timers, which are kept in self.pending and never run.""" + timer = mock.patch.object( + threading, "Timer", + lambda interval, function, args=None: ScheduledCheck( + self.pending, function, args or [])) + timer.start() + self.addCleanup(timer.stop) + + +# If-Modified-Since is an HTTP date (RFC 9110 section 5.6.7), set to the +# publication date, or the file system last modified time without one. +class IfModifiedSinceTests(UpdateServiceTestCase): + + def testPublishedDateIsSentAsHttpDate(self): + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertEqual( + "Tue, 03 Nov 2026 10:30:00 GMT", + self.sent[0]["If-Modified-Since"]) + + def testLastModifiedTimeIsUsedWithoutPublishedDate(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "data.dat") + with open(path, "w") as file: + file.write("x") + os.utime(path, (1700000000, 1700000000)) + self.data_file = HttpDataFile(path) + self.data_file.published = Exception("no published date") + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertEqual( + formatdate(1700000000, usegmt=True), + self.sent[0]["If-Modified-Since"]) + self.assertEqual( + "Tue, 14 Nov 2023 22:13:20 GMT", + self.sent[0]["If-Modified-Since"]) + + def testNoHeaderWhenDisabled(self): + self.data_file.verify_if_modified_since = False + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertNotIn("If-Modified-Since", self.sent[0]) + + +# Polling does not start until after the time the data update is expected, +# then waits the random offset as well. +class UpdateExpectedTimeTests(UpdateServiceTestCase): + + def setUp(self): + super(UpdateExpectedTimeTests, self).setUp() + self.intervals = [] + timer = mock.patch.object( + threading, "Timer", + lambda interval, function, args=None: self.record(interval)) + timer.start() + self.addCleanup(timer.stop) + randomise = mock.patch.object(random, "randint", lambda a, b: 5) + randomise.start() + self.addCleanup(randomise.stop) + self.data_file.auto_update = True + + def record(self, interval): + self.intervals.append(interval) + return ScheduledCheck([], None, []) + + @staticmethod + def utc_now(): + return datetime.datetime.now(datetime.timezone.utc).replace( + tzinfo=None) + + def testFirstCheckWaitsUntilExpectedTime(self): + self.data_file.next_update = ( + self.utc_now() + datetime.timedelta(hours=2)) + self.service.register_data_file(self.data_file) + self.assertEqual(1, len(self.intervals)) + self.assertAlmostEqual(7200 + 5, self.intervals[0], delta=10) + + def testPollingIntervalIsUsedOnceExpectedTimeHasPassed(self): + self.data_file.next_update = ( + self.utc_now() - datetime.timedelta(hours=2)) + self.service.register_data_file(self.data_file) + self.assertEqual([30 * 60 + 5], self.intervals) + + def testPlainDateOfTodayIsNotInTheFuture(self): + self.data_file.next_update = datetime.date.today() + self.service.register_data_file(self.data_file) + self.assertEqual([30 * 60 + 5], self.intervals) + + +# A memory data file can be refreshed from the bytes given. +class UpdateFromMemoryTests(UpdateServiceTestCase): + + def testMemoryDataFileIsRefreshedFromTheBytes(self): + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) + self.assertEqual(b"fresh", self.data_file.data) + self.assertEqual([b"fresh"], self.data_file.refreshed) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + + def testFailedRefreshIsReportedAndLogged(self): + self.data_file.refresh_error = Exception("boom") + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual(UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, status) + self.assertIn( + ("error", + "An error occurred while applying a data file update to " + "FakeEngine. Error detail: boom"), + self.pipeline.logs) + + def testFileDataFileIsRefused(self): + self.data_file.download = True + with self.assertRaises(ValueError): + self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual([], self.data_file.refreshed) + + def testUnregisteredDataFileIsLeftAlone(self): + self.service.unregister_data_file(self.data_file) + self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual([], self.data_file.refreshed) + + +# The messages and levels are those of the Logging section of the data +# updates specification. +class LoggingTests(UpdateServiceTestCase): + + def levels(self): + return [level for level, message in self.pipeline.logs] + + def testSuccessfulUpdateLogsEachAction(self): + self.get(response(200)) + self.service.update_data_file(self.data_file) + self.assertEqual([ + ("info", "Checking for update"), + ("info", + "Checking for update from '" + URL + "' for Engine " + "'FakeEngine'"), + ("info", + "Downloaded new data from '" + URL + "' for Engine " + "'FakeEngine'"), + ("info", + "Attempting to refresh Engine 'FakeEngine' with new data"), + ], self.pipeline.logs) + + def testNotModifiedIsInformationNotWarning(self): + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertIn( + ("info", + "No data newer than Tue, 03 Nov 2026 10:30:00 GMT found at '" + + URL + "' for Engine 'FakeEngine'"), + self.pipeline.logs) + for level in ("warn", "warning", "error"): + self.assertNotIn(level, self.levels()) + + def testRefusedDownloadIsWarning(self): + self.get(response(403)) + self.service.update_data_file(self.data_file) + self.assertIn( + ("warning", + "An error occurred while downloading a data file update for " + "FakeEngine from " + URL + ". Error detail: HTTP status code " + "403"), + self.pipeline.logs) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN], self.statuses) + + def testAutomaticUpdateSaysItWillBeTriedAgain(self): + self.data_file.auto_update = True + self.hold_timers() + self.get(response(500)) + self.service.update_data_file(self.data_file) + self.assertIn( + ("warning", + "An error occurred while downloading a data file update for " + "FakeEngine from " + URL + ". Update will be attempted again " + "later. Error detail: HTTP status code 500"), + self.pipeline.logs) + + def testConnectionFailureIsWarning(self): + self.get(requests.ConnectionError("boom")) + self.assertFalse(self.service.update_data_file(self.data_file)) + self.assertIn( + ("warning", + "An error occurred when connecting to " + URL + " in order to " + "check for data file updates for FakeEngine. Error detail: " + "boom"), + self.pipeline.logs) + self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) + + def testFailedRefreshIsError(self): + self.get(response(200)) + self.data_file.refresh_error = Exception("boom") + self.service.update_data_file(self.data_file) + self.assertIn( + ("error", + "An error occurred while applying a data file update to " + "FakeEngine. Error detail: boom"), + self.pipeline.logs) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) + + def testUpdateOnStartIsLogged(self): + self.data_file.update_on_start = True + self.data_file.updated_on_start = False + with mock.patch.object(self.service, "update_data_file"): + self.service.register_data_file(self.data_file) + self.assertIn(("info", "Updating on start-up"), self.pipeline.logs) + + def testFileSystemWatcherIsLogged(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + data_file = HttpDataFile(os.path.join(folder, "data.dat")) + with open(data_file.path, "w") as file: + file.write("x") + data_file.file_system_watcher = True + self.hold_timers() + self.service.register_data_file(data_file) + self.assertIn( + ("info", "Creating file system watcher"), self.pipeline.logs) + + +if __name__ == "__main__": + unittest.main() diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index 5fd9c872..b1f87363 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -130,9 +130,9 @@ def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): self.assertEqual("version 1", data_file.data) self.assertEqual( [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) - self.assertEqual(1, len(self.pipeline.logs)) - self.assertEqual("error", self.pipeline.logs[0][0]) - self.assertIn("The data file cannot be used.", self.pipeline.logs[0][1]) + errors = [log for log in self.pipeline.logs if log[0] == "error"] + self.assertEqual(1, len(errors)) + self.assertIn("The data file cannot be used.", errors[0][1]) self.assertEqual(1, len(self.pending)) self.write("version 2") From 18d4db32bcaf56e2c50dd044680f839af2beda7f Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 16:34:39 +0000 Subject: [PATCH 02/30] Treat a data file that is already current as a finished update in the update examples With If-Modified-Since sent as an HTTP date, the Distributor answers 304 when the file is current, which the examples reported as a failure. --- .../onpremise/datafileupdate_console.py | 3 +++ .../onpremise/datafileupdate_console.py | 3 +++ 2 files changed, 6 insertions(+) diff --git a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py index c6d0e6a5..9f7be5f2 100644 --- a/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py +++ b/device-detection/fiftyone_devicedetection_examples/src/fiftyone_devicedetection_examples/onpremise/datafileupdate_console.py @@ -354,6 +354,9 @@ def run(self, data_file, license_key, interactive, logger, output): output(f"Update on file modification complete, status: {update_event.status}") else: output("Update on file modification timed out") + elif update_event.status == UpdateStatus.AUTO_UPDATE_NOT_NEEDED: + output("The data file is already the latest published, " + "so there is nothing to download") else: logger.log("error", "Auto update was not successful, abandoning example") error_message = f"Auto update failed: {update_event.status}" diff --git a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/datafileupdate_console.py b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/datafileupdate_console.py index 54971693..b93b74b3 100644 --- a/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/datafileupdate_console.py +++ b/ip-intelligence/fiftyone_ipintelligence_examples/src/fiftyone_ipintelligence_examples/onpremise/datafileupdate_console.py @@ -231,6 +231,9 @@ def run(self, data_file, license_key, interactive, logger, output): f"{update_event.status}") else: output("Update on file modification timed out") + elif update_event.status == UpdateStatus.AUTO_UPDATE_NOT_NEEDED: + output("The data file is already the latest published, " + "so there is nothing to download") else: logger.log("error", "Auto update was not successful, abandoning example") error_message = f"Auto update failed: {update_event.status}" From a233b7f40bbf48a4bf04d416f8ba07737181ecf3 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 10:06:13 +0000 Subject: [PATCH 03/30] Test that no update message holds the licence key, the watcher needs automatic updates, a file data file is written, and content checks carry on --- .../tests/test_datafile_update_http.py | 174 +++++++++++++++++- .../tests/test_licence_key_masking.py | 18 +- 2 files changed, 182 insertions(+), 10 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 2aa00668..42ad1e2c 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -1,5 +1,6 @@ import datetime +import gzip import os import random import shutil @@ -19,6 +20,13 @@ URL = "https://example.com/update" +# An update URL that carries a licence key, which no message may hold in +# full. Only its last four characters may be shown. +KEY = "ABCDEFGHIJKL" +KEYED_URL = "https://example.com/update?licenseKeys=" + KEY + "&Type=HashV41" +MASKED_URL = ( + "https://example.com/update?licenseKeys=********IJKL&Type=HashV41") + class FakeEngine: """Stands in for the flow element of a data file.""" @@ -48,11 +56,12 @@ def __init__(self, path=None): self.published = datetime.datetime(2026, 11, 3, 10, 30, 0) self.next_update = datetime.datetime(2026, 11, 3, 10, 30, 0) self.refresh_error = None + self.url = URL if path is not None: self.path = path def get_update_url(self): - return URL + return self.url def get_date_published(self): if isinstance(self.published, Exception): @@ -212,11 +221,36 @@ def testFailedRefreshIsReportedAndLogged(self): "FakeEngine. Error detail: boom"), self.pipeline.logs) - def testFileDataFileIsRefused(self): + def testFileDataFileIsWrittenThenRefreshed(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "data.dat") + with open(path, "wb") as file: + file.write(b"old") + self.data_file = HttpDataFile(path) + self.data_file.download = True + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) + with open(path, "rb") as file: + self.assertEqual(b"fresh", file.read()) + self.assertEqual(1, len(self.data_file.refreshed)) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + + def testFailedWriteIsLoggedAndNotRefreshed(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "missing", "data.dat") + self.data_file = HttpDataFile(path) self.data_file.download = True - with self.assertRaises(ValueError): - self.service.update_from_memory(self.data_file, b"fresh") + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual(UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM, status) self.assertEqual([], self.data_file.refreshed) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual(1, len(errors), self.pipeline.logs) + self.assertTrue(errors[0].startswith( + "An error occurred when writing to " + path + + ". Error detail: "), errors[0]) def testUnregisteredDataFileIsLeftAlone(self): self.service.unregister_data_file(self.data_file) @@ -318,11 +352,143 @@ def testFileSystemWatcherIsLogged(self): with open(data_file.path, "w") as file: file.write("x") data_file.file_system_watcher = True + data_file.auto_update = True self.hold_timers() self.service.register_data_file(data_file) self.assertIn( ("info", "Creating file system watcher"), self.pipeline.logs) + def testNoFileSystemWatcherWithoutAutomaticUpdates(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + data_file = HttpDataFile(os.path.join(folder, "data.dat")) + with open(data_file.path, "w") as file: + file.write("x") + data_file.file_system_watcher = True + data_file.auto_update = False + self.hold_timers() + self.service.register_data_file(data_file) + self.assertNotIn( + ("info", "Creating file system watcher"), self.pipeline.logs) + self.assertEqual([], self.pending) + self.assertIsNone(getattr(data_file, "watcher_timer", None)) + + +# No message holds a licence key in full, as the Logging section of the data +# updates specification and the License keys section of logging.md say. +class LicenceKeyTests(UpdateServiceTestCase): + + def setUp(self): + super(LicenceKeyTests, self).setUp() + self.data_file.url = KEYED_URL + + def assertNoKey(self): + self.assertTrue(self.pipeline.logs) + for level, message in self.pipeline.logs: + self.assertNotIn(KEY[:8], message) + + def testSuccessfulUpdateShowsOnlyTheMaskedKey(self): + self.get(response(200)) + self.service.update_data_file(self.data_file) + self.assertNoKey() + self.assertIn( + ("info", + "Checking for update from '" + MASKED_URL + "' for Engine " + "'FakeEngine'"), + self.pipeline.logs) + self.assertIn( + ("info", + "Downloaded new data from '" + MASKED_URL + "' for Engine " + "'FakeEngine'"), + self.pipeline.logs) + + def testEachFailedAnswerShowsOnlyTheMaskedKey(self): + for status in (304, 403, 429, 500): + self.pipeline.logs.clear() + self.get(response(status)) + self.service.update_data_file(self.data_file) + self.assertNoKey() + self.assertTrue(any(MASKED_URL in message + for level, message in self.pipeline.logs), + status) + + def testFailedConnectionShowsOnlyTheMaskedKey(self): + # The message of a failed request names the URL, key included. + self.get(requests.ConnectionError("failed for url: " + KEYED_URL)) + self.service.update_data_file(self.data_file) + self.assertNoKey() + self.assertIn( + ("warning", + "An error occurred when connecting to " + MASKED_URL + + " in order to check for data file updates for FakeEngine. " + "Error detail: failed for url: " + MASKED_URL), + self.pipeline.logs) + + +# A missing MD5 header, content that is not compressed and content that +# cannot be read each end as the data updates specification says, and the +# checks carry on. +class ContentTests(UpdateServiceTestCase): + + def testMissingMd5HeaderIsWarnedAndTheUpdateCarriesOn(self): + self.data_file.verify_md5 = True + self.data_file.url = KEYED_URL + self.get(response(200)) + self.service.update_data_file(self.data_file) + self.assertEqual([b"new"], self.data_file.refreshed) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + self.assertIn( + ("warning", + "No MD5 hash included in data update response for '" + + MASKED_URL + "'. Unable to verify data integrity"), + self.pipeline.logs) + + def testCompressedContentIsDecompressed(self): + self.data_file.decompress = True + self.get(response(200, gzip.compress(b"new"))) + self.service.update_data_file(self.data_file) + self.assertEqual([b"new"], self.data_file.refreshed) + + def testContentThatIsNotCompressedIsUsedAsItIs(self): + self.data_file.decompress = True + self.get(response(200, b"plain")) + self.service.update_data_file(self.data_file) + self.assertEqual([b"plain"], self.data_file.refreshed) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + + def testCorruptCompressedContentIsWarnedAndCheckedAgainLater(self): + self.data_file.decompress = True + self.data_file.auto_update = True + self.hold_timers() + self.get(response(200, gzip.compress(b"new")[:12])) + self.assertFalse(self.service.update_data_file(self.data_file)) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM], self.statuses) + warnings = [message for level, message in self.pipeline.logs + if level == "warning"] + self.assertEqual(1, len(warnings), self.pipeline.logs) + self.assertTrue(warnings[0].startswith( + "An error occurred during the integrity check of new data file " + "for FakeEngine. Update will be attempted again later. Error " + "detail: "), warnings[0]) + self.assertEqual(1, len(self.pending)) + + def testFailedMd5CheckOnStartUpIsNotRepeatedAtOnce(self): + reply = response(200) + reply.headers = {"content-md5": "not the hash"} + reply.iter_content = lambda size: [b"new"] + self.data_file.verify_md5 = True + self.data_file.update_on_start = True + self.data_file.updated_on_start = False + self.hold_timers() + self.get(reply) + self.service.update_data_file(self.data_file) + self.assertEqual(1, len(self.sent)) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED], + self.statuses) + if __name__ == "__main__": unittest.main() diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py b/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py index 535ac41c..d42e1e85 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py @@ -112,16 +112,22 @@ def test_the_masked_url_is_in_the_message(self): self.assertTrue(any(MASKED_URL in message for level, message in logs), logs) - def test_an_error_raised_by_the_request_does_not_hold_the_key(self): + def test_a_failed_request_logs_only_the_masked_key(self): + # The update service logs a failed request rather than raising it, + # and the message of the request's error names the URL. def fail(*args, **kwargs): raise requests.ConnectionError("failed for url: " + URL) - service = DataFileUpdateService(RecordingPipeline()) + pipeline = RecordingPipeline() + service = DataFileUpdateService(pipeline) with mock.patch.object(requests, "get", fail): - with self.assertRaises(requests.RequestException) as caught: - service.update_data_file(StandInDataFile()) - self.assertNotIn(KEY, str(caught.exception)) - self.assertIn(MASKED_URL, str(caught.exception)) + service.update_data_file(StandInDataFile()) + self.assertTrue(pipeline.logs) + for level, message in pipeline.logs: + self.assertNotIn(KEY, message) + self.assertTrue(any(MASKED_URL in message + for level, message in pipeline.logs), + pipeline.logs) if __name__ == "__main__": From 68df725610e24e42d8c587c33c8b47eec5780f48 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 10:06:29 +0000 Subject: [PATCH 04/30] Register the watched data files in the watcher tests with automatic updates on --- .../tests/test_datafile_update_service.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index b1f87363..1ed18b0a 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -119,11 +119,19 @@ def run_next_check(self): check = self.pending.pop(0) check.function(*check.args) + def register_watched(self, data_file): + # The watcher runs only with automatic updates on. The update checks + # those would schedule are left out, so only the watcher's are + # pending. + data_file.auto_update = True + with mock.patch.object(self.service, "check_next_update"): + self.service.register_data_file(data_file) + # A refresh that fails is logged and reported, the data in use is kept, # and a later good file is picked up. def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): data_file = StandInDataFile(self.path) - self.service.register_data_file(data_file) + self.register_watched(data_file) self.write("unusable") self.run_next_check() @@ -143,7 +151,7 @@ def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): # replaced, is looked for again at the next check. def testFileSystemWatcherKeepsWatchingWhileFileIsMissing(self): data_file = StandInDataFile(self.path) - self.service.register_data_file(data_file) + self.register_watched(data_file) os.remove(self.path) self.run_next_check() @@ -160,7 +168,7 @@ def testFileSystemWatcherKeepsWatchingWhileFileIsMissing(self): # nothing is left holding the data file. def testUnregisterStopsFileSystemWatcher(self): data_file = StandInDataFile(self.path) - self.service.register_data_file(data_file) + self.register_watched(data_file) check = self.pending[0] self.service.unregister_data_file(data_file) From 5d80b133f7fcab9de4f8b983a641c2ee2a1f057b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 10:08:32 +0000 Subject: [PATCH 05/30] Mask the licence key in every update message, create the watcher only with automatic updates, write update_from_memory's bytes to a data file kept on disk, and carry on without an MD5 header or with content that is not compressed --- .../datafile_update_service.py | 175 ++++++++++-------- 1 file changed, 100 insertions(+), 75 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index c68e2bd7..821ce898 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -30,10 +30,9 @@ import random import threading import time +import zlib import requests from fiftyone_pipeline_core.keymask import mask_licence_keys_in_text -# gzip decompress alternative for python 2 -from io import StringIO class UpdateStatus(Enum): @@ -60,6 +59,10 @@ class UpdateStatus(Enum): # Refreshing the engine with the new data caused an error to occur. AUTO_UPDATE_REFRESH_FAILED = 10 +# The first two bytes of gzip compressed content (RFC 1952 section 2.3.1). +GZIP_MAGIC = b"\x1f\x8b" + + class DataFileUpdateService(): """! @@ -108,6 +111,8 @@ def update_data_file(self, data_file): # Get the update url url = data_file.get_update_url() + # The URL carries the licence key, so only a masked copy is logged. + shown_url = mask_licence_keys_in_text(url) engine = self._engine_type(data_file) headers = {} @@ -123,7 +128,7 @@ def update_data_file(self, data_file): self._log( "info", - "Checking for update from '" + url + "' for Engine '" + + "Checking for update from '" + shown_url + "' for Engine '" + engine + "'") retry = self._retry_text(data_file) @@ -131,21 +136,24 @@ def update_data_file(self, data_file): try: response = requests.get(url, headers=headers) except requests.RequestException as e: + # The message of a failed request names the URL, key included. self._log( "warning", - "An error occurred when connecting to " + url + + "An error occurred when connecting to " + shown_url + " in order to check for data file updates for " + engine + - "." + retry + " Error detail: " + str(e)) + "." + retry + " Error detail: " + + mask_licence_keys_in_text(e)) return self._update_failed( - data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR, True) + data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) if response.status_code == 304: self._log( "info", "No data newer than " + (since or "the data in use") + - " found at '" + url + "' for Engine '" + engine + "'") + " found at '" + shown_url + "' for Engine '" + engine + + "'") return self._update_failed( - data_file, UpdateStatus.AUTO_UPDATE_NOT_NEEDED, True) + data_file, UpdateStatus.AUTO_UPDATE_NOT_NEEDED) if response.status_code != 200: @@ -161,15 +169,20 @@ def update_data_file(self, data_file): self._log( "warning", "An error occurred while downloading a data file update " - "for " + engine + " from " + url + "." + retry + + "for " + engine + " from " + shown_url + "." + retry + " Error detail: HTTP status code " + str(response.status_code)) - return self._update_failed(data_file, status, True) + return self._update_failed(data_file, status) - # Verify MD5 hash if enabled + # Verify the MD5 hash if enabled and the answer gives one. - if data_file.verify_md5: - header_md5 = response.headers["content-md5"] + header_md5 = response.headers.get("content-md5") + if data_file.verify_md5 and header_md5 is None: + self._log( + "warning", + "No MD5 hash included in data update response for '" + + shown_url + "'. Unable to verify data integrity") + elif data_file.verify_md5: import hashlib hash_md5 = hashlib.md5() @@ -186,64 +199,33 @@ def update_data_file(self, data_file): "match the content-md5 header.") return self._update_failed( data_file, - UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED, - False) + UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED) self._log( "info", - "Downloaded new data from '" + url + "' for Engine '" + + "Downloaded new data from '" + shown_url + "' for Engine '" + engine + "'") data = response.content - # Decompress if set - if data_file.decompress: - # Python 3 version - if(hasattr(gzip, "decompress")): - data = gzip.decompress(response.content) - else: - compressed_stream = StringIO(response.content) - gzipper = gzip.GzipFile(fileobj=compressed_stream) - data = gzipper.read() - - # Download datafile if needed, or pass through memory - if data_file.download: - - if data_file.create_temp_copy: - - filename = data_file.temp_directory + "/" + data_file.identifier + str(time.time()) - - else: - - filename = data_file.path - + # Decompress if set and the content is compressed. + if data_file.decompress and data[:2] == GZIP_MAGIC: try: - with open(filename, 'wb') as f: - f.write(data) - except OSError as e: + data = gzip.decompress(data) + except (OSError, EOFError, zlib.error) as e: self._log( - "error", - "An error occurred when writing to " + filename + - ". Error detail: " + str(e)) + "warning", + "An error occurred during the integrity check of new " + "data file for " + engine + "." + retry + + " Error detail: " + str(e)) return self._update_failed( - data_file, UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM, - True) - - if data_file.create_temp_copy: + data_file, UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM) - # Move to real file - - try: - shutil.move(filename, data_file.path) - except (OSError, shutil.Error): - self._log( - "error", - "An error occurred when copying a data file to " - "replace the existing one at '" + data_file.path + - "'.") - return self._update_failed( - data_file, - UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, True) + # Download datafile if needed, or pass through memory + if data_file.download: + failed = self._write_data_file(data_file, data) + if failed is not None: + return self._update_failed(data_file, failed) else: @@ -268,27 +250,35 @@ def update_data_file(self, data_file): def update_from_memory(self, data_file, data): """! - Refresh a data file that is kept in memory from the bytes given, - without downloading anything. + Refresh a data file from the bytes given, without downloading + anything. A data file kept on disk has the bytes written to its path + first, as the update route writes a download, and the engine is then + refreshed from the file. @type data_file: DataFile - @param data_file: Datafile to update, created with download set to - False + @param data_file: Datafile to update @type data: bytes @param data: The new content of the data file @rtype: UpdateStatus @return: AUTO_UPDATE_SUCCESS, AUTO_UPDATE_REFRESH_FAILED if the - engine could not use the data (the data it had is kept), or + engine could not use the data (the data it had is kept), the status + of a failed write for a data file kept on disk, or AUTO_UPDATE_NOT_NEEDED if the data file's engine has been closed """ - if data_file.download: - raise ValueError( - "update_from_memory needs a data file that is kept in " - "memory, created with download set to False.") - if not self.__is_registered(data_file): return UpdateStatus.AUTO_UPDATE_NOT_NEEDED + if data_file.download and getattr(data_file, "path", None): + data_file.updating = True + try: + status = self._write_data_file(data_file, data) + if status is None: + status = self._refresh(data_file) + finally: + data_file.updating = False + self.__emit(status, data_file) + return status + previous = data_file.data data_file.data = data data_file.updating = True @@ -304,6 +294,39 @@ def update_from_memory(self, data_file, data): return status + def _write_data_file(self, data_file, data): + # Write new data to the data file's path, through a temporary copy + # when the data file asks for one. None when written, or the status + # of the failure, which is logged. + if data_file.create_temp_copy: + filename = os.path.join( + data_file.temp_directory, + data_file.identifier + str(time.time())) + else: + filename = data_file.path + + try: + with open(filename, 'wb') as f: + f.write(data) + except OSError as e: + self._log( + "error", + "An error occurred when writing to " + filename + + ". Error detail: " + str(e)) + return UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM + + if data_file.create_temp_copy: + try: + shutil.move(filename, data_file.path) + except (OSError, shutil.Error): + self._log( + "error", + "An error occurred when copying a data file to " + "replace the existing one at '" + data_file.path + + "'.") + return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME + return None + def _refresh(self, data_file): # Refresh the engine with the data file's new content. engine = self._engine_type(data_file) @@ -320,12 +343,12 @@ def _refresh(self, data_file): return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED return UpdateStatus.AUTO_UPDATE_SUCCESS - def _update_failed(self, data_file, status, run_after_error): - # An update that ends without new data. With run_after_error False, a - # check on start-up that has not yet succeeded is made again at once. + def _update_failed(self, data_file, status): + # An update that ends without new data. The next check is the next + # scheduled one, never an immediate repeat of a check on start-up. data_file.updating = False self.__emit(status, data_file) - self.check_next_update(data_file, run_after_error=run_after_error) + self.check_next_update(data_file, run_after_error=True) return False def _log(self, level, message): @@ -454,7 +477,9 @@ def register_data_file(self, data_file): # Watch for changes in file and refresh if # file_system_watcher is enabled on data_file - if data_file.file_system_watcher: + # The watcher is an automatic update, so it only runs with + # automatic updates on. + if data_file.auto_update and data_file.file_system_watcher: self._log("info", "Creating file system watcher") data_file.last_changed_time = os.stat(data_file.path).st_mtime self.track_local_data_file_changes(data_file) From d0cbbd61e6956c967ee448b3010a485db26978fd Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 20:28:39 +0000 Subject: [PATCH 06/30] Test that a callback that raises does not stop the checks, and a connection failure keeps polling --- .../tests/test_datafile_update_http.py | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 42ad1e2c..5e17de3e 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -376,6 +376,35 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): # No message holds a licence key in full, as the Logging section of the data # updates specification and the License keys section of logging.md say. +# A callback that raises is logged, and the other callbacks and the next +# check still run, so polling carries on. +class CallbackTests(UpdateServiceTestCase): + + def testACallbackThatRaisesDoesNotStopTheChecks(self): + def fails(status, data_file): + raise RuntimeError("callback failed") + + self.service.on_complete_methods.insert(0, fails) + self.data_file.auto_update = True + self.hold_timers() + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertEqual([UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) + self.assertEqual(1, len(self.pending)) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual(1, len(errors), self.pipeline.logs) + self.assertIn("callback failed", errors[0]) + + def testAConnectionFailureKeepsPolling(self): + self.data_file.auto_update = True + self.hold_timers() + self.get(requests.ConnectionError("boom")) + self.assertFalse(self.service.update_data_file(self.data_file)) + self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) + self.assertEqual(1, len(self.pending)) + + class LicenceKeyTests(UpdateServiceTestCase): def setUp(self): From b2ab2216812335e23077b6f8d62e0a1f631086cd Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 10 Nov 2026 20:28:41 +0000 Subject: [PATCH 07/30] Log a callback that raises, so the other callbacks and the next check still run --- .../datafile_update_service.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 821ce898..03319ad6 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -85,8 +85,17 @@ def __init__(self, pipeline = None): self.on_complete_methods = [] def __emit(self, status, data_file): + # A callback that raises is logged, so that the other callbacks and + # the next check still run. for method in self.on_complete_methods: - method(status, data_file) + try: + method(status, data_file) + except Exception as e: + self._log( + "error", + "An update complete callback raised an error for " + + self._engine_type(data_file) + ". Error detail: " + + str(e)) def update_data_file(self, data_file): From bf8ac545c1029ad36d6a01e92399a2e20e2454f8 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 00:26:58 +0000 Subject: [PATCH 08/30] Test that the watcher refresh is routed as the polling refresh is --- .../tests/test_datafile_update_service.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index 1ed18b0a..6cbe5f29 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -147,6 +147,38 @@ def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): self.run_next_check() self.assertEqual("version 2", data_file.data) + # The watcher's refresh takes the same route as the polling one, so it + # logs the attempt, uses the table's failure message and level, and + # reports success to the update complete callbacks. + def testFileSystemWatcherRefreshIsRoutedAsPollingIs(self): + data_file = StandInDataFile(self.path) + self.register_watched(data_file) + + self.write("version 2") + self.run_next_check() + + self.assertEqual("version 2", data_file.data) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + attempts = [log for log in self.pipeline.logs + if "Attempting to refresh Engine" in log[1]] + self.assertEqual(1, len(attempts)) + self.assertEqual("info", attempts[0][0]) + + # A refresh that fails from the watcher carries the message and level of + # the specification's errors table, as the polling route's does. + def testFileSystemWatcherFailedRefreshUsesTheTableMessage(self): + data_file = StandInDataFile(self.path) + self.register_watched(data_file) + + self.write("unusable") + self.run_next_check() + + errors = [log for log in self.pipeline.logs if log[0] == "error"] + self.assertEqual(1, len(errors)) + self.assertIn( + "An error occurred while applying a data file update to", + errors[0][1]) + # A file that is missing at a check, for example part way through being # replaced, is looked for again at the next check. def testFileSystemWatcherKeepsWatchingWhileFileIsMissing(self): From 705cf590ea50e87abbcfdb3d878c4f96d17935ec Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 00:26:59 +0000 Subject: [PATCH 09/30] Route the watcher refresh through the polling refresh --- .../datafile_update_service.py | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 03319ad6..dbd36544 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -517,10 +517,12 @@ def __is_registered(self, data_file): def track_local_data_file_changes(self, data_file): """! Check for local changes to a datafile. Used by - the file system watcher. A refresh that fails is logged, and - reported with AUTO_UPDATE_REFRESH_FAILED as on the update route. - The next check is scheduled whatever happens, so that a later file - is still picked up, until the data file is unregistered. + the file system watcher. The refresh takes the same route as the + polling one, so the attempt is logged, a failure carries the + errors table's message and level, and the result is reported to + the update complete callbacks either way. The next check is + scheduled whatever happens, so that a later file is still picked + up, until the data file is unregistered. @type data_file: DataFile @param data_file: Datafile to check updates for """ @@ -538,13 +540,7 @@ def track_local_data_file_changes(self, data_file): changed_time > data_file.last_changed_time and not data_file.updating): data_file.last_changed_time = changed_time - data_file.refresh() - except Exception as e: - self.pipeline.log( - "error", - "Refreshing from the changed data file '" + data_file.path + - "' failed. " + str(e)) - self.__emit(UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, data_file) + self.__emit(self._refresh(data_file), data_file) finally: if self.__is_registered(data_file): t = threading.Timer( From 9314d95413974217e9628e63384fdb919a1d14ac Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 00:33:25 +0000 Subject: [PATCH 10/30] Test the data update messages and levels against .NET --- .../tests/test_datafile_update_http.py | 176 +++++++++++++++--- 1 file changed, 147 insertions(+), 29 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 5e17de3e..d57d2131 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -236,21 +236,25 @@ def testFileDataFileIsWrittenThenRefreshed(self): self.assertEqual(1, len(self.data_file.refreshed)) self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) - def testFailedWriteIsLoggedAndNotRefreshed(self): + # A write that fails whilst updating from memory is a warning, and the + # Engine is still refreshed with the new data, as .NET does. + def testFailedWriteIsAWarningAndTheEngineIsStillRefreshed(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder, ignore_errors=True) path = os.path.join(folder, "missing", "data.dat") self.data_file = HttpDataFile(path) self.data_file.download = True status = self.service.update_from_memory(self.data_file, b"fresh") - self.assertEqual(UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM, status) - self.assertEqual([], self.data_file.refreshed) - errors = [message for level, message in self.pipeline.logs - if level == "error"] - self.assertEqual(1, len(errors), self.pipeline.logs) - self.assertTrue(errors[0].startswith( - "An error occurred when writing to " + path + - ". Error detail: "), errors[0]) + self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) + self.assertEqual([b"fresh"], self.data_file.refreshed) + self.assertEqual([], [message for level, message + in self.pipeline.logs if level == "error"]) + self.assertIn( + ("warning", + "An error occurred when writing to '" + path + + "'. The engine will be updated to use the new data but the " + "file on disk will still contain old data."), + self.pipeline.logs) def testUnregisteredDataFileIsLeftAlone(self): self.service.unregister_data_file(self.data_file) @@ -258,8 +262,8 @@ def testUnregisteredDataFileIsLeftAlone(self): self.assertEqual([], self.data_file.refreshed) -# The messages and levels are those of the Logging section of the data -# updates specification. +# The messages and levels are those of the .NET data update service, which +# the Errors table of the data updates specification follows. class LoggingTests(UpdateServiceTestCase): def levels(self): @@ -291,38 +295,47 @@ def testNotModifiedIsInformationNotWarning(self): for level in ("warn", "warning", "error"): self.assertNotIn(level, self.levels()) - def testRefusedDownloadIsWarning(self): + def testRefusedDownloadIsError(self): self.get(response(403)) self.service.update_data_file(self.data_file) self.assertIn( - ("warning", - "An error occurred while downloading a data file update for " - "FakeEngine from " + URL + ". Error detail: HTTP status code " - "403"), + ("error", + "Access denied to data update service at '" + URL + + "' for engine 'FakeEngine'"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN], self.statuses) - def testAutomaticUpdateSaysItWillBeTriedAgain(self): + def testTooManyRequestsIsError(self): + self.get(response(429)) + self.service.update_data_file(self.data_file) + self.assertIn( + ("error", + "Too many requests to '" + URL + "' for engine 'FakeEngine'"), + self.pipeline.logs) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS], + self.statuses) + + def testAnyOtherStatusNamesTheCode(self): self.data_file.auto_update = True self.hold_timers() self.get(response(500)) self.service.update_data_file(self.data_file) self.assertIn( - ("warning", - "An error occurred while downloading a data file update for " - "FakeEngine from " + URL + ". Update will be attempted again " - "later. Error detail: HTTP status code 500"), + ("error", + "HTTP status code '500' from data update service at '" + URL + + "' for engine 'FakeEngine'"), self.pipeline.logs) + self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) - def testConnectionFailureIsWarning(self): + def testConnectionFailureIsError(self): self.get(requests.ConnectionError("boom")) self.assertFalse(self.service.update_data_file(self.data_file)) self.assertIn( - ("warning", - "An error occurred when connecting to " + URL + " in order to " - "check for data file updates for FakeEngine. Error detail: " - "boom"), + ("error", + "Error accessing data update service at '" + URL + + "' for engine 'FakeEngine'"), self.pipeline.logs) self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) @@ -332,18 +345,36 @@ def testFailedRefreshIsError(self): self.service.update_data_file(self.data_file) self.assertIn( ("error", - "An error occurred while applying a data file update to " - "FakeEngine. Error detail: boom"), + "An error occurred when applying a data update to engine " + "'FakeEngine'"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) + def testIntegrityCheckFailureIsError(self): + reply = response(200) + reply.headers = {"content-md5": "not the hash"} + reply.iter_content = lambda size: [b"new"] + self.get(reply) + self.data_file.verify_md5 = True + self.service.update_data_file(self.data_file) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual(1, len(errors), self.pipeline.logs) + self.assertTrue(errors[0].startswith( + "Integrity check failed. MD5 hash in HTTP response " + "'not the hash' for 'FakeEngine' data update does not match " + "calculated hash for the downloaded file '"), errors[0]) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED], + self.statuses) + def testUpdateOnStartIsLogged(self): self.data_file.update_on_start = True self.data_file.updated_on_start = False with mock.patch.object(self.service, "update_data_file"): self.service.register_data_file(self.data_file) - self.assertIn(("info", "Updating on start-up"), self.pipeline.logs) + self.assertIn(("info", "Updating on startup"), self.pipeline.logs) def testFileSystemWatcherIsLogged(self): folder = tempfile.mkdtemp() @@ -374,6 +405,93 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): self.assertIsNone(getattr(data_file, "watcher_timer", None)) +# A data file kept on disk that cannot put the new file in place, and one +# with no temporary file path, carry .NET's message and status. +class FileReplacementTests(UpdateServiceTestCase): + + def setUp(self): + super(FileReplacementTests, self).setUp() + self.folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.folder, ignore_errors=True) + self.path = os.path.join(self.folder, "data.dat") + with open(self.path, "w") as file: + file.write("old") + self.data_file = HttpDataFile(self.path) + self.data_file.download = True + self.data_file.create_temp_copy = True + self.data_file.temp_directory = self.folder + + def testFailedCopyNamesThePath(self): + move = mock.patch.object( + shutil, "move", + mock.Mock(side_effect=OSError("in use"))) + move.start() + self.addCleanup(move.stop) + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual( + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) + self.assertIn( + ("error", + "An error occurred when copying a data file to replace the " + "existing one at '" + self.path + "'."), + self.pipeline.logs) + + def testNoTemporaryPathIsReported(self): + self.data_file.temp_directory = None + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual( + UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET, status) + self.assertIn( + ("error", + "The data file '" + self.data_file.identifier + + "' is checking for updates but does not have a temporary " + "file path configured."), + self.pipeline.logs) + + +# Every check is reported to the update complete callbacks once it ends, and +# a check that needed no update is reported as not needed, as the Reporting +# each check section says and as .NET does. +class ReportingEachCheckTests(UpdateServiceTestCase): + + def testNotModifiedIsReportedAsNotNeeded(self): + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) + + def testAWatcherEventWithNothingToDoIsReportedAsNotNeeded(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "data.dat") + with open(path, "w") as file: + file.write("x") + data_file = HttpDataFile(path) + data_file.last_changed_time = os.stat(path).st_mtime + data_file.registered = True + self.service.on_complete( + lambda status, df: None) + self.hold_timers() + self.service.track_local_data_file_changes(data_file) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) + + def testAWatcherEventWhilstUpdatingIsReportedAsInProgress(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "data.dat") + with open(path, "w") as file: + file.write("x") + data_file = HttpDataFile(path) + data_file.last_changed_time = os.stat(path).st_mtime - 10 + data_file.registered = True + data_file.updating = True + self.hold_timers() + self.service.track_local_data_file_changes(data_file) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_IN_PROGRESS], self.statuses) + + # No message holds a licence key in full, as the Logging section of the data # updates specification and the License keys section of logging.md say. # A callback that raises is logged, and the other callbacks and the next From 6795685b39298425f9e5416a5e1b4c07c1604cd2 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 00:36:30 +0000 Subject: [PATCH 11/30] Align the data update messages and levels to .NET Each event in the data update service now carries the message and level of the .NET data update service, which the Errors table of data-updates.md follows. A failed connection, a 429, a 403, any other status, a failed integrity check and a failed refresh move from Warning to Error and take .NET wording. A failed write to the data file becomes a Warning and the engine is still refreshed from the data in memory. Every check is reported to the update complete callbacks once it ends. A watcher event with no newer file reports not needed, and one that arrived whilst another was being applied reports in progress. --- .../datafile_update_service.py | 137 +++++++++++------- .../tests/test_datafile_update_http.py | 29 ++-- .../tests/test_datafile_update_service.py | 10 +- 3 files changed, 105 insertions(+), 71 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index dbd36544..29113b40 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -58,6 +58,14 @@ class UpdateStatus(Enum): AUTO_UPDATE_NEW_FILE_CANT_RENAME = 9 # Refreshing the engine with the new data caused an error to occur. AUTO_UPDATE_REFRESH_FAILED = 10 + # The data file is checking for updates but has no temporary file path. + AUTO_UPDATE_TEMP_PATH_NOT_SET = 11 + # Writing the new data to the data file failed. The engine is still + # refreshed from the data, so this status never reaches a caller and + # marks only that the file on disk still holds the old data. + AUTO_UPDATE_WRITE_FAILED = 14 + # Any other error during a check for an update. + AUTO_UPDATE_UNKNOWN_ERROR = 15 # The first two bytes of gzip compressed content (RFC 1952 section 2.3.1). GZIP_MAGIC = b"\x1f\x8b" @@ -140,18 +148,21 @@ def update_data_file(self, data_file): "Checking for update from '" + shown_url + "' for Engine '" + engine + "'") - retry = self._retry_text(data_file) - try: response = requests.get(url, headers=headers) - except requests.RequestException as e: - # The message of a failed request names the URL, key included. + except requests.RequestException: self._log( - "warning", - "An error occurred when connecting to " + shown_url + - " in order to check for data file updates for " + engine + - "." + retry + " Error detail: " + - mask_licence_keys_in_text(e)) + "error", + "Error accessing data update service at '" + shown_url + + "' for engine '" + engine + "'") + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) + + if response is None: + self._log( + "error", + "No response from data update service at '" + shown_url + + "' for engine '" + engine + "'") return self._update_failed( data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) @@ -166,21 +177,25 @@ def update_data_file(self, data_file): if response.status_code != 200: - # Handle errors + # Each response that is not a success and not a 304 carries the + # message and status of the matching row of the errors table. if response.status_code == 429: status = UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS + message = ("Too many requests to '" + shown_url + + "' for engine '" + engine + "'") elif response.status_code == 403: status = UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN + message = ("Access denied to data update service at '" + + shown_url + "' for engine '" + engine + "'") else: status = UpdateStatus.AUTO_UPDATE_HTTPS_ERR + message = ("HTTP status code '" + + str(response.status_code) + + "' from data update service at '" + shown_url + + "' for engine '" + engine + "'") - self._log( - "warning", - "An error occurred while downloading a data file update " - "for " + engine + " from " + shown_url + "." + retry + - " Error detail: HTTP status code " + - str(response.status_code)) + self._log("error", message) return self._update_failed(data_file, status) # Verify the MD5 hash if enabled and the answer gives one. @@ -201,11 +216,11 @@ def update_data_file(self, data_file): if header_md5 != hash_md5.hexdigest(): self._log( - "warning", - "An error occurred during the integrity check of new " - "data file for " + engine + "." + retry + - " Error detail: The MD5 hash of the data does not " - "match the content-md5 header.") + "error", + "Integrity check failed. MD5 hash in HTTP response '" + + str(header_md5) + "' for '" + engine + "' data update " + "does not match calculated hash for the downloaded " + "file '" + hash_md5.hexdigest() + "'.") return self._update_failed( data_file, UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED) @@ -221,20 +236,26 @@ def update_data_file(self, data_file): if data_file.decompress and data[:2] == GZIP_MAGIC: try: data = gzip.decompress(data) - except (OSError, EOFError, zlib.error) as e: + except (OSError, EOFError, zlib.error): + # Content that cannot be read is any other error, which + # carries the last row of the errors table. self._log( - "warning", - "An error occurred during the integrity check of new " - "data file for " + engine + "." + retry + - " Error detail: " + str(e)) + "error", + "An unhandled error occurred while checking for " + "automatic updates for engine '" + engine + "'") return self._update_failed( - data_file, UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM) + data_file, UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR) # Download datafile if needed, or pass through memory if data_file.download: failed = self._write_data_file(data_file, data) - if failed is not None: + # A write that failed is only a warning, so the engine is still + # refreshed from the data just downloaded. + if (failed is not None and + failed != UpdateStatus.AUTO_UPDATE_WRITE_FAILED): return self._update_failed(data_file, failed) + if failed == UpdateStatus.AUTO_UPDATE_WRITE_FAILED: + data_file.data = data else: @@ -281,7 +302,11 @@ def update_from_memory(self, data_file, data): data_file.updating = True try: status = self._write_data_file(data_file, data) - if status is None: + # A write that failed is only a warning, so the engine is + # still refreshed from the data held in memory. + if (status is None or + status == UpdateStatus.AUTO_UPDATE_WRITE_FAILED): + data_file.data = data status = self._refresh(data_file) finally: data_file.updating = False @@ -306,8 +331,18 @@ def update_from_memory(self, data_file, data): def _write_data_file(self, data_file, data): # Write new data to the data file's path, through a temporary copy # when the data file asks for one. None when written, or the status - # of the failure, which is logged. + # of the failure, which is logged. A write that fails is a warning + # and leaves the caller free to refresh the Engine from the data in + # memory, whilst a temporary path that is not set and a copy that + # cannot replace the file stop the update. if data_file.create_temp_copy: + if not getattr(data_file, "temp_directory", None): + self._log( + "error", + "The data file '" + str(data_file.identifier) + + "' is checking for updates but does not have a " + "temporary file path configured.") + return UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET filename = os.path.join( data_file.temp_directory, data_file.identifier + str(time.time())) @@ -317,12 +352,13 @@ def _write_data_file(self, data_file, data): try: with open(filename, 'wb') as f: f.write(data) - except OSError as e: + except OSError: self._log( - "error", - "An error occurred when writing to " + filename + - ". Error detail: " + str(e)) - return UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM + "warning", + "An error occurred when writing to '" + str(filename) + + "'. The engine will be updated to use the new data but " + "the file on disk will still contain old data.") + return UpdateStatus.AUTO_UPDATE_WRITE_FAILED if data_file.create_temp_copy: try: @@ -344,11 +380,11 @@ def _refresh(self, data_file): "Attempting to refresh Engine '" + engine + "' with new data") try: data_file.refresh() - except Exception as e: + except Exception: self._log( "error", - "An error occurred while applying a data file update to " + - engine + ". Error detail: " + str(e)) + "An error occurred when applying a data update to engine '" + + engine + "'") return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED return UpdateStatus.AUTO_UPDATE_SUCCESS @@ -369,13 +405,6 @@ def _log(self, level, message): def _engine_type(data_file): return type(getattr(data_file, "flow_element", data_file)).__name__ - @staticmethod - def _retry_text(data_file): - # Only an automatic update is tried again. - if getattr(data_file, "auto_update", False): - return " Update will be attempted again later." - return "" - @staticmethod def _http_date(moment): # An HTTP date, as RFC 9110 section 5.6.7 requires. A time with no @@ -451,7 +480,7 @@ def check_next_update(self, data_file, run_after_error = False): return if data_file.update_on_start and not data_file.updated_on_start and not run_after_error: - self._log("info", "Updating on start-up") + self._log("info", "Updating on startup") self.update_data_file(data_file) elif data_file.auto_update: @@ -535,12 +564,18 @@ def track_local_data_file_changes(self, data_file): # replaced, is looked for again at the next check. changed_time = None try: - # Don't change if data_file is updating - if (changed_time is not None and - changed_time > data_file.last_changed_time and - not data_file.updating): + # Every event is reported once it ends. An event that arrived + # whilst another was being applied is reported as in progress, + # and one with no newer file as not needed. + if data_file.updating: + status = UpdateStatus.AUTO_UPDATE_IN_PROGRESS + elif (changed_time is not None and + changed_time > data_file.last_changed_time): data_file.last_changed_time = changed_time - self.__emit(self._refresh(data_file), data_file) + status = self._refresh(data_file) + else: + status = UpdateStatus.AUTO_UPDATE_NOT_NEEDED + self.__emit(status, data_file) finally: if self.__is_registered(data_file): t = threading.Timer( diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index d57d2131..256c9177 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -217,8 +217,8 @@ def testFailedRefreshIsReportedAndLogged(self): self.assertEqual(UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, status) self.assertIn( ("error", - "An error occurred while applying a data file update to " - "FakeEngine. Error detail: boom"), + "An error occurred when applying a data update to engine " + "'FakeEngine'"), self.pipeline.logs) def testFileDataFileIsWrittenThenRefreshed(self): @@ -560,15 +560,14 @@ def testEachFailedAnswerShowsOnlyTheMaskedKey(self): status) def testFailedConnectionShowsOnlyTheMaskedKey(self): - # The message of a failed request names the URL, key included. + # The message of a failed request names the URL, masked. self.get(requests.ConnectionError("failed for url: " + KEYED_URL)) self.service.update_data_file(self.data_file) self.assertNoKey() self.assertIn( - ("warning", - "An error occurred when connecting to " + MASKED_URL + - " in order to check for data file updates for FakeEngine. " - "Error detail: failed for url: " + MASKED_URL), + ("error", + "Error accessing data update service at '" + MASKED_URL + + "' for engine 'FakeEngine'"), self.pipeline.logs) @@ -603,7 +602,7 @@ def testContentThatIsNotCompressedIsUsedAsItIs(self): self.assertEqual([b"plain"], self.data_file.refreshed) self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) - def testCorruptCompressedContentIsWarnedAndCheckedAgainLater(self): + def testCorruptCompressedContentIsAnErrorAndCheckedAgainLater(self): self.data_file.decompress = True self.data_file.auto_update = True self.hold_timers() @@ -611,14 +610,12 @@ def testCorruptCompressedContentIsWarnedAndCheckedAgainLater(self): self.assertFalse(self.service.update_data_file(self.data_file)) self.assertEqual([], self.data_file.refreshed) self.assertEqual( - [UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM], self.statuses) - warnings = [message for level, message in self.pipeline.logs - if level == "warning"] - self.assertEqual(1, len(warnings), self.pipeline.logs) - self.assertTrue(warnings[0].startswith( - "An error occurred during the integrity check of new data file " - "for FakeEngine. Update will be attempted again later. Error " - "detail: "), warnings[0]) + [UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR], self.statuses) + self.assertIn( + ("error", + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'"), + self.pipeline.logs) self.assertEqual(1, len(self.pending)) def testFailedMd5CheckOnStartUpIsNotRepeatedAtOnce(self): diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index 6cbe5f29..2d5b2b1b 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -137,10 +137,12 @@ def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): self.run_next_check() self.assertEqual("version 1", data_file.data) self.assertEqual( - [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) + UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, self.statuses[-1]) errors = [log for log in self.pipeline.logs if log[0] == "error"] self.assertEqual(1, len(errors)) - self.assertIn("The data file cannot be used.", errors[0][1]) + self.assertIn( + "An error occurred when applying a data update to engine", + errors[0][1]) self.assertEqual(1, len(self.pending)) self.write("version 2") @@ -158,7 +160,7 @@ def testFileSystemWatcherRefreshIsRoutedAsPollingIs(self): self.run_next_check() self.assertEqual("version 2", data_file.data) - self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, self.statuses[-1]) attempts = [log for log in self.pipeline.logs if "Attempting to refresh Engine" in log[1]] self.assertEqual(1, len(attempts)) @@ -176,7 +178,7 @@ def testFileSystemWatcherFailedRefreshUsesTheTableMessage(self): errors = [log for log in self.pipeline.logs if log[0] == "error"] self.assertEqual(1, len(errors)) self.assertIn( - "An error occurred while applying a data file update to", + "An error occurred when applying a data update to engine", errors[0][1]) # A file that is missing at a check, for example part way through being From 18b1bfaa57705573b121675c4dcc7e57e290c0b4 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 08:36:28 +0000 Subject: [PATCH 12/30] Test that a callback's error is logged with .NET's message for any other error DataUpdateService.cs logs an error from an update complete callback with MessageAutoUpdateUnhandledError at Error level. The test now asserts that text, so it fails while Python writes its own sentence. --- .../tests/test_datafile_update_http.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 5e17de3e..6085269d 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -394,7 +394,11 @@ def fails(status, data_file): errors = [message for level, message in self.pipeline.logs if level == "error"] self.assertEqual(1, len(errors), self.pipeline.logs) - self.assertIn("callback failed", errors[0]) + # The callback's error is any other error, logged with .NET's + # message for it. + self.assertEqual( + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'", errors[0]) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True From 03a3dba6c0374d8e317f093510c8675f76d2dfd4 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:03:05 +0000 Subject: [PATCH 13/30] Test that a callback that raises stops the callbacks after it data-updates.md at e89679e9 says the callbacks after the one that raised are not called. On update polling the error is logged with the last row's message, which is .NET's MessageAutoUpdateUnhandledError, and on a programmatic check it passes to the caller. The tests cover polling, a programmatic check and an update from memory, and that polling still schedules its next check. This commit is the tests alone, so they fail. --- .../tests/test_datafile_update_http.py | 46 ++++++++++++++----- 1 file changed, 34 insertions(+), 12 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 6085269d..518fa7a8 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -376,29 +376,51 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): # No message holds a licence key in full, as the Logging section of the data # updates specification and the License keys section of logging.md say. -# A callback that raises is logged, and the other callbacks and the next -# check still run, so polling carries on. +# A callback that raises stops the callbacks after it, as data-updates.md +# says. On a check the service runs in the background the error is logged +# with .NET's message for any other error, and polling carries on. On a +# check the caller started it passes to the caller. class CallbackTests(UpdateServiceTestCase): - def testACallbackThatRaisesDoesNotStopTheChecks(self): + def fail_first(self): def fails(status, data_file): raise RuntimeError("callback failed") self.service.on_complete_methods.insert(0, fails) + + def errors(self): + return [message for level, message in self.pipeline.logs + if level == "error"] + + def testACallbackErrorOnPollingIsLoggedAndStopsTheCallbacks(self): + self.fail_first() self.data_file.auto_update = True self.hold_timers() self.get(response(304)) - self.service.update_data_file(self.data_file) - self.assertEqual([UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) + self.service.async_update_data_file(self.data_file).join() + self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - errors = [message for level, message in self.pipeline.logs - if level == "error"] - self.assertEqual(1, len(errors), self.pipeline.logs) - # The callback's error is any other error, logged with .NET's - # message for it. self.assertEqual( - "An unhandled error occurred while checking for automatic " - "updates for engine 'FakeEngine'", errors[0]) + ["An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'"], self.errors()) + + def testACallbackErrorOnAProgrammaticCheckPassesToTheCaller(self): + self.fail_first() + self.data_file.auto_update = True + self.hold_timers() + self.get(response(304)) + with self.assertRaises(RuntimeError): + self.service.update_data_file(self.data_file) + self.assertEqual([], self.statuses) + self.assertEqual(1, len(self.pending)) + self.assertEqual([], self.errors()) + + def testACallbackErrorOnAnUpdateFromMemoryPassesToTheCaller(self): + self.fail_first() + with self.assertRaises(RuntimeError): + self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual([], self.statuses) + self.assertEqual([], self.errors()) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True From 80694f229422a5e8b65cac1daa881637144ea751 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:06:31 +0000 Subject: [PATCH 14/30] Test that a callback that raises is logged and stops the callbacks after it data-updates.md at e89679e9 says the callbacks after the one that raised are not called, and that on update polling and the file watcher the error is logged with the last row's message, .NET's MessageAutoUpdateUnhandledError. Passing the error to the caller of a programmatic check, an update from memory or the update on start-up waits for next week, so on those routes it is logged in the same way for now. This commit is the tests alone, so they fail. --- .../tests/test_datafile_update_http.py | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 518fa7a8..c2f10ac4 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -377,11 +377,13 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): # No message holds a licence key in full, as the Logging section of the data # updates specification and the License keys section of logging.md say. # A callback that raises stops the callbacks after it, as data-updates.md -# says. On a check the service runs in the background the error is logged -# with .NET's message for any other error, and polling carries on. On a -# check the caller started it passes to the caller. +# says, and its error is logged with .NET's message for any other error. +# Polling carries on. class CallbackTests(UpdateServiceTestCase): + UNHANDLED = ("An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'") + def fail_first(self): def fails(status, data_file): raise RuntimeError("callback failed") @@ -400,27 +402,25 @@ def testACallbackErrorOnPollingIsLoggedAndStopsTheCallbacks(self): self.service.async_update_data_file(self.data_file).join() self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual( - ["An unhandled error occurred while checking for automatic " - "updates for engine 'FakeEngine'"], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) - def testACallbackErrorOnAProgrammaticCheckPassesToTheCaller(self): + def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( + self): self.fail_first() self.data_file.auto_update = True self.hold_timers() self.get(response(304)) - with self.assertRaises(RuntimeError): - self.service.update_data_file(self.data_file) + self.service.update_data_file(self.data_file) self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) - def testACallbackErrorOnAnUpdateFromMemoryPassesToTheCaller(self): + def testACallbackErrorOnAnUpdateFromMemoryIsLoggedAndStopsTheCallbacks( + self): self.fail_first() - with self.assertRaises(RuntimeError): - self.service.update_from_memory(self.data_file, b"fresh") + self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual([], self.statuses) - self.assertEqual([], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True From 22342c6316c68e502a010b1d50beba7f4769941a Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:07:32 +0000 Subject: [PATCH 15/30] Stop at the first callback that raises, and log it with .NET's message DataUpdateService.cs wraps the callback loop in one try, so the first callback that raises stops the rest, and LogUpdateException logs it with MessageAutoUpdateUnhandledError, "An unhandled error occurred while checking for automatic updates for engine '{0}'", at Error. data-updates.md says the same. Python now does both, on every route, and the next check is still scheduled. --- .../datafile_update_service.py | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index dbd36544..43cdf135 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -85,17 +85,17 @@ def __init__(self, pipeline = None): self.on_complete_methods = [] def __emit(self, status, data_file): - # A callback that raises is logged, so that the other callbacks and - # the next check still run. - for method in self.on_complete_methods: - try: + # A callback that raises stops the callbacks after it, as .NET's service + # does, and its error is logged with .NET's message for any other + # error, so that the next check still runs. + try: + for method in self.on_complete_methods: method(status, data_file) - except Exception as e: - self._log( - "error", - "An update complete callback raised an error for " + - self._engine_type(data_file) + ". Error detail: " + - str(e)) + except Exception: + self._log( + "error", + "An unhandled error occurred while checking for automatic " + "updates for engine '" + self._engine_type(data_file) + "'") def update_data_file(self, data_file): From 2f15773be94bace96e5fa5df7c40db6e0885a0fe Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:13:53 +0000 Subject: [PATCH 16/30] Test that an error from the errors table is logged as .NET logs it data-updates.md at e89679e9, lines 476 to 485, says an error from a row at Error level is logged at Error as "Error during check for data update" with the error attached, and the error carries the row's text. An error of any other kind, a callback's included, is logged with the last row's message, also with the error attached. Python's logger takes a string, so the error follows the entry text, its cause after it as "Error detail:". This commit is the tests alone, so they fail. --- .../tests/test_datafile_update_http.py | 57 +++++++++++-------- 1 file changed, 34 insertions(+), 23 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index eb070b89..1b7f6096 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -85,6 +85,12 @@ def response(status, content=b"new"): return reply +# The entry an error from a row of the errors table is logged under, with +# the error, being the row's text and then its cause, after it, as .NET logs +# a DataUpdateException. +CHECK_ERROR = "Error during check for data update. " + + class UpdateServiceTestCase(unittest.TestCase): def setUp(self): @@ -217,8 +223,8 @@ def testFailedRefreshIsReportedAndLogged(self): self.assertEqual(UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, status) self.assertIn( ("error", - "An error occurred when applying a data update to engine " - "'FakeEngine'"), + CHECK_ERROR + "An error occurred when applying a data update " + "to engine 'FakeEngine'. Error detail: boom"), self.pipeline.logs) def testFileDataFileIsWrittenThenRefreshed(self): @@ -300,8 +306,8 @@ def testRefusedDownloadIsError(self): self.service.update_data_file(self.data_file) self.assertIn( ("error", - "Access denied to data update service at '" + URL + - "' for engine 'FakeEngine'"), + CHECK_ERROR + "Access denied to data update service at '" + + URL + "' for engine 'FakeEngine'"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN], self.statuses) @@ -311,7 +317,8 @@ def testTooManyRequestsIsError(self): self.service.update_data_file(self.data_file) self.assertIn( ("error", - "Too many requests to '" + URL + "' for engine 'FakeEngine'"), + CHECK_ERROR + "Too many requests to '" + URL + + "' for engine 'FakeEngine'"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS], @@ -324,8 +331,8 @@ def testAnyOtherStatusNamesTheCode(self): self.service.update_data_file(self.data_file) self.assertIn( ("error", - "HTTP status code '500' from data update service at '" + URL + - "' for engine 'FakeEngine'"), + CHECK_ERROR + "HTTP status code '500' from data update service " + "at '" + URL + "' for engine 'FakeEngine'"), self.pipeline.logs) self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) @@ -334,8 +341,8 @@ def testConnectionFailureIsError(self): self.assertFalse(self.service.update_data_file(self.data_file)) self.assertIn( ("error", - "Error accessing data update service at '" + URL + - "' for engine 'FakeEngine'"), + CHECK_ERROR + "Error accessing data update service at '" + + URL + "' for engine 'FakeEngine'. Error detail: boom"), self.pipeline.logs) self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) @@ -345,8 +352,8 @@ def testFailedRefreshIsError(self): self.service.update_data_file(self.data_file) self.assertIn( ("error", - "An error occurred when applying a data update to engine " - "'FakeEngine'"), + CHECK_ERROR + "An error occurred when applying a data update " + "to engine 'FakeEngine'. Error detail: boom"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) @@ -362,7 +369,7 @@ def testIntegrityCheckFailureIsError(self): if level == "error"] self.assertEqual(1, len(errors), self.pipeline.logs) self.assertTrue(errors[0].startswith( - "Integrity check failed. MD5 hash in HTTP response " + CHECK_ERROR + "Integrity check failed. MD5 hash in HTTP response " "'not the hash' for 'FakeEngine' data update does not match " "calculated hash for the downloaded file '"), errors[0]) self.assertEqual( @@ -432,8 +439,9 @@ def testFailedCopyNamesThePath(self): UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) self.assertIn( ("error", - "An error occurred when copying a data file to replace the " - "existing one at '" + self.path + "'."), + CHECK_ERROR + "An error occurred when copying a data file to " + "replace the existing one at '" + self.path + + "'. Error detail: in use"), self.pipeline.logs) def testNoTemporaryPathIsReported(self): @@ -443,7 +451,7 @@ def testNoTemporaryPathIsReported(self): UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET, status) self.assertIn( ("error", - "The data file '" + self.data_file.identifier + + CHECK_ERROR + "The data file '" + self.data_file.identifier + "' is checking for updates but does not have a temporary " "file path configured."), self.pipeline.logs) @@ -500,7 +508,8 @@ def testAWatcherEventWhilstUpdatingIsReportedAsInProgress(self): class CallbackTests(UpdateServiceTestCase): UNHANDLED = ("An unhandled error occurred while checking for automatic " - "updates for engine 'FakeEngine'") + "updates for engine 'FakeEngine'. Error detail: " + "callback failed") def fail_first(self): def fails(status, data_file): @@ -592,8 +601,9 @@ def testFailedConnectionShowsOnlyTheMaskedKey(self): self.assertNoKey() self.assertIn( ("error", - "Error accessing data update service at '" + MASKED_URL + - "' for engine 'FakeEngine'"), + CHECK_ERROR + "Error accessing data update service at '" + + MASKED_URL + "' for engine 'FakeEngine'. Error detail: " + "failed for url: " + MASKED_URL), self.pipeline.logs) @@ -637,11 +647,12 @@ def testCorruptCompressedContentIsAnErrorAndCheckedAgainLater(self): self.assertEqual([], self.data_file.refreshed) self.assertEqual( [UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR], self.statuses) - self.assertIn( - ("error", - "An unhandled error occurred while checking for automatic " - "updates for engine 'FakeEngine'"), - self.pipeline.logs) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual(1, len(errors), self.pipeline.logs) + self.assertTrue(errors[0].startswith( + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'. Error detail: "), errors[0]) self.assertEqual(1, len(self.pending)) def testFailedMd5CheckOnStartUpIsNotRepeatedAtOnce(self): From 81929958b7048a18335cf676692c8dc31c4478e7 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:16:04 +0000 Subject: [PATCH 17/30] Log an error from the errors table as .NET does, with its cause An error from a row at Error level is now logged as "Error during check for data update", with the error after it, being the row's text and then its cause, as .NET logs a DataUpdateException with the exception attached. Any other error, a callback's included, is logged with the last row's message and the error after it. Python's logger takes a string, so the error follows as "Error detail: ", as the other Python messages put a cause. The cause of a failed connection is masked, as it can name the URL with its licence key. --- .../datafile_update_service.py | 74 +++++++++++-------- 1 file changed, 45 insertions(+), 29 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index e564edfd..d67be9b4 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -99,11 +99,8 @@ def __emit(self, status, data_file): try: for method in self.on_complete_methods: method(status, data_file) - except Exception: - self._log( - "error", - "An unhandled error occurred while checking for automatic " - "updates for engine '" + self._engine_type(data_file) + "'") + except Exception as e: + self._log_unhandled(data_file, e) def update_data_file(self, data_file): @@ -150,17 +147,17 @@ def update_data_file(self, data_file): try: response = requests.get(url, headers=headers) - except requests.RequestException: - self._log( - "error", + except requests.RequestException as e: + # The message of a failed request names the URL, key included. + self._log_check_error( "Error accessing data update service at '" + shown_url + - "' for engine '" + engine + "'") + "' for engine '" + engine + "'", + mask_licence_keys_in_text(e)) return self._update_failed( data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) if response is None: - self._log( - "error", + self._log_check_error( "No response from data update service at '" + shown_url + "' for engine '" + engine + "'") return self._update_failed( @@ -195,7 +192,7 @@ def update_data_file(self, data_file): "' from data update service at '" + shown_url + "' for engine '" + engine + "'") - self._log("error", message) + self._log_check_error(message) return self._update_failed(data_file, status) # Verify the MD5 hash if enabled and the answer gives one. @@ -215,8 +212,7 @@ def update_data_file(self, data_file): hash_md5.update(data) if header_md5 != hash_md5.hexdigest(): - self._log( - "error", + self._log_check_error( "Integrity check failed. MD5 hash in HTTP response '" + str(header_md5) + "' for '" + engine + "' data update " "does not match calculated hash for the downloaded " @@ -236,13 +232,10 @@ def update_data_file(self, data_file): if data_file.decompress and data[:2] == GZIP_MAGIC: try: data = gzip.decompress(data) - except (OSError, EOFError, zlib.error): + except (OSError, EOFError, zlib.error) as e: # Content that cannot be read is any other error, which # carries the last row of the errors table. - self._log( - "error", - "An unhandled error occurred while checking for " - "automatic updates for engine '" + engine + "'") + self._log_unhandled(data_file, e) return self._update_failed( data_file, UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR) @@ -337,8 +330,7 @@ def _write_data_file(self, data_file, data): # cannot replace the file stop the update. if data_file.create_temp_copy: if not getattr(data_file, "temp_directory", None): - self._log( - "error", + self._log_check_error( "The data file '" + str(data_file.identifier) + "' is checking for updates but does not have a " "temporary file path configured.") @@ -363,12 +355,11 @@ def _write_data_file(self, data_file, data): if data_file.create_temp_copy: try: shutil.move(filename, data_file.path) - except (OSError, shutil.Error): - self._log( - "error", + except (OSError, shutil.Error) as e: + self._log_check_error( "An error occurred when copying a data file to " "replace the existing one at '" + data_file.path + - "'.") + "'.", e) return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME return None @@ -380,11 +371,10 @@ def _refresh(self, data_file): "Attempting to refresh Engine '" + engine + "' with new data") try: data_file.refresh() - except Exception: - self._log( - "error", + except Exception as e: + self._log_check_error( "An error occurred when applying a data update to engine '" + - engine + "'") + engine + "'", e) return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED return UpdateStatus.AUTO_UPDATE_SUCCESS @@ -401,6 +391,32 @@ def _log(self, level, message): if pipeline is not None: pipeline.log(level, message) + def _log_check_error(self, text, cause=None): + # An error from a row of the errors table at Error level is logged + # as .NET logs a DataUpdateException, under .NET's entry text with + # the error, being the row's text and then its cause, after it. + self._log( + "error", + "Error during check for data update. " + + self._with_cause(text, cause)) + + def _log_unhandled(self, data_file, cause): + # Any other error is logged with the last row's message, and the + # error after it. + self._log( + "error", + self._with_cause( + "An unhandled error occurred while checking for automatic " + "updates for engine '" + self._engine_type(data_file) + "'", + cause)) + + @staticmethod + def _with_cause(text, cause): + if cause is None: + return text + return (text + ("" if text.endswith(".") else ".") + + " Error detail: " + str(cause)) + @staticmethod def _engine_type(data_file): return type(getattr(data_file, "flow_element", data_file)).__name__ From 4c19c242c4cca1b0ee13ed8becc5299ce80b9ba5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:56:32 +0000 Subject: [PATCH 18/30] Test that the error entry is written exactly as 51degrees-node writes it 51degrees-node writes an error from a row of the errors table as "Error during check for data update. Error detail: ", with ". Error detail: " after it when there is a cause, and nothing more when there is none. The entry constant now holds that text, and a new test writes out two entries in full, one without a cause and one with. This commit is the tests alone, so they fail. --- .../tests/test_datafile_update_http.py | 22 +++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 1b7f6096..f250e29f 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -87,8 +87,8 @@ def response(status, content=b"new"): # The entry an error from a row of the errors table is logged under, with # the error, being the row's text and then its cause, after it, as .NET logs -# a DataUpdateException. -CHECK_ERROR = "Error during check for data update. " +# a DataUpdateException. The text is the one 51degrees-node writes. +CHECK_ERROR = "Error during check for data update. Error detail: " class UpdateServiceTestCase(unittest.TestCase): @@ -301,6 +301,24 @@ def testNotModifiedIsInformationNotWarning(self): for level in ("warn", "warning", "error"): self.assertNotIn(level, self.levels()) + def testTheErrorEntryIsWrittenExactly(self): + # Written out in full, so a change to any character fails. + self.get(response(403)) + self.service.update_data_file(self.data_file) + self.get(requests.ConnectionError("boom")) + self.service.update_data_file(self.data_file) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual( + ["Error during check for data update. Error detail: Access " + "denied to data update service at " + "'https://example.com/update' for engine 'FakeEngine'", + "Error during check for data update. Error detail: Error " + "accessing data update service at " + "'https://example.com/update' for engine 'FakeEngine'. " + "Error detail: boom"], + errors) + def testRefusedDownloadIsError(self): self.get(response(403)) self.service.update_data_file(self.data_file) From 082c572bc482b520278119f4172b10023b2020ad Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 09:57:12 +0000 Subject: [PATCH 19/30] Write the error entry as 51degrees-node does An error from a row of the errors table is now logged as "Error during check for data update. Error detail: ", followed by ". Error detail: " when there is a cause, which is the text 51degrees-node writes at f471d65d (checkErrorText and withCause in dataFileUpdateService.js), so the two languages' logs match character for character. --- .../fiftyone_pipeline_engines/datafile_update_service.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index d67be9b4..9512ad73 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -394,11 +394,13 @@ def _log(self, level, message): def _log_check_error(self, text, cause=None): # An error from a row of the errors table at Error level is logged # as .NET logs a DataUpdateException, under .NET's entry text with - # the error, being the row's text and then its cause, after it. + # the error, being the row's text and then its cause, after it, in + # the words 51degrees-node writes. self._log( "error", - "Error during check for data update. " + - self._with_cause(text, cause)) + self._with_cause( + self._with_cause("Error during check for data update", text), + cause)) def _log_unhandled(self, data_file, cause): # Any other error is logged with the last row's message, and the From 5bf1ec98659f05c7275dcb348dac48f084ed3918 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 10:06:26 +0000 Subject: [PATCH 20/30] Test the start-up, refresh and integrity texts as 51degrees-node writes them 51degrees-node at f471d65d writes "Updating on start-up", as the data-updates.md table at e89679e9 does, names the one try in a refresh after a check ("... after 1 tries.") and ends a refresh from memory with a full stop, and writes the integrity row as the table does, with no space before "data update". The tests now expect those texts, so they fail. --- .../tests/test_datafile_update_http.py | 8 +++++--- .../tests/test_datafile_update_service.py | 8 ++++++-- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index f250e29f..fb54b9b4 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -368,10 +368,12 @@ def testFailedRefreshIsError(self): self.get(response(200)) self.data_file.refresh_error = Exception("boom") self.service.update_data_file(self.data_file) + # A refresh after a check names the one try made, as the row for a + # refresh from the data file does. self.assertIn( ("error", CHECK_ERROR + "An error occurred when applying a data update " - "to engine 'FakeEngine'. Error detail: boom"), + "to engine 'FakeEngine' after 1 tries. Error detail: boom"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) @@ -388,7 +390,7 @@ def testIntegrityCheckFailureIsError(self): self.assertEqual(1, len(errors), self.pipeline.logs) self.assertTrue(errors[0].startswith( CHECK_ERROR + "Integrity check failed. MD5 hash in HTTP response " - "'not the hash' for 'FakeEngine' data update does not match " + "'not the hash' for 'FakeEngine'data update does not match " "calculated hash for the downloaded file '"), errors[0]) self.assertEqual( [UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED], @@ -399,7 +401,7 @@ def testUpdateOnStartIsLogged(self): self.data_file.updated_on_start = False with mock.patch.object(self.service, "update_data_file"): self.service.register_data_file(self.data_file) - self.assertIn(("info", "Updating on startup"), self.pipeline.logs) + self.assertIn(("info", "Updating on start-up"), self.pipeline.logs) def testFileSystemWatcherIsLogged(self): folder = tempfile.mkdtemp() diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index 2d5b2b1b..96dd178f 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -141,7 +141,9 @@ def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): errors = [log for log in self.pipeline.logs if log[0] == "error"] self.assertEqual(1, len(errors)) self.assertIn( - "An error occurred when applying a data update to engine", + "Error during check for data update. Error detail: An error " + "occurred when applying a data update to engine 'StandInDataFile' " + "after 1 tries.", errors[0][1]) self.assertEqual(1, len(self.pending)) @@ -178,7 +180,9 @@ def testFileSystemWatcherFailedRefreshUsesTheTableMessage(self): errors = [log for log in self.pipeline.logs if log[0] == "error"] self.assertEqual(1, len(errors)) self.assertIn( - "An error occurred when applying a data update to engine", + "Error during check for data update. Error detail: An error " + "occurred when applying a data update to engine 'StandInDataFile' " + "after 1 tries.", errors[0][1]) # A file that is missing at a check, for example part way through being From 77ca8160f259100695e439368976194eee541e9c Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 10:07:32 +0000 Subject: [PATCH 21/30] Write the start-up, refresh and integrity texts as 51degrees-node does "Updating on start-up" takes the hyphen the data-updates.md table at e89679e9 and 51degrees-node write. A failed refresh after a check or a watched change names its one try ("... after 1 tries."), and one from update_from_memory ends with a full stop, as the table's two rows and 51degrees-node's refreshDataFile do. The integrity row has no space before "data update", as the table writes it. --- .../datafile_update_service.py | 23 +++++++++++++------ 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 9512ad73..fada00bb 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -70,6 +70,10 @@ class UpdateStatus(Enum): # The first two bytes of gzip compressed content (RFC 1952 section 2.3.1). GZIP_MAGIC = b"\x1f\x8b" +# The tries made to refresh an engine from a data file, which the message +# for a failed refresh names. +REFRESH_TRIES = 1 + class DataFileUpdateService(): @@ -214,7 +218,7 @@ def update_data_file(self, data_file): if header_md5 != hash_md5.hexdigest(): self._log_check_error( "Integrity check failed. MD5 hash in HTTP response '" + - str(header_md5) + "' for '" + engine + "' data update " + str(header_md5) + "' for '" + engine + "'data update " "does not match calculated hash for the downloaded " "file '" + hash_md5.hexdigest() + "'.") return self._update_failed( @@ -300,7 +304,7 @@ def update_from_memory(self, data_file, data): if (status is None or status == UpdateStatus.AUTO_UPDATE_WRITE_FAILED): data_file.data = data - status = self._refresh(data_file) + status = self._refresh(data_file, from_file=False) finally: data_file.updating = False self.__emit(status, data_file) @@ -310,7 +314,7 @@ def update_from_memory(self, data_file, data): data_file.data = data data_file.updating = True try: - status = self._refresh(data_file) + status = self._refresh(data_file, from_file=False) finally: data_file.updating = False @@ -363,8 +367,11 @@ def _write_data_file(self, data_file, data): return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME return None - def _refresh(self, data_file): - # Refresh the engine with the data file's new content. + def _refresh(self, data_file, from_file=True): + # Refresh the engine with the data file's new content. A refresh + # after a check or a watched change is from the data file, and one + # from update_from_memory is not, as the two rows of the errors + # table say. engine = self._engine_type(data_file) self._log( "info", @@ -374,7 +381,9 @@ def _refresh(self, data_file): except Exception as e: self._log_check_error( "An error occurred when applying a data update to engine '" + - engine + "'", e) + engine + "'" + + (" after " + str(REFRESH_TRIES) + " tries." if from_file + else "."), e) return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED return UpdateStatus.AUTO_UPDATE_SUCCESS @@ -498,7 +507,7 @@ def check_next_update(self, data_file, run_after_error = False): return if data_file.update_on_start and not data_file.updated_on_start and not run_after_error: - self._log("info", "Updating on startup") + self._log("info", "Updating on start-up") self.update_data_file(data_file) elif data_file.auto_update: From 3a24a3b18c4afc9cccfe8b6f6e11398afab3304c Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 07:32:47 +0000 Subject: [PATCH 22/30] Test that a failed write ends the update and that an untouched watched file is not reported A data file kept on disk is read back by its engine, so the tests now use a data file whose refresh reads the file, as the on-premise engines do. They cover a download whose write fails, a temporary copy that cannot be written, an update from memory whose write fails, and a watched file nobody has written to. --- .../tests/test_datafile_update_http.py | 131 +++++++++++++++--- 1 file changed, 108 insertions(+), 23 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index fb54b9b4..c9348ed0 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -77,6 +77,22 @@ def refresh(self): self.refreshed.append(self.data) +class FileDataFile(HttpDataFile): + """Stands in for a data file kept on disk, whose engine reads the file + rather than the data held in memory, as the on-premise engines do.""" + + def __init__(self, path): + super(FileDataFile, self).__init__(path) + self.download = True + + def refresh(self): + if self.refresh_error: + raise self.refresh_error + with open(self.path, "rb") as file: + self.data = file.read() + self.refreshed.append(self.data) + + def response(status, content=b"new"): reply = mock.Mock() reply.status_code = status @@ -233,33 +249,36 @@ def testFileDataFileIsWrittenThenRefreshed(self): path = os.path.join(folder, "data.dat") with open(path, "wb") as file: file.write(b"old") - self.data_file = HttpDataFile(path) - self.data_file.download = True + self.data_file = FileDataFile(path) status = self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) with open(path, "rb") as file: self.assertEqual(b"fresh", file.read()) - self.assertEqual(1, len(self.data_file.refreshed)) + self.assertEqual([b"fresh"], self.data_file.refreshed) self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) - # A write that fails whilst updating from memory is a warning, and the - # Engine is still refreshed with the new data, as .NET does. - def testFailedWriteIsAWarningAndTheEngineIsStillRefreshed(self): + # The engine of a data file kept on disk reads that file, so a write + # that failed leaves it with the data it had. The update ends with the + # failure rather than reporting a success the engine did not have. + def testAFailedWriteFromMemoryEndsTheUpdate(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder, ignore_errors=True) path = os.path.join(folder, "missing", "data.dat") - self.data_file = HttpDataFile(path) - self.data_file.download = True + self.data_file = FileDataFile(path) status = self.service.update_from_memory(self.data_file, b"fresh") - self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) - self.assertEqual([b"fresh"], self.data_file.refreshed) - self.assertEqual([], [message for level, message - in self.pipeline.logs if level == "error"]) - self.assertIn( - ("warning", - "An error occurred when writing to '" + path + - "'. The engine will be updated to use the new data but the " - "file on disk will still contain old data."), + self.assertEqual( + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) + self.assertEqual( + [], [message for level, message + in self.pipeline.logs if level == "warning"]) + self.assertTrue( + [message for level, message in self.pipeline.logs + if level == "error" and message.startswith( + CHECK_ERROR + "An error occurred when copying a data file " + "to replace the existing one at '" + path + "'.")], self.pipeline.logs) def testUnregisteredDataFileIsLeftAlone(self): @@ -477,9 +496,71 @@ def testNoTemporaryPathIsReported(self): self.pipeline.logs) +# A download is written to the data file's path and the engine then reads +# that file, so a write that failed leaves the engine with the data it had +# and the update ends there, as it does in .NET and 51degrees-node. +class DownloadedFileTests(UpdateServiceTestCase): + + def setUp(self): + super(DownloadedFileTests, self).setUp() + self.folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.folder, ignore_errors=True) + self.path = os.path.join(self.folder, "data.dat") + with open(self.path, "wb") as file: + file.write(b"old") + self.data_file = FileDataFile(self.path) + + def errors(self): + return [message for level, message in self.pipeline.logs + if level == "error"] + + def testADownloadedFileIsWrittenAndReadBackByTheEngine(self): + self.get(response(200, b"new")) + self.service.update_data_file(self.data_file) + with open(self.path, "rb") as file: + self.assertEqual(b"new", file.read()) + self.assertEqual([b"new"], self.data_file.refreshed) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + + def testAFailedWriteEndsTheUpdateAndLeavesTheEngineAlone(self): + self.data_file = FileDataFile( + os.path.join(self.folder, "missing", "data.dat")) + self.get(response(200, b"new")) + self.service.update_data_file(self.data_file) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) + self.assertTrue( + [message for message in self.errors() + if message.startswith( + CHECK_ERROR + "An error occurred when copying a data file " + "to replace the existing one at '" + + self.data_file.path + "'.")], + self.pipeline.logs) + + def testAFailedTemporaryCopyIsAnyOtherError(self): + self.data_file.create_temp_copy = True + self.data_file.temp_directory = os.path.join(self.folder, "missing") + self.get(response(200, b"new")) + self.service.update_data_file(self.data_file) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) + self.assertTrue( + [message for message in self.errors() + if message.startswith( + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'.")], + self.pipeline.logs) + with open(self.path, "rb") as file: + self.assertEqual(b"old", file.read()) + + # Every check is reported to the update complete callbacks once it ends, and # a check that needed no update is reported as not needed, as the Reporting -# each check section says and as .NET does. +# each check section says and as .NET does. The watcher's checks are the +# changes it finds, as they are in .NET and 51degrees-node, so a look that +# finds the file untouched is not a check and is not reported. class ReportingEachCheckTests(UpdateServiceTestCase): def testNotModifiedIsReportedAsNotNeeded(self): @@ -488,7 +569,12 @@ def testNotModifiedIsReportedAsNotNeeded(self): self.assertEqual( [UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) - def testAWatcherEventWithNothingToDoIsReportedAsNotNeeded(self): + # A watched file nobody has written to is not an event, so the watcher + # reports nothing, as the watchers of .NET and 51degrees-node do. Both + # are told by the operating system, where this one looks every ten + # seconds, so reporting each look would tell every callback that no + # update was needed six times a minute for ever. + def testAnUnchangedWatchedFileIsNotReported(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder, ignore_errors=True) path = os.path.join(folder, "data.dat") @@ -497,12 +583,11 @@ def testAWatcherEventWithNothingToDoIsReportedAsNotNeeded(self): data_file = HttpDataFile(path) data_file.last_changed_time = os.stat(path).st_mtime data_file.registered = True - self.service.on_complete( - lambda status, df: None) self.hold_timers() self.service.track_local_data_file_changes(data_file) - self.assertEqual( - [UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) + self.service.track_local_data_file_changes(data_file) + self.assertEqual([], self.statuses) + self.assertEqual([], self.pipeline.logs) def testAWatcherEventWhilstUpdatingIsReportedAsInProgress(self): folder = tempfile.mkdtemp() From 3b5a065c2655669f79023bedc66c9f4be03380f2 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 07:32:56 +0000 Subject: [PATCH 23/30] End the update when the data file cannot be written, and report only the changes the watcher finds The engine of a data file kept on disk reads that file, so letting a failed write through reported a success with the old data still in use. Both routes now end with the failure, under the errors table row for the copy that replaces the file, or the row for any other error when the temporary copy is the one that cannot be written, which is what 51degrees-node writes. The watcher looks every ten seconds where .NET and 51degrees-node are told by the operating system, so a look that finds the file untouched is not an event and is not reported. The statuses take the numbers .NET gives them. --- .../datafile_update_service.py | 96 ++++++++++--------- 1 file changed, 49 insertions(+), 47 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index fada00bb..d189d51d 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -58,14 +58,12 @@ class UpdateStatus(Enum): AUTO_UPDATE_NEW_FILE_CANT_RENAME = 9 # Refreshing the engine with the new data caused an error to occur. AUTO_UPDATE_REFRESH_FAILED = 10 + # 11 is .NET's AUTO_UPDATE_NO_CONFIGURATION, which this service has no + # route to, so the two below keep the numbers .NET gives them. # The data file is checking for updates but has no temporary file path. - AUTO_UPDATE_TEMP_PATH_NOT_SET = 11 - # Writing the new data to the data file failed. The engine is still - # refreshed from the data, so this status never reaches a caller and - # marks only that the file on disk still holds the old data. - AUTO_UPDATE_WRITE_FAILED = 14 + AUTO_UPDATE_TEMP_PATH_NOT_SET = 12 # Any other error during a check for an update. - AUTO_UPDATE_UNKNOWN_ERROR = 15 + AUTO_UPDATE_UNKNOWN_ERROR = 13 # The first two bytes of gzip compressed content (RFC 1952 section 2.3.1). GZIP_MAGIC = b"\x1f\x8b" @@ -245,14 +243,13 @@ def update_data_file(self, data_file): # Download datafile if needed, or pass through memory if data_file.download: + # The engine of a data file kept on disk reads that file, so a + # write that failed leaves it with the data it already had and + # the update ends there, as it does in .NET and + # 51degrees-node. failed = self._write_data_file(data_file, data) - # A write that failed is only a warning, so the engine is still - # refreshed from the data just downloaded. - if (failed is not None and - failed != UpdateStatus.AUTO_UPDATE_WRITE_FAILED): + if failed is not None: return self._update_failed(data_file, failed) - if failed == UpdateStatus.AUTO_UPDATE_WRITE_FAILED: - data_file.data = data else: @@ -288,7 +285,8 @@ def update_from_memory(self, data_file, data): @rtype: UpdateStatus @return: AUTO_UPDATE_SUCCESS, AUTO_UPDATE_REFRESH_FAILED if the engine could not use the data (the data it had is kept), the status - of a failed write for a data file kept on disk, or + of a failed write for a data file kept on disk, which is not + refreshed because its engine reads the file, or AUTO_UPDATE_NOT_NEEDED if the data file's engine has been closed """ @@ -299,12 +297,8 @@ def update_from_memory(self, data_file, data): data_file.updating = True try: status = self._write_data_file(data_file, data) - # A write that failed is only a warning, so the engine is - # still refreshed from the data held in memory. - if (status is None or - status == UpdateStatus.AUTO_UPDATE_WRITE_FAILED): - data_file.data = data - status = self._refresh(data_file, from_file=False) + if status is None: + status = self._refresh(data_file) finally: data_file.updating = False self.__emit(status, data_file) @@ -328,10 +322,8 @@ def update_from_memory(self, data_file, data): def _write_data_file(self, data_file, data): # Write new data to the data file's path, through a temporary copy # when the data file asks for one. None when written, or the status - # of the failure, which is logged. A write that fails is a warning - # and leaves the caller free to refresh the Engine from the data in - # memory, whilst a temporary path that is not set and a copy that - # cannot replace the file stop the update. + # of the failure, which is logged. Every failure leaves the file on + # disk as it was, so the caller ends the update. if data_file.create_temp_copy: if not getattr(data_file, "temp_directory", None): self._log_check_error( @@ -348,13 +340,19 @@ def _write_data_file(self, data_file, data): try: with open(filename, 'wb') as f: f.write(data) - except OSError: - self._log( - "warning", - "An error occurred when writing to '" + str(filename) + - "'. The engine will be updated to use the new data but " - "the file on disk will still contain old data.") - return UpdateStatus.AUTO_UPDATE_WRITE_FAILED + except OSError as e: + if filename == data_file.path: + # Writing straight to the data file is the copy that + # replaces it, so it carries that row of the errors table. + self._log_check_error( + "An error occurred when copying a data file to replace " + "the existing one at '" + str(data_file.path) + "'.", e) + else: + # A temporary copy that cannot be written has no row of its + # own, so it is any other error, as 51degrees-node writes + # it. + self._log_unhandled(data_file, e) + return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME if data_file.create_temp_copy: try: @@ -573,12 +571,13 @@ def __is_registered(self, data_file): def track_local_data_file_changes(self, data_file): """! Check for local changes to a datafile. Used by - the file system watcher. The refresh takes the same route as the - polling one, so the attempt is logged, a failure carries the - errors table's message and level, and the result is reported to - the update complete callbacks either way. The next check is - scheduled whatever happens, so that a later file is still picked - up, until the data file is unregistered. + the file system watcher. A file that has been written to is an + event, and the refresh takes the same route as the polling one, so + the attempt is logged, a failure carries the errors table's message + and level, and the result is reported to the update complete + callbacks either way. The next check is scheduled whatever happens, + so that a later file is still picked up, until the data file is + unregistered. @type data_file: DataFile @param data_file: Datafile to check updates for """ @@ -591,18 +590,21 @@ def track_local_data_file_changes(self, data_file): # replaced, is looked for again at the next check. changed_time = None try: - # Every event is reported once it ends. An event that arrived - # whilst another was being applied is reported as in progress, - # and one with no newer file as not needed. + if (changed_time is None or + changed_time <= data_file.last_changed_time): + # A file nobody has written to is not an event, so nothing + # is reported. The watchers of .NET and 51degrees-node are + # told by the operating system and say nothing whilst a + # file is untouched, and this check stands in for that. + return if data_file.updating: - status = UpdateStatus.AUTO_UPDATE_IN_PROGRESS - elif (changed_time is not None and - changed_time > data_file.last_changed_time): - data_file.last_changed_time = changed_time - status = self._refresh(data_file) - else: - status = UpdateStatus.AUTO_UPDATE_NOT_NEEDED - self.__emit(status, data_file) + # A newer file found whilst an update is being applied is + # reported as in progress, and the time is not taken, so + # the next check looks at the file again. + self.__emit(UpdateStatus.AUTO_UPDATE_IN_PROGRESS, data_file) + return + data_file.last_changed_time = changed_time + self.__emit(self._refresh(data_file), data_file) finally: if self.__is_registered(data_file): t = threading.Timer( From 9453fc5746d98e2ec400aeae663496907baab67b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 07:43:03 +0000 Subject: [PATCH 24/30] Attach the error a callback raised to the log entry The errors table in the data updates specification says an error of any other kind is logged with the error attached, and that a callback's error is logged in the same way. __emit wrote the message and bound no variable, so the error the callback raised was lost. It is now written after the message, which is what .NET, Java, Node.js and Go each do. The new test file takes the licence header every other file in the repository carries. --- .../datafile_update_service.py | 8 +++-- .../tests/test_datafile_update_http.py | 35 ++++++++++++++++--- 2 files changed, 35 insertions(+), 8 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 43cdf135..4f70b555 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -87,15 +87,17 @@ def __init__(self, pipeline = None): def __emit(self, status, data_file): # A callback that raises stops the callbacks after it, as .NET's service # does, and its error is logged with .NET's message for any other - # error, so that the next check still runs. + # error and the error itself after it, so that the next check still + # runs. try: for method in self.on_complete_methods: method(status, data_file) - except Exception: + except Exception as e: self._log( "error", "An unhandled error occurred while checking for automatic " - "updates for engine '" + self._engine_type(data_file) + "'") + "updates for engine '" + self._engine_type(data_file) + + "'. Error detail: " + str(e)) def update_data_file(self, data_file): diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index c2f10ac4..f69eb205 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -1,3 +1,24 @@ +# ********************************************************************* +# This Original Work is copyright of 51 Degrees Mobile Experts Limited. +# Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, +# Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. +# +# This Original Work is licensed under the European Union Public Licence +# (EUPL) v.1.2 and is subject to its terms as set out below. +# +# If a copy of the EUPL was not distributed with this file, You can obtain +# one at https://opensource.org/licenses/EUPL-1.2. +# +# The 'Compatible Licences' set out in the Appendix to the EUPL (as may be +# amended by the European Commission) shall be deemed incompatible for +# the purposes of the Work and the provisions of the compatibility +# clause in Article 5 of the EUPL shall not apply. +# +# If using the Work as, or as part of, a network application, by +# including the attribution notice(s) required under Article 5 of the EUPL +# in the end user terms of the application under an appropriate heading, +# such notice(s) shall fulfill the requirements of that article. +# ********************************************************************* import datetime import gzip @@ -377,13 +398,17 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): # No message holds a licence key in full, as the Logging section of the data # updates specification and the License keys section of logging.md say. # A callback that raises stops the callbacks after it, as data-updates.md -# says, and its error is logged with .NET's message for any other error. -# Polling carries on. +# says, and its error is logged with .NET's message for any other error, +# with the error the callback raised attached. Polling carries on. class CallbackTests(UpdateServiceTestCase): UNHANDLED = ("An unhandled error occurred while checking for automatic " "updates for engine 'FakeEngine'") + # The error the callback raised is attached, as the errors table asks + # and as .NET, Java, Node.js and Go all write it. + UNHANDLED_WITH_CAUSE = UNHANDLED + ". Error detail: callback failed" + def fail_first(self): def fails(status, data_file): raise RuntimeError("callback failed") @@ -402,7 +427,7 @@ def testACallbackErrorOnPollingIsLoggedAndStopsTheCallbacks(self): self.service.async_update_data_file(self.data_file).join() self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([self.UNHANDLED], self.errors()) + self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self): @@ -413,14 +438,14 @@ def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self.service.update_data_file(self.data_file) self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([self.UNHANDLED], self.errors()) + self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) def testACallbackErrorOnAnUpdateFromMemoryIsLoggedAndStopsTheCallbacks( self): self.fail_first() self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual([], self.statuses) - self.assertEqual([self.UNHANDLED], self.errors()) + self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True From 08e8a7bf3e18a8b84ea2aaccd50ada75d85f6131 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 08:36:45 +0000 Subject: [PATCH 25/30] Name only the languages that attach a callback's error The comment on the expected message said .NET, Node.js and Go write the error the callback raised, and Java as well. Java's data update service calls each onUpdateComplete in a bare loop with no catch (DataUpdateServiceDefault.java on its rc, the loops after the update and after the file watcher's refresh), so a callback that raises there is not caught and nothing is logged. The comment now names the three that do. --- .../tests/test_datafile_update_http.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index f69eb205..a1e938eb 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -406,7 +406,7 @@ class CallbackTests(UpdateServiceTestCase): "updates for engine 'FakeEngine'") # The error the callback raised is attached, as the errors table asks - # and as .NET, Java, Node.js and Go all write it. + # and as .NET, Node.js and Go each write it. UNHANDLED_WITH_CAUSE = UNHANDLED + ". Error detail: callback failed" def fail_first(self): From 3395e52ae58c44d03f7c5d9d3dc26d0aa926bcc1 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 08:57:11 +0000 Subject: [PATCH 26/30] One constant for the unhandled error entry, with the cause written once The merge of the data update alignment brought a second constant for the same entry. This branch already built the text with the cause attached, so appending the cause again made the expectation read it twice. The one constant carries the whole entry, and the comment saying why the cause is attached stays with it. --- .../tests/test_datafile_update_http.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index ddd8b368..ddf5d26d 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -633,14 +633,12 @@ def testAWatcherEventWhilstUpdatingIsReportedAsInProgress(self): # with the error the callback raised attached. Polling carries on. class CallbackTests(UpdateServiceTestCase): + # The error the callback raised is attached, as the errors table asks + # and as .NET, Node.js and Go each write it. UNHANDLED = ("An unhandled error occurred while checking for automatic " "updates for engine 'FakeEngine'. Error detail: " "callback failed") - # The error the callback raised is attached, as the errors table asks - # and as .NET, Node.js and Go each write it. - UNHANDLED_WITH_CAUSE = UNHANDLED + ". Error detail: callback failed" - def fail_first(self): def fails(status, data_file): raise RuntimeError("callback failed") @@ -659,7 +657,7 @@ def testACallbackErrorOnPollingIsLoggedAndStopsTheCallbacks(self): self.service.async_update_data_file(self.data_file).join() self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self): @@ -670,14 +668,14 @@ def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self.service.update_data_file(self.data_file) self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testACallbackErrorOnAnUpdateFromMemoryIsLoggedAndStopsTheCallbacks( self): self.fail_first() self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual([], self.statuses) - self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True From eb1085dbab6ae89c3cc45335cb21c368855a8ec5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 09:22:50 +0000 Subject: [PATCH 27/30] Test that a callback's error is masked in the log entry --- .../tests/test_licence_key_masking.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py b/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py index d72b93a2..4f437f50 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py @@ -150,6 +150,24 @@ def fail(*args, **kwargs): for level, message in pipeline.logs), pipeline.logs) + def test_a_callback_error_logs_only_the_masked_key(self): + # A callback's error is written into the log entry, and a callback + # that passes on a request error carries the update URL with it. + def fails(status, data_file): + raise requests.ConnectionError("failed for url: " + URL) + + pipeline = RecordingPipeline() + service = DataFileUpdateService(pipeline) + service.on_complete(fails) + with mock.patch.object(requests, "get", lambda *a, **k: answer(304)): + service.update_data_file(StandInDataFile()) + self.assertTrue(pipeline.logs) + for level, message in pipeline.logs: + self.assertNotIn(KEY, message) + self.assertTrue(any(MASKED_URL in message + for level, message in pipeline.logs), + pipeline.logs) + if __name__ == "__main__": unittest.main() From e770a88116587f91f98676ba6ed84fbcc08012fa Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 09:23:17 +0000 Subject: [PATCH 28/30] Mask every licence key in a callback's error before logging it --- .../src/fiftyone_pipeline_engines/datafile_update_service.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 4f70b555..93568b01 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -88,7 +88,8 @@ def __emit(self, status, data_file): # A callback that raises stops the callbacks after it, as .NET's service # does, and its error is logged with .NET's message for any other # error and the error itself after it, so that the next check still - # runs. + # runs. The error is a caller's, so it is masked as any other text + # that may name the update URL is. try: for method in self.on_complete_methods: method(status, data_file) @@ -97,7 +98,7 @@ def __emit(self, status, data_file): "error", "An unhandled error occurred while checking for automatic " "updates for engine '" + self._engine_type(data_file) + - "'. Error detail: " + str(e)) + "'. Error detail: " + mask_licence_keys_in_text(e)) def update_data_file(self, data_file): From 5fe319399126976eba241e099dee50a55dd32529 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 10:12:10 +0000 Subject: [PATCH 29/30] Pair a failed temporary copy with the status its message names The errors table pairs the any other error message with AUTO_UPDATE_UNKNOWN_ERROR, and a temporary copy that cannot be written has no row of its own, so it is that error. The test asserted the copy-over-the-existing-file status instead. --- .../tests/test_datafile_update_http.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index ddf5d26d..fa30d679 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -566,7 +566,7 @@ def testAFailedTemporaryCopyIsAnyOtherError(self): self.service.update_data_file(self.data_file) self.assertEqual([], self.data_file.refreshed) self.assertEqual( - [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) + [UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR], self.statuses) self.assertTrue( [message for message in self.errors() if message.startswith( From cc916dbac4ccd558c5ea87c7f07ba631cd883ae5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 10:13:04 +0000 Subject: [PATCH 30/30] A failed temporary copy reports the unknown error status The write failure logged the any other error message and returned the copy-over-the-existing-file status, which the errors table pairs with a different message. Each branch now returns the status its own row gives. --- .../datafile_update_service.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 9c2d16a4..7d5c6972 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -349,12 +349,12 @@ def _write_data_file(self, data_file, data): self._log_check_error( "An error occurred when copying a data file to replace " "the existing one at '" + str(data_file.path) + "'.", e) - else: - # A temporary copy that cannot be written has no row of its - # own, so it is any other error, as 51degrees-node writes - # it. - self._log_unhandled(data_file, e) - return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME + return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME + # A temporary copy that cannot be written has no row of its own, + # so it is any other error, as 51degrees-node writes it, with + # the status that row of the table gives. + self._log_unhandled(data_file, e) + return UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR if data_file.create_temp_copy: try: