|
| 1 | +"""Tests for transitioning Manager from 'idle' to 'connected' state.""" |
| 2 | + |
| 3 | +import pytest |
| 4 | + |
| 5 | +from manager.manager.manager import Manager |
| 6 | + |
| 7 | + |
| 8 | +class DummyConsumer: |
| 9 | + """A dummy consumer to capture messages sent by the Manager.""" |
| 10 | + |
| 11 | + def __init__(self): |
| 12 | + """Initialize the DummyConsumer with empty message storage.""" |
| 13 | + self.messages = [] |
| 14 | + self.last_message = None |
| 15 | + |
| 16 | + def send_message(self, *args, **kwargs): |
| 17 | + """Capture the message sent by the Manager.""" |
| 18 | + self.messages.append((args, kwargs)) |
| 19 | + # Store the last message for verification |
| 20 | + self.last_message = (args, kwargs) |
| 21 | + |
| 22 | + |
| 23 | +@pytest.fixture |
| 24 | +def manager(monkeypatch): |
| 25 | + """Fixture to provide a Manager instance with patched dependencies for testing.""" |
| 26 | + # Patch subprocess.check_output for ROS_DISTRO and IMAGE_TAG |
| 27 | + def fake_check_output(cmd, *a, **k): |
| 28 | + if "ROS_DISTRO" in cmd[-1]: |
| 29 | + return b"humble" |
| 30 | + if "IMAGE_TAG" in cmd[-1]: |
| 31 | + return b"test_image_tag" |
| 32 | + return b"" |
| 33 | + |
| 34 | + monkeypatch.setattr("subprocess.check_output", fake_check_output) |
| 35 | + |
| 36 | + # Patch check_gpu_acceleration where it is used |
| 37 | + monkeypatch.setattr( |
| 38 | + "manager.manager.manager.check_gpu_acceleration", lambda x=None: "OFF" |
| 39 | + ) |
| 40 | + |
| 41 | + # Patch os.makedirs and os.path.isdir to avoid real FS operations |
| 42 | + monkeypatch.setattr("os.makedirs", lambda path, exist_ok=False: None) |
| 43 | + monkeypatch.setattr("os.path.isdir", lambda path: True) |
| 44 | + |
| 45 | + # Setup Manager with dummy consumer |
| 46 | + m = Manager(host="localhost", port=12345) |
| 47 | + m.consumer = DummyConsumer() |
| 48 | + return m |
| 49 | + |
| 50 | + |
| 51 | +def test_idle_to_connected(manager): |
| 52 | + """Test transitioning Manager from 'idle' to 'connected' state.""" |
| 53 | + # Initial state should be 'idle' |
| 54 | + assert manager.state == "idle" |
| 55 | + # Simulate the 'connect' event |
| 56 | + manager.trigger("connect", event=None) |
| 57 | + # State should now be 'connected' |
| 58 | + assert manager.state == "connected" |
| 59 | + # Check that the consumer received the expected message |
| 60 | + msgs = manager.consumer.messages |
| 61 | + on_connect_msg = msgs[0][0] |
| 62 | + state_change_msg = msgs[1] |
| 63 | + # print(msgs) |
| 64 | + # Verify the first message (on connect) |
| 65 | + assert on_connect_msg[0]["robotics_backend_version"] == b"test_image_tag" |
| 66 | + assert on_connect_msg[0]["ros_version"] == b"humble" |
| 67 | + assert on_connect_msg[0]["gpu_avaliable"] == "OFF" |
| 68 | + |
| 69 | + # Verify the state change message |
| 70 | + assert state_change_msg[0][0]["state"] == "connected" |
| 71 | + assert state_change_msg[1]["command"] == "state-changed" |
0 commit comments