109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
import threading
|
|
import signal
|
|
import pytest
|
|
import mock
|
|
|
|
from robot_interface.state import State
|
|
|
|
|
|
def test_initialize_does_not_reinitialize():
|
|
"""
|
|
Check that calling `initialize` on an already initialized state does not change existing
|
|
attributes.
|
|
"""
|
|
state = State()
|
|
|
|
# Mock qi_session to avoid real session creation
|
|
mock_session = mock.MagicMock()
|
|
state.qi_session = mock_session
|
|
|
|
# Set state as already initialized
|
|
state.is_initialized = True
|
|
old_exit_event = state.exit_event
|
|
|
|
# Call initialize
|
|
state.initialize()
|
|
|
|
# Ensure existing attributes were not overwritten
|
|
assert state.exit_event == old_exit_event # exit_event should not be recreated
|
|
assert state.qi_session == mock_session # qi_session should not be replaced
|
|
assert state.is_initialized is True # is_initialized should remain True
|
|
|
|
|
|
def test_deinitialize_behavior():
|
|
"""Check that deinitialize closes sockets and updates the initialization state correctly."""
|
|
state = State()
|
|
|
|
# Case 1: Initialized with sockets
|
|
state.is_initialized = True
|
|
mock_socket_1 = mock.Mock()
|
|
mock_socket_2 = mock.Mock()
|
|
state.sockets = [mock_socket_1, mock_socket_2]
|
|
state.deinitialize()
|
|
|
|
# Sockets should be closed
|
|
mock_socket_1.close.assert_called_once()
|
|
mock_socket_2.close.assert_called_once()
|
|
# State should be marked as not initialized
|
|
assert not state.is_initialized
|
|
|
|
# Case 2: Not initialized, should not raise
|
|
state.is_initialized = False
|
|
state.sockets = []
|
|
state.deinitialize()
|
|
assert not state.is_initialized
|
|
|
|
|
|
def test_access_control_before_initialization():
|
|
"""Verify that accessing certain attributes before initialization raises RuntimeError."""
|
|
state = State()
|
|
|
|
with pytest.raises(RuntimeError, match=".*sockets.*"):
|
|
_ = state.sockets
|
|
|
|
with pytest.raises(RuntimeError, match=".*qi_session.*"):
|
|
_ = state.qi_session
|
|
|
|
|
|
def test_exit_event_before_initialized_returns_if_set():
|
|
"""Check that exit_event can be accessed even if state is not initialized,
|
|
but only if it is set."""
|
|
state = State()
|
|
|
|
# Manually create and set the exit_event
|
|
object.__setattr__(state, "exit_event", threading.Event())
|
|
object.__getattribute__(state, "exit_event").set()
|
|
|
|
# Should return the event without raising
|
|
assert state.exit_event.is_set()
|
|
|
|
|
|
def test_getattribute_allowed_attributes_before_init():
|
|
"""Ensure attributes allowed before initialization can be accessed without error."""
|
|
state = State()
|
|
|
|
assert callable(state.initialize)
|
|
assert callable(state.deinitialize)
|
|
assert state.is_initialized is False
|
|
assert state.__dict__ is not None
|
|
assert state.__class__.__name__ == "State"
|
|
assert state.__doc__ is not None
|
|
|
|
|
|
def test_signal_handler_sets_exit_event(monkeypatch):
|
|
"""Ensure SIGINT triggers the exit_event via signal handler."""
|
|
state = State()
|
|
|
|
# Patch get_qi_session to prevent real session creation
|
|
monkeypatch.setattr("robot_interface.state.get_qi_session", lambda: "dummy_session")
|
|
|
|
# Initialize state to set up signal handlers
|
|
state.initialize()
|
|
|
|
# Simulate SIGINT
|
|
signal_handler = signal.getsignal(signal.SIGINT)
|
|
signal_handler(None, None)
|
|
|
|
# Exit event should be set
|
|
assert state.exit_event.is_set()
|