# FakeElement/FakeUi deliberately mirror nicegui's real API signatures (text=''
# before *args) and build fake elements dynamically, so pylint's static checks
# for keyword-before-vararg and attribute-defined-outside-init are expected
# false positives here, not real issues.
# pylint: disable=keyword-arg-before-vararg,attribute-defined-outside-init
import sys
import unittest
from dataclasses import dataclass, field
from types import ModuleType, SimpleNamespace
from unittest.mock import Mock, patch
[docs]
class FakeElement:
ui = None
def __init__(self, text='', *args, **kwargs) -> None:
self.text = text
self.args = args
self.kwargs = kwargs
self.value = kwargs.get('value')
self.enabled = True
self.children = []
self.handlers = {}
self.binding = None
self.class_names = ''
self.properties = ''
self.styles = []
self.is_deleted = False
self.is_open = False
def __enter__(self):
self.ui.context.append(self)
return self
def __exit__(self, *_args) -> None:
self.ui.context.pop()
[docs]
def classes(self, value):
self.class_names = value
return self
[docs]
def props(self, value):
self.properties = value
return self
[docs]
def style(self, value):
self.styles.append(value)
return self
[docs]
def bind_value(self, source, attribute, forward=None):
self.binding = SimpleNamespace(
kind='value', source=source, attribute=attribute,
transform=forward,
)
self.value = getattr(source, attribute)
return self
[docs]
def bind_value_from(self, source, attribute, backward=None):
self.binding = SimpleNamespace(
kind='value', source=source, attribute=attribute,
transform=backward,
)
self.refresh_binding()
return self
[docs]
def bind_text_from(self, source, attribute, backward=None):
self.binding = SimpleNamespace(
kind='text', source=source, attribute=attribute,
transform=backward,
)
self.refresh_binding()
return self
[docs]
def bind_enabled_from(self, source, attribute, backward=None):
self.binding = SimpleNamespace(
kind='enabled', source=source, attribute=attribute,
transform=backward,
)
self.refresh_binding()
return self
[docs]
def refresh_binding(self):
value = getattr(self.binding.source, self.binding.attribute)
if self.binding.transform:
value = self.binding.transform(value)
setattr(self, self.binding.kind, value)
return value
[docs]
def on(self, event, handler):
self.handlers[event] = handler
return self
[docs]
def set_value(self, value) -> None:
self.value = value
[docs]
def set_text(self, value) -> None:
self.text = value
[docs]
def clear(self) -> None:
self.children.clear()
[docs]
def open(self) -> None:
self.is_open = True
[docs]
def close(self) -> None:
self.is_open = False
[docs]
class FakeDialog(FakeElement):
def __init__(self, result) -> None:
super().__init__()
self.result = result
def __await__(self):
async def wait_for_result():
return self.result
return wait_for_result().__await__()
[docs]
class FakeTimer:
def __init__(self, interval, callback, once=False) -> None:
self.interval = interval
self.callback = callback
self.once = once
[docs]
class FakeUi:
def __init__(self) -> None:
self.context = []
self.elements = []
self.timers = []
self.dialog_results = []
[docs]
def reset(self) -> None:
self.context.clear()
self.elements.clear()
self.timers.clear()
self.dialog_results.clear()
def _element(self, kind, text='', *args, **kwargs):
element = FakeElement(text, *args, **kwargs)
element.kind = kind
self.elements.append(element)
if self.context:
self.context[-1].children.append(element)
return element
[docs]
def row(self, *args, **kwargs):
return self._element('row', *args, **kwargs)
[docs]
def label(self, text='', *args, **kwargs):
return self._element('label', text, *args, **kwargs)
[docs]
def number(self, *args, **kwargs):
return self._element('number', *args, **kwargs)
[docs]
def toggle(self, *args, **kwargs):
return self._element('toggle', *args, **kwargs)
[docs]
def checkbox(self, text='', *args, **kwargs):
return self._element('checkbox', text, *args, **kwargs)
[docs]
def html(self, text='', *args, **kwargs):
return self._element('html', text, *args, **kwargs)
[docs]
def separator(self, *args, **kwargs):
return self._element('separator', *args, **kwargs)
[docs]
def column(self, *args, **kwargs):
return self._element('column', *args, **kwargs)
[docs]
def element(self, tag, *args, **kwargs):
return self._element(tag, *args, **kwargs)
[docs]
def dialog(self):
result = self.dialog_results.pop(0) if self.dialog_results else None
dialog = FakeDialog(result)
dialog.kind = 'dialog'
self.elements.append(dialog)
if self.context:
self.context[-1].children.append(dialog)
return dialog
[docs]
def timer(self, interval, callback, once=False):
timer = FakeTimer(interval, callback, once)
self.timers.append(timer)
return timer
[docs]
def find(self, kind, text=None):
return next(
element for element in self.elements
if element.kind == kind and (text is None or element.text == text)
)
fake_ui = FakeUi()
FakeElement.ui = fake_ui
nicegui = ModuleType('nicegui')
nicegui.ui = fake_ui
fake_ui.card = FakeElement
with patch.dict(sys.modules, {'nicegui': nicegui}):
from devkit_ui.constants import NAV_ACTION, ROW_ACTION
from devkit_ui.pages.run.drop_node_card import DropNodeCard
from devkit_ui.pages.run.navigation_sidebar import NavigationSidebar
from devkit_ui.pages.run.row_discovery_card import RowDiscoveryCard
from devkit_ui.view_models.global_view_model import GlobalViewModel
from devkit_ui.view_models.run_view_model import RunViewModel
geometry_msgs = ModuleType('geometry_msgs')
geometry_msgs_msg = ModuleType('geometry_msgs.msg')
geometry_msgs_msg.Point32 = type('Point32', (), {})
geometry_msgs_msg.PolygonStamped = type('PolygonStamped', (), {})
rclpy = ModuleType('rclpy')
rclpy_qos = ModuleType('rclpy.qos')
rclpy_qos.DurabilityPolicy = SimpleNamespace(TRANSIENT_LOCAL='transient_local')
rclpy_qos.HistoryPolicy = SimpleNamespace(KEEP_LAST='keep_last')
rclpy_qos.ReliabilityPolicy = SimpleNamespace(RELIABLE='reliable')
rclpy_qos.QoSProfile = lambda **_kwargs: object()
yaml = ModuleType('yaml')
yaml.safe_load = Mock()
yaml.dump = Mock()
with patch.dict(sys.modules, {
'geometry_msgs': geometry_msgs,
'geometry_msgs.msg': geometry_msgs_msg,
'nicegui': nicegui,
'rclpy': rclpy,
'rclpy.qos': rclpy_qos,
'yaml': yaml,
}):
from devkit_ui.obstacles import attach_nav_card
[docs]
@dataclass
class Node:
name: str
meta: dict = field(default_factory=dict)
[docs]
class TestRunViewModel(unittest.TestCase):
[docs]
def test_defaults_match_idle_run_state(self) -> None:
state = RunViewModel()
self.assertEqual(state.joystick.pose_lbl, 'no odom')
self.assertEqual((state.node_map.map_svg, state.node_map.robot_svg), ('', ''))
self.assertEqual(
(state.track.interval, state.track.row_role, state.track.running),
(5.0, 'entry', False),
)
self.assertEqual((state.topo.current_node, state.topo.nav_status), ('—', 'idle'))
self.assertEqual((state.discovery.active, state.discovery.status), (False, 'idle'))
[docs]
def test_instances_do_not_share_nested_state(self) -> None:
first = RunViewModel()
second = RunViewModel()
first.topo.selected_node = 'ROW_1'
first.discovery.active = True
self.assertIsNone(second.topo.selected_node)
self.assertFalse(second.discovery.active)
[docs]
class TestDropNodeCard(unittest.TestCase):
[docs]
def setUp(self) -> None:
fake_ui.reset()
self.state = RunViewModel.DropNode()
self.topo = RunViewModel.Topo()
self.on_drop = Mock()
self.card = DropNodeCard(self.state, self.topo, self.on_drop)
[docs]
def test_row_id_binding_normalizes_numbers_and_blank_values(self) -> None:
row_id_input = fake_ui.find('number')
self.assertEqual(row_id_input.binding.transform('4'), 4)
self.assertIsNone(row_id_input.binding.transform(''))
self.assertIsNone(row_id_input.binding.transform(None))
[docs]
def test_hint_current_node_and_status_bindings_cover_boundary_states(self) -> None:
self.assertEqual(self.card.row_hint.text, NAV_ACTION)
self.assertIn('#8c959f', self.card.row_hint.styles[-1])
self.assertEqual(self.card.current_node_lbl.text, 'no current node')
self.state.row_id = 1
self.assertEqual(self.card.row_hint.refresh_binding(), ROW_ACTION)
self.assertIn('#0969da', self.card.row_hint.styles[-1])
self.topo.current_node = 'ROW_1_IN'
self.assertEqual(self.card.current_node_lbl.refresh_binding(), '→ ROW_1_IN')
self.assertIn('#1a7f37', self.card.current_node_lbl.styles[-1])
self.state.status = 'ERROR: no odometry'
self.card.status_lbl.refresh_binding()
self.assertIn('#cf222e', self.card.status_lbl.styles[-1])
[docs]
def test_drop_uses_latest_bound_state(self) -> None:
self.state.name = 'ROW_2_OUT'
self.state.row_id = 2
self.state.row_role = 'exit'
fake_ui.find('button', 'Drop').on_click()
self.on_drop.assert_called_once_with('ROW_2_OUT', 2, 'exit')
[docs]
class TestRowDiscoveryCard(unittest.IsolatedAsyncioTestCase):
[docs]
def setUp(self) -> None:
fake_ui.reset()
self.state = RunViewModel.Discovery()
self.on_start = Mock()
self.on_stop = Mock()
RowDiscoveryCard(self.state, self.on_start, self.on_stop)
self.checkbox = fake_ui.find('checkbox', 'Discovery mode')
self.on_change = self.checkbox.handlers['update:model-value']
[docs]
async def test_start_requires_explicit_confirmation(self) -> None:
fake_ui.dialog_results.append('cancel')
await self.on_change(SimpleNamespace(args=True))
self.assertFalse(self.checkbox.value)
self.on_start.assert_not_called()
[docs]
async def test_confirm_starts_and_unticking_stops_discovery(self) -> None:
fake_ui.dialog_results.append('go')
await self.on_change(SimpleNamespace(args=True))
await self.on_change(SimpleNamespace(args=False))
self.on_start.assert_called_once_with()
self.on_stop.assert_called_once_with()
[docs]
class TestObstacleUndo(unittest.TestCase):
[docs]
def setUp(self) -> None:
fake_ui.reset()
self.node = SimpleNamespace(
latest_gps=SimpleNamespace(latitude=51.45, longitude=-2.58),
obstacle_status='',
)
self.manager = Mock()
[docs]
def test_marked_obstacle_can_be_undone(self) -> None:
self.manager.add.return_value = 'OBS_1'
attach_nav_card(self.node, self.manager)
fake_ui.find('input').value = 'gate post'
fake_ui.find('button', 'Mark here').on_click()
self.manager.add.assert_called_once_with(
'circle',
lat=51.45,
lon=-2.58,
radius_m=0.5,
name='gate post',
)
self.assertEqual(fake_ui.find('input').value, '')
dialog = fake_ui.find('dialog')
self.assertTrue(dialog.is_open)
self.assertIn('position=bottom seamless', dialog.properties)
fake_ui.find('button', 'Undo').on_click()
self.manager.delete.assert_called_once_with('OBS_1')
self.assertFalse(dialog.is_open)
[docs]
def test_undo_prompt_times_out_once_but_ignores_deleted_dialog(self) -> None:
self.manager.add.return_value = 'OBS_2'
attach_nav_card(self.node, self.manager)
fake_ui.find('button', 'Mark here').on_click()
dialog = fake_ui.find('dialog')
timeout = next(timer for timer in fake_ui.timers if timer.once)
self.assertEqual(timeout.interval, 5.0)
timeout.callback()
self.assertFalse(dialog.is_open)
dialog.is_open = True
dialog.is_deleted = True
timeout.callback()
self.assertTrue(dialog.is_open)
[docs]
def test_missing_gps_reports_error_without_creating_undo_prompt(self) -> None:
self.node.latest_gps = None
attach_nav_card(self.node, self.manager)
fake_ui.find('button', 'Mark here').on_click()
self.assertEqual(self.node.obstacle_status, 'ERROR: no GPS message yet')
self.manager.add.assert_not_called()
self.assertFalse(any(element.kind == 'dialog' for element in fake_ui.elements))
if __name__ == '__main__':
unittest.main()