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 93568b01..7d5c6972 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -58,10 +58,20 @@ class UpdateStatus(Enum): AUTO_UPDATE_NEW_FILE_CANT_RENAME = 9 # Refreshing the engine with the new data caused an error to occur. AUTO_UPDATE_REFRESH_FAILED = 10 + # 11 is .NET's AUTO_UPDATE_NO_CONFIGURATION, which this service has no + # route to, so the two below keep the numbers .NET gives them. + # The data file is checking for updates but has no temporary file path. + AUTO_UPDATE_TEMP_PATH_NOT_SET = 12 + # Any other error during a check for an update. + AUTO_UPDATE_UNKNOWN_ERROR = 13 # The first two bytes of gzip compressed content (RFC 1952 section 2.3.1). GZIP_MAGIC = b"\x1f\x8b" +# The tries made to refresh an engine from a data file, which the message +# for a failed refresh names. +REFRESH_TRIES = 1 + class DataFileUpdateService(): @@ -94,11 +104,7 @@ def __emit(self, status, data_file): 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)) + self._log_unhandled(data_file, e) def update_data_file(self, data_file): @@ -143,21 +149,24 @@ def update_data_file(self, data_file): "Checking for update from '" + shown_url + "' for Engine '" + engine + "'") - retry = self._retry_text(data_file) - try: response = requests.get(url, headers=headers) except requests.RequestException as e: # The message of a failed request names the URL, key included. - self._log( - "warning", - "An error occurred when connecting to " + shown_url + - " in order to check for data file updates for " + engine + - "." + retry + " Error detail: " + + self._log_check_error( + "Error accessing data update service at '" + shown_url + + "' for engine '" + engine + "'", mask_licence_keys_in_text(e)) return self._update_failed( data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) + if response is None: + self._log_check_error( + "No response from data update service at '" + shown_url + + "' for engine '" + engine + "'") + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) + if response.status_code == 304: self._log( "info", @@ -169,21 +178,25 @@ def update_data_file(self, data_file): if response.status_code != 200: - # Handle errors + # Each response that is not a success and not a 304 carries the + # message and status of the matching row of the errors table. if response.status_code == 429: status = UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS + message = ("Too many requests to '" + shown_url + + "' for engine '" + engine + "'") elif response.status_code == 403: status = UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN + message = ("Access denied to data update service at '" + + shown_url + "' for engine '" + engine + "'") else: status = UpdateStatus.AUTO_UPDATE_HTTPS_ERR + message = ("HTTP status code '" + + str(response.status_code) + + "' from data update service at '" + shown_url + + "' for engine '" + engine + "'") - self._log( - "warning", - "An error occurred while downloading a data file update " - "for " + engine + " from " + shown_url + "." + retry + - " Error detail: HTTP status code " + - str(response.status_code)) + self._log_check_error(message) return self._update_failed(data_file, status) # Verify the MD5 hash if enabled and the answer gives one. @@ -203,12 +216,11 @@ def update_data_file(self, data_file): hash_md5.update(data) 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.") + self._log_check_error( + "Integrity check failed. MD5 hash in HTTP response '" + + str(header_md5) + "' for '" + engine + "'data update " + "does not match calculated hash for the downloaded " + "file '" + hash_md5.hexdigest() + "'.") return self._update_failed( data_file, UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED) @@ -225,16 +237,18 @@ def update_data_file(self, data_file): 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)) + # Content that cannot be read is any other error, which + # carries the last row of the errors table. + self._log_unhandled(data_file, e) return self._update_failed( - data_file, UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM) + data_file, UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR) # Download datafile if needed, or pass through memory if data_file.download: + # The engine of a data file kept on disk reads that file, so a + # write that failed leaves it with the data it already had and + # the update ends there, as it does in .NET and + # 51degrees-node. failed = self._write_data_file(data_file, data) if failed is not None: return self._update_failed(data_file, failed) @@ -273,7 +287,8 @@ def update_from_memory(self, data_file, data): @rtype: UpdateStatus @return: AUTO_UPDATE_SUCCESS, AUTO_UPDATE_REFRESH_FAILED if the engine could not use the data (the data it had is kept), the status - of a failed write for a data file kept on disk, or + of a failed write for a data file kept on disk, which is not + refreshed because its engine reads the file, or AUTO_UPDATE_NOT_NEEDED if the data file's engine has been closed """ @@ -295,7 +310,7 @@ def update_from_memory(self, data_file, data): data_file.data = data data_file.updating = True try: - status = self._refresh(data_file) + status = self._refresh(data_file, from_file=False) finally: data_file.updating = False @@ -309,8 +324,15 @@ def update_from_memory(self, data_file, data): def _write_data_file(self, data_file, data): # Write new data to the data file's path, through a temporary copy # when the data file asks for one. None when written, or the status - # of the failure, which is logged. + # of the failure, which is logged. Every failure leaves the file on + # disk as it was, so the caller ends the update. if data_file.create_temp_copy: + if not getattr(data_file, "temp_directory", None): + self._log_check_error( + "The data file '" + str(data_file.identifier) + + "' is checking for updates but does not have a " + "temporary file path configured.") + return UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET filename = os.path.join( data_file.temp_directory, data_file.identifier + str(time.time())) @@ -321,26 +343,35 @@ def _write_data_file(self, data_file, data): 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 filename == data_file.path: + # Writing straight to the data file is the copy that + # replaces it, so it carries that row of the errors table. + self._log_check_error( + "An error occurred when copying a data file to replace " + "the existing one at '" + str(data_file.path) + "'.", e) + return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME + # A temporary copy that cannot be written has no row of its own, + # so it is any other error, as 51degrees-node writes it, with + # the status that row of the table gives. + self._log_unhandled(data_file, e) + return UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR if data_file.create_temp_copy: try: shutil.move(filename, data_file.path) - except (OSError, shutil.Error): - self._log( - "error", + except (OSError, shutil.Error) as e: + self._log_check_error( "An error occurred when copying a data file to " "replace the existing one at '" + data_file.path + - "'.") + "'.", e) return UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME return None - def _refresh(self, data_file): - # Refresh the engine with the data file's new content. + def _refresh(self, data_file, from_file=True): + # Refresh the engine with the data file's new content. A refresh + # after a check or a watched change is from the data file, and one + # from update_from_memory is not, as the two rows of the errors + # table say. engine = self._engine_type(data_file) self._log( "info", @@ -348,10 +379,11 @@ def _refresh(self, data_file): 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)) + 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) return UpdateStatus.AUTO_UPDATE_REFRESH_FAILED return UpdateStatus.AUTO_UPDATE_SUCCESS @@ -368,16 +400,40 @@ def _log(self, level, message): if pipeline is not None: pipeline.log(level, message) + def _log_check_error(self, text, cause=None): + # An error from a row of the errors table at Error level is logged + # as .NET logs a DataUpdateException, under .NET's entry text with + # the error, being the row's text and then its cause, after it, in + # the words 51degrees-node writes. + self._log( + "error", + self._with_cause( + self._with_cause("Error during check for data update", text), + cause)) + + def _log_unhandled(self, data_file, cause): + # Any other error is logged with the last row's message, and the + # error after it. + self._log( + "error", + self._with_cause( + "An unhandled error occurred while checking for automatic " + "updates for engine '" + self._engine_type(data_file) + "'", + cause)) + @staticmethod - def _engine_type(data_file): - return type(getattr(data_file, "flow_element", data_file)).__name__ + def _with_cause(text, cause): + # The cause may be a caller's error or a request's, either of which + # can name the update URL, so every key in it is masked here rather + # than at each call. + if cause is None: + return text + return (text + ("" if text.endswith(".") else ".") + + " Error detail: " + mask_licence_keys_in_text(cause)) @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 "" + def _engine_type(data_file): + return type(getattr(data_file, "flow_element", data_file)).__name__ @staticmethod def _http_date(moment): @@ -520,12 +576,13 @@ def __is_registered(self, data_file): def track_local_data_file_changes(self, data_file): """! Check for local changes to a datafile. Used by - the file system watcher. The refresh takes the same route as the - polling one, so the attempt is logged, a failure carries the - errors table's message and level, and the result is reported to - the update complete callbacks either way. The next check is - scheduled whatever happens, so that a later file is still picked - up, until the data file is unregistered. + the file system watcher. A file that has been written to is an + event, and the refresh takes the same route as the polling one, so + the attempt is logged, a failure carries the errors table's message + and level, and the result is reported to the update complete + callbacks either way. The next check is scheduled whatever happens, + so that a later file is still picked up, until the data file is + unregistered. @type data_file: DataFile @param data_file: Datafile to check updates for """ @@ -538,12 +595,21 @@ def track_local_data_file_changes(self, data_file): # replaced, is looked for again at the next check. changed_time = None try: - # Don't change if data_file is updating - if (changed_time is not None and - changed_time > data_file.last_changed_time and - not data_file.updating): - data_file.last_changed_time = changed_time - self.__emit(self._refresh(data_file), data_file) + if (changed_time is None or + changed_time <= data_file.last_changed_time): + # A file nobody has written to is not an event, so nothing + # is reported. The watchers of .NET and 51degrees-node are + # told by the operating system and say nothing whilst a + # file is untouched, and this check stands in for that. + return + if data_file.updating: + # A newer file found whilst an update is being applied is + # reported as in progress, and the time is not taken, so + # the next check looks at the file again. + self.__emit(UpdateStatus.AUTO_UPDATE_IN_PROGRESS, data_file) + return + data_file.last_changed_time = changed_time + self.__emit(self._refresh(data_file), data_file) finally: if self.__is_registered(data_file): t = threading.Timer( 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 a1e938eb..fa30d679 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -98,6 +98,22 @@ def refresh(self): self.refreshed.append(self.data) +class FileDataFile(HttpDataFile): + """Stands in for a data file kept on disk, whose engine reads the file + rather than the data held in memory, as the on-premise engines do.""" + + def __init__(self, path): + super(FileDataFile, self).__init__(path) + self.download = True + + def refresh(self): + if self.refresh_error: + raise self.refresh_error + with open(self.path, "rb") as file: + self.data = file.read() + self.refreshed.append(self.data) + + def response(status, content=b"new"): reply = mock.Mock() reply.status_code = status @@ -106,6 +122,12 @@ def response(status, content=b"new"): return reply +# The entry an error from a row of the errors table is logged under, with +# the error, being the row's text and then its cause, after it, as .NET logs +# a DataUpdateException. The text is the one 51degrees-node writes. +CHECK_ERROR = "Error during check for data update. Error detail: " + + class UpdateServiceTestCase(unittest.TestCase): def setUp(self): @@ -238,8 +260,8 @@ def testFailedRefreshIsReportedAndLogged(self): self.assertEqual(UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, status) self.assertIn( ("error", - "An error occurred while applying a data file update to " - "FakeEngine. Error detail: boom"), + CHECK_ERROR + "An error occurred when applying a data update " + "to engine 'FakeEngine'. Error detail: boom"), self.pipeline.logs) def testFileDataFileIsWrittenThenRefreshed(self): @@ -248,30 +270,37 @@ def testFileDataFileIsWrittenThenRefreshed(self): path = os.path.join(folder, "data.dat") with open(path, "wb") as file: file.write(b"old") - self.data_file = HttpDataFile(path) - self.data_file.download = True + self.data_file = FileDataFile(path) status = self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, status) with open(path, "rb") as file: self.assertEqual(b"fresh", file.read()) - self.assertEqual(1, len(self.data_file.refreshed)) + self.assertEqual([b"fresh"], self.data_file.refreshed) self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) - def testFailedWriteIsLoggedAndNotRefreshed(self): + # The engine of a data file kept on disk reads that file, so a write + # that failed leaves it with the data it had. The update ends with the + # failure rather than reporting a success the engine did not have. + def testAFailedWriteFromMemoryEndsTheUpdate(self): folder = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, folder, ignore_errors=True) path = os.path.join(folder, "missing", "data.dat") - self.data_file = HttpDataFile(path) - self.data_file.download = True + self.data_file = FileDataFile(path) status = self.service.update_from_memory(self.data_file, b"fresh") - self.assertEqual(UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM, status) + self.assertEqual( + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) self.assertEqual([], self.data_file.refreshed) - errors = [message for level, message in self.pipeline.logs - if level == "error"] - self.assertEqual(1, len(errors), self.pipeline.logs) - self.assertTrue(errors[0].startswith( - "An error occurred when writing to " + path + - ". Error detail: "), errors[0]) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) + self.assertEqual( + [], [message for level, message + in self.pipeline.logs if level == "warning"]) + self.assertTrue( + [message for level, message in self.pipeline.logs + if level == "error" and message.startswith( + CHECK_ERROR + "An error occurred when copying a data file " + "to replace the existing one at '" + path + "'.")], + self.pipeline.logs) def testUnregisteredDataFileIsLeftAlone(self): self.service.unregister_data_file(self.data_file) @@ -279,8 +308,8 @@ def testUnregisteredDataFileIsLeftAlone(self): self.assertEqual([], self.data_file.refreshed) -# The messages and levels are those of the Logging section of the data -# updates specification. +# The messages and levels are those of the .NET data update service, which +# the Errors table of the data updates specification follows. class LoggingTests(UpdateServiceTestCase): def levels(self): @@ -312,38 +341,66 @@ def testNotModifiedIsInformationNotWarning(self): for level in ("warn", "warning", "error"): self.assertNotIn(level, self.levels()) - def testRefusedDownloadIsWarning(self): + def testTheErrorEntryIsWrittenExactly(self): + # Written out in full, so a change to any character fails. + self.get(response(403)) + self.service.update_data_file(self.data_file) + self.get(requests.ConnectionError("boom")) + self.service.update_data_file(self.data_file) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual( + ["Error during check for data update. Error detail: Access " + "denied to data update service at " + "'https://example.com/update' for engine 'FakeEngine'", + "Error during check for data update. Error detail: Error " + "accessing data update service at " + "'https://example.com/update' for engine 'FakeEngine'. " + "Error detail: boom"], + errors) + + def testRefusedDownloadIsError(self): self.get(response(403)) self.service.update_data_file(self.data_file) self.assertIn( - ("warning", - "An error occurred while downloading a data file update for " - "FakeEngine from " + URL + ". Error detail: HTTP status code " - "403"), + ("error", + CHECK_ERROR + "Access denied to data update service at '" + + URL + "' for engine 'FakeEngine'"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_ERR_403_FORBIDDEN], self.statuses) - def testAutomaticUpdateSaysItWillBeTriedAgain(self): + def testTooManyRequestsIsError(self): + self.get(response(429)) + self.service.update_data_file(self.data_file) + self.assertIn( + ("error", + CHECK_ERROR + "Too many requests to '" + URL + + "' for engine 'FakeEngine'"), + self.pipeline.logs) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_429_TOO_MANY_ATTEMPTS], + self.statuses) + + def testAnyOtherStatusNamesTheCode(self): self.data_file.auto_update = True self.hold_timers() self.get(response(500)) self.service.update_data_file(self.data_file) self.assertIn( - ("warning", - "An error occurred while downloading a data file update for " - "FakeEngine from " + URL + ". Update will be attempted again " - "later. Error detail: HTTP status code 500"), + ("error", + CHECK_ERROR + "HTTP status code '500' from data update service " + "at '" + URL + "' for engine 'FakeEngine'"), self.pipeline.logs) + self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) - def testConnectionFailureIsWarning(self): + def testConnectionFailureIsError(self): self.get(requests.ConnectionError("boom")) self.assertFalse(self.service.update_data_file(self.data_file)) self.assertIn( - ("warning", - "An error occurred when connecting to " + URL + " in order to " - "check for data file updates for FakeEngine. Error detail: " - "boom"), + ("error", + CHECK_ERROR + "Error accessing data update service at '" + + URL + "' for engine 'FakeEngine'. Error detail: boom"), self.pipeline.logs) self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], self.statuses) @@ -351,14 +408,34 @@ def testFailedRefreshIsError(self): self.get(response(200)) self.data_file.refresh_error = Exception("boom") self.service.update_data_file(self.data_file) + # A refresh after a check names the one try made, as the row for a + # refresh from the data file does. self.assertIn( ("error", - "An error occurred while applying a data file update to " - "FakeEngine. Error detail: boom"), + CHECK_ERROR + "An error occurred when applying a data update " + "to engine 'FakeEngine' after 1 tries. Error detail: boom"), self.pipeline.logs) self.assertEqual( [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) + def testIntegrityCheckFailureIsError(self): + reply = response(200) + reply.headers = {"content-md5": "not the hash"} + reply.iter_content = lambda size: [b"new"] + self.get(reply) + self.data_file.verify_md5 = True + self.service.update_data_file(self.data_file) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual(1, len(errors), self.pipeline.logs) + self.assertTrue(errors[0].startswith( + CHECK_ERROR + "Integrity check failed. MD5 hash in HTTP response " + "'not the hash' for 'FakeEngine'data update does not match " + "calculated hash for the downloaded file '"), errors[0]) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_ERR_MD5_VALIDATION_FAILED], + self.statuses) + def testUpdateOnStartIsLogged(self): self.data_file.update_on_start = True self.data_file.updated_on_start = False @@ -395,6 +472,160 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): self.assertIsNone(getattr(data_file, "watcher_timer", None)) +# A data file kept on disk that cannot put the new file in place, and one +# with no temporary file path, carry .NET's message and status. +class FileReplacementTests(UpdateServiceTestCase): + + def setUp(self): + super(FileReplacementTests, self).setUp() + self.folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.folder, ignore_errors=True) + self.path = os.path.join(self.folder, "data.dat") + with open(self.path, "w") as file: + file.write("old") + self.data_file = HttpDataFile(self.path) + self.data_file.download = True + self.data_file.create_temp_copy = True + self.data_file.temp_directory = self.folder + + def testFailedCopyNamesThePath(self): + move = mock.patch.object( + shutil, "move", + mock.Mock(side_effect=OSError("in use"))) + move.start() + self.addCleanup(move.stop) + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual( + UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME, status) + self.assertIn( + ("error", + CHECK_ERROR + "An error occurred when copying a data file to " + "replace the existing one at '" + self.path + + "'. Error detail: in use"), + self.pipeline.logs) + + def testNoTemporaryPathIsReported(self): + self.data_file.temp_directory = None + status = self.service.update_from_memory(self.data_file, b"fresh") + self.assertEqual( + UpdateStatus.AUTO_UPDATE_TEMP_PATH_NOT_SET, status) + self.assertIn( + ("error", + CHECK_ERROR + "The data file '" + self.data_file.identifier + + "' is checking for updates but does not have a temporary " + "file path configured."), + self.pipeline.logs) + + +# A download is written to the data file's path and the engine then reads +# that file, so a write that failed leaves the engine with the data it had +# and the update ends there, as it does in .NET and 51degrees-node. +class DownloadedFileTests(UpdateServiceTestCase): + + def setUp(self): + super(DownloadedFileTests, self).setUp() + self.folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.folder, ignore_errors=True) + self.path = os.path.join(self.folder, "data.dat") + with open(self.path, "wb") as file: + file.write(b"old") + self.data_file = FileDataFile(self.path) + + def errors(self): + return [message for level, message in self.pipeline.logs + if level == "error"] + + def testADownloadedFileIsWrittenAndReadBackByTheEngine(self): + self.get(response(200, b"new")) + self.service.update_data_file(self.data_file) + with open(self.path, "rb") as file: + self.assertEqual(b"new", file.read()) + self.assertEqual([b"new"], self.data_file.refreshed) + self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + + def testAFailedWriteEndsTheUpdateAndLeavesTheEngineAlone(self): + self.data_file = FileDataFile( + os.path.join(self.folder, "missing", "data.dat")) + self.get(response(200, b"new")) + self.service.update_data_file(self.data_file) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME], self.statuses) + self.assertTrue( + [message for message in self.errors() + if message.startswith( + CHECK_ERROR + "An error occurred when copying a data file " + "to replace the existing one at '" + + self.data_file.path + "'.")], + self.pipeline.logs) + + def testAFailedTemporaryCopyIsAnyOtherError(self): + self.data_file.create_temp_copy = True + self.data_file.temp_directory = os.path.join(self.folder, "missing") + self.get(response(200, b"new")) + self.service.update_data_file(self.data_file) + self.assertEqual([], self.data_file.refreshed) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR], self.statuses) + self.assertTrue( + [message for message in self.errors() + if message.startswith( + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'.")], + self.pipeline.logs) + with open(self.path, "rb") as file: + self.assertEqual(b"old", file.read()) + + +# Every check is reported to the update complete callbacks once it ends, and +# a check that needed no update is reported as not needed, as the Reporting +# each check section says and as .NET does. The watcher's checks are the +# changes it finds, as they are in .NET and 51degrees-node, so a look that +# finds the file untouched is not a check and is not reported. +class ReportingEachCheckTests(UpdateServiceTestCase): + + def testNotModifiedIsReportedAsNotNeeded(self): + self.get(response(304)) + self.service.update_data_file(self.data_file) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_NOT_NEEDED], self.statuses) + + # A watched file nobody has written to is not an event, so the watcher + # reports nothing, as the watchers of .NET and 51degrees-node do. Both + # are told by the operating system, where this one looks every ten + # seconds, so reporting each look would tell every callback that no + # update was needed six times a minute for ever. + def testAnUnchangedWatchedFileIsNotReported(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "data.dat") + with open(path, "w") as file: + file.write("x") + data_file = HttpDataFile(path) + data_file.last_changed_time = os.stat(path).st_mtime + data_file.registered = True + self.hold_timers() + self.service.track_local_data_file_changes(data_file) + self.service.track_local_data_file_changes(data_file) + self.assertEqual([], self.statuses) + self.assertEqual([], self.pipeline.logs) + + def testAWatcherEventWhilstUpdatingIsReportedAsInProgress(self): + folder = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, folder, ignore_errors=True) + path = os.path.join(folder, "data.dat") + with open(path, "w") as file: + file.write("x") + data_file = HttpDataFile(path) + data_file.last_changed_time = os.stat(path).st_mtime - 10 + data_file.registered = True + data_file.updating = True + self.hold_timers() + self.service.track_local_data_file_changes(data_file) + self.assertEqual( + [UpdateStatus.AUTO_UPDATE_IN_PROGRESS], self.statuses) + + # No message holds a licence key in full, as the Logging section of the data # updates specification and the License keys section of logging.md say. # A callback that raises stops the callbacks after it, as data-updates.md @@ -402,12 +633,11 @@ def testNoFileSystemWatcherWithoutAutomaticUpdates(self): # 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" + UNHANDLED = ("An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'. Error detail: " + "callback failed") def fail_first(self): def fails(status, data_file): @@ -427,7 +657,7 @@ def testACallbackErrorOnPollingIsLoggedAndStopsTheCallbacks(self): self.service.async_update_data_file(self.data_file).join() self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self): @@ -438,14 +668,14 @@ def testACallbackErrorOnAProgrammaticCheckIsLoggedAndStopsTheCallbacks( self.service.update_data_file(self.data_file) self.assertEqual([], self.statuses) self.assertEqual(1, len(self.pending)) - self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testACallbackErrorOnAnUpdateFromMemoryIsLoggedAndStopsTheCallbacks( self): self.fail_first() self.service.update_from_memory(self.data_file, b"fresh") self.assertEqual([], self.statuses) - self.assertEqual([self.UNHANDLED_WITH_CAUSE], self.errors()) + self.assertEqual([self.UNHANDLED], self.errors()) def testAConnectionFailureKeepsPolling(self): self.data_file.auto_update = True @@ -493,15 +723,15 @@ def testEachFailedAnswerShowsOnlyTheMaskedKey(self): status) def testFailedConnectionShowsOnlyTheMaskedKey(self): - # The message of a failed request names the URL, key included. + # The message of a failed request names the URL, masked. self.get(requests.ConnectionError("failed for url: " + KEYED_URL)) self.service.update_data_file(self.data_file) self.assertNoKey() self.assertIn( - ("warning", - "An error occurred when connecting to " + MASKED_URL + - " in order to check for data file updates for FakeEngine. " - "Error detail: failed for url: " + MASKED_URL), + ("error", + CHECK_ERROR + "Error accessing data update service at '" + + MASKED_URL + "' for engine 'FakeEngine'. Error detail: " + "failed for url: " + MASKED_URL), self.pipeline.logs) @@ -536,7 +766,7 @@ def testContentThatIsNotCompressedIsUsedAsItIs(self): self.assertEqual([b"plain"], self.data_file.refreshed) self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) - def testCorruptCompressedContentIsWarnedAndCheckedAgainLater(self): + def testCorruptCompressedContentIsAnErrorAndCheckedAgainLater(self): self.data_file.decompress = True self.data_file.auto_update = True self.hold_timers() @@ -544,14 +774,13 @@ def testCorruptCompressedContentIsWarnedAndCheckedAgainLater(self): self.assertFalse(self.service.update_data_file(self.data_file)) self.assertEqual([], self.data_file.refreshed) self.assertEqual( - [UpdateStatus.AUTO_UPDATE_ERR_READING_STREAM], self.statuses) - warnings = [message for level, message in self.pipeline.logs - if level == "warning"] - self.assertEqual(1, len(warnings), self.pipeline.logs) - self.assertTrue(warnings[0].startswith( - "An error occurred during the integrity check of new data file " - "for FakeEngine. Update will be attempted again later. Error " - "detail: "), warnings[0]) + [UpdateStatus.AUTO_UPDATE_UNKNOWN_ERROR], self.statuses) + errors = [message for level, message in self.pipeline.logs + if level == "error"] + self.assertEqual(1, len(errors), self.pipeline.logs) + self.assertTrue(errors[0].startswith( + "An unhandled error occurred while checking for automatic " + "updates for engine 'FakeEngine'. Error detail: "), errors[0]) self.assertEqual(1, len(self.pending)) def testFailedMd5CheckOnStartUpIsNotRepeatedAtOnce(self): diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py index 6cbe5f29..96dd178f 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_service.py @@ -137,10 +137,14 @@ def testFileSystemWatcherKeepsWatchingAfterFailedRefresh(self): self.run_next_check() self.assertEqual("version 1", data_file.data) self.assertEqual( - [UpdateStatus.AUTO_UPDATE_REFRESH_FAILED], self.statuses) + UpdateStatus.AUTO_UPDATE_REFRESH_FAILED, self.statuses[-1]) errors = [log for log in self.pipeline.logs if log[0] == "error"] self.assertEqual(1, len(errors)) - self.assertIn("The data file cannot be used.", errors[0][1]) + self.assertIn( + "Error during check for data update. Error detail: An error " + "occurred when applying a data update to engine 'StandInDataFile' " + "after 1 tries.", + errors[0][1]) self.assertEqual(1, len(self.pending)) self.write("version 2") @@ -158,7 +162,7 @@ def testFileSystemWatcherRefreshIsRoutedAsPollingIs(self): self.run_next_check() self.assertEqual("version 2", data_file.data) - self.assertEqual([UpdateStatus.AUTO_UPDATE_SUCCESS], self.statuses) + self.assertEqual(UpdateStatus.AUTO_UPDATE_SUCCESS, self.statuses[-1]) attempts = [log for log in self.pipeline.logs if "Attempting to refresh Engine" in log[1]] self.assertEqual(1, len(attempts)) @@ -176,7 +180,9 @@ def testFileSystemWatcherFailedRefreshUsesTheTableMessage(self): errors = [log for log in self.pipeline.logs if log[0] == "error"] self.assertEqual(1, len(errors)) self.assertIn( - "An error occurred while applying a data file update to", + "Error during check for data update. Error detail: An error " + "occurred when applying a data update to engine 'StandInDataFile' " + "after 1 tries.", errors[0][1]) # A file that is missing at a check, for example part way through being