1import sys 2from unittest.mock import patch, MagicMock 3 4import pytest 5 6import Arcus 7 8from UM.Backend.Backend import Backend 9 10@pytest.fixture 11def backend(): 12 with patch("UM.Application.Application.getInstance"): 13 backend = Backend() 14 return backend 15 16@pytest.fixture 17def process(): 18 mocked_process = MagicMock() 19 mocked_process.stderr.readline = MagicMock() 20 mocked_process.stderr.readline.side_effect= [b"blarg", b"", b""] 21 22 mocked_process.stdout.readline = MagicMock() 23 mocked_process.stdout.readline.side_effect = [b"blarg", b"", b""] 24 return mocked_process 25 26 27def test_setState(backend): 28 backend.backendStateChange.emit = MagicMock() 29 30 backend.setState("BEEP") 31 backend.backendStateChange.emit.assert_called_once_with("BEEP") 32 33 # Calling it again should not cause another emit 34 backend.setState("BEEP") 35 backend.backendStateChange.emit.assert_called_once_with("BEEP") 36 37 38def test_startEngine(backend, process): 39 backend.getEngineCommand = MagicMock(return_value = "blarg") 40 41 backend._runEngineProcess = MagicMock(return_value = process) 42 43 backend.startEngine() 44 backend._runEngineProcess.assert_called_once_with("blarg") 45 46 backend.startEngine() 47 process.terminate.assert_called_once_with() 48 49 50def test_startEngineWithoutCommand(backend): 51 backend.getEngineCommand = MagicMock(return_value = None) 52 53 backend._createSocket = MagicMock() 54 55 backend.startEngine() 56 backend._createSocket.assert_called_once_with() 57 58 59def test__onSocketStateChanged_listening(backend): 60 backend.startEngine = MagicMock() 61 with patch("UM.Application.Application.getInstance"): 62 backend._onSocketStateChanged(Arcus.SocketState.Listening) 63 assert backend.startEngine.called_once_with() 64 65 66def test_onSocketStateChanged_connected(backend): 67 backend.backendConnected = MagicMock() 68 backend._onSocketStateChanged(Arcus.SocketState.Connected) 69 assert backend.backendConnected.emit.called_once_with() 70 71 72def test_handleKnownMessage(backend): 73 handler = MagicMock() 74 backend._message_handlers = {"beep": handler} 75 socket = MagicMock() 76 message = MagicMock() 77 message.getTypeName = MagicMock(return_value = "beep") 78 socket.takeNextMessage = MagicMock(return_value = message) 79 backend._socket = socket 80 backend._onMessageReceived() 81 82 handler.assert_called_once_with(message) 83 84 85def test_onSocketBindFailed(backend): 86 port = backend._port 87 backend._createSocket = MagicMock() 88 bind_failed_error = MagicMock() 89 bind_failed_error.getErrorCode = MagicMock(return_value=Arcus.ErrorCode.BindFailedError) 90 backend._onSocketError(bind_failed_error) 91 assert backend._createSocket.call_count == 1 92 assert port + 1 == backend._port 93 94 95def test_getLog(backend): 96 backend._backendLog(b"blooop") 97 98 assert backend.getLog() == [b"blooop"] 99 100 backend._backend_log_max_lines = 3 101 102 backend._backendLog(b"omgzomg") 103 backend._backendLog(b"omgzomg2") 104 # The max lines keeps deleting until the number of entries left is lower than the max (2 in this case) 105 assert backend.getLog() == [b"omgzomg", b"omgzomg2"] 106 107 108@pytest.mark.parametrize("exception", [PermissionError, FileNotFoundError, BlockingIOError]) 109def test_runEngineProcessException(backend, exception): 110 # It should be able to handle a number of exceptions without problems 111 with patch('subprocess.Popen', side_effect = exception): 112 assert backend._runEngineProcess([""]) is None 113 114 115def test_createSocket(backend): 116 # We're not testing the signal socket here, so mock it 117 mocked_signal_socket = MagicMock() 118 with patch("UM.Backend.Backend.SignalSocket", MagicMock(return_value = mocked_signal_socket)): 119 with patch("UM.Application.Application.getInstance"): 120 backend._createSocket("beep") 121 122 if sys.platform == "win32": 123 mocked_signal_socket.registerAllMessageTypes.assert_called_once_with(b"beep") 124 else: 125 mocked_signal_socket.registerAllMessageTypes.assert_called_once_with("beep") 126 127 # Try to create it again. 128 backend._createSocket("beep") 129 mocked_signal_socket.close.assert_called_once_with() 130 131 backend.close() 132 assert mocked_signal_socket.close.call_count == 2