From c21743137c830954c0568a00a8f932e2f2025a76 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 6 Sep 2025 11:50:05 -0700 Subject: [PATCH] Move tests directory to local docs folder - Move entire tests/ directory to docs/local/tests/ - Remove all test files from git tracking (20 files removed) - Add tests/ to .gitignore to prevent future commits of test files - Tests are development/experimental code not needed for public repository - Repository is now cleaner and more focused on production code --- .gitignore | 3 +- tests/README.md | 56 ----- tests/__init__.py | 1 - tests/discover_ble_uuids.py | 185 ---------------- tests/example_usage.py | 336 ------------------------------ tests/meshcore_packet_analyzer.py | 214 ------------------- tests/repeater_management_demo.py | 158 -------------- tests/test_ble_connection.py | 201 ------------------ tests/test_channels.py | 57 ----- tests/test_config.py | 164 --------------- tests/test_config_parsing.py | 33 --- tests/test_contacts.py | 102 --------- tests/test_dynamic_channels.py | 60 ------ tests/test_event_structure.py | 58 ------ tests/test_get_channel.py | 65 ------ tests/test_installation.py | 251 ---------------------- tests/test_meshcore_official.py | 48 ----- tests/test_path_extraction.py | 113 ---------- tests/test_path_info.py | 123 ----------- tests/test_race_condition_fix.py | 188 ----------------- 20 files changed, 2 insertions(+), 2414 deletions(-) delete mode 100644 tests/README.md delete mode 100644 tests/__init__.py delete mode 100755 tests/discover_ble_uuids.py delete mode 100755 tests/example_usage.py delete mode 100755 tests/meshcore_packet_analyzer.py delete mode 100644 tests/repeater_management_demo.py delete mode 100755 tests/test_ble_connection.py delete mode 100644 tests/test_channels.py delete mode 100755 tests/test_config.py delete mode 100644 tests/test_config_parsing.py delete mode 100644 tests/test_contacts.py delete mode 100644 tests/test_dynamic_channels.py delete mode 100644 tests/test_event_structure.py delete mode 100644 tests/test_get_channel.py delete mode 100755 tests/test_installation.py delete mode 100644 tests/test_meshcore_official.py delete mode 100644 tests/test_path_extraction.py delete mode 100644 tests/test_path_info.py delete mode 100644 tests/test_race_condition_fix.py diff --git a/.gitignore b/.gitignore index d5bbbe9..c2e363b 100644 --- a/.gitignore +++ b/.gitignore @@ -121,5 +121,6 @@ config.ini *.db mctomqtt.py -# Local documentation (internal development notes) +# Local documentation and development files docs/local/ +tests/ diff --git a/tests/README.md b/tests/README.md deleted file mode 100644 index 63a1eb6..0000000 --- a/tests/README.md +++ /dev/null @@ -1,56 +0,0 @@ -# MeshCore Bot Tests - -This directory contains all test files for the MeshCore Bot project. - -## Test Files - -### Standard Test Files -- `test_ble_connection.py` - Tests for BLE connection functionality -- `test_channels.py` - Tests for channel management -- `test_config_parsing.py` - Tests for configuration parsing -- `test_config.py` - General configuration tests -- `test_contacts.py` - Tests for contact management -- `test_dynamic_channels.py` - Tests for dynamic channel functionality -- `test_event_structure.py` - Tests for event structure handling -- `test_get_channel.py` - Tests for channel retrieval -- `test_installation.py` - Tests for installation/setup -- `test_meshcore_official.py` - Tests for official meshcore integration -- `test_path_extraction.py` - Tests for path extraction functionality -- `test_path_info.py` - Tests for path information handling - -### Utility and Analysis Tools -- `meshcore_packet_analyzer.py` - Packet analysis and testing tool -- `discover_ble_uuids.py` - BLE UUID discovery utility -- `example_usage.py` - Example usage and demo code - -## Running Tests - -To run tests from the project root: - -```bash -# Run a specific test -python3 -m pytest tests/test_config.py - -# Run all tests -python3 -m pytest tests/ - -# Run with verbose output -python3 -m pytest tests/ -v -``` - -## Running Utility Tools - -```bash -# Run packet analyzer -python3 tests/meshcore_packet_analyzer.py - -# Run BLE UUID discovery -python3 tests/discover_ble_uuids.py - -# Run example usage demo -python3 tests/example_usage.py demo -``` - -## Note - -These test files were moved from the project root to keep the main directory clean and organized. diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index ccef8ba..0000000 --- a/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# Tests package for MeshCore Bot diff --git a/tests/discover_ble_uuids.py b/tests/discover_ble_uuids.py deleted file mode 100755 index 8b68d9d..0000000 --- a/tests/discover_ble_uuids.py +++ /dev/null @@ -1,185 +0,0 @@ -#!/usr/bin/env python3 -""" -BLE UUID Discovery Script for MeshCore -This script connects to your MeshCore device and discovers all available services and characteristics. -""" - -import asyncio -import configparser -from pathlib import Path - - -async def discover_device_uuids(): - """Discover all UUIDs on the MeshCore device""" - print("MeshCore BLE UUID Discovery") - print("=" * 40) - - # Load config - config_file = "config.ini" - if not Path(config_file).exists(): - print(f"Config file {config_file} not found!") - return - - config = configparser.ConfigParser() - config.read(config_file) - - device_name = config.get('Connection', 'ble_device_name', fallback='') - if not device_name: - print("No BLE device name configured!") - return - - print(f"Target device: {device_name}") - print() - - try: - from bleak import BleakScanner, BleakClient - - # Clean device name - device_name = device_name.strip().strip('"').strip("'") - - # Scan for device - print("Scanning for device...") - devices = await BleakScanner.discover(timeout=10) - - target_device = None - for device in devices: - if device.name and device.name.strip() == device_name: - target_device = device - break - - if not target_device: - print(f"Device '{device_name}' not found!") - return - - print(f"Found device: {target_device.name}") - print(f"Address: {target_device.address}") - print() - - # Connect to device - print("Connecting to device...") - client = BleakClient(target_device.address) - await client.connect() - - if not client.is_connected: - print("Failed to connect!") - return - - print("✓ Connected successfully!") - print() - - # Discover all services - print("Discovering services and characteristics...") - print("-" * 50) - - services = client.services - service_count = len(list(services)) - print(f"Found {service_count} service(s):") - print() - - all_uuids = { - 'services': [], - 'characteristics': [] - } - - for i, service in enumerate(services, 1): - print(f"Service {i}:") - print(f" UUID: {service.uuid}") - print(f" Description: {service.description}") - - all_uuids['services'].append({ - 'uuid': service.uuid, - 'description': service.description - }) - - # Get characteristics for this service - char_count = len(service.characteristics) - print(f" Characteristics ({char_count}):") - - for j, char in enumerate(service.characteristics, 1): - print(f" {j}. UUID: {char.uuid}") - print(f" Properties: {char.properties}") - print(f" Description: {char.description}") - - all_uuids['characteristics'].append({ - 'uuid': char.uuid, - 'properties': char.properties, - 'description': char.description, - 'service_uuid': service.uuid - }) - - print() - - # Analyze the discovered UUIDs - print("UUID Analysis:") - print("-" * 50) - - # Check for Nordic UART Service - nus_service_found = False - nus_tx_found = False - nus_rx_found = False - - for service in all_uuids['services']: - if service['uuid'].lower() == '6e400001-b5a3-f393-e0a9-e50e24dcca9e': - nus_service_found = True - print("✓ Nordic UART Service found") - break - - for char in all_uuids['characteristics']: - if char['uuid'].lower() == '6e400002-b5a3-f393-e0a9-e50e24dcca9e': - nus_tx_found = True - print("✓ Nordic UART TX characteristic found") - elif char['uuid'].lower() == '6e400003-b5a3-f393-e0a9-e50e24dcca9e': - nus_rx_found = True - print("✓ Nordic UART RX characteristic found") - - if not nus_service_found: - print("✗ Nordic UART Service not found") - if not nus_tx_found: - print("✗ Nordic UART TX characteristic not found") - if not nus_rx_found: - print("✗ Nordic UART RX characteristic not found") - - print() - - # Look for alternative MeshCore UUIDs - print("Looking for MeshCore-specific UUIDs...") - meshcore_uuids = [] - - for char in all_uuids['characteristics']: - if 'mesh' in char['description'].lower() or 'core' in char['description'].lower(): - meshcore_uuids.append(char) - print(f"Potential MeshCore characteristic: {char['uuid']} ({char['description']})") - - if not meshcore_uuids: - print("No obvious MeshCore-specific characteristics found") - - print() - - # Summary - print("Summary:") - print("-" * 50) - if nus_service_found and nus_tx_found and nus_rx_found: - print("✓ Device uses standard Nordic UART Service") - print("✓ Current UUIDs in meshcore_protocol.py are correct") - else: - print("✗ Device does not use standard Nordic UART Service") - print("✗ UUIDs in meshcore_protocol.py may need updating") - print("\nAlternative UUIDs found:") - for char in all_uuids['characteristics']: - if 'write' in char['properties']: - print(f" Write characteristic: {char['uuid']} ({char['description']})") - if 'read' in char['properties'] or 'notify' in char['properties']: - print(f" Read/Notify characteristic: {char['uuid']} ({char['description']})") - - # Disconnect - await client.disconnect() - print("\n✓ Disconnected successfully") - - except ImportError: - print("BLE support not available. Install bleak: pip install bleak") - except Exception as e: - print(f"Error: {e}") - - -if __name__ == "__main__": - asyncio.run(discover_device_uuids()) diff --git a/tests/example_usage.py b/tests/example_usage.py deleted file mode 100755 index d7372ea..0000000 --- a/tests/example_usage.py +++ /dev/null @@ -1,336 +0,0 @@ -#!/usr/bin/env python3 -""" -Example usage of the MeshCore Bot Framework -This script demonstrates common use cases and how to extend the bot. -""" - -import asyncio -import configparser -from meshcore_bot import MeshCoreBot -from meshcore_protocol import MeshCoreMessage, MessageType - - -class ExtendedMeshCoreBot(MeshCoreBot): - """Extended bot with additional features""" - - def __init__(self, config_file: str = "config.ini"): - super().__init__(config_file) - self.message_count = 0 - self.custom_handlers = [] - - def add_custom_handler(self, handler_func): - """Add a custom message handler""" - self.custom_handlers.append(handler_func) - self.logger.info(f"Added custom handler: {handler_func.__name__}") - - async def process_message(self, message: MeshCoreMessage): - """Override to add custom processing""" - # Call parent processing first - await super().process_message(message) - - # Run custom handlers - for handler in self.custom_handlers: - try: - await handler(message) - except Exception as e: - self.logger.error(f"Custom handler error: {e}") - - # Track message count - self.message_count += 1 - if self.message_count % 10 == 0: - self.logger.info(f"Processed {self.message_count} messages") - - -async def weather_handler(message: MeshCoreMessage): - """Custom handler for weather requests""" - if "weather" in message.content.lower(): - # Simulate weather API call - weather_data = "Sunny, 22°C, Humidity: 65%" - response = f"Weather update: {weather_data}" - - # Send response (this would need access to the bot instance) - # await bot.send_message(message.channel, response) - print(f"Weather request from {message.sender_id}: {response}") - - -async def stats_handler(message: MeshCoreMessage): - """Custom handler for statistics requests""" - if "stats" in message.content.lower(): - stats = "Bot Statistics:\n- Messages processed: 42\n- Active users: 15\n- Network health: Good" - print(f"Stats request from {message.sender_id}: {stats}") - - -async def echo_handler(message: MeshCoreMessage): - """Custom handler for echo functionality""" - if message.content.startswith("echo "): - echo_text = message.content[5:] # Remove "echo " prefix - response = f"Echo: {echo_text}" - print(f"Echo from {message.sender_id}: {response}") - - -def create_example_config(): - """Create an example configuration file""" - config = configparser.ConfigParser() - - config['Connection'] = { - 'connection_type': 'serial', - 'serial_port': '/dev/ttyUSB0', - 'serial_baudrate': '115200', - 'ble_device_name': 'MeshCore', - 'timeout': '30' - } - - config['Bot'] = { - 'bot_name': 'ExampleBot', - 'node_id': '', - 'enabled': 'true', - 'passive_mode': 'false', - 'rate_limit_seconds': '10' - } - - config['Keywords'] = { - 'test': 'Message received! Hops: {hops}, Path: {path}, From: {sender}', - 'hello': 'Hello {sender}! Welcome to the MeshCore network.', - 'help': 'Available commands: test, hello, help, weather, stats, echo ', - 'ping': 'Pong! Response time: {timestamp}', - 'info': 'Bot Info: {sender} sent "{content}" via {channel}' - } - - config['Channels'] = { - 'monitor_channels': 'general,test,emergency,weather', - 'respond_to_dms': 'true' - } - - config['Banned_Users'] = { - 'banned_users': '' - } - - config['Scheduled_Messages'] = { - '08:00': 'general:Good morning! Weather update coming soon.', - '12:00': 'general:Lunch time reminder - stay hydrated!', - '18:00': 'general:Evening update - network status: Good' - } - - config['Logging'] = { - 'log_level': 'INFO', - 'log_file': 'example_bot.log', - 'colored_output': 'true' - } - - config['External_Data'] = { - 'weather_api_key': '', - 'weather_update_interval': '3600', - 'tide_api_key': '', - 'tide_update_interval': '1800' - } - - with open('example_config.ini', 'w') as f: - config.write(f) - - print("Created example_config.ini") - - -async def run_example_bot(): - """Run the example bot with custom handlers""" - print("Starting Example MeshCore Bot...") - - # Create example config if it doesn't exist - try: - with open('example_config.ini', 'r'): - pass - except FileNotFoundError: - create_example_config() - - # Create extended bot - bot = ExtendedMeshCoreBot('example_config.ini') - - # Add custom handlers - bot.add_custom_handler(weather_handler) - bot.add_custom_handler(stats_handler) - bot.add_custom_handler(echo_handler) - - # Add some example keywords - bot.add_keyword("weather", "Current weather: Sunny, 22°C") - bot.add_keyword("stats", "Network stats: 15 nodes, 3 hops max") - bot.add_keyword("time", "Current time: {timestamp}") - - # Start the bot - try: - await bot.start() - except KeyboardInterrupt: - print("\nShutting down example bot...") - bot.stop() - except Exception as e: - print(f"Error running example bot: {e}") - bot.stop() - - -async def demo_offline_mode(): - """Demonstrate bot functionality without actual connection""" - print("Running offline demo...") - - bot = ExtendedMeshCoreBot('example_config.ini') - - # Add custom handlers - bot.add_custom_handler(weather_handler) - bot.add_custom_handler(stats_handler) - bot.add_custom_handler(echo_handler) - - # Simulate some messages - from datetime import datetime - - test_messages = [ - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node1", - channel="general", - content="Hello everyone!", - hops=1, - path="AB", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node2", - channel="general", - content="What's the weather like?", - hops=2, - path="CD", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node3", - channel="test", - content="test message", - hops=1, - path="EF", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node4", - channel="general", - content="echo Hello world!", - hops=3, - path="GH", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node5", - channel="general", - content="Show me the stats", - hops=1, - path="IJ", - timestamp=datetime.now() - ) - ] - - print("Processing test messages...") - for message in test_messages: - print(f"\n--- Processing message from {message.sender_id} ---") - print(f"Channel: {message.channel}") - print(f"Content: {message.content}") - print(f"Hops: {message.hops}, Path: {message.path}") - - # Check if message should be processed - if bot.should_process_message(message): - print("✓ Message will be processed") - - # Check for keywords - keyword_matches = bot.check_keywords(message) - if keyword_matches: - print("✓ Keywords matched:") - for keyword, response in keyword_matches: - print(f" '{keyword}' -> '{response}'") - else: - print("✗ No keywords matched") - - # Run custom handlers - for handler in bot.custom_handlers: - try: - await handler(message) - except Exception as e: - print(f"Handler error: {e}") - else: - print("✗ Message will not be processed") - - print(f"\nDemo complete. Processed {len(test_messages)} messages.") - - -def show_configuration_examples(): - """Show examples of different configuration options""" - print("=== Configuration Examples ===\n") - - print("1. Basic Bot Configuration:") - print(""" -[Bot] -bot_name = MyMeshBot -enabled = true -passive_mode = false -rate_limit_seconds = 5 -""") - - print("2. Keyword Response Examples:") - print(""" -[Keywords] -hello = "Hello {sender}! Welcome to the network." -weather = "Weather for {sender}: {content}" -help = "Available commands: hello, weather, help, ping" -ping = "Pong! Response time: {timestamp}" -info = "Message info: {sender} -> {channel} via {path} ({hops} hops)" -""") - - print("3. Scheduled Messages:") - print(""" -[Scheduled_Messages] -08:00 = general:Good morning! Network status check. -12:00 = general:Midday reminder - stay connected! -18:00 = general:Evening update - network running smoothly. -22:00 = general:Good night! Network will continue monitoring. -""") - - print("4. Channel Management:") - print(""" -[Channels] -monitor_channels = general,emergency,weather,announcements -respond_to_dms = true -""") - - print("5. User Management:") - print(""" -[Banned_Users] -banned_users = spam_user,malicious_node,test_user -""") - - -if __name__ == "__main__": - import sys - - if len(sys.argv) > 1: - command = sys.argv[1].lower() - - if command == "demo": - asyncio.run(demo_offline_mode()) - elif command == "config": - show_configuration_examples() - elif command == "create-config": - create_example_config() - elif command == "run": - asyncio.run(run_example_bot()) - else: - print("Unknown command. Available commands:") - print(" demo - Run offline demo") - print(" config - Show configuration examples") - print(" create-config - Create example config file") - print(" run - Run the bot (requires connection)") - else: - print("MeshCore Bot Example Usage") - print("=========================") - print("Available commands:") - print(" python example_usage.py demo - Run offline demo") - print(" python example_usage.py config - Show configuration examples") - print(" python example_usage.py create-config - Create example config file") - print(" python example_usage.py run - Run the bot (requires connection)") - print("\nRun 'python example_usage.py demo' to see the bot in action!") diff --git a/tests/meshcore_packet_analyzer.py b/tests/meshcore_packet_analyzer.py deleted file mode 100755 index 7dd6c12..0000000 --- a/tests/meshcore_packet_analyzer.py +++ /dev/null @@ -1,214 +0,0 @@ -#!/usr/bin/env python3 -""" -MeshCore Packet Analyzer -This script helps analyze and decode MeshCore binary packets -""" - -import struct -from datetime import datetime -from typing import Dict, Any, Optional - - -class MeshCorePacketAnalyzer: - """Analyzes MeshCore binary packets""" - - def __init__(self): - self.packet_count = 0 - - def analyze_packet(self, raw_data: bytes) -> Dict[str, Any]: - """Analyze a binary packet and extract information""" - self.packet_count += 1 - - analysis = { - 'packet_number': self.packet_count, - 'raw_hex': raw_data.hex(), - 'length': len(raw_data), - 'timestamp': datetime.now(), - 'analysis': {} - } - - if len(raw_data) < 4: - analysis['analysis']['error'] = 'Packet too short' - return analysis - - # Extract basic packet structure - # This is speculative - adapt based on actual MeshCore protocol - - # First byte might be packet type/header - header = raw_data[0] - analysis['analysis']['header'] = f"0x{header:02x}" - - # Second byte might be length or flags - second_byte = raw_data[1] - analysis['analysis']['second_byte'] = f"0x{second_byte:02x}" - - # Look for patterns - analysis['analysis']['patterns'] = self._find_patterns(raw_data) - - # Try to extract potential fields - analysis['analysis']['potential_fields'] = self._extract_potential_fields(raw_data) - - # Check for common MeshCore patterns - analysis['analysis']['meshcore_patterns'] = self._check_meshcore_patterns(raw_data) - - return analysis - - def _find_patterns(self, data: bytes) -> Dict[str, Any]: - """Find patterns in the binary data""" - patterns = {} - - # Check for repeated bytes - byte_counts = {} - for byte in data: - byte_counts[byte] = byte_counts.get(byte, 0) + 1 - - # Find most common bytes - common_bytes = sorted(byte_counts.items(), key=lambda x: x[1], reverse=True)[:5] - patterns['common_bytes'] = [(f"0x{b:02x}", count) for b, count in common_bytes] - - # Check for sequences - sequences = [] - for i in range(len(data) - 2): - seq = data[i:i+3] - if seq.count(seq[0]) == len(seq): # All same byte - sequences.append(f"0x{seq[0]:02x} repeated {len(seq)} times at position {i}") - - patterns['sequences'] = sequences - - return patterns - - def _extract_potential_fields(self, data: bytes) -> Dict[str, Any]: - """Extract potential packet fields""" - fields = {} - - if len(data) >= 4: - # First 4 bytes might be a header - fields['header_4bytes'] = data[:4].hex() - - # Next 4 bytes might be length or timestamp - if len(data) >= 8: - fields['next_4bytes'] = data[4:8].hex() - - # Try to interpret as length - try: - length = struct.unpack('= 3: # Only meaningful text - fields[f'text_at_{i-len(text)}'] = text - text_candidates = [] - - # Check for any remaining text - if text_candidates: - text = ''.join(text_candidates) - if len(text) >= 3: - fields['text_at_end'] = text - - return fields - - def _check_meshcore_patterns(self, data: bytes) -> Dict[str, Any]: - """Check for common MeshCore protocol patterns""" - patterns = {} - - # Check for common MeshCore packet types - if len(data) > 0: - header = data[0] - - # These are speculative based on common packet protocols - if header == 0x88: - patterns['packet_type'] = 'Possible MeshCore data packet' - elif header == 0x83: - patterns['packet_type'] = 'Possible MeshCore control packet' - elif header == 0x81: - patterns['packet_type'] = 'Possible MeshCore acknowledgment' - else: - patterns['packet_type'] = f'Unknown packet type: 0x{header:02x}' - - # Check for potential node IDs or addresses - if len(data) >= 8: - # Look for potential 4-byte node ID - potential_node_id = data[4:8] - patterns['potential_node_id'] = potential_node_id.hex() - - # Check for potential message content - if len(data) > 8: - content_start = 8 - content = data[content_start:] - - # Try to find text content - text_content = "" - for byte in content: - if 32 <= byte <= 126: # Printable ASCII - text_content += chr(byte) - else: - break - - if text_content: - patterns['text_content'] = text_content - - return patterns - - def print_analysis(self, analysis: Dict[str, Any]): - """Print packet analysis in a readable format""" - print(f"\n=== Packet Analysis #{analysis['packet_number']} ===") - print(f"Timestamp: {analysis['timestamp']}") - print(f"Length: {analysis['length']} bytes") - print(f"Raw Hex: {analysis['raw_hex']}") - - print("\n--- Header Analysis ---") - if 'header' in analysis['analysis']: - print(f"Header: {analysis['analysis']['header']}") - if 'second_byte' in analysis['analysis']: - print(f"Second Byte: {analysis['analysis']['second_byte']}") - - print("\n--- Patterns ---") - patterns = analysis['analysis'].get('patterns', {}) - if 'common_bytes' in patterns: - print("Most common bytes:") - for byte, count in patterns['common_bytes']: - print(f" {byte}: {count} times") - - print("\n--- Potential Fields ---") - fields = analysis['analysis'].get('potential_fields', {}) - for field, value in fields.items(): - print(f" {field}: {value}") - - print("\n--- MeshCore Patterns ---") - meshcore = analysis['analysis'].get('meshcore_patterns', {}) - for pattern, value in meshcore.items(): - print(f" {pattern}: {value}") - - -def test_packet_analyzer(): - """Test the packet analyzer with sample data""" - analyzer = MeshCorePacketAnalyzer() - - # Test with the packets we've seen - test_packets = [ - bytes.fromhex('8830d405037e5fbd54c559a4df32702e'), - bytes.fromhex('8830e505037e5f0154c559a4df32702e'), - bytes.fromhex('882ef70a00f54621da57be35db1e5108'), - # Add the packet from your DM - bytes.fromhex('888388888188888388888188888388') - ] - - print("MeshCore Packet Analyzer") - print("=" * 50) - - for packet in test_packets: - analysis = analyzer.analyze_packet(packet) - analyzer.print_analysis(analysis) - - -if __name__ == "__main__": - test_packet_analyzer() diff --git a/tests/repeater_management_demo.py b/tests/repeater_management_demo.py deleted file mode 100644 index a419dad..0000000 --- a/tests/repeater_management_demo.py +++ /dev/null @@ -1,158 +0,0 @@ -#!/usr/bin/env python3 -""" -Repeater Management Demo Script -Demonstrates the repeater contact management functionality -""" - -import asyncio -import sys -from pathlib import Path - -# Add the project root to the Python path -sys.path.insert(0, str(Path(__file__).parent)) - -from modules.repeater_manager import RepeaterManager - - -class MockBot: - """Mock bot class for demonstration purposes""" - - def __init__(self): - self.logger = self._create_mock_logger() - - def _create_mock_logger(self): - """Create a mock logger for demonstration""" - import logging - logger = logging.getLogger("MockBot") - logger.setLevel(logging.INFO) - - # Create console handler - handler = logging.StreamHandler() - formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') - handler.setFormatter(formatter) - logger.addHandler(handler) - - return logger - - -async def demo_repeater_management(): - """Demonstrate repeater management functionality""" - print("🔧 Repeater Management System Demo") - print("=" * 50) - - # Create mock bot and repeater manager - bot = MockBot() - repeater_manager = RepeaterManager(bot, "demo_repeater_contacts.db") - - print("\n1. 📡 Adding sample repeater contacts...") - - # Simulate adding some repeater contacts - sample_contacts = [ - { - 'public_key': '15a24fcbc0dd1234567890abcdef1234567890abcdef1234567890abcdef12', - 'adv_name': 'Hillcrest Repeater', - 'type': 'repeater', - 'name': 'Hillcrest Repeater' - }, - { - 'public_key': '25b35fddc1ee2345678901bcdef2345678901bcdef2345678901bcdef23456', - 'adv_name': 'Downtown Room Server', - 'type': 'roomserver', - 'name': 'Downtown Room Server' - }, - { - 'public_key': '35c46feed2ff3456789012cdef3456789012cdef3456789012cdef34567890', - 'adv_name': 'Westside Repeater', - 'type': 'repeater', - 'name': 'Westside Repeater' - } - ] - - # Add contacts to database (simulating the detection process) - for contact in sample_contacts: - # In the real implementation, this would use repeater_manager._is_repeater_device(contact) - # The detection is now synchronous and LoRa-aware (no network communication needed) - device_type = 'RoomServer' if 'room' in contact['adv_name'].lower() else 'Repeater' - - # Simulate database insertion - import sqlite3 - with sqlite3.connect(repeater_manager.db_path) as conn: - cursor = conn.cursor() - cursor.execute(''' - INSERT OR REPLACE INTO repeater_contacts - (public_key, name, device_type, contact_data) - VALUES (?, ?, ?, ?) - ''', ( - contact['public_key'], - contact['adv_name'], - device_type, - '{"demo": "sample_contact"}' - )) - conn.commit() - - print("✅ Added 3 sample repeater contacts") - - print("\n2. 📋 Listing all repeater contacts...") - repeaters = await repeater_manager.get_repeater_contacts(active_only=False) - for repeater in repeaters: - status = "🟢 Active" if repeater['is_active'] else "🔴 Purged" - print(f" {status} {repeater['name']} ({repeater['device_type']})") - - print("\n3. 🗑️ Purging old repeaters (simulating 30+ days old)...") - # Simulate old timestamps - import sqlite3 - from datetime import datetime, timedelta - old_date = (datetime.now() - timedelta(days=35)).isoformat() - - with sqlite3.connect(repeater_manager.db_path) as conn: - cursor = conn.cursor() - cursor.execute( - 'UPDATE repeater_contacts SET last_seen = ? WHERE name = ?', - (old_date, 'Hillcrest Repeater') - ) - conn.commit() - - purged_count = await repeater_manager.purge_old_repeaters(days=30, reason="Demo: Auto-purge old repeaters") - print(f"✅ Purged {purged_count} old repeaters") - - print("\n4. 📊 Getting management statistics...") - stats = await repeater_manager.get_purging_stats() - print(f" Total repeaters: {stats.get('total_repeaters', 0)}") - print(f" Active repeaters: {stats.get('active_repeaters', 0)}") - print(f" Purged repeaters: {stats.get('purged_repeaters', 0)}") - - print("\n5. 🔄 Restoring a purged repeater...") - # Find a purged repeater to restore - purged_repeaters = [r for r in repeaters if not r['is_active']] - if purged_repeaters: - success = await repeater_manager.restore_repeater( - purged_repeaters[0]['public_key'], - "Demo: Manual restore" - ) - if success: - print(f"✅ Restored: {purged_repeaters[0]['name']}") - else: - print(f"❌ Failed to restore: {purged_repeaters[0]['name']}") - - print("\n6. 🧹 Cleaning up demo database...") - # Clean up the demo database - Path("demo_repeater_contacts.db").unlink(missing_ok=True) - print("✅ Demo database cleaned up") - - print("\n🎉 Demo completed successfully!") - print("\nTo use this in your bot:") - print("1. The RepeaterManager is automatically initialized in your bot") - print("2. Use the !repeater command to manage repeaters") - print("3. Run '!repeater scan' to catalog repeaters from your contacts") - print("4. Run '!repeater list' to see all repeaters") - print("5. Run '!repeater purge 30' to purge old repeaters") - print("\nKey Features:") - print("• LoRa-aware: Uses local contact data for detection (no network overhead)") - print("• Actually removes contacts from device using remove_contact command") - print("• 30-second timeouts and batch processing for LoRa communication") - print("• Maintains audit trail of all operations") - print("• Provides statistics and monitoring capabilities") - - -if __name__ == "__main__": - asyncio.run(demo_repeater_management()) diff --git a/tests/test_ble_connection.py b/tests/test_ble_connection.py deleted file mode 100755 index 78d4048..0000000 --- a/tests/test_ble_connection.py +++ /dev/null @@ -1,201 +0,0 @@ -#!/usr/bin/env python3 -""" -BLE Connection Test Script for MeshCore Bot -This script helps debug BLE connectivity issues. -""" - -import asyncio -import sys -import configparser -from pathlib import Path - - -async def test_ble_scan(): - """Test BLE scanning functionality""" - print("Testing BLE Scanning...") - print("=" * 40) - - try: - from bleak import BleakScanner - - print("Scanning for BLE devices...") - devices = await BleakScanner.discover(timeout=10) - - if not devices: - print("No BLE devices found!") - return False - - print(f"Found {len(devices)} BLE devices:") - for i, device in enumerate(devices, 1): - print(f"{i:2d}. Name: {device.name or 'Unknown'}") - print(f" Address: {device.address}") - print(f" RSSI: {device.rssi}") - if device.metadata: - print(f" Metadata: {device.metadata}") - print() - - return True - - except ImportError: - print("BLE support not available. Install bleak:") - print(" pip install bleak") - return False - except Exception as e: - print(f"BLE scanning failed: {e}") - return False - - -async def test_specific_device(device_name): - """Test connection to a specific device""" - print(f"Testing connection to: {device_name}") - print("=" * 40) - - try: - from bleak import BleakScanner, BleakClient - - # Clean device name - device_name = device_name.strip().strip('"').strip("'") - print(f"Looking for device: '{device_name}'") - - # Scan for devices - print("Scanning...") - devices = await BleakScanner.discover(timeout=10) - - # Find our device - target_device = None - for device in devices: - if device.name and device.name.strip() == device_name: - target_device = device - break - - if not target_device: - print(f"Device '{device_name}' not found!") - print("Available devices:") - for device in devices: - if device.name: - print(f" - {device.name}") - return False - - print(f"Found device: {target_device.name}") - print(f"Address: {target_device.address}") - # Use advertisement data for RSSI to avoid deprecation warning - if hasattr(target_device, 'advertisement') and target_device.advertisement: - print(f"RSSI: {target_device.advertisement.rssi}") - else: - print("RSSI: Not available") - - # Try to connect - print("Attempting to connect...") - client = BleakClient(target_device.address) - - try: - await client.connect() - print("✓ Successfully connected!") - - # Get services - print("Discovering services...") - services = client.services - service_count = len(list(services)) - print(f"Found {service_count} services:") - - for service in services: - print(f" Service: {service.uuid}") - print(f" Description: {service.description}") - - # Get characteristics - for char in service.characteristics: - print(f" Characteristic: {char.uuid}") - print(f" Properties: {char.properties}") - print(f" Description: {char.description}") - - # Disconnect - await client.disconnect() - print("✓ Disconnected successfully") - return True - - except Exception as e: - print(f"✗ Connection failed: {e}") - return False - - except Exception as e: - print(f"Test failed: {e}") - return False - - -def load_config(): - """Load configuration from config.ini""" - config_file = "config.ini" - if not Path(config_file).exists(): - print(f"Config file {config_file} not found!") - return None - - config = configparser.ConfigParser() - config.read(config_file) - return config - - -async def main(): - """Main test function""" - print("MeshCore Bot - BLE Connection Test") - print("=" * 50) - - # Check if bleak is available - try: - import bleak - print("✓ BLE support available") - except ImportError: - print("✗ BLE support not available") - print("Install bleak: pip install bleak") - return - - # Load config - config = load_config() - if not config: - return - - # Check connection type - connection_type = config.get('Connection', 'connection_type', fallback='serial') - if connection_type != 'ble': - print(f"Connection type is set to '{connection_type}', not 'ble'") - print("Update config.ini to use BLE connection") - return - - # Get device name - device_name = config.get('Connection', 'ble_device_name', fallback='') - if not device_name: - print("No BLE device name configured!") - return - - print(f"Configured device: {device_name}") - print() - - # Run tests - print("1. Testing BLE scanning...") - scan_success = await test_ble_scan() - print() - - if scan_success: - print("2. Testing specific device connection...") - device_success = await test_specific_device(device_name) - print() - - if device_success: - print("✓ All BLE tests passed!") - print("Your MeshCore node should be ready for the bot.") - else: - print("✗ Device connection test failed") - print("Check that:") - print(" - Your MeshCore node is powered on") - print(" - BLE is enabled on the node") - print(" - The device name is correct") - print(" - You're close enough to the device") - else: - print("✗ BLE scanning failed") - print("Check that:") - print(" - BLE is enabled on your computer") - print(" - You have permission to access BLE") - print(" - There are BLE devices nearby") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/tests/test_channels.py b/tests/test_channels.py deleted file mode 100644 index fd950ba..0000000 --- a/tests/test_channels.py +++ /dev/null @@ -1,57 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to fetch and display channel information from MeshCore device -""" - -import asyncio -import meshcore -from meshcore_cli.meshcore_cli import next_cmd - -async def test_channels(): - """Test fetching channels from MeshCore device""" - print("Connecting to MeshCore device...") - - try: - # Connect to MeshCore device - mc = await meshcore.MeshCore.create_ble(debug=True) - - if mc.is_connected: - print(f"Connected to: {mc.self_info}") - - # Try to fetch channels - print("\nFetching channels...") - try: - result = await next_cmd(mc, ["channels"]) - print(f"Channels command result: {result}") - - if result: - print(f"Found {len(result)} channels:") - for i, channel in enumerate(result): - print(f" Channel {i}: {channel}") - else: - print("No channels returned") - - except Exception as e: - print(f"Error fetching channels: {e}") - - # Try individual channel queries - print("\nTrying individual channel queries...") - for channel_num in range(5): - try: - result = await next_cmd(mc, ["get_channel", str(channel_num)]) - print(f"Channel {channel_num}: {result}") - except Exception as e: - print(f"Channel {channel_num}: Error - {e}") - - # Disconnect - await mc.disconnect() - print("\nDisconnected") - - else: - print("Failed to connect") - - except Exception as e: - print(f"Error: {e}") - -if __name__ == "__main__": - asyncio.run(test_channels()) diff --git a/tests/test_config.py b/tests/test_config.py deleted file mode 100755 index bbd4d5a..0000000 --- a/tests/test_config.py +++ /dev/null @@ -1,164 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to verify the updated configuration for the Testing channel -""" - -import configparser -from meshcore_bot import MeshCoreBot -from meshcore_protocol import MeshCoreMessage, MessageType -from datetime import datetime - - -def test_configuration(): - """Test the updated configuration""" - print("Testing MeshCore Bot Configuration") - print("=" * 40) - - # Load configuration - bot = MeshCoreBot() - - print(f"Bot Name: {bot.config.get('Bot', 'bot_name') or 'Will be auto-detected'}") - print(f"Node ID: {bot.config.get('Bot', 'node_id') or 'Will be auto-detected'}") - print(f"Enabled: {bot.config.getboolean('Bot', 'enabled')}") - print(f"Passive Mode: {bot.config.getboolean('Bot', 'passive_mode')}") - print(f"Rate Limit: {bot.config.getint('Bot', 'rate_limit_seconds')} seconds") - - print(f"\nMonitor Channels: {', '.join(bot.monitor_channels)}") - print(f"Respond to DMs: {bot.config.getboolean('Channels', 'respond_to_dms')}") - - if bot.config.has_option('Channels', 'channel_public_key'): - print(f"Channel Public Key: {bot.config.get('Channels', 'channel_public_key')}") - - print(f"\nKeywords ({len(bot.keywords)}):") - for keyword, response in bot.keywords.items(): - print(f" '{keyword}' -> '{response}'") - - print(f"\nBanned Users ({len(bot.banned_users)}):") - if bot.banned_users: - for user in bot.banned_users: - print(f" {user}") - else: - print(" None") - - # Test scheduled messages - if bot.config.has_section('Scheduled_Messages'): - scheduled = dict(bot.config.items('Scheduled_Messages')) - print(f"\nScheduled Messages ({len(scheduled)}):") - for time, message_info in scheduled.items(): - try: - channel, message = message_info.split(':', 1) - print(f" {time}: {channel} -> {message}") - except ValueError: - print(f" {time}: {message_info} (invalid format)") - - print("\n" + "=" * 40) - - -def test_message_processing(): - """Test message processing with the new configuration""" - print("Testing Message Processing") - print("=" * 40) - - bot = MeshCoreBot() - - # Create test messages - test_messages = [ - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node1", - channel="Testing", - content="test message", - hops=1, - path="AB", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node2", - channel="general", - content="test message", - hops=2, - path="CD", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node3", - channel="Testing", - content="ping", - hops=1, - path="EF", - timestamp=datetime.now() - ), - MeshCoreMessage( - message_type=MessageType.TEXT, - sender_id="node4", - channel="@bot", - content="test message", - hops=1, - path="GH", - timestamp=datetime.now(), - is_dm=True - ) - ] - - print("Testing message processing:") - for i, message in enumerate(test_messages, 1): - print(f"\nMessage {i}:") - print(f" From: {message.sender_id}") - print(f" Channel: {message.channel}") - print(f" Content: {message.content}") - print(f" Is DM: {message.is_dm}") - - # Check if message should be processed - should_process = bot.should_process_message(message) - print(f" Should Process: {'✓' if should_process else '✗'}") - - if should_process: - # Check for keywords - keyword_matches = bot.check_keywords(message) - if keyword_matches: - print(" Keywords Matched:") - for keyword, response in keyword_matches: - print(f" '{keyword}' -> '{response}'") - else: - print(" No keywords matched") - - print("\n" + "=" * 40) - - -def test_node_detection_simulation(): - """Simulate node detection process""" - print("Testing Node Detection Simulation") - print("=" * 40) - - bot = MeshCoreBot() - - print("Node detection will attempt to:") - print("1. Send 'get_node_info' command to node") - print("2. Parse response for node_name and node_id") - print("3. Update config.ini with detected values") - print("4. Fall back to defaults if detection fails") - - print("\nExpected behavior:") - print("- Bot name will be auto-detected from node") - print("- Node ID will be auto-detected from node") - print("- If detection fails, bot name defaults to 'MeshCoreBot'") - print("- Node ID will be auto-assigned by MeshCore if not detected") - - print("\n" + "=" * 40) - - -if __name__ == "__main__": - test_configuration() - test_message_processing() - test_node_detection_simulation() - - print("\nConfiguration Summary:") - print("✓ Bot configured for 'Testing' channel only") - print("✓ Channel public key configured") - print("✓ DM responses enabled") - print("✓ Node name/ID will be auto-detected") - print("✓ Keywords configured for testing") - print("✓ Scheduled messages set for Testing channel") - print("\nReady to connect to MeshCore node!") diff --git a/tests/test_config_parsing.py b/tests/test_config_parsing.py deleted file mode 100644 index dbb354a..0000000 --- a/tests/test_config_parsing.py +++ /dev/null @@ -1,33 +0,0 @@ -#!/usr/bin/env python3 -""" -Test config parsing to debug the scheduled messages issue -""" - -import configparser - -def test_config_parsing(): - config = configparser.ConfigParser() - config.read('config.ini') - - print("=== Config Parsing Test ===") - - if config.has_section('Scheduled_Messages'): - print("Scheduled_Messages section found:") - for key, value in config.items('Scheduled_Messages'): - print(f" Key: '{key}' -> Value: '{value}'") - - # Test our time format validation - if len(key) == 4: - try: - hour = int(key[:2]) - minute = int(key[2:]) - print(f" Parsed as: {hour:02d}:{minute:02d}") - except ValueError: - print(f" Invalid time format") - else: - print(f" Wrong length: {len(key)}") - else: - print("Scheduled_Messages section not found!") - -if __name__ == "__main__": - test_config_parsing() diff --git a/tests/test_contacts.py b/tests/test_contacts.py deleted file mode 100644 index a13110e..0000000 --- a/tests/test_contacts.py +++ /dev/null @@ -1,102 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to examine contacts in the meshcore package -""" - -import asyncio -import meshcore -from meshcore import EventType -from meshcore_cli.meshcore_cli import send_cmd - -async def test_contacts(): - """Test contact discovery and examination""" - try: - print("Connecting to MeshCore node...") - mc = await meshcore.MeshCore.create_ble(debug=True) - - if mc.is_connected: - print(f"Connected to: {mc.self_info}") - - # Try to manually load contacts - print("Attempting to load contacts...") - - # Check if there are any commands to load contacts - print("Available methods on meshcore object:") - for method in dir(mc): - if not method.startswith('_') and 'contact' in method.lower(): - print(f" {method}") - - # Try to ensure contacts are loaded - if hasattr(mc, 'ensure_contacts'): - print("Calling ensure_contacts()...") - await mc.ensure_contacts() - - # Wait a bit more - print("Waiting for contacts to load (10 seconds)...") - await asyncio.sleep(10) - - # Examine contacts - print(f"\nContacts ({len(mc.contacts)}):") - for key, contact in mc.contacts.items(): - print(f" Key: {key}") - print(f" Contact: {contact}") - print(f" Name: {contact.get('adv_name', 'N/A')}") - print(f" Type: {contact.get('type', 'N/A')}") - print(f" Pubkey: {contact.get('pubkey', 'N/A')}") - print(f" Pubkey prefix: {contact.get('pubkey_prefix', 'N/A')}") - print(" ---") - - # Check pending contacts - print(f"\nPending contacts ({len(mc.pending_contacts)}):") - for contact in mc.pending_contacts: - print(f" Contact: {contact}") - print(f" Name: {contact.get('adv_name', 'N/A')}") - print(f" Pubkey: {contact.get('pubkey', 'N/A')}") - print(f" Pubkey prefix: {contact.get('pubkey_prefix', 'N/A')}") - print(" ---") - - # Try to manually request contacts from the device - print("\nTrying to request contacts from device...") - try: - # Send a command to get contacts - from meshcore_cli.meshcore_cli import next_cmd - result = await next_cmd(mc, ["contacts"]) - print(f"Contacts command result: {result}") - except Exception as e: - print(f"Error requesting contacts: {e}") - - # Test get_contact_by_key_prefix - test_prefix = "460728508c17" - print(f"\nTesting get_contact_by_key_prefix('{test_prefix}'):") - contact = mc.get_contact_by_key_prefix(test_prefix) - if contact: - print(f"Found contact: {contact}") - else: - print("Contact not found") - - # Try to find by name - print(f"\nTrying to find by name...") - for key, contact in mc.contacts.items(): - if contact.get('adv_name') == test_prefix: - print(f"Found by name: {contact}") - break - else: - print("Not found by name either") - - # Test get_contact_by_name - print(f"\nTesting get_contact_by_name('{test_prefix}'):") - contact = mc.get_contact_by_name(test_prefix) - if contact: - print(f"Found contact: {contact}") - else: - print("Contact not found by name") - - await mc.disconnect() - else: - print("Failed to connect") - - except Exception as e: - print(f"Error: {e}") - -if __name__ == "__main__": - asyncio.run(test_contacts()) diff --git a/tests/test_dynamic_channels.py b/tests/test_dynamic_channels.py deleted file mode 100644 index 6df83f7..0000000 --- a/tests/test_dynamic_channels.py +++ /dev/null @@ -1,60 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to verify dynamic channel fetching using get_channel commands -""" - -import asyncio -import meshcore -from meshcore_cli.meshcore_cli import next_cmd - -async def test_dynamic_channels(): - """Test dynamic channel fetching""" - print("Connecting to MeshCore device...") - - try: - # Connect to MeshCore device - mc = await meshcore.MeshCore.create_ble(debug=True) - - if mc.is_connected: - print(f"Connected to: {mc.self_info}") - - # Test dynamic channel fetching - print("\nTesting dynamic channel fetching...") - channels = {} - - for channel_num in range(10): # Check channels 0-9 - try: - print(f"Fetching channel {channel_num}...") - result = await next_cmd(mc, ["get_channel", str(channel_num)]) - if result and len(result) > 0: - channel_info = result[0] - if isinstance(channel_info, dict) and 'channel_name' in channel_info: - channels[channel_num] = { - 'number': channel_num, - 'name': channel_info['channel_name'], - 'secret': channel_info.get('channel_secret', '') - } - print(f" ✓ Channel {channel_num}: {channel_info['channel_name']}") - else: - print(f" ✗ Channel {channel_num}: Invalid format") - else: - print(f" ✗ Channel {channel_num}: No result") - except Exception as e: - print(f" ✗ Channel {channel_num}: Error - {e}") - - print(f"\nFound {len(channels)} channels:") - for num, info in channels.items(): - print(f" Channel {num}: {info['name']}") - - # Disconnect - await mc.disconnect() - print("\nDisconnected") - - else: - print("Failed to connect") - - except Exception as e: - print(f"Error: {e}") - -if __name__ == "__main__": - asyncio.run(test_dynamic_channels()) diff --git a/tests/test_event_structure.py b/tests/test_event_structure.py deleted file mode 100644 index 8d8ffe7..0000000 --- a/tests/test_event_structure.py +++ /dev/null @@ -1,58 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to examine the actual event structure for channel messages -""" - -import asyncio -import meshcore -from meshcore import EventType - -async def test_event_structure(): - """Test the event structure for channel messages""" - print("Testing event structure for channel messages...") - - try: - # Create a BLE connection - mc = await meshcore.MeshCore.create_ble(debug=True) - print(f"Connected to: {mc.self_info}") - - # Subscribe to channel message events - async def on_channel_msg(event): - print(f"\n=== CHANNEL MESSAGE EVENT ===") - print(f"Event type: {event.type}") - print(f"Event payload: {event.payload}") - print(f"Event payload type: {type(event.payload)}") - print(f"Event payload keys: {list(event.payload.keys()) if hasattr(event.payload, 'keys') else 'No keys'}") - print(f"Event metadata: {getattr(event, 'metadata', 'No metadata')}") - print(f"Event dir: {[x for x in dir(event) if not x.startswith('_')]}") - - # Try to access all possible attributes - for attr in dir(event): - if not attr.startswith('_'): - try: - value = getattr(event, attr) - print(f" {attr}: {value}") - except Exception as e: - print(f" {attr}: Error accessing - {e}") - - print("=== END EVENT ===\n") - - subscription = mc.subscribe(EventType.CHANNEL_MSG_RECV, on_channel_msg) - - # Start auto message fetching - await mc.start_auto_message_fetching() - - print("Listening for channel messages... Press Ctrl+C to stop") - print("Send a test message to a channel to see the event structure") - - # Keep running - while True: - await asyncio.sleep(1) - - except Exception as e: - print(f"Error: {e}") - import traceback - traceback.print_exc() - -if __name__ == "__main__": - asyncio.run(test_event_structure()) diff --git a/tests/test_get_channel.py b/tests/test_get_channel.py deleted file mode 100644 index 828fdb0..0000000 --- a/tests/test_get_channel.py +++ /dev/null @@ -1,65 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to see what get_channel command returns -""" - -import asyncio -import meshcore -from meshcore_cli.meshcore_cli import next_cmd - -async def test_get_channel(): - """Test get_channel command""" - print("Connecting to MeshCore device...") - - try: - # Connect to MeshCore device - mc = await meshcore.MeshCore.create_ble(debug=True) - - if mc.is_connected: - print(f"Connected to: {mc.self_info}") - - # Test get_channel for channel 5 (MyTest) - print("\nTesting get_channel for channel 5...") - try: - result = await next_cmd(mc, ["get_channel", "5"]) - print(f"Result type: {type(result)}") - print(f"Result: {result}") - if result: - print(f"Result length: {len(result)}") - for i, item in enumerate(result): - print(f" Item {i}: {item} (type: {type(item)})") - if isinstance(item, dict): - for key, value in item.items(): - print(f" {key}: {value}") - except Exception as e: - print(f"Error: {e}") - - # Test get_channel for channel 4 (Testing) - print("\nTesting get_channel for channel 4...") - try: - result = await next_cmd(mc, ["get_channel", "4"]) - print(f"Result type: {type(result)}") - print(f"Result: {result}") - if result: - print(f"Result length: {len(result)}") - for i, item in enumerate(result): - print(f" Item {i}: {item} (type: {type(item)})") - if isinstance(item, dict): - for key, value in item.items(): - print(f" {key}: {value}") - except Exception as e: - print(f"Error: {e}") - - # Disconnect - await mc.disconnect() - print("\nDisconnected") - - else: - print("Failed to connect") - - except Exception as e: - print(f"Error: {e}") - -if __name__ == "__main__": - asyncio.run(test_get_channel()) - diff --git a/tests/test_installation.py b/tests/test_installation.py deleted file mode 100755 index 52ac4c0..0000000 --- a/tests/test_installation.py +++ /dev/null @@ -1,251 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to verify MeshCore Bot installation -Run this script to check if all dependencies are installed correctly. -""" - -import sys -import importlib -from pathlib import Path - - -def test_imports(): - """Test if all required modules can be imported""" - print("Testing imports...") - - required_modules = [ - 'asyncio', - 'configparser', - 'logging', - 'time', - 're', - 'serial', - 'serial.tools.list_ports', - 'datetime', - 'typing', - 'dataclasses', - 'pathlib', - 'colorlog', - 'schedule', - 'threading' - ] - - optional_modules = [ - 'bleak' - ] - - failed_imports = [] - - # Test required modules - for module in required_modules: - try: - importlib.import_module(module) - print(f"✓ {module}") - except ImportError as e: - print(f"✗ {module} - {e}") - failed_imports.append(module) - - # Test optional modules - print("\nOptional modules:") - for module in optional_modules: - try: - importlib.import_module(module) - print(f"✓ {module} (optional)") - except ImportError: - print(f"✗ {module} (optional) - not installed") - - return failed_imports - - -def test_bot_modules(): - """Test if bot modules can be imported""" - print("\nTesting bot modules...") - - bot_modules = [ - 'meshcore_bot', - 'meshcore_protocol' - ] - - failed_imports = [] - - for module in bot_modules: - try: - importlib.import_module(module) - print(f"✓ {module}") - except ImportError as e: - print(f"✗ {module} - {e}") - failed_imports.append(module) - - return failed_imports - - -def test_basic_functionality(): - """Test basic bot functionality""" - print("\nTesting basic functionality...") - - try: - from meshcore_bot import MeshCoreBot - from meshcore_protocol import MeshCoreProtocol, MeshCoreMessage, MessageType - - # Test bot creation - bot = MeshCoreBot() - print("✓ Bot creation successful") - - # Test protocol creation - protocol = MeshCoreProtocol() - print("✓ Protocol creation successful") - - # Test message parsing - test_message = '{"type": "text", "sender": "test", "channel": "general", "content": "test", "hops": 1, "path": "AB"}' - parsed = protocol.parse_message(test_message) - if parsed: - print("✓ Message parsing successful") - else: - print("✗ Message parsing failed") - - # Test keyword loading - keywords = bot.load_keywords() - print(f"✓ Keyword loading successful ({len(keywords)} keywords)") - - # Test banned users loading - banned = bot.load_banned_users() - print(f"✓ Banned users loading successful ({len(banned)} banned users)") - - return True - - except Exception as e: - print(f"✗ Basic functionality test failed: {e}") - return False - - -def test_config_file(): - """Test configuration file handling""" - print("\nTesting configuration file...") - - try: - from meshcore_bot import MeshCoreBot - - # Test with non-existent config (should create default) - test_config = "test_config.ini" - bot = MeshCoreBot(test_config) - - if Path(test_config).exists(): - print("✓ Default config creation successful") - - # Clean up - Path(test_config).unlink() - print("✓ Test config cleanup successful") - else: - print("✗ Default config creation failed") - return False - - return True - - except Exception as e: - print(f"✗ Config file test failed: {e}") - return False - - -def test_serial_ports(): - """Test serial port detection""" - print("\nTesting serial port detection...") - - try: - import serial.tools.list_ports - - ports = list(serial.tools.list_ports.comports()) - print(f"✓ Found {len(ports)} serial ports:") - - for port in ports: - print(f" - {port.device}: {port.description}") - - return True - - except Exception as e: - print(f"✗ Serial port detection failed: {e}") - return False - - -def test_ble_support(): - """Test BLE support""" - print("\nTesting BLE support...") - - try: - import bleak - print("✓ BLE support available (bleak installed)") - return True - except ImportError: - print("✗ BLE support not available (bleak not installed)") - print(" Install with: pip install bleak") - return False - - -def main(): - """Run all tests""" - print("MeshCore Bot Installation Test") - print("==============================\n") - - # Test Python version - print(f"Python version: {sys.version}") - if sys.version_info < (3, 7): - print("⚠ Warning: Python 3.7+ recommended") - else: - print("✓ Python version OK") - - print() - - # Run tests - failed_imports = test_imports() - failed_bot_imports = test_bot_modules() - - basic_ok = test_basic_functionality() - config_ok = test_config_file() - serial_ok = test_serial_ports() - ble_ok = test_ble_support() - - # Summary - print("\n" + "="*50) - print("TEST SUMMARY") - print("="*50) - - if not failed_imports and not failed_bot_imports and basic_ok and config_ok: - print("✓ All core tests passed!") - print("✓ MeshCore Bot is ready to use") - - if serial_ok: - print("✓ Serial communication ready") - else: - print("⚠ Serial communication may have issues") - - if ble_ok: - print("✓ BLE communication ready") - else: - print("⚠ BLE communication not available (install bleak)") - - print("\nNext steps:") - print("1. Configure your device in config.ini") - print("2. Run: python meshcore_bot.py") - print("3. Or try: python example_usage.py demo") - - else: - print("✗ Some tests failed:") - - if failed_imports: - print(f" - Missing required modules: {', '.join(failed_imports)}") - print(" Install with: pip install -r requirements.txt") - - if failed_bot_imports: - print(f" - Bot modules not found: {', '.join(failed_bot_imports)}") - print(" Make sure you're in the correct directory") - - if not basic_ok: - print(" - Basic functionality test failed") - - if not config_ok: - print(" - Configuration file test failed") - - print("\nPlease fix the issues above before using the bot.") - - -if __name__ == "__main__": - main() diff --git a/tests/test_meshcore_official.py b/tests/test_meshcore_official.py deleted file mode 100644 index 6ba8a43..0000000 --- a/tests/test_meshcore_official.py +++ /dev/null @@ -1,48 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to understand how the official meshcore package works -""" - -import asyncio -import meshcore -from meshcore import EventType - -async def test_meshcore(): - """Test the official meshcore package""" - print("Testing official meshcore package...") - - # Try to create a BLE connection - try: - # This will scan for devices - mc = await meshcore.MeshCore.create_ble(debug=True) - print(f"Connected to: {mc.self_info}") - - # Subscribe to message events - async def on_message(event): - print(f"Received message event: {event}") - - async def on_contact_msg(event): - print(f"Received contact message: {event}") - - async def on_channel_msg(event): - print(f"Received channel message: {event}") - - subscription1 = mc.subscribe(EventType.CONTACT_MSG_RECV, on_contact_msg) - subscription2 = mc.subscribe(EventType.CHANNEL_MSG_RECV, on_channel_msg) - - # Start auto message fetching - await mc.start_auto_message_fetching() - - print("Listening for messages... Press Ctrl+C to stop") - - # Keep running - while True: - await asyncio.sleep(1) - - except Exception as e: - print(f"Error: {e}") - import traceback - traceback.print_exc() - -if __name__ == "__main__": - asyncio.run(test_meshcore()) diff --git a/tests/test_path_extraction.py b/tests/test_path_extraction.py deleted file mode 100644 index 915ad6a..0000000 --- a/tests/test_path_extraction.py +++ /dev/null @@ -1,113 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to verify path extraction logic -""" - -def test_path_extraction(): - """Test the path extraction logic with sample contact data""" - - # Sample contact data from the debug output - sample_contacts = { - '460728508c17ef336412a223144d3a623215162682045c44fef7241af0161923': { - 'public_key': '460728508c17ef336412a223144d3a623215162682045c44fef7241af0161923', - 'out_path': '5f', - 'out_path_len': 1, - 'adv_name': 'HOWL' - }, - '5fbfb6c8937581fa427082a7ae60fa57aa59798ea39836c7e0aeccfc5f0274c3': { - 'public_key': '5fbfb6c8937581fa427082a7ae60fa57aa59798ea39836c7e0aeccfc5f0274c3', - 'out_path': '01', - 'out_path_len': 1, - 'adv_name': 'N7JMV-PAINE-RP' - }, - '7e7662676f7f0850a8a355baafbfc1eb7b4174c340442d7d7161c9474a2c9400': { - 'public_key': '7e7662676f7f0850a8a355baafbfc1eb7b4174c340442d7d7161c9474a2c9400', - 'out_path': '015f', - 'out_path_len': 2, - 'adv_name': 'WW7STR/PugetMesh Cougar' - }, - '15a24fcbc0dd2d2a4f80c7930cbb1de2139883bdd42b678afb19a4fa1ee1a6c8': { - 'public_key': '15a24fcbc0dd2d2a4f80c7930cbb1de2139883bdd42b678afb19a4fa1ee1a6c8', - 'out_path': '015f', - 'out_path_len': 2, - 'adv_name': 'Hillcrest Repeater' - }, - '2c3d703f6649e613639c12ba97a399d16ec9f112a16089eedaca3ad450566dc8': { - 'public_key': '2c3d703f6649e613639c12ba97a399d16ec9f112a16089eedaca3ad450566dc8', - 'out_path': '015fd0', - 'out_path_len': 3, - 'adv_name': 'Lower Capitol Hill' - }, - '1ffbd69aa03faadc0f40b2146665f2435fc8915b21e910e7125dfeba547646d9': { - 'public_key': '1ffbd69aa03faadc0f40b2146665f2435fc8915b21e910e7125dfeba547646d9', - 'out_path': '015f7e', - 'out_path_len': 3, - 'adv_name': 'First Hill Skyline' - }, - 'eaf3a101c6b2ff28b21f60439bfacbbfb7b24ad46c65761dc0ac0d76bcf888d3': { - 'public_key': 'eaf3a101c6b2ff28b21f60439bfacbbfb7b24ad46c65761dc0ac0d76bcf888d3', - 'out_path': '015f4a8094', - 'out_path_len': 5, - 'adv_name': 'Cowen West' - } - } - - def extract_path_info(pubkey_prefix, contacts): - """Extract path information from contacts using pubkey_prefix""" - path_info = "Unknown" - - for contact_key, contact_data in contacts.items(): - if contact_data.get('public_key', '').startswith(pubkey_prefix): - out_path = contact_data.get('out_path', '') - out_path_len = contact_data.get('out_path_len', -1) - - if out_path and out_path_len > 0: - # Convert hex path to readable node IDs using first 2 chars of pubkey - try: - path_bytes = bytes.fromhex(out_path) - path_nodes = [] - for i in range(0, len(path_bytes), 2): - if i + 1 < len(path_bytes): - node_id = int.from_bytes(path_bytes[i:i+2], byteorder='little') - # Convert to 2-character hex representation - path_nodes.append(f"{node_id:02x}") - - path_info = f"{','.join(path_nodes)} ({out_path_len} hops)" - print(f"Found path info: {path_info}") - except Exception as e: - print(f"Error converting path: {e}") - path_info = f"Path: {out_path} ({out_path_len} hops)" - break - elif out_path_len == 0: - path_info = "Direct" - print(f"Direct connection: {path_info}") - break - else: - path_info = "Unknown path" - print(f"No path info available: {path_info}") - break - - return path_info - - # Test with different pubkey prefixes - test_cases = [ - ('460728508c17', 'HOWL'), - ('5fbfb6c89375', 'N7JMV-PAINE-RP'), - ('7e7662676f7f', 'WW7STR/PugetMesh Cougar'), - ('15a24fcbc0dd', 'Hillcrest Repeater'), - ('2c3d703f6649', 'Lower Capitol Hill'), - ('1ffbd69aa03f', 'First Hill Skyline'), - ('eaf3a101c6b2', 'Cowen West'), - ('nonexistent', 'Non-existent contact') - ] - - print("Testing path extraction logic:") - print("=" * 50) - - for pubkey_prefix, expected_name in test_cases: - print(f"\nTesting {expected_name} (prefix: {pubkey_prefix}):") - path_info = extract_path_info(pubkey_prefix, sample_contacts) - print(f"Result: {path_info}") - -if __name__ == "__main__": - test_path_extraction() diff --git a/tests/test_path_info.py b/tests/test_path_info.py deleted file mode 100644 index d4ae0c5..0000000 --- a/tests/test_path_info.py +++ /dev/null @@ -1,123 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to investigate path information availability in meshcore-cli -""" - -import asyncio -import meshcore -from meshcore import EventType - -async def test_path_information(): - """Test to see what path information is available""" - print("Testing path information availability in meshcore-cli...") - - try: - # Create a BLE connection - mc = await meshcore.MeshCore.create_ble(debug=True) - print(f"Connected to: {mc.self_info}") - - # Wait for contacts to load - print("Waiting for contacts to load...") - await asyncio.sleep(10) - - # Try to manually request contacts if none loaded - if len(mc.contacts) == 0: - print("No contacts loaded automatically, trying manual request...") - try: - from meshcore_cli.meshcore_cli import next_cmd - result = await next_cmd(mc, ["contacts"]) - print(f"Manual contacts request result: {result}") - await asyncio.sleep(5) # Wait for contacts to process - except Exception as e: - print(f"Error requesting contacts: {e}") - - # Examine contacts for path information - print(f"\nExamining {len(mc.contacts)} contacts for path information:") - path_contacts = [] - for key, contact in mc.contacts.items(): - out_path = contact.get('out_path', '') - out_path_len = contact.get('out_path_len', -1) - if out_path and out_path_len > 0: - path_contacts.append({ - 'name': contact.get('adv_name', 'Unknown'), - 'pubkey': contact.get('public_key', '')[:16] + '...', - 'out_path': out_path, - 'out_path_len': out_path_len - }) - print(f" {contact.get('adv_name', 'Unknown')}: {out_path} ({out_path_len} hops)") - - print(f"\nFound {len(path_contacts)} contacts with path information") - - # Test path discovery for a specific contact - if path_contacts: - test_contact = path_contacts[0] - print(f"\nTesting path discovery for: {test_contact['name']}") - print(f" Path: {test_contact['out_path']}") - print(f" Hops: {test_contact['out_path_len']}") - - # Convert hex path to readable format - path_bytes = bytes.fromhex(test_contact['out_path']) - path_nodes = [] - for i in range(0, len(path_bytes), 2): - if i + 1 < len(path_bytes): - node_id = int.from_bytes(path_bytes[i:i+2], byteorder='little') - path_nodes.append(f"{node_id:04x}") - - print(f" Path nodes: {' -> '.join(path_nodes)}") - - # Set up event handlers to capture path information - path_events = [] - - async def on_path_update(event): - print(f"PATH_UPDATE event: {event.payload}") - path_events.append(('PATH_UPDATE', event.payload)) - - async def on_path_response(event): - print(f"PATH_RESPONSE event: {event.payload}") - path_events.append(('PATH_RESPONSE', event.payload)) - - async def on_channel_message(event): - print(f"CHANNEL_MSG_RECV event: {event.payload}") - print(f" Path length: {event.payload.get('path_len', 'N/A')}") - print(f" Sender: {event.payload.get('text', 'N/A')[:20]}...") - - # Try to look up path information from contacts - if hasattr(event, 'metadata') and event.metadata: - pubkey_prefix = event.metadata.get('pubkey_prefix', '') - if pubkey_prefix: - print(f" Looking for contact with pubkey_prefix: {pubkey_prefix}") - for key, contact in mc.contacts.items(): - if contact.get('public_key', '').startswith(pubkey_prefix): - out_path = contact.get('out_path', '') - out_path_len = contact.get('out_path_len', -1) - if out_path and out_path_len > 0: - print(f" Found path: {out_path} ({out_path_len} hops)") - # Convert to readable format - path_bytes = bytes.fromhex(out_path) - path_nodes = [] - for i in range(0, len(path_bytes), 2): - if i + 1 < len(path_bytes): - node_id = int.from_bytes(path_bytes[i:i+2], byteorder='little') - path_nodes.append(f"{node_id:04x}") - print(f" Path nodes: {' -> '.join(path_nodes)}") - break - - # Subscribe to events - mc.subscribe(EventType.PATH_UPDATE, on_path_update) - mc.subscribe(EventType.PATH_RESPONSE, on_path_response) - mc.subscribe(EventType.CHANNEL_MSG_RECV, on_channel_message) - - print("\nListening for messages and path events (30 seconds)...") - await asyncio.sleep(30) - - print(f"\nCaptured {len(path_events)} path events:") - for event_type, payload in path_events: - print(f" {event_type}: {payload}") - - await mc.disconnect() - - except Exception as e: - print(f"Error: {e}") - -if __name__ == "__main__": - asyncio.run(test_path_information()) diff --git a/tests/test_race_condition_fix.py b/tests/test_race_condition_fix.py deleted file mode 100644 index 97a593a..0000000 --- a/tests/test_race_condition_fix.py +++ /dev/null @@ -1,188 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to verify the race condition fixes in RF data correlation -""" - -import asyncio -import time -import sys -import os - -# Add the parent directory to the path so we can import modules -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from modules.message_handler import MessageHandler -from modules.core import MeshCoreBot -import configparser - -class MockBot: - """Mock bot for testing""" - def __init__(self): - self.logger = self - self.config = configparser.ConfigParser() - self.config.read('config.ini') - self.channel_manager = MockChannelManager() - - def info(self, msg): - print(f"INFO: {msg}") - - def debug(self, msg): - print(f"DEBUG: {msg}") - - def warning(self, msg): - print(f"WARNING: {msg}") - - def error(self, msg): - print(f"ERROR: {msg}") - -class MockChannelManager: - """Mock channel manager""" - def get_channel_name(self, channel_idx): - return f"TestChannel{channel_idx}" - -async def test_race_condition_fix(): - """Test the race condition fixes""" - print("Testing RF Data Correlation Race Condition Fixes") - print("=" * 50) - - # Create mock bot and message handler - bot = MockBot() - handler = MessageHandler(bot) - - print(f"RF Data Timeout: {handler.rf_data_timeout}s") - print(f"Message Timeout: {handler.message_timeout}s") - print(f"Enhanced Correlation: {handler.enhanced_correlation}") - print() - - # Test 1: Basic RF data storage and retrieval - print("Test 1: Basic RF data storage and retrieval") - import time - current_time = time.time() - - # Simulate RF data - rf_data = { - 'timestamp': current_time, - 'pubkey_prefix': 'f2981503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'snr': -3.5, - 'rssi': -104, - 'raw_hex': 'f2981503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'payload': '1503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'payload_length': 24, - 'routing_info': { - 'path_length': 3, - 'path_hex': '387e5f', - 'path_nodes': ['38', '7e', '5f'], - 'route_type': 'ROUTE_TYPE_DIRECT', - 'transport_size': 2, - 'payload_type': 'CHANNEL_ACK' - } - } - - # Store RF data - handler.recent_rf_data.append(rf_data) - handler.rf_data_by_timestamp[current_time] = rf_data - handler.rf_data_by_pubkey[rf_data['pubkey_prefix']] = [rf_data] - - # Test immediate correlation - found_data = handler.find_recent_rf_data(rf_data['pubkey_prefix']) - if found_data: - print("✅ Immediate correlation successful") - print(f" Found SNR: {found_data['snr']}, RSSI: {found_data['rssi']}") - else: - print("❌ Immediate correlation failed") - - print() - - # Test 2: Message correlation system - print("Test 2: Message correlation system") - - # Simulate a message payload - message_payload = { - 'channel_idx': 1, - 'text': 'Jade: Test', - 'pubkey_prefix': 'f2981503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'path_len': 3 - } - - # Store message for correlation - message_id = f"test_{int(time.time() * 1000)}" - handler.store_message_for_correlation(message_id, message_payload) - - # Try to correlate - correlated_data = handler.correlate_message_with_rf_data(message_id) - if correlated_data: - print("✅ Message correlation successful") - print(f" Correlated SNR: {correlated_data['snr']}, RSSI: {correlated_data['rssi']}") - else: - print("❌ Message correlation failed") - - print() - - # Test 3: Extended timeout correlation - print("Test 3: Extended timeout correlation") - - # Create older RF data (within extended timeout) - older_time = current_time - 20 # 20 seconds ago - older_rf_data = { - 'timestamp': older_time, - 'pubkey_prefix': 'f2981503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'snr': -5.0, - 'rssi': -110, - 'raw_hex': 'f2981503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'payload': '1503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', - 'payload_length': 24, - 'routing_info': None - } - - handler.recent_rf_data.append(older_rf_data) - - # Test with extended timeout - extended_data = handler.find_recent_rf_data('f2981503387e5fd5dfaeeb0cdb920b2c409387a92d77a5dc8706', max_age_seconds=30.0) - if extended_data: - print("✅ Extended timeout correlation successful") - print(f" Found SNR: {extended_data['snr']}, RSSI: {extended_data['rssi']}") - else: - print("❌ Extended timeout correlation failed") - - print() - - # Test 4: Partial pubkey matching - print("Test 4: Partial pubkey matching") - - # Test with partial pubkey - partial_pubkey = 'f2981503387e5fd5' # First 16 characters - partial_data = handler.find_recent_rf_data(partial_pubkey) - if partial_data: - print("✅ Partial pubkey matching successful") - print(f" Found SNR: {partial_data['snr']}, RSSI: {partial_data['rssi']}") - else: - print("❌ Partial pubkey matching failed") - - print() - - # Test 5: Cleanup functionality - print("Test 5: Cleanup functionality") - - # Add some old pending messages - old_time = time.time() - 15 # 15 seconds ago - handler.pending_messages['old_message'] = { - 'data': message_payload, - 'timestamp': old_time, - 'processed': False - } - - print(f"Pending messages before cleanup: {len(handler.pending_messages)}") - handler.cleanup_old_messages() - print(f"Pending messages after cleanup: {len(handler.pending_messages)}") - - if len(handler.pending_messages) == 1: # Only the test message should remain - print("✅ Cleanup functionality working") - else: - print("❌ Cleanup functionality failed") - - print() - print("Race Condition Fix Tests Complete!") - print("=" * 50) - -if __name__ == "__main__": - asyncio.run(test_race_condition_fix())