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 2a8d0fd2..572c7a82 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 6e7a8ebe..1ef6fcf4 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}" 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 7ded5ad9..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 @@ -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 @@ -27,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): @@ -57,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(): """! @@ -79,8 +85,20 @@ def __init__(self, pipeline = None): self.on_complete_methods = [] def __emit(self, status, data_file): - for method in self.on_complete_methods: - method(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. 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) + except Exception as e: + self._log( + "error", + "An unhandled error occurred while checking for automatic " + "updates for engine '" + self._engine_type(data_file) + + "'. Error detail: " + mask_licence_keys_in_text(e)) def update_data_file(self, data_file): @@ -100,103 +118,126 @@ 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() # 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 = {} 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 - + since = self._if_modified_since(data_file) + if since: + headers["If-Modified-Since"] = since + + self._log( + "info", + "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. - raise requests.RequestException( - mask_licence_keys_in_text(e)) from None + 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)) + return self._update_failed( + 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 '" + shown_url + "' for Engine '" + engine + + "'") + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_NOT_NEEDED) if response.status_code != 200: # Handle errors if response.status_code == 429: - self.pipeline.log("error", "Too many requests to " + shown_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 " + shown_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 " + shown_url + "for engine " + data_file.flow_element.datakey) status = UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN else: - self.pipeline.log("error", str(response.status_code) + shown_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 " + shown_url + "." + retry + + " Error detail: HTTP status code " + + str(response.status_code)) + 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() 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 " + shown_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) + + self._log( + "info", + "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() + # Decompress if set and the content is compressed. + if data_file.decompress and data[:2] == GZIP_MAGIC: + try: + data = gzip.decompress(data) + except (OSError, EOFError, zlib.error) as e: + self._log( + "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) # 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 - - with open(filename, 'wb') as f: - f.write(data) - - if data_file.create_temp_copy: - - # Move to real file - - shutil.move(filename, data_file.path) + failed = self._write_data_file(data_file, data) + if failed is not None: + return self._update_failed(data_file, failed) else: @@ -209,19 +250,184 @@ 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 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 + @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), 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 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 + 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 _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) + 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): + # 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=True) + 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): """! @@ -237,7 +443,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 """ @@ -246,13 +454,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" @@ -277,7 +489,10 @@ 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) @@ -305,10 +520,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 """ @@ -326,13 +543,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( 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..a1e938eb --- /dev/null +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -0,0 +1,574 @@ +# ********************************************************************* +# 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 +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" + +# 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.""" + 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 + self.url = URL + if path is not None: + self.path = path + + def get_update_url(self): + return self.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 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 + 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) + 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 + 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. +# 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, +# 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, 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") + + 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.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()) + + def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( + 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([], self.statuses) + self.assertEqual(1, len(self.pending)) + 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_WITH_CAUSE], self.errors()) + + 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): + 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_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index 5fd9c872..6cbe5f29 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -119,31 +119,71 @@ 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() 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") 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): 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 +200,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) 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 8f87c1a6..4f437f50 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_licence_key_masking.py @@ -133,16 +133,40 @@ 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) + + 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__":