💚 Fix unit tests

Author: foosel

src/octoprint/util/comm.py

@@ -452,6 +452,7 @@ def __init__(self, port = None, baudrate=None, callbackObject=None, printerProfi
 		self._sdAlwaysAvailable = settings().getBoolean(["serial", "sdAlwaysAvailable"])
 		self._sdRelativePath = settings().getBoolean(["serial", "sdRelativePath"])
 		self._blockWhileDwelling = settings().getBoolean(["serial", "blockWhileDwelling"])
+		self._send_m112_on_error = settings().getBoolean(["serial", "sendM112OnError"])
 		self._currentLine = 1
 		self._line_mutex = threading.RLock()
 		self._resendDelta = None
@@ -2768,8 +2769,8 @@ def _trigger_error(self, text, reason, close=True):
 		self._changeState(self.STATE_ERROR)
 		eventManager().fire(Events.ERROR, {"error": self.getErrorString(), "reason": reason})
 		if close:
-			if settings().getBoolean(["serial", "sendM112OnError"]) and not self.isSdPrinting():
-				self._gcode_M112_queuing("M112", close=False)
+			if self._send_m112_on_error and not self.isSdPrinting():
+				self._trigger_emergency_stop(close=False)
 			self.close(is_error=True)
 
 	def _readline(self):
@@ -3699,7 +3700,7 @@ def _gcode_M110_sending(self, cmd, cmd_type=None, gcode=None, subcode=None, *arg
 			self._lastLines.clear()
 		self._resendDelta = None
 
-	def _gcode_M112_queuing(self, cmd, cmd_type=None, gcode=None, subcode=None, *args, **kwargs):
+	def _trigger_emergency_stop(self, close=True):
 		self._logger.info(u"Force-sending M112 to the printer")
 
 		# emergency stop, jump the queue with the M112, regardless of whether the EMERGENCY_PARSER capability is
@@ -3723,7 +3724,7 @@ def _gcode_M112_queuing(self, cmd, cmd_type=None, gcode=None, subcode=None, *arg
 		error_text = "Closing serial port due to emergency stop M112."
 		self._log(error_text)
 
-		if kwargs.get("close", True):
+		if close:
 			self._errorValue = error_text
 			self.close(is_error=True)
 
@@ -3732,8 +3733,8 @@ def _gcode_M112_queuing(self, cmd, cmd_type=None, gcode=None, subcode=None, *arg
 		if gcode in gcodeToEvent:
 			eventManager().fire(gcodeToEvent[gcode])
 
-		# return None 1-tuple to eat the one that is queuing because we don't want to send it twice
-		# I hope it got it the first time because as far as I can tell, there is no way to know
+	def _gcode_M112_queuing(self, *args, **kwargs):
+		self._trigger_emergency_stop()
 		return None,
 
 	def _gcode_M114_queued(self, *args, **kwargs):

tests/util/test_comm.py

@@ -21,11 +21,14 @@ def setUp(self):
 		self._comm._errorValue = None
 		self._comm._clear_to_send = mock.Mock()
 		self._comm._error_message_hooks = dict()
+		self._comm._trigger_emergency_stop = mock.Mock()
 
 		# settings
 		self._comm._ignore_errors = False
 		self._comm._disconnect_on_errors = True
+		self._comm._send_m112_on_error = True
 		self._comm.isPrinting.return_value = True
+		self._comm.isSdPrinting.return_value = False
 		self._comm.isError.return_value = False
 
 	@ddt.data(
@@ -132,6 +135,7 @@ def test_other_error_disconnect(self, line):
 		self.assertEqual(line, result)
 
 		# what should have happened
+		self.assert_m112_sent()
 		self.assert_disconnected()
 
 		# what should not have happened
@@ -140,6 +144,24 @@ def test_other_error_disconnect(self, line):
 		self.assert_not_print_cancelled()
 		self.assert_not_cleared_to_send()
 
+	@ddt.data("Error: Printer on fire")
+	def test_other_error_no_m112(self, line):
+		"""Should trigger escalation"""
+		self._comm._send_m112_on_error = False
+
+		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()
+		self.assert_not_m112_sent()
+
 	@ddt.data("Error: Printer on fire")
 	def test_other_error_cancel(self, line):
 		"""Should trigger print cancel"""
@@ -155,6 +177,7 @@ def test_other_error_cancel(self, line):
 		# what should not have happened
 		self.assert_not_handle_ok()
 		self.assert_not_last_comm_error()
+		self.assert_not_m112_sent()
 		self.assert_not_disconnected()
 
 	@ddt.data("Error: Printer on fire")
@@ -172,6 +195,8 @@ def test_other_error_ignored(self, line):
 		self.assert_not_handle_ok()
 		self.assert_not_last_comm_error()
 		self.assert_not_print_cancelled()
+		self.assert_not_m112_sent()
+		self.assert_not_disconnected()
 
 	def test_not_an_error(self):
 		"""Should pass"""
@@ -205,6 +230,12 @@ def assert_last_comm_error(self):
 	def assert_not_last_comm_error(self):
 		self.assertIsNone(self._comm._lastCommError)
 
+	def assert_m112_sent(self):
+		self._comm._trigger_emergency_stop.assert_called_once_with(close=False)
+
+	def assert_not_m112_sent(self):
+		self._comm._trigger_emergency_stop.assert_not_called()
+
 	def assert_disconnected(self):
 		self.assertIsNotNone(self._comm._errorValue)
 		self._comm._changeState.assert_called_with(self._comm.STATE_ERROR)