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 44a9b9df..5dea9793 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 @@ -65,6 +65,26 @@ class UpdateStatus(Enum): # Any other error during a check for an update. AUTO_UPDATE_UNKNOWN_ERROR = 13 + +class DataUpdateException(Exception): + """! + A data update that failed, which an update from memory raises to its + caller. The message is the one the errors table of the data updates + specification gives for the failure, and the error that caused it, if + any, is chained as its cause. + """ + + def __init__(self, message, status): + """! + @type message: str + @param message: What failed, naming the engine or the path + @type status: UpdateStatus + @param status: The status of the failure + """ + super(DataUpdateException, self).__init__(message) + self.status = status + + # The first two bytes of gzip compressed content (RFC 1952 section 2.3.1). GZIP_MAGIC = b"\x1f\x8b" @@ -285,9 +305,11 @@ def _update_data_file(self, data_file): # write that failed leaves it with the data it already had and # the update ends there, as it does in .NET and # 51degrees-node. - failed = self._write_data_file(data_file, data) - if failed is not None: - return self._update_failed(data_file, failed) + try: + self._write_data_file(data_file, data) + except DataUpdateException as e: + self._log_failed_write(data_file, e) + return self._update_failed(data_file, e.status) else: @@ -313,17 +335,23 @@ 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. + refreshed from the file. The result goes to the caller alone, so + nothing is reported to the update complete callbacks, and a failure + is raised rather than logged. @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, which is not - refreshed because its engine reads the file, or - AUTO_UPDATE_NOT_NEEDED if the data file's engine has been closed + @return: AUTO_UPDATE_SUCCESS when the engine was refreshed, or + AUTO_UPDATE_NOT_NEEDED, with nothing done, when the data file's + engine has been closed + @raises DataUpdateException: When the engine could not use the + data, with the status AUTO_UPDATE_REFRESH_FAILED, and the engine + keeps the data it had. Also when a data file kept on disk could not + be written, with the status of that failure, and the engine is not + refreshed because it reads the file. The error that caused the + failure is chained as its cause. """ if not self.__is_registered(data_file): @@ -339,41 +367,39 @@ def _update_from_memory(self, data_file, data): 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) + self._write_data_file(data_file, data) + self._refresh_from_memory(data_file) finally: data_file.updating = was_updating - self.__emit(status, data_file) - return status + return UpdateStatus.AUTO_UPDATE_SUCCESS previous = data_file.data data_file.data = data data_file.updating = True try: - status = self._refresh(data_file, from_file=False) + self._refresh_from_memory(data_file) + except DataUpdateException: + # The engine keeps the data it had, so the data file does too. + data_file.data = previous + raise finally: data_file.updating = was_updating - if status != UpdateStatus.AUTO_UPDATE_SUCCESS: - data_file.data = previous - - self.__emit(status, data_file) - - return status + return UpdateStatus.AUTO_UPDATE_SUCCESS 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. Every failure leaves the file on - # disk as it was, so the caller ends the update. + # when the data file asks for one. A failure raises a + # DataUpdateException with the message and status of its row of the + # errors table, and leaves the file on disk as it was, so the caller + # ends the update. if data_file.create_temp_copy: if not getattr(data_file, "temp_directory", None): - self._log_check_error( + raise DataUpdateException( "The data file '" + str(data_file.identifier) + "' is checking for updates but does not have a " - "temporary file path configured.") - return UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET + "temporary file path configured.", + UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET) filename = os.path.join( data_file.temp_directory, data_file.identifier + str(time.time())) @@ -387,32 +413,39 @@ def _write_data_file(self, data_file, data): if filename == data_file.path: # Writing straight to the data file is the copy that # replaces it, so it carries that row of the errors table. - self._log_check_error( + raise DataUpdateException( "An error occurred when copying a data file to replace " - "the existing one at '" + str(data_file.path) + "'.", e) - return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME + "the existing one at '" + str(data_file.path) + "'.", + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME) from e # A temporary copy that cannot be written has no row of its own, # so it is any other error, as 51degrees-node writes it, with # the status that row of the table gives. - self._log_unhandled(data_file, e) - return UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR + raise DataUpdateException( + self._unhandled_message(data_file), + UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR) from e if data_file.create_temp_copy: try: shutil.move(filename, data_file.path) except (OSError, shutil.Error) as e: - self._log_check_error( + raise DataUpdateException( "An error occurred when copying a data file to " "replace the existing one at '" + data_file.path + - "'.", e) - return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME - return None - - def _refresh(self, data_file, from_file=True): - # Refresh the engine with the data file's new content. A refresh - # after a check or a watched change is from the data file, and one - # from update_from_memory is not, as the two rows of the errors - # table say. + "'.", + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME) from e + + def _log_failed_write(self, data_file, error): + # A write that failed during a check is logged with the message of + # its row of the errors table and the error that caused it. + if error.status == UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR: + self._log_unhandled(data_file, error.__cause__) + else: + self._log_check_error(str(error), error.__cause__) + + def _refresh(self, data_file): + # Refresh the engine from the data file after a check or a watched + # change. A failure is logged with the errors table's message for a + # refresh from the data file, and its status returned. engine = self._engine_type(data_file) self._log( "info", @@ -422,12 +455,26 @@ def _refresh(self, data_file, from_file=True): except Exception as e: self._log_check_error( "An error occurred when applying a data update to engine '" + - engine + "'" + - (" after " + str(REFRESH_TRIES) + " tries." if from_file - else "."), e) + engine + "' after " + str(REFRESH_TRIES) + " tries.", e) return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED return UpdateStatus.AUTO_UPDATE_SUCCESS + def _refresh_from_memory(self, data_file): + # Refresh the engine with data given to update_from_memory. A + # failure is raised to the caller, not logged, with the errors + # table's message for a refresh from data held in memory. + 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: + raise DataUpdateException( + "An error occurred when applying a data update to engine '" + + engine + "'.", + UpdateStatus.AUTO_UPDATE_REFRESH_FAILED) from e + 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. @@ -456,10 +503,12 @@ def _log_unhandled(self, data_file, cause): # error after it. self._log( "error", - self._with_cause( - "An unhandled error occurred while checking for automatic " - "updates for engine '" + self._engine_type(data_file) + "'", - cause)) + self._with_cause(self._unhandled_message(data_file), cause)) + + def _unhandled_message(self, data_file): + # The message of the errors table's last row, for any other error. + return ("An unhandled error occurred while checking for automatic " + "updates for engine '" + self._engine_type(data_file) + "'") @staticmethod def _with_cause(text, cause): 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 47a74f19..48901f8b 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -34,6 +34,7 @@ import requests +from fiftyone_pipeline_engines import datafile_update_service from fiftyone_pipeline_engines.datafile_update_service import DataFileUpdateService from fiftyone_pipeline_engines.datafile_update_service import UpdateStatus @@ -163,6 +164,20 @@ def hold_timers(self): timer.start() self.addCleanup(timer.stop) + def assertUpdateFromMemoryFails(self, status, message): + """Runs an update from memory that should fail, checks the error it + raises and that the data file is no longer updating, and returns the + error.""" + with self.assertRaises(Exception) as caught: + self.service.update_from_memory(self.data_file, b"fresh") + error = caught.exception + self.assertIsInstance( + error, datafile_update_service.DataUpdateException) + self.assertEqual(message, str(error)) + self.assertEqual(status, error.status) + self.assertFalse(self.data_file.updating) + return error + # 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. @@ -245,63 +260,106 @@ def testPlainDateOfTodayIsNotInTheFuture(self): self.assertEqual([30 * 60 + 5], self.intervals) -# A memory data file can be refreshed from the bytes given. +# An update from memory gives its result to its caller and reports nothing to +# the update complete callbacks, as the Reporting each check section of the +# data updates specification says. It returns success, or raises an error +# that is not logged, as the Errors section says. class UpdateFromMemoryTests(UpdateServiceTestCase): + # The errors table's message for a refresh from data held in memory, + # which an update from memory gives whichever kind of data file it has. + REFRESH_FAILED = ("An error occurred when applying a data update to " + "engine 'FakeEngine'.") + # The one entry an update from memory that refreshes the engine logs. + ATTEMPT = ("info", + "Attempting to refresh Engine 'FakeEngine' with new data") + + def file_data_file(self): + """A data file kept on disk, whose file holds the old data.""" + 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") + return FileDataFile(path) + 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", - CHECK_ERROR + "An error occurred when applying a data update " - "to engine 'FakeEngine'. Error detail: boom"), - self.pipeline.logs) + self.assertEqual([], self.statuses) + self.assertEqual([self.ATTEMPT], self.pipeline.logs) + + def testAFailedRefreshRaisesAndIsNeitherReportedNorLogged(self): + cause = ValueError("boom") + self.data_file.refresh_error = cause + error = self.assertUpdateFromMemoryFails( + UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, self.REFRESH_FAILED) + self.assertIs(cause, error.__cause__) + # The engine, and so the data file, keep the data they had. + self.assertEqual(b"old", self.data_file.data) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual([], self.statuses) + self.assertEqual([self.ATTEMPT], 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 = FileDataFile(path) + self.data_file = self.file_data_file() status = self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) - with open(path, "rb") as file: + with open(self.data_file.path, "rb") as file: self.assertEqual(b"fresh", file.read()) self.assertEqual([b"fresh"], self.data_file.refreshed) - self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + self.assertEqual([], self.statuses) + self.assertEqual([self.ATTEMPT], self.pipeline.logs) + + def testAFailedRefreshOfAFileDataFileRaisesAndIsNeitherReportedNorLogged( + self): + self.data_file = self.file_data_file() + cause = ValueError("boom") + self.data_file.refresh_error = cause + error = self.assertUpdateFromMemoryFails( + UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, self.REFRESH_FAILED) + self.assertIs(cause, error.__cause__) + self.assertEqual(b"old", self.data_file.data) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual([], self.statuses) + self.assertEqual([self.ATTEMPT], self.pipeline.logs) # The engine of a data file kept on disk reads that file, so a write - # that failed leaves it with the data it had. The update ends with the - # failure rather than reporting a success the engine did not have. + # that failed leaves it with the data it had. The update ends with an + # error rather than a success the engine did not have. def testAFailedWriteFromMemoryEndsTheUpdate(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder, ignore_errors=True) path = os.path.join(folder, "missing", "data.dat") self.data_file = FileDataFile(path) - status = self.service.update_from_memory(self.data_file, b"fresh") - self.assertEqual( - UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) + error = self.assertUpdateFromMemoryFails( + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, + "An error occurred when copying a data file to replace the " + "existing one at '" + path + "'.") + self.assertIsInstance(error.__cause__, OSError) self.assertEqual([], self.data_file.refreshed) - self.assertEqual( - [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) - self.assertEqual( - [], [message for level, message - in self.pipeline.logs if level == "warning"]) - self.assertTrue( - [message for level, message in self.pipeline.logs - if level == "error" and message.startswith( - CHECK_ERROR + "An error occurred when copying a data file " - "to replace the existing one at '" + path + "'.")], - self.pipeline.logs) + self.assertEqual([], self.statuses) + self.assertEqual([], self.pipeline.logs) + + # A temporary copy that cannot be written is any other error, as it is + # for a download, and the file on disk is left as it was. + def testAFailedTemporaryCopyFromMemoryRaisesAsAnyOtherError(self): + self.data_file = self.file_data_file() + self.data_file.create_temp_copy = True + self.data_file.temp_directory = os.path.join( + os.path.dirname(self.data_file.path), "missing") + error = self.assertUpdateFromMemoryFails( + UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR, + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'") + self.assertIsInstance(error.__cause__, OSError) + with open(self.data_file.path, "rb") as file: + self.assertEqual(b"old", file.read()) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual([], self.statuses) + self.assertEqual([], self.pipeline.logs) def testUnregisteredDataFileIsLeftAlone(self): self.service.unregister_data_file(self.data_file) @@ -458,7 +516,9 @@ def testFileSystemWatcherIsLogged(self): # A data file kept on disk that cannot put the new file in place, and one -# with no temporary file path, carry .NET's message and status. +# with no temporary file path, carry .NET's message and status. A check logs +# the message and reports the status, and an update from memory raises an +# error that carries both and is not logged. class FileReplacementTests(UpdateServiceTestCase): def setUp(self): @@ -473,15 +533,19 @@ def setUp(self): self.data_file.create_temp_copy = True self.data_file.temp_directory = self.folder - def testFailedCopyNamesThePath(self): + def fail_move(self): move = mock.patch.object( shutil, "move", mock.Mock(side_effect=OSError("in use"))) move.start() self.addCleanup(move.stop) - status = self.service.update_from_memory(self.data_file, b"fresh") + + def testFailedCopyNamesThePath(self): + self.fail_move() + self.get(response(200, b"fresh")) + self.service.update_data_file(self.data_file) self.assertEqual( - UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) + [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) self.assertIn( ("error", CHECK_ERROR + "An error occurred when copying a data file to " @@ -489,11 +553,22 @@ def testFailedCopyNamesThePath(self): "'. Error detail: in use"), self.pipeline.logs) + def testFailedCopyFromMemoryRaisesAnErrorNamingThePath(self): + self.fail_move() + error = self.assertUpdateFromMemoryFails( + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, + "An error occurred when copying a data file to replace the " + "existing one at '" + self.path + "'.") + self.assertEqual("in use", str(error.__cause__)) + self.assertEqual([], self.statuses) + self.assertEqual([], self.pipeline.logs) + def testNoTemporaryPathIsReported(self): self.data_file.temp_directory = None - status = self.service.update_from_memory(self.data_file, b"fresh") + self.get(response(200, b"fresh")) + self.service.update_data_file(self.data_file) self.assertEqual( - UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET, status) + [UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET], self.statuses) self.assertIn( ("error", CHECK_ERROR + "The data file '" + self.data_file.identifier + @@ -501,6 +576,17 @@ def testNoTemporaryPathIsReported(self): "file path configured."), self.pipeline.logs) + def testNoTemporaryPathFromMemoryRaises(self): + self.data_file.temp_directory = None + error = self.assertUpdateFromMemoryFails( + UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET, + "The data file '" + self.data_file.identifier + + "' is checking for updates but does not have a temporary file " + "path configured.") + self.assertIsNone(error.__cause__) + self.assertEqual([], self.statuses) + self.assertEqual([], self.pipeline.logs) + # A download is written to the data file's path and the engine then reads # that file, so a write that failed leaves the engine with the data it had @@ -655,12 +741,14 @@ def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self.assertEqual(1, len(self.pending)) self.assertEqual([self.UNHANDLED], self.errors()) - def testACallbackErrorOnAnUpdateFromMemoryIsLoggedAndStopsTheCallbacks( - self): + # An update from memory reports nothing to the callbacks, so one that + # would raise is never called. + def testAnUpdateFromMemoryCallsNoCallback(self): self.fail_first() - 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_SUCCESS, status) self.assertEqual([], self.statuses) - self.assertEqual([self.UNHANDLED], self.errors()) + self.assertEqual([], self.errors()) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True