dandri/meshcore-bot
1
0
Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-08-06 18:59:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move tests directory to local docs folder

Browse Source 
- 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
This commit is contained in:
agessaman
2025-09-06 11:50:05 -07:00
parent 8803070d78
commit c21743137c
20 changed files with 2 additions and 2414 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -121,5 +121,6 @@ config.ini
*.db
mctomqtt.py
# Local documentation (internal development notes)
# Local documentation and development files
docs/local/
tests/
tests/README.md
-56
View File
@@ -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.
tests/__init__.py
-1
View File
@@ -1 +0,0 @@
# Tests package for MeshCore Bot
tests/discover_ble_uuids.py
-185
View File
@@ -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())
tests/example_usage.py
-336
View File
@@ -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 <message>',
'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!")
tests/meshcore_packet_analyzer.py
-214
View File
@@ -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('<I', data[4:8])[0]
fields['length_interpretation'] = length
except:
pass
# Look for text-like data
text_candidates = []
for i in range(len(data)):
if 32 <= data[i] <= 126: # Printable ASCII
text_candidates.append(chr(data[i]))
else:
if text_candidates:
text = ''.join(text_candidates)
if len(text) >= 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()
tests/repeater_management_demo.py
-158
View File
@@ -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())
tests/test_ble_connection.py
-201
View File
@@ -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())
tests/test_channels.py
-57
View File
@@ -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())
tests/test_config.py
-164
View File
@@ -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!")
tests/test_config_parsing.py
-33
View File
@@ -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()
tests/test_contacts.py
-102
View File
@@ -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())
tests/test_dynamic_channels.py
-60
View File
@@ -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())
tests/test_event_structure.py
-58
View File
@@ -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())
tests/test_get_channel.py
-65
View File
@@ -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())
tests/test_installation.py
-251
View File
@@ -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()
tests/test_meshcore_official.py
-48
View File
@@ -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())
tests/test_path_extraction.py
-113
View File
@@ -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()
tests/test_path_info.py
-123
View File
@@ -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())
tests/test_race_condition_fix.py
-188
View File
@@ -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())