From df11c1c0c505b2be8bbb18ed1c1b7a54d17d9e25 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 16:13:33 +0000 Subject: [PATCH 01/17] 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/17] 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/17] 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/17] 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/17] 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/17] 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/17] 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/17] 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/17] 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 18b1bfaa57705573b121675c4dcc7e57e290c0b4 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Wed, 11 Nov 2026 08:36:28 +0000 Subject: [PATCH 10/17] 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 11/17] 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 12/17] 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 13/17] 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 9453fc5746d98e2ec400aeae663496907baab67b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 07:43:03 +0000 Subject: [PATCH 14/17] 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 15/17] 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 eb1085dbab6ae89c3cc45335cb21c368855a8ec5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 09:22:50 +0000 Subject: [PATCH 16/17] 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 17/17] 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):