diff --git a/README.md b/README.md index b47c627..2699d6c 100644 --- a/README.md +++ b/README.md @@ -61,19 +61,19 @@ If `priority_nodes` is empty in `config.yaml`, the monitor will automatically se ### Basic Run (USB/Serial) Connect your Meshtastic device via USB and run: ```bash -python3 -m mesh_monitor.monitor +python3 main.py ``` ### Network Connection (TCP) If your node is on the network (e.g., WiFi): ```bash -python3 -m mesh_monitor.monitor --tcp 192.168.1.10 +python3 main.py --tcp 192.168.1.10 ``` ### Options * `--ignore-no-position`: Suppress warnings about routers without a position (useful for portable routers or privacy). ```bash - python3 -m mesh_monitor.monitor --ignore-no-position + python3 main.py --ignore-no-position ``` ## Configuration (Priority Testing) diff --git a/main.py b/main.py new file mode 100755 index 0000000..40f65b5 --- /dev/null +++ b/main.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +""" +Entry point for the Meshtastic Network Monitor. +""" +from mesh_monitor.monitor import main + +if __name__ == "__main__": + main() diff --git a/mesh_monitor/active_tests.py b/mesh_monitor/active_tests.py index 208d450..0a8f192 100644 --- a/mesh_monitor/active_tests.py +++ b/mesh_monitor/active_tests.py @@ -334,8 +334,10 @@ class ActiveTester: 'timestamp': time.time() }) self._check_cycle_completion(node_id) + self._check_cycle_completion(node_id) if self.pending_traceroute == node_id: self.pending_traceroute = None # Clear pending if this was the node we were waiting for + self.last_test_time = time.time() # Start cooldown def record_timeout(self, node_id): """ @@ -350,6 +352,7 @@ class ActiveTester: self._check_cycle_completion(node_id) if self.pending_traceroute == node_id: self.pending_traceroute = None # Clear pending if this was the node we were waiting for + self.last_test_time = time.time() # Start cooldown def _check_cycle_completion(self, node_id): """ diff --git a/mesh_monitor/monitor.py b/mesh_monitor/monitor.py index b4c4745..baa3fc0 100644 --- a/mesh_monitor/monitor.py +++ b/mesh_monitor/monitor.py @@ -354,7 +354,7 @@ class MeshMonitor: logger.error(f"Error in main loop: {e}") time.sleep(10) -if __name__ == "__main__": +def main(): # Simple CLI for testing import argparse parser = argparse.ArgumentParser(description='Meshtastic Network Monitor') @@ -367,4 +367,10 @@ if __name__ == "__main__": else: monitor = MeshMonitor(interface_type='serial', ignore_no_position=args.ignore_no_position) - monitor.start() + try: + monitor.start() + except KeyboardInterrupt: + monitor.stop() + +if __name__ == "__main__": + main() diff --git a/tests/mock_test.py b/tests/mock_test.py index 4a64024..a2ed910 100644 --- a/tests/mock_test.py +++ b/tests/mock_test.py @@ -168,6 +168,55 @@ class TestNetworkMonitor(unittest.TestCase): self.assertEqual(selected, expected) print("Stratified Discovery Test Passed!") + + def test_cooldown_logic(self): + print("\nRunning Cooldown Logic Test...") + mock_interface = MagicMock() + + # Setup + tester = ActiveTester(mock_interface, test_interval=30, traceroute_timeout=60) + tester.priority_nodes = ["!n1", "!n2"] + + # 1. Start Test 1 + tester.run_next_test() + self.assertEqual(tester.pending_traceroute, "!n1") + start_time = tester.last_test_time + + # 2. Simulate Timeout (at T+61) + # We need to mock time.time() to control flow + with patch('time.time') as mock_time: + # Initial call was at T0. + # Advance to T+61 + mock_time.return_value = start_time + 61 + + # Run next test -> Should trigger timeout recording + tester.run_next_test() + + # Verify timeout recorded + self.assertIsNone(tester.pending_traceroute) + self.assertEqual(tester.test_results[-1]['status'], 'timeout') + + # Verify last_test_time updated to T+61 (Cooldown start) + self.assertEqual(tester.last_test_time, start_time + 61) + + # 3. Try to run next test immediately (at T+62) + mock_time.return_value = start_time + 62 + mock_interface.reset_mock() + tester.run_next_test() + + # Should NOT send because 62 - 61 = 1 < 30 + mock_interface.sendTraceRoute.assert_not_called() + print(" [Pass] Cooldown enforced after timeout") + + # 4. Advance past cooldown (at T+92) + mock_time.return_value = start_time + 92 + tester.run_next_test() + + # Should send now + mock_interface.sendTraceRoute.assert_called_with("!n2", hopLimit=7) + print(" [Pass] Next test sent after cooldown") + + print("Cooldown Logic Test Passed!") print("\nRunning Test Interval Config Test...") mock_interface = MagicMock()