""" Unit tests for rnode_interface.py Tests KISS protocol escaping/unescaping, configuration validation, and firmware version validation. """ import pytest import sys from unittest.mock import MagicMock, patch # Mock RNS before importing rnode_interface sys.modules['RNS'] = MagicMock() from rnode_interface import KISS, ColumbaRNodeInterface class TestKISSEscape: """Tests for KISS.escape() byte escaping.""" def test_escape_fend(self): """FEND (0xC0) should be escaped to 0xDB 0xDC.""" data = bytes([0xC0]) escaped = KISS.escape(data) assert escaped == bytes([0xDB, 0xDC]) def test_escape_fesc(self): """FESC (0xDB) should be escaped to 0xDB 0xDD.""" data = bytes([0xDB]) escaped = KISS.escape(data) assert escaped == bytes([0xDB, 0xDD]) def test_escape_mixed_special_bytes(self): """Multiple special bytes should all be escaped.""" data = bytes([0xC0, 0xDB, 0xC0]) escaped = KISS.escape(data) # 0xDB must be escaped first, then 0xC0 assert escaped == bytes([0xDB, 0xDC, 0xDB, 0xDD, 0xDB, 0xDC]) def test_escape_no_special_bytes(self): """Data without special bytes should pass through unchanged.""" data = bytes([0x00, 0x01, 0x02, 0xFF]) escaped = KISS.escape(data) assert escaped == data def test_escape_empty_data(self): """Empty data should return empty.""" data = bytes() escaped = KISS.escape(data) assert escaped == bytes() def test_escape_all_fend(self): """All FEND bytes should be escaped.""" data = bytes([0xC0, 0xC0, 0xC0]) escaped = KISS.escape(data) assert escaped == bytes([0xDB, 0xDC, 0xDB, 0xDC, 0xDB, 0xDC]) class TestKISSUnescape: """Tests for KISS.unescape() byte unescaping.""" def test_unescape_fend(self): """0xDB 0xDC should be unescaped to FEND (0xC0).""" data = bytes([0xDB, 0xDC]) unescaped = KISS.unescape(data) assert unescaped == bytes([0xC0]) def test_unescape_fesc(self): """0xDB 0xDD should be unescaped to FESC (0xDB).""" data = bytes([0xDB, 0xDD]) unescaped = KISS.unescape(data) assert unescaped == bytes([0xDB]) def test_unescape_mixed(self): """Multiple escape sequences should all be unescaped.""" data = bytes([0xDB, 0xDC, 0xDB, 0xDD, 0xDB, 0xDC]) unescaped = KISS.unescape(data) assert unescaped == bytes([0xC0, 0xDB, 0xC0]) def test_unescape_no_escape_sequences(self): """Data without escape sequences should pass through unchanged.""" data = bytes([0x00, 0x01, 0x02, 0xFF]) unescaped = KISS.unescape(data) assert unescaped == data def test_unescape_empty_data(self): """Empty data should return empty.""" data = bytes() unescaped = KISS.unescape(data) assert unescaped == bytes() def test_unescape_invalid_escape_skipped(self): """Invalid escape sequences (0xDB followed by invalid byte) should be skipped. Per KISS protocol, 0xDB should only be followed by 0xDC (TFEND) or 0xDD (TFESC). If followed by any other byte, both bytes should be skipped to avoid data corruption. """ # 0xDB 0xAA is invalid - should skip both bytes data = bytes([0x01, 0xDB, 0xAA, 0x02]) unescaped = KISS.unescape(data) # Invalid escape (0xDB 0xAA) should be skipped assert unescaped == bytes([0x01, 0x02]) def test_unescape_trailing_fesc_skipped(self): """Trailing FESC (0xDB) at end of data should be skipped. If data ends with 0xDB, there's no following byte to complete the escape sequence. The trailing 0xDB should be skipped. """ data = bytes([0x01, 0x02, 0xDB]) unescaped = KISS.unescape(data) # Trailing 0xDB should be skipped assert unescaped == bytes([0x01, 0x02]) class TestKISSRoundTrip: """Tests for escape/unescape round-trip consistency.""" def test_roundtrip_special_bytes(self): """escape(unescape(x)) should equal x for escaped data.""" original_unescaped = bytes([0xC0, 0xDB, 0x00, 0xFF]) escaped = KISS.escape(original_unescaped) roundtrip = KISS.unescape(escaped) assert roundtrip == original_unescaped def test_roundtrip_random_data(self): """Round-trip should work for arbitrary byte sequences.""" original = bytes(range(256)) escaped = KISS.escape(original) roundtrip = KISS.unescape(escaped) assert roundtrip == original class TestKISSErrorMessages: """Tests for KISS.get_error_message().""" def test_error_initradio(self): """Error 0x01 should return radio initialization message.""" msg = KISS.get_error_message(0x01) assert "Radio initialization failed" in msg def test_error_txfailed(self): """Error 0x02 should return transmission failed message.""" msg = KISS.get_error_message(0x02) assert "Transmission failed" in msg def test_error_queue_full(self): """Error 0x04 should return queue overflow message.""" msg = KISS.get_error_message(0x04) assert "queue" in msg.lower() def test_error_invalid_config(self): """Error 0x40 should return config error with TX power hint.""" msg = KISS.get_error_message(0x40) assert "Invalid configuration" in msg assert "TX power" in msg def test_error_unknown(self): """Unknown error codes should return hex representation.""" msg = KISS.get_error_message(0xFF) assert "0xFF" in msg or "Unknown" in msg class TestValidateConfig: """Tests for ColumbaRNodeInterface._validate_config().""" def create_interface(self, **overrides): """Create a test interface with mocked dependencies.""" config = { 'target_device_name': 'RNode Test', 'connection_mode': 'ble', 'frequency': 915000000, 'bandwidth': 250000, 'txpower': 17, 'sf': 11, 'cr': 5, 'st_alock': None, 'lt_alock': None, 'mode': 'full', 'enable_framebuffer': True, } config.update(overrides) with patch.object(ColumbaRNodeInterface, '__init__', lambda x, *args: None): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.FREQ_MIN = 137000000 iface.FREQ_MAX = 3000000000 iface.frequency = config['frequency'] iface.txpower = config['txpower'] iface.bandwidth = config['bandwidth'] iface.sf = config['sf'] iface.cr = config['cr'] iface.st_alock = config['st_alock'] iface.lt_alock = config['lt_alock'] return iface def test_valid_config(self): """Valid configuration should not raise.""" iface = self.create_interface() iface._validate_config() # Should not raise def test_frequency_below_min(self): """Frequency below 137 MHz should raise ValueError.""" iface = self.create_interface(frequency=100000000) with pytest.raises(ValueError, match="frequency"): iface._validate_config() def test_frequency_above_max(self): """Frequency above 3000 MHz should raise ValueError.""" iface = self.create_interface(frequency=4000000000) with pytest.raises(ValueError, match="frequency"): iface._validate_config() def test_frequency_at_min_boundary(self): """Frequency at 137 MHz (min) should be valid.""" iface = self.create_interface(frequency=137000000) iface._validate_config() # Should not raise def test_frequency_at_max_boundary(self): """Frequency at 3000 MHz (max) should be valid.""" iface = self.create_interface(frequency=3000000000) iface._validate_config() # Should not raise def test_txpower_negative(self): """Negative TX power should raise ValueError.""" iface = self.create_interface(txpower=-1) with pytest.raises(ValueError, match="TX power"): iface._validate_config() def test_txpower_above_max(self): """TX power above 36 dBm should raise ValueError.""" iface = self.create_interface(txpower=40) with pytest.raises(ValueError, match="TX power"): iface._validate_config() def test_txpower_at_max(self): """TX power at 36 dBm (NZ regional max) should be valid.""" iface = self.create_interface(txpower=36) iface._validate_config() # Should not raise def test_txpower_us_regional_max(self): """TX power at 30 dBm (US regional max) should be valid.""" iface = self.create_interface(txpower=30) iface._validate_config() # Should not raise def test_txpower_eu_regional_max(self): """TX power at 27 dBm (EU 868-P regional max) should be valid.""" iface = self.create_interface(txpower=27) iface._validate_config() # Should not raise def test_txpower_common_device_max(self): """TX power at 22 dBm (common RNode device max) should be valid.""" iface = self.create_interface(txpower=22) iface._validate_config() # Should not raise def test_bandwidth_below_min(self): """Bandwidth below 7800 Hz should raise ValueError.""" iface = self.create_interface(bandwidth=5000) with pytest.raises(ValueError, match="bandwidth"): iface._validate_config() def test_bandwidth_above_max(self): """Bandwidth above 1625000 Hz should raise ValueError.""" iface = self.create_interface(bandwidth=2000000) with pytest.raises(ValueError, match="bandwidth"): iface._validate_config() def test_sf_below_min(self): """Spreading factor below 5 should raise ValueError.""" iface = self.create_interface(sf=4) with pytest.raises(ValueError, match="spreading factor"): iface._validate_config() def test_sf_above_max(self): """Spreading factor above 12 should raise ValueError.""" iface = self.create_interface(sf=13) with pytest.raises(ValueError, match="spreading factor"): iface._validate_config() def test_cr_below_min(self): """Coding rate below 5 should raise ValueError.""" iface = self.create_interface(cr=4) with pytest.raises(ValueError, match="coding rate"): iface._validate_config() def test_cr_above_max(self): """Coding rate above 8 should raise ValueError.""" iface = self.create_interface(cr=9) with pytest.raises(ValueError, match="coding rate"): iface._validate_config() def test_st_alock_negative(self): """Negative short-term airtime limit should raise ValueError.""" iface = self.create_interface(st_alock=-1.0) with pytest.raises(ValueError, match="short-term airtime"): iface._validate_config() def test_st_alock_above_100(self): """Short-term airtime limit above 100% should raise ValueError.""" iface = self.create_interface(st_alock=101.0) with pytest.raises(ValueError, match="short-term airtime"): iface._validate_config() def test_st_alock_valid(self): """Valid short-term airtime limit should not raise.""" iface = self.create_interface(st_alock=10.0) iface._validate_config() # Should not raise def test_lt_alock_negative(self): """Negative long-term airtime limit should raise ValueError.""" iface = self.create_interface(lt_alock=-0.1) with pytest.raises(ValueError, match="long-term airtime"): iface._validate_config() def test_lt_alock_above_100(self): """Long-term airtime limit above 100% should raise ValueError.""" iface = self.create_interface(lt_alock=100.1) with pytest.raises(ValueError, match="long-term airtime"): iface._validate_config() def test_alock_none_is_valid(self): """None airtime limits should be valid (disabled).""" iface = self.create_interface(st_alock=None, lt_alock=None) iface._validate_config() # Should not raise class TestValidateFirmware: """Tests for ColumbaRNodeInterface._validate_firmware().""" def create_interface_for_firmware(self, maj_version, min_version): """Create a test interface for firmware validation.""" with patch.object(ColumbaRNodeInterface, '__init__', lambda x, *args: None): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.REQUIRED_FW_VER_MAJ = 1 iface.REQUIRED_FW_VER_MIN = 52 iface.maj_version = maj_version iface.min_version = min_version iface.firmware_ok = False return iface def test_firmware_valid_exact_minimum(self): """Firmware 1.52 (exact minimum) should be valid.""" iface = self.create_interface_for_firmware(1, 52) iface._validate_firmware() assert iface.firmware_ok is True def test_firmware_valid_higher_minor(self): """Firmware 1.100 should be valid.""" iface = self.create_interface_for_firmware(1, 100) iface._validate_firmware() assert iface.firmware_ok is True def test_firmware_valid_higher_major(self): """Firmware 2.0 should be valid (major > required).""" iface = self.create_interface_for_firmware(2, 0) iface._validate_firmware() assert iface.firmware_ok is True def test_firmware_valid_much_higher_major(self): """Firmware 5.10 should be valid.""" iface = self.create_interface_for_firmware(5, 10) iface._validate_firmware() assert iface.firmware_ok is True def test_firmware_invalid_low_minor(self): """Firmware 1.51 should be invalid (below 1.52).""" iface = self.create_interface_for_firmware(1, 51) iface._validate_firmware() assert iface.firmware_ok is False def test_firmware_invalid_low_major(self): """Firmware 0.99 should be invalid.""" iface = self.create_interface_for_firmware(0, 99) iface._validate_firmware() assert iface.firmware_ok is False def test_firmware_invalid_zero(self): """Firmware 0.0 should be invalid.""" iface = self.create_interface_for_firmware(0, 0) iface._validate_firmware() assert iface.firmware_ok is False def test_firmware_invalid_1_0(self): """Firmware 1.0 should be invalid.""" iface = self.create_interface_for_firmware(1, 0) iface._validate_firmware() assert iface.firmware_ok is False class TestThreadSafety: """Tests for thread-safe read loop control using threading.Event.""" def create_interface(self): """Create a test interface with mocked dependencies.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): config = { 'frequency': 915000000, 'bandwidth': 125000, 'txpower': 17, 'sf': 8, 'cr': 5, } iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.frequency = config['frequency'] iface.bandwidth = config['bandwidth'] iface.txpower = config['txpower'] iface.sf = config['sf'] iface.cr = config['cr'] iface.st_alock = None iface.lt_alock = None iface.name = "test-rnode" iface.target_device_name = "TestRNode" iface.connection_mode = 0 iface.kotlin_bridge = None iface.enable_framebuffer = False iface.framebuffer_enabled = False iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface._reconnect_thread = None iface._reconnecting = False iface._max_reconnect_attempts = 30 iface._reconnect_interval = 10.0 iface._on_error_callback = None iface._on_online_status_changed = None return iface def test_running_flag_is_threading_event(self): """_running should be a threading.Event for thread-safe signaling.""" import threading iface = self.create_interface() assert isinstance(iface._running, threading.Event) def test_running_event_initially_not_set(self): """_running event should not be set on initialization.""" iface = self.create_interface() assert not iface._running.is_set() def test_running_event_can_be_set_and_cleared(self): """_running event should support set() and clear() operations.""" iface = self.create_interface() # Initially not set assert not iface._running.is_set() # Set the event iface._running.set() assert iface._running.is_set() # Clear the event iface._running.clear() assert not iface._running.is_set() def test_running_event_thread_safe_signaling(self): """Verify threading.Event provides thread-safe stop signaling.""" import threading import time iface = self.create_interface() loop_iterations = [] def mock_loop(): """Simulate read loop behavior.""" while iface._running.is_set(): loop_iterations.append(1) time.sleep(0.01) # Start the event and thread iface._running.set() thread = threading.Thread(target=mock_loop) thread.start() # Let it run a bit time.sleep(0.05) # Clear event to stop the loop iface._running.clear() thread.join(timeout=1.0) # Thread should have stopped cleanly assert not thread.is_alive() assert len(loop_iterations) > 0 # Loop ran at least once class TestThreadSafeAccessors: """Tests for thread-safe RSSI/SNR accessors.""" def create_interface(self): """Create a test interface with mocked dependencies.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface._read_lock = threading.Lock() iface.r_stat_rssi = -75 iface.r_stat_snr = 8.5 return iface def test_get_rssi_uses_lock(self): """get_rssi() should acquire _read_lock for thread safety.""" iface = self.create_interface() mock_lock = MagicMock() mock_lock.__enter__ = MagicMock(return_value=None) mock_lock.__exit__ = MagicMock(return_value=None) iface._read_lock = mock_lock result = iface.get_rssi() mock_lock.__enter__.assert_called_once() mock_lock.__exit__.assert_called_once() assert result == -75 def test_get_snr_uses_lock(self): """get_snr() should acquire _read_lock for thread safety.""" iface = self.create_interface() mock_lock = MagicMock() mock_lock.__enter__ = MagicMock(return_value=None) mock_lock.__exit__ = MagicMock(return_value=None) iface._read_lock = mock_lock result = iface.get_snr() mock_lock.__enter__.assert_called_once() mock_lock.__exit__.assert_called_once() assert result == 8.5 def test_get_rssi_returns_correct_value_under_lock(self): """get_rssi() should return correct value while holding lock.""" iface = self.create_interface() iface.r_stat_rssi = -90 result = iface.get_rssi() assert result == -90 def test_get_snr_returns_correct_value_under_lock(self): """get_snr() should return correct value while holding lock.""" iface = self.create_interface() iface.r_stat_snr = 12.0 result = iface.get_snr() assert result == 12.0 class TestWriteRetry: """Tests for write retry with exponential backoff.""" def create_interface_for_write(self): """Create a test interface with mocked dependencies.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.name = "test-rnode" iface.kotlin_bridge = MagicMock() iface.connection_mode = ColumbaRNodeInterface.MODE_CLASSIC return iface def test_write_uses_exponential_backoff(self): """Write retry should use exponential backoff delays.""" import time iface = self.create_interface_for_write() test_data = b"test" # Mock writeSync to always fail (return 0 bytes written) iface.kotlin_bridge.writeSync.return_value = 0 sleep_calls = [] def mock_sleep(seconds): sleep_calls.append(seconds) # Don't actually sleep in tests with patch('time.sleep', mock_sleep): with pytest.raises(IOError): iface._write(test_data) # With 3 retries, we get 2 sleep calls between attempts # Should have exponential backoff: 0.3s, 1.0s assert len(sleep_calls) == 2 assert sleep_calls[0] == pytest.approx(0.3, rel=0.01) assert sleep_calls[1] == pytest.approx(1.0, rel=0.01) def test_write_succeeds_on_first_attempt(self): """Write should not retry if first attempt succeeds.""" import time iface = self.create_interface_for_write() test_data = b"test" # Mock writeSync to succeed iface.kotlin_bridge.writeSync.return_value = len(test_data) sleep_calls = [] with patch('time.sleep', lambda s: sleep_calls.append(s)): iface._write(test_data) # Should not raise # No retries needed assert len(sleep_calls) == 0 assert iface.kotlin_bridge.writeSync.call_count == 1 def test_write_succeeds_on_retry(self): """Write should succeed if a retry works.""" import time iface = self.create_interface_for_write() test_data = b"test" # Mock writeSync to fail first, then succeed iface.kotlin_bridge.writeSync.side_effect = [0, len(test_data)] sleep_calls = [] with patch('time.sleep', lambda s: sleep_calls.append(s)): iface._write(test_data) # Should not raise # One retry delay assert len(sleep_calls) == 1 assert iface.kotlin_bridge.writeSync.call_count == 2 class TestRadioStateThreadSafety: """Tests for thread-safe access to radio state variables (Issue 3).""" def create_interface(self): """Create a test interface with mocked dependencies.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.frequency = 915000000 iface.bandwidth = 125000 iface.txpower = 17 iface.sf = 8 iface.cr = 5 iface.st_alock = None iface.lt_alock = None iface.name = "test-rnode" iface.target_device_name = "TestRNode" iface.connection_mode = 0 iface.kotlin_bridge = None iface.enable_framebuffer = False iface.framebuffer_enabled = False iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface._reconnect_thread = None iface._reconnecting = False iface._max_reconnect_attempts = 30 iface._reconnect_interval = 10.0 iface._on_error_callback = None iface._on_online_status_changed = None # Radio state readback variables (these need thread safety) iface.r_frequency = None iface.r_bandwidth = None iface.r_txpower = None iface.r_sf = None iface.r_cr = None iface.r_state = None iface.r_stat_rssi = None iface.r_stat_snr = None return iface def test_radio_state_lock_exists(self): """Interface should have a _read_lock for thread-safe radio state access.""" import threading iface = self.create_interface() assert hasattr(iface, '_read_lock') assert isinstance(iface._read_lock, type(threading.Lock())) def test_radio_state_can_be_updated_safely(self): """Radio state variables should be accessible under lock.""" iface = self.create_interface() # Simulate what the read loop does - update state under lock with iface._read_lock: iface.r_frequency = 915000000 iface.r_bandwidth = 250000 iface.r_sf = 11 iface.r_cr = 5 iface.r_state = 0x01 # RADIO_STATE_ON # Read values (should be consistent) with iface._read_lock: assert iface.r_frequency == 915000000 assert iface.r_bandwidth == 250000 assert iface.r_sf == 11 assert iface.r_cr == 5 assert iface.r_state == 0x01 def test_concurrent_radio_state_access(self): """Concurrent reads and writes to radio state should not cause errors.""" import threading import time iface = self.create_interface() errors = [] iterations = 100 def writer(): """Simulate read loop updating state.""" for i in range(iterations): with iface._read_lock: iface.r_frequency = 915000000 + i iface.r_bandwidth = 250000 iface.r_sf = 11 iface.r_cr = 5 time.sleep(0.001) def reader(): """Simulate _validate_radio_state reading state.""" for _ in range(iterations): try: with iface._read_lock: # Read all values atomically freq = iface.r_frequency bw = iface.r_bandwidth sf = iface.r_sf cr = iface.r_cr # Values should be consistent (all set or all None) if freq is not None: assert bw is not None assert sf is not None assert cr is not None except Exception as e: errors.append(e) time.sleep(0.001) # Run concurrent threads writer_thread = threading.Thread(target=writer) reader_thread = threading.Thread(target=reader) writer_thread.start() reader_thread.start() writer_thread.join(timeout=5.0) reader_thread.join(timeout=5.0) assert len(errors) == 0, f"Concurrent access errors: {errors}" class TestOnlineStatusThreadSafety: """Tests for thread-safe online status access (PR #36 critical fix).""" def create_interface(self): """Create a test interface with mocked dependencies.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.frequency = 915000000 iface.bandwidth = 125000 iface.txpower = 17 iface.sf = 8 iface.cr = 5 iface.st_alock = None iface.lt_alock = None iface.name = "test-rnode" iface.target_device_name = "TestRNode" iface.connection_mode = 0 iface.kotlin_bridge = MagicMock() iface.kotlin_bridge.writeSync.return_value = 10 iface.enable_framebuffer = False iface.framebuffer_enabled = False iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface._reconnect_thread = None iface._reconnecting = False iface._max_reconnect_attempts = 30 iface._reconnect_interval = 10.0 iface._on_error_callback = None iface._on_online_status_changed = None iface.online = False iface.txb = 0 iface.r_stat_rssi = None iface.r_stat_snr = None return iface def test_set_online_uses_lock(self): """_set_online should use _read_lock for thread-safe updates.""" iface = self.create_interface() # Verify initial state assert iface.online is False # Set online iface._set_online(True) assert iface.online is True # Set offline iface._set_online(False) assert iface.online is False def test_process_outgoing_uses_lock(self): """process_outgoing should use _read_lock when checking online status.""" iface = self.create_interface() iface.online = True # Should not raise - online status should be readable iface.process_outgoing(b"test data") # Verify write was attempted iface.kotlin_bridge.writeSync.assert_called() def test_concurrent_online_status_access(self): """Concurrent reads/writes to online status should be thread-safe.""" import threading import time iface = self.create_interface() errors = [] iterations = 200 def status_changer(): """Simulate connection state changes.""" for i in range(iterations): iface._set_online(i % 2 == 0) time.sleep(0.001) def sender(): """Simulate sending data.""" for _ in range(iterations): try: iface.process_outgoing(b"test") except IOError: pass # Expected when offline except Exception as e: errors.append(e) time.sleep(0.001) changer_thread = threading.Thread(target=status_changer) sender_thread = threading.Thread(target=sender) changer_thread.start() sender_thread.start() changer_thread.join(timeout=5.0) sender_thread.join(timeout=5.0) assert len(errors) == 0, f"Thread safety errors: {errors}" def test_online_status_callback_called_on_change(self): """_on_online_status_changed callback should be called when status changes.""" iface = self.create_interface() callback_calls = [] def callback(is_online): callback_calls.append(is_online) iface._on_online_status_changed = callback # Change from offline to online iface._set_online(True) assert callback_calls == [True] # Change from online to offline iface._set_online(False) assert callback_calls == [True, False] # Same status should not trigger callback iface._set_online(False) assert callback_calls == [True, False] # No new call class TestRssiSnrThreadSafety: """Tests for thread-safe RSSI/SNR access (PR #36 critical fix).""" def create_interface(self): """Create a test interface with mocked dependencies.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface._read_lock = threading.Lock() iface.r_stat_rssi = None iface.r_stat_snr = None return iface def test_rssi_snr_can_be_updated_under_lock(self): """RSSI and SNR should be updatable under lock.""" iface = self.create_interface() with iface._read_lock: iface.r_stat_rssi = -80 iface.r_stat_snr = 10.5 assert iface.r_stat_rssi == -80 assert iface.r_stat_snr == 10.5 class TestMalformedEscapeSequenceLogging: """Tests for malformed KISS escape sequence warning (PR #36 critical fix).""" def test_escape_sequence_logging_documented(self): """Verify the fix documents invalid escape sequence handling. The read loop at line 666-669 now logs a warning when FESC is followed by an unexpected byte (not TFEND or TFESC). This test documents the expected behavior. """ # This is a documentation test - the actual logging happens in _read_loop # which requires a full mock of the read loop. We verify the KISS constants # that define valid escape sequences. assert KISS.FESC == 0xDB assert KISS.TFEND == 0xDC assert KISS.TFESC == 0xDD # Any byte after FESC that is NOT TFEND (0xDC) or TFESC (0xDD) # is considered invalid and will now be logged class TestKISSConstants: """Tests for KISS protocol constants.""" def test_fend_value(self): """FEND should be 0xC0.""" assert KISS.FEND == 0xC0 def test_fesc_value(self): """FESC should be 0xDB.""" assert KISS.FESC == 0xDB def test_tfend_value(self): """TFEND (escaped FEND) should be 0xDC.""" assert KISS.TFEND == 0xDC def test_tfesc_value(self): """TFESC (escaped FESC) should be 0xDD.""" assert KISS.TFESC == 0xDD def test_cmd_data_is_zero(self): """CMD_DATA should be 0x00.""" assert KISS.CMD_DATA == 0x00 def test_radio_states(self): """Radio states should have expected values.""" assert KISS.RADIO_STATE_OFF == 0x00 assert KISS.RADIO_STATE_ON == 0x01 assert KISS.RADIO_STATE_ASK == 0xFF def test_bt_ctrl_command(self): """CMD_BT_CTRL should be 0x46.""" assert KISS.CMD_BT_CTRL == 0x46 def test_bt_pin_command(self): """CMD_BT_PIN should be 0x62 for Bluetooth PIN response.""" assert KISS.CMD_BT_PIN == 0x62 def test_bt_ctrl_pairing_mode(self): """BT_CTRL_PAIRING_MODE should be 0x02.""" assert KISS.BT_CTRL_PAIRING_MODE == 0x02 class TestUSBModeConstants: """Tests for USB mode constants.""" def test_mode_usb_exists(self): """MODE_USB should be defined.""" assert hasattr(ColumbaRNodeInterface, 'MODE_USB') assert ColumbaRNodeInterface.MODE_USB == "usb" def test_mode_classic_exists(self): """MODE_CLASSIC should be defined for Bluetooth Classic.""" assert hasattr(ColumbaRNodeInterface, 'MODE_CLASSIC') assert ColumbaRNodeInterface.MODE_CLASSIC == "classic" def test_mode_ble_exists(self): """MODE_BLE should be defined for Bluetooth LE.""" assert hasattr(ColumbaRNodeInterface, 'MODE_BLE') assert ColumbaRNodeInterface.MODE_BLE == "ble" class TestUSBModeInterface: """Tests for USB mode interface functionality.""" def create_usb_interface(self): """Create a test interface configured for USB mode.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.frequency = 915000000 iface.bandwidth = 125000 iface.txpower = 17 iface.sf = 8 iface.cr = 5 iface.st_alock = None iface.lt_alock = None iface.name = "test-rnode-usb" iface.target_device_name = "" # Empty for USB mode iface.usb_device_id = 123 iface.connection_mode = ColumbaRNodeInterface.MODE_USB iface.kotlin_bridge = None iface.usb_bridge = None iface.enable_framebuffer = False iface.framebuffer_enabled = False iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface._reconnect_thread = None iface._reconnecting = False iface._max_reconnect_attempts = 30 iface._reconnect_interval = 10.0 iface._on_error_callback = None iface._on_online_status_changed = None iface.online = False return iface def test_usb_mode_stored_correctly(self): """USB mode should be stored in connection_mode.""" iface = self.create_usb_interface() assert iface.connection_mode == ColumbaRNodeInterface.MODE_USB def test_usb_device_id_stored(self): """USB device ID should be stored.""" iface = self.create_usb_interface() assert iface.usb_device_id == 123 def test_enter_bluetooth_pairing_mode_requires_usb_mode(self): """enter_bluetooth_pairing_mode should return False if not in USB mode.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.connection_mode = ColumbaRNodeInterface.MODE_CLASSIC iface.usb_bridge = None iface._read_lock = threading.Lock() iface.online = True result = iface.enter_bluetooth_pairing_mode() assert result is False def test_enter_bluetooth_pairing_mode_requires_usb_bridge(self): """enter_bluetooth_pairing_mode should return False if USB bridge is None.""" iface = self.create_usb_interface() iface.usb_bridge = None result = iface.enter_bluetooth_pairing_mode() assert result is False def test_enter_bluetooth_pairing_mode_requires_connection(self): """enter_bluetooth_pairing_mode should return False if not connected.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.isConnected.return_value = False iface.usb_bridge = mock_bridge result = iface.enter_bluetooth_pairing_mode() assert result is False def test_enter_bluetooth_pairing_mode_sends_kiss_command(self): """enter_bluetooth_pairing_mode should send correct KISS command.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.isConnected.return_value = True mock_bridge.write.return_value = 4 iface.usb_bridge = mock_bridge result = iface.enter_bluetooth_pairing_mode() assert result is True # Verify the KISS command sent: FEND CMD_BT_CTRL BT_CTRL_PAIRING_MODE FEND expected_cmd = bytes([KISS.FEND, KISS.CMD_BT_CTRL, KISS.BT_CTRL_PAIRING_MODE, KISS.FEND]) mock_bridge.write.assert_called_once() actual_cmd = mock_bridge.write.call_args[0][0] assert actual_cmd == expected_cmd class TestUSBBridgeIntegration: """Tests for USB bridge integration.""" def create_usb_interface(self): """Create a test interface configured for USB mode.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.frequency = 915000000 iface.bandwidth = 125000 iface.txpower = 17 iface.sf = 8 iface.cr = 5 iface.st_alock = None iface.lt_alock = None iface.name = "test-rnode-usb" iface.target_device_name = "" iface.usb_device_id = 123 iface.usb_vendor_id = 0x0403 iface.usb_product_id = 0x6001 iface.connection_mode = ColumbaRNodeInterface.MODE_USB iface.kotlin_bridge = None iface.usb_bridge = None iface.enable_framebuffer = False iface.framebuffer_enabled = False iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface._reconnect_thread = None iface._reconnecting = False iface._max_reconnect_attempts = 30 iface._reconnect_interval = 10.0 iface._on_error_callback = None iface._on_online_status_changed = None iface.online = False iface.detected = False iface.firmware_ok = False iface.interface_ready = False iface.txb = 0 iface.r_stat_rssi = None iface.r_stat_snr = None return iface def test_get_usb_bridge_success(self): """_get_usb_bridge should get bridge from usb_bridge module.""" iface = self.create_usb_interface() mock_bridge = MagicMock() with patch.dict('sys.modules', {'usb_bridge': MagicMock()}): import sys sys.modules['usb_bridge'].get_usb_bridge.return_value = mock_bridge iface._get_usb_bridge() assert iface.usb_bridge == mock_bridge def test_get_usb_bridge_returns_none(self): """_get_usb_bridge should handle None return gracefully.""" iface = self.create_usb_interface() with patch.dict('sys.modules', {'usb_bridge': MagicMock()}): import sys sys.modules['usb_bridge'].get_usb_bridge.return_value = None iface._get_usb_bridge() assert iface.usb_bridge is None def test_get_usb_bridge_handles_import_error(self): """_get_usb_bridge should handle import errors gracefully.""" iface = self.create_usb_interface() with patch.dict('sys.modules', {'usb_bridge': None}): # This should not raise an exception iface._get_usb_bridge() assert iface.usb_bridge is None def test_start_usb_no_bridge_returns_false(self): """_start_usb should return False if bridge is not available.""" iface = self.create_usb_interface() with patch.object(iface, '_get_usb_bridge'): iface.usb_bridge = None result = iface._start_usb() assert result is False def test_start_usb_finds_device_by_vid_pid(self): """_start_usb should find device by VID/PID and update device ID.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.findDeviceByVidPid.return_value = 456 # Different device ID mock_bridge.isConnected.return_value = False mock_bridge.connect.return_value = True with patch.object(iface, '_get_usb_bridge'): with patch.object(iface, '_configure_device'): iface.usb_bridge = mock_bridge result = iface._start_usb() assert result is True assert iface.usb_device_id == 456 mock_bridge.findDeviceByVidPid.assert_called_once_with(0x0403, 0x6001) def test_start_usb_device_not_found_by_vid_pid(self): """_start_usb should return False if device not found by VID/PID.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.findDeviceByVidPid.return_value = -1 # Not found with patch.object(iface, '_get_usb_bridge'): iface.usb_bridge = mock_bridge result = iface._start_usb() assert result is False def test_start_usb_no_device_id_or_vid_pid(self): """_start_usb should return False if no device ID and no VID/PID.""" iface = self.create_usb_interface() iface.usb_device_id = None iface.usb_vendor_id = None iface.usb_product_id = None mock_bridge = MagicMock() with patch.object(iface, '_get_usb_bridge'): iface.usb_bridge = mock_bridge result = iface._start_usb() assert result is False def test_start_usb_clears_stale_connection(self): """_start_usb should disconnect if bridge thinks it's connected but interface is offline.""" iface = self.create_usb_interface() iface.online = False mock_bridge = MagicMock() mock_bridge.findDeviceByVidPid.return_value = 123 mock_bridge.isConnected.return_value = True # Bridge thinks it's connected mock_bridge.connect.return_value = True with patch.object(iface, '_get_usb_bridge'): with patch.object(iface, '_configure_device'): iface.usb_bridge = mock_bridge iface._start_usb() mock_bridge.disconnect.assert_called_once() def test_start_usb_connect_failure(self): """_start_usb should return False if connect fails.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.findDeviceByVidPid.return_value = 123 mock_bridge.isConnected.return_value = False mock_bridge.connect.return_value = False # Connect fails with patch.object(iface, '_get_usb_bridge'): iface.usb_bridge = mock_bridge result = iface._start_usb() assert result is False def test_start_usb_sets_up_callbacks(self): """_start_usb should set up data and connection state callbacks.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.findDeviceByVidPid.return_value = 123 mock_bridge.isConnected.return_value = False mock_bridge.connect.return_value = True with patch.object(iface, '_get_usb_bridge'): with patch.object(iface, '_configure_device'): iface.usb_bridge = mock_bridge iface._start_usb() mock_bridge.setOnDataReceived.assert_called_once() mock_bridge.setOnConnectionStateChanged.assert_called_once() def test_start_usb_configure_device_exception(self): """_start_usb should return False and stop if configure_device raises.""" iface = self.create_usb_interface() mock_bridge = MagicMock() mock_bridge.findDeviceByVidPid.return_value = 123 mock_bridge.isConnected.return_value = False mock_bridge.connect.return_value = True with patch.object(iface, '_get_usb_bridge'): with patch.object(iface, '_configure_device', side_effect=Exception("Config failed")): with patch.object(iface, 'stop') as mock_stop: iface.usb_bridge = mock_bridge result = iface._start_usb() assert result is False mock_stop.assert_called_once() class TestUSBConnectionStateCallback: """Tests for USB connection state change callback.""" def create_usb_interface(self): """Create a test interface configured for USB mode.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.name = "test-rnode-usb" iface.usb_device_id = 123 iface.connection_mode = ColumbaRNodeInterface.MODE_USB iface._read_lock = threading.Lock() iface._running = threading.Event() iface.online = True iface.detected = True iface._on_online_status_changed = None return iface def test_on_usb_disconnected_sets_offline(self): """Disconnection callback should set interface offline.""" iface = self.create_usb_interface() iface.online = True iface.detected = True iface._on_usb_connection_state_changed(connected=False, device_id=123) assert iface.online is False assert iface.detected is False def test_on_usb_connected_does_not_change_online(self): """Connection callback should not automatically set online (config needed first).""" iface = self.create_usb_interface() iface.online = False iface._on_usb_connection_state_changed(connected=True, device_id=123) # online should still be False - configuration step sets it to True assert iface.online is False def test_on_usb_disconnected_calls_status_callback(self): """Disconnection should call the online status changed callback.""" iface = self.create_usb_interface() iface.online = True mock_callback = MagicMock() iface._on_online_status_changed = mock_callback iface._on_usb_connection_state_changed(connected=False, device_id=123) mock_callback.assert_called_once_with(False) class TestUSBVidPidConfig: """Tests for USB VID/PID configuration attributes.""" def test_usb_vendor_id_attribute_exists(self): """USB interface should have usb_vendor_id attribute.""" with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): config = { 'connection_mode': 'usb', 'usb_device_id': 123, 'usb_vendor_id': 0x0403, 'usb_product_id': 0x6001, 'frequency': 915000000, 'bandwidth': 125000, 'txpower': 17, 'sf': 8, 'cr': 5, } iface = ColumbaRNodeInterface(None, "test", config) assert iface.usb_vendor_id == 0x0403 assert iface.usb_product_id == 0x6001 def test_usb_config_defaults_to_none(self): """USB VID/PID should default to None if not provided.""" with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): config = { 'connection_mode': 'usb', 'usb_device_id': 123, 'frequency': 915000000, 'bandwidth': 125000, 'txpower': 17, 'sf': 8, 'cr': 5, } iface = ColumbaRNodeInterface(None, "test", config) assert iface.usb_vendor_id is None assert iface.usb_product_id is None class TestStartRouting: """Tests for start() method routing to correct implementation.""" def create_interface(self, mode): """Create interface with specified connection mode.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.frequency = 915000000 iface.bandwidth = 125000 iface.txpower = 17 iface.sf = 8 iface.cr = 5 iface.st_alock = None iface.lt_alock = None iface.name = "test-rnode" iface.target_device_name = "TestRNode" iface.usb_device_id = 123 iface.usb_vendor_id = 0x0403 iface.usb_product_id = 0x6001 iface.connection_mode = mode iface.kotlin_bridge = MagicMock() iface.usb_bridge = MagicMock() iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface.online = False iface.detected = False iface.firmware_ok = False iface.interface_ready = False return iface def test_start_routes_to_usb_for_usb_mode(self): """start() should call _start_usb() for USB mode.""" iface = self.create_interface(ColumbaRNodeInterface.MODE_USB) with patch.object(iface, '_start_usb', return_value=True) as mock_start_usb: result = iface.start() mock_start_usb.assert_called_once() assert result is True def test_start_uses_bluetooth_for_classic_mode(self): """start() should use Bluetooth path for classic mode.""" iface = self.create_interface(ColumbaRNodeInterface.MODE_CLASSIC) iface.kotlin_bridge.connectSync.return_value = True with patch.object(iface, '_start_usb') as mock_start_usb: with patch.object(iface, '_configure_device'): iface.start() mock_start_usb.assert_not_called() class TestKISSBluetoothCommands: """Tests for Bluetooth-related KISS commands.""" def test_pairing_mode_kiss_frame(self): """Verify the correct KISS frame for entering Bluetooth pairing mode.""" # KISS frame: FEND CMD_BT_CTRL BT_CTRL_PAIRING_MODE FEND kiss_frame = bytes([KISS.FEND, KISS.CMD_BT_CTRL, KISS.BT_CTRL_PAIRING_MODE, KISS.FEND]) assert len(kiss_frame) == 4 assert kiss_frame[0] == 0xC0 # FEND assert kiss_frame[1] == 0x46 # CMD_BT_CTRL assert kiss_frame[2] == 0x02 # BT_CTRL_PAIRING_MODE assert kiss_frame[3] == 0xC0 # FEND def test_bt_pin_command_value(self): """CMD_BT_PIN should be the expected value for PIN response.""" # When RNode responds with a Bluetooth PIN, it uses CMD_BT_PIN (0x62) assert KISS.CMD_BT_PIN == 0x62 def test_bt_pin_response_frame_structure(self): """Test the expected structure of a Bluetooth PIN response frame.""" # A PIN response is: FEND CMD_BT_PIN <4-byte big-endian integer> FEND # Example: PIN 123456 -> bytes [0x00, 0x01, 0xE2, 0x40] pin_value = 123456 pin_bytes = pin_value.to_bytes(4, byteorder='big') kiss_frame = bytes([KISS.FEND, KISS.CMD_BT_PIN]) + pin_bytes + bytes([KISS.FEND]) assert len(kiss_frame) == 7 # 1 + 1 + 4 + 1 assert kiss_frame[0] == KISS.FEND assert kiss_frame[1] == KISS.CMD_BT_PIN assert kiss_frame[2:6] == pin_bytes assert kiss_frame[6] == KISS.FEND # Verify the PIN value can be decoded back decoded_pin = int.from_bytes(kiss_frame[2:6], byteorder='big') assert decoded_pin == 123456 class TestInterfaceMode: """Tests for interface mode configuration (lines 224-237 in rnode_interface.py).""" def create_interface_with_mode(self, mode_str): """Create a test interface with the specified mode.""" with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): config = { 'target_device_name': 'RNode Test', 'connection_mode': 'ble', 'frequency': 915000000, 'bandwidth': 250000, 'txpower': 17, 'sf': 11, 'cr': 5, 'mode': mode_str, } iface = ColumbaRNodeInterface(None, "test", config) return iface def test_interface_mode_full(self): """Mode 'full' should set MODE_FULL.""" iface = self.create_interface_with_mode('full') # MODE_FULL is the default mode (value 1 in RNS.Interfaces.Interface.Interface) # We can't directly test the RNS constant since it's mocked, but we can verify # the mode is set and consistent with the config assert iface.mode is not None def test_interface_mode_gateway(self): """Mode 'gateway' should set MODE_GATEWAY.""" iface = self.create_interface_with_mode('gateway') assert iface.mode is not None def test_interface_mode_access_point(self): """Mode 'access_point' should set MODE_ACCESS_POINT.""" iface = self.create_interface_with_mode('access_point') assert iface.mode is not None def test_interface_mode_roaming(self): """Mode 'roaming' should set MODE_ROAMING.""" iface = self.create_interface_with_mode('roaming') assert iface.mode is not None def test_interface_mode_boundary(self): """Mode 'boundary' should set MODE_BOUNDARY.""" iface = self.create_interface_with_mode('boundary') assert iface.mode is not None def test_interface_mode_invalid_defaults_to_full(self): """Invalid mode string should default to MODE_FULL.""" iface = self.create_interface_with_mode('invalid_mode') assert iface.mode is not None # The mode should be set to the same value as 'full' mode iface_full = self.create_interface_with_mode('full') assert iface.mode == iface_full.mode def test_interface_mode_defaults_to_full_when_not_specified(self): """Mode should default to 'full' when not specified in config.""" with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): config = { 'target_device_name': 'RNode Test', 'connection_mode': 'ble', 'frequency': 915000000, 'bandwidth': 250000, 'txpower': 17, 'sf': 11, 'cr': 5, # 'mode' not specified } iface = ColumbaRNodeInterface(None, "test", config) assert iface.mode is not None # Should match the 'full' mode iface_full = self.create_interface_with_mode('full') assert iface.mode == iface_full.mode class TestKISSCommandParsing: """Tests for KISS command parsing logic (lines 840-895 in rnode_interface.py).""" def create_minimal_interface(self): """Create a minimal interface for command parsing tests.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.name = "test-rnode" iface._read_lock = threading.Lock() iface._running = threading.Event() iface.r_frequency = None iface.r_bandwidth = None iface.r_txpower = None iface.r_sf = None iface.r_cr = None iface.r_state = None iface.r_stat_rssi = None iface.r_stat_snr = None iface.maj_version = 0 iface.min_version = 0 iface.platform = None iface.mcu = None iface.detected = False iface._on_error_callback = None return iface def test_cmd_frequency_updates_r_frequency(self): """CMD_FREQUENCY command should update r_frequency.""" iface = self.create_minimal_interface() # Simulate parsing frequency command: 4 bytes in big-endian # Example: 915 MHz = 915000000 Hz = 0x369D6440 frequency_bytes = (915000000).to_bytes(4, byteorder='big') # Simulate the read loop receiving frequency command # In actual code: elif command == KISS.CMD_FREQUENCY for i, byte in enumerate(frequency_bytes): if i == 3: # Last byte iface.r_frequency = int.from_bytes(frequency_bytes, byteorder='big') assert iface.r_frequency == 915000000 def test_cmd_stat_rssi_calculation(self): """CMD_STAT_RSSI should calculate RSSI as byte - 157.""" iface = self.create_minimal_interface() # RSSI byte value 80 should become -77 dBm (80 - 157 = -77) rssi_byte = 80 iface.r_stat_rssi = rssi_byte - 157 assert iface.r_stat_rssi == -77 def test_cmd_stat_rssi_range(self): """Test RSSI calculation for typical range of values.""" iface = self.create_minimal_interface() # Test several typical RSSI values # RSSI byte 157 = 0 dBm (157 - 157 = 0) iface.r_stat_rssi = 157 - 157 assert iface.r_stat_rssi == 0 # RSSI byte 100 = -57 dBm (100 - 157 = -57) iface.r_stat_rssi = 100 - 157 assert iface.r_stat_rssi == -57 # RSSI byte 50 = -107 dBm (50 - 157 = -107) iface.r_stat_rssi = 50 - 157 assert iface.r_stat_rssi == -107 def test_cmd_stat_snr_calculation(self): """CMD_STAT_SNR should calculate SNR as signed byte / 4.0.""" iface = self.create_minimal_interface() # SNR byte value 40 (0x28) should become 10.0 dB (40 / 4.0 = 10.0) snr_byte = 40 iface.r_stat_snr = int.from_bytes([snr_byte], 'big', signed=True) / 4.0 assert iface.r_stat_snr == 10.0 def test_cmd_stat_snr_negative_value(self): """CMD_STAT_SNR should handle negative SNR values correctly.""" iface = self.create_minimal_interface() # SNR byte -40 (0xD8 = 216 unsigned, -40 signed) should become -10.0 dB snr_byte = 216 # -40 as unsigned byte iface.r_stat_snr = int.from_bytes([snr_byte], 'big', signed=True) / 4.0 assert iface.r_stat_snr == -10.0 def test_cmd_txpower_updates_r_txpower(self): """CMD_TXPOWER command should update r_txpower.""" iface = self.create_minimal_interface() # TX power byte value 17 (typical for 915 MHz) txpower_byte = 17 iface.r_txpower = txpower_byte assert iface.r_txpower == 17 def test_cmd_sf_updates_r_sf(self): """CMD_SF command should update r_sf (spreading factor).""" iface = self.create_minimal_interface() # Spreading factor 7 (common LoRa value) sf_byte = 7 iface.r_sf = sf_byte assert iface.r_sf == 7 def test_cmd_cr_updates_r_cr(self): """CMD_CR command should update r_cr (coding rate).""" iface = self.create_minimal_interface() # Coding rate 5 (4/5 coding rate) cr_byte = 5 iface.r_cr = cr_byte assert iface.r_cr == 5 def test_cmd_radio_state_updates_r_state(self): """CMD_RADIO_STATE command should update r_state.""" iface = self.create_minimal_interface() # Radio state ON state_byte = KISS.RADIO_STATE_ON iface.r_state = state_byte assert iface.r_state == KISS.RADIO_STATE_ON def test_cmd_detect_response(self): """CMD_DETECT with DETECT_RESP should set detected flag.""" iface = self.create_minimal_interface() # Simulate detect response detect_byte = KISS.DETECT_RESP if detect_byte == KISS.DETECT_RESP: iface.detected = True assert iface.detected is True def test_cmd_error_callback_invoked(self): """CMD_ERROR should invoke error callback if set.""" iface = self.create_minimal_interface() # Set up error callback error_received = [] def error_callback(error_code, error_message): error_received.append((error_code, error_message)) iface._on_error_callback = error_callback # Simulate error code 0x40 (TX power exceeds limits) error_code = 0x40 error_message = KISS.get_error_message(error_code) if iface._on_error_callback: iface._on_error_callback(error_code, error_message) assert len(error_received) == 1 assert error_received[0][0] == 0x40 assert "TX" in error_received[0][1].upper() class TestStartErrorHandling: """Tests for start() method error conditions (lines 351-390 in rnode_interface.py).""" def create_interface_for_start_test(self, **overrides): """Create a test interface for start() method testing.""" import threading with patch.object(ColumbaRNodeInterface, '_get_kotlin_bridge'): with patch.object(ColumbaRNodeInterface, '_validate_config'): config = { 'target_device_name': 'RNode Test', 'connection_mode': 'ble', 'frequency': 915000000, 'bandwidth': 250000, 'txpower': 17, 'sf': 11, 'cr': 5, } config.update(overrides) iface = ColumbaRNodeInterface.__new__(ColumbaRNodeInterface) iface.name = "test-rnode" iface.kotlin_bridge = MagicMock() iface.target_device_name = config.get('target_device_name') iface.connection_mode = config.get('connection_mode', 'ble') iface.frequency = config['frequency'] iface.bandwidth = config['bandwidth'] iface.txpower = config['txpower'] iface.sf = config['sf'] iface.cr = config['cr'] iface.st_alock = None iface.lt_alock = None iface._read_thread = None iface._running = threading.Event() iface._read_lock = threading.Lock() iface.MODE_USB = "usb" iface.MODE_BLE = "ble" iface.MODE_CLASSIC = "classic" return iface def test_start_fails_without_kotlin_bridge(self): """start() should return False when kotlin_bridge is None.""" iface = self.create_interface_for_start_test() iface.kotlin_bridge = None result = iface.start() assert result is False def test_start_fails_without_target_device_name(self): """start() should return False when target_device_name is missing or empty.""" # Test with None iface = self.create_interface_for_start_test(target_device_name=None) result = iface.start() assert result is False # Test with empty string iface = self.create_interface_for_start_test(target_device_name="") result = iface.start() assert result is False def test_start_handles_connection_failure(self): """start() should return False and log error when connection fails.""" iface = self.create_interface_for_start_test() iface.kotlin_bridge.connect.return_value = False result = iface.start() assert result is False iface.kotlin_bridge.connect.assert_called_once_with('RNode Test', 'ble') def test_start_handles_configuration_failure(self): """start() should return False and call stop() when device configuration fails.""" iface = self.create_interface_for_start_test() iface.kotlin_bridge.connect.return_value = True with patch.object(iface, '_configure_device', side_effect=Exception("Config failed")): with patch.object(iface, 'stop') as mock_stop: with patch('time.sleep'): # Skip the sleep result = iface.start() assert result is False mock_stop.assert_called_once() def test_start_usb_mode_skips_bluetooth_path(self): """start() should route to _start_usb() for USB mode.""" iface = self.create_interface_for_start_test(connection_mode='usb') with patch.object(iface, '_start_usb', return_value=True) as mock_start_usb: result = iface.start() mock_start_usb.assert_called_once() # Bluetooth connection methods should not be called iface.kotlin_bridge.connect.assert_not_called() assert result is True