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