""" 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] (123456 in big-endian) pin_int = 123456 pin_bytes = bytes([ (pin_int >> 24) & 0xFF, (pin_int >> 16) & 0xFF, (pin_int >> 8) & 0xFF, pin_int & 0xFF, ]) 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] == bytes([0x00, 0x01, 0xE2, 0x40]) assert kiss_frame[6] == KISS.FEND