From cc0a17f2ac81ab31506257639099f9576227d09d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gina=20H=C3=A4u=C3=9Fge?= Date: Mon, 4 Dec 2017 15:56:31 +0100 Subject: [PATCH] Unit tests for MachineCom._handle_errors With various Marlin and Repetier error variants. --- setup.py | 2 +- src/octoprint/util/comm.py | 1 + tests/util/test_comm.py | 237 +++++++++++++++++++++++++++++++++++++ 3 files changed, 239 insertions(+), 1 deletion(-) create mode 100644 tests/util/test_comm.py diff --git a/setup.py b/setup.py index dd0ec9b6..2f22d210 100644 --- a/setup.py +++ b/setup.py @@ -63,7 +63,7 @@ EXTRA_REQUIRES = dict( # Dependencies for developing OctoPrint develop=[ # Testing dependencies - "mock>=1.0.1,<1.1", + "mock>=2.0.0,<3", "nose>=1.3.0,<1.4", "ddt", diff --git a/src/octoprint/util/comm.py b/src/octoprint/util/comm.py index d4199fe7..e3fb219d 100644 --- a/src/octoprint/util/comm.py +++ b/src/octoprint/util/comm.py @@ -1965,6 +1965,7 @@ class MachineCom(object): "workdir", "error writing to file", "cannot open", + "open failed", "cannot enter") def _handle_errors(self, line): if line is None: diff --git a/tests/util/test_comm.py b/tests/util/test_comm.py new file mode 100644 index 00000000..e606db42 --- /dev/null +++ b/tests/util/test_comm.py @@ -0,0 +1,237 @@ + +import unittest +import mock +import ddt + +import octoprint.util.comm + +@ddt.ddt +class TestCommErrorHandling(unittest.TestCase): + + def setUp(self): + self._comm = mock.create_autospec(octoprint.util.comm.MachineCom) + + # mocks + self._comm._handle_errors = lambda *args, **kwargs: octoprint.util.comm.MachineCom._handle_errors(self._comm, *args, **kwargs) + self._comm._recoverable_communication_errors = octoprint.util.comm.MachineCom._recoverable_communication_errors + self._comm._resend_request_communication_errors = octoprint.util.comm.MachineCom._resend_request_communication_errors + self._comm._sd_card_errors = octoprint.util.comm.MachineCom._sd_card_errors + self._comm._lastCommError = None + self._comm._errorValue = None + self._comm._clear_to_send = mock.Mock() + + # settings + self._comm._ignore_errors = False + self._comm._disconnect_on_errors = True + self._comm.isPrinting.return_value = True + self._comm.isError.return_value = False + + @ddt.data( + # Marlin + "Error: Line Number is not Last Line Number+1, Last Line: 1", + + # Repetier + "Error: Expected Line 1 got 2", + + # !! error type for good measure + "!! expected line 1 got 2" + ) + def test_lineno_mismatch(self, line): + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + self.assert_resend() + + @ddt.data( + # Marlin + "Error: No Line Number with checksum, Last Line: 1", + ) + def test_lineno_missing(self, line): + """Should simulate OK to force resend request""" + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + self.assert_recoverable() + + @ddt.data( + # Marlin + "Error: checksum mismatch", + + # Repetier + "Error: Wrong checksum", + ) + def test_checksum_mismatch(self, line): + """Should prepare receiving resend request""" + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + self.assert_resend() + + @ddt.data( + # Marlin + "Error: No Checksum with line number, Last Line: 1", + + # Repetier + "Error: Checksum missing", + ) + def test_checksum_missing(self, line): + """Should prepare receiving resend request""" + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + self.assert_resend() + + @ddt.data( + # Marlin + "Error: volume.init failed", + "Error: openRoot failed", + "Error: workDir open failed", + "Error: Cannot enter subdir: folder", + + # Repetier + "Error: file.open failed", + + # Marlin & Repetier (halleluja!) + "Error: error writing to file", + "Error: open failed, File: foo.gco", + + # Legacy? + "Error: Cannot open foo.gco", + ) + def test_sd_error(self, line): + """Should pass""" + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + self.assert_nop() + + @ddt.data( + # Marlin + "Error: Unknown command: \"ABC\"", + + # Repetier + "Error: Unknown command:ABC", + ) + def test_unknown_command(self, line): + """Should pass""" + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + self.assert_nop() + + @ddt.data("Error: Printer on fire") + def test_other_error_disconnect(self, line): + """Should trigger escalation""" + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + + # what should have happened + self.assert_disconnected() + + # what should not have happened + self.assert_not_handle_ok() + self.assert_not_last_comm_error() + self.assert_not_print_cancelled() + self.assert_not_cleared_to_send() + + @ddt.data("Error: Printer on fire") + def test_other_error_cancel(self, line): + """Should trigger print cancel""" + self._comm._disconnect_on_errors = False + + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + + # what should have happened + self.assert_print_cancelled() + self.assert_cleared_to_send() + + # what should not have happened + self.assert_not_handle_ok() + self.assert_not_last_comm_error() + self.assert_not_disconnected() + + @ddt.data("Error: Printer on fire") + def test_other_error_ignored(self, line): + """Should only log""" + self._comm._ignore_errors = True + + result = self._comm._handle_errors(line) + self.assertEqual(line, result) + + # what should have happened + self.assert_cleared_to_send() + + # what should not have happened + self.assert_not_handle_ok() + self.assert_not_last_comm_error() + self.assert_not_print_cancelled() + + def test_not_an_error(self): + """Should pass""" + result = self._comm._handle_errors("Not an error") + self.assertEqual("Not an error", result) + self.assert_nop() + + def test_already_error(self): + """Should pass""" + self._comm.isError.return_value = True + + result = self._comm._handle_errors("Error: Printer on fire") + self.assertEqual("Error: Printer on fire", result) + self.assert_nop() + + def test_line_none(self): + """Should pass""" + self.assertIsNone(self._comm._handle_errors(None)) + + ##~~ assertion helpers + + def assert_handle_ok(self): + self._comm._handle_ok.assert_called_once() + + def assert_not_handle_ok(self): + self._comm._handle_ok.assert_not_called() + + def assert_last_comm_error(self): + self.assertIsNotNone(self._comm._lastCommError) + + def assert_not_last_comm_error(self): + self.assertIsNone(self._comm._lastCommError) + + def assert_disconnected(self): + self.assertIsNotNone(self._comm._errorValue) + self._comm._changeState.assert_called_once() + + def assert_not_disconnected(self): + self.assertIsNone(self._comm._errorValue) + self._comm._changeState.assert_not_called() + + def assert_print_cancelled(self): + self._comm.cancelPrint.assert_called_once() + + def assert_not_print_cancelled(self): + self._comm.cancelPrint.assert_not_called() + + def assert_cleared_to_send(self): + self._comm._clear_to_send.set.assert_called_once() + + def assert_not_cleared_to_send(self): + self._comm._clear_to_send.set.assert_not_called() + + def assert_nop(self): + self.assert_not_handle_ok() + self.assert_not_last_comm_error() + self.assert_not_disconnected() + self.assert_not_print_cancelled() + self.assert_not_cleared_to_send() + + def assert_recoverable(self): + self.assert_handle_ok() + + self.assert_not_last_comm_error() + self.assert_not_disconnected() + self.assert_not_print_cancelled() + self.assert_not_cleared_to_send() + + def assert_resend(self): + self.assert_last_comm_error() + + self.assert_not_handle_ok() + self.assert_not_disconnected() + self.assert_not_print_cancelled() + self.assert_not_cleared_to_send()