mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-04-01 21:15:38 +00:00
6c8151389b
- Updated the MeshGraph class to prevent promoting a 1-byte edge to a 3-byte edge when the existing 1-byte edge lacks public keys, ensuring accurate observation attribution. - Added a new test case to verify the behavior of edge promotion under specific conditions. - Modified the BotDataViewer API to return the prefix length dynamically based on the edges, improving data consistency and user experience in the web viewer. - Enhanced the mesh.html template to support displaying prefix byte counts, providing clearer information on node connections.
253 lines
11 KiB
Python
253 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for MeshGraph edge management
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
from tests.helpers import create_test_edge
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestMeshGraphEdges:
|
|
"""Test MeshGraph edge management functionality."""
|
|
|
|
def test_add_new_edge(self, mesh_graph):
|
|
"""Test adding a new edge to the graph."""
|
|
mesh_graph.add_edge('01', '7e')
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge is not None
|
|
assert edge['from_prefix'] == '01'
|
|
assert edge['to_prefix'] == '7e'
|
|
assert edge['observation_count'] == 1
|
|
assert edge['first_seen'] is not None
|
|
assert edge['last_seen'] is not None
|
|
|
|
def test_add_edge_with_public_keys(self, mesh_graph):
|
|
"""Test adding edge with public keys."""
|
|
from_key = '0101010101010101010101010101010101010101010101010101010101010101'
|
|
to_key = '7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e'
|
|
|
|
mesh_graph.add_edge('01', '7e', from_public_key=from_key, to_public_key=to_key)
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['from_public_key'] == from_key
|
|
assert edge['to_public_key'] == to_key
|
|
|
|
def test_add_edge_with_hop_position(self, mesh_graph):
|
|
"""Test adding edge with hop position."""
|
|
mesh_graph.add_edge('01', '7e', hop_position=2)
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['avg_hop_position'] == 2.0
|
|
|
|
def test_add_edge_with_geographic_distance(self, mesh_graph):
|
|
"""Test adding edge with geographic distance."""
|
|
mesh_graph.add_edge('01', '7e', geographic_distance=15.5)
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['geographic_distance'] == 15.5
|
|
|
|
def test_update_existing_edge(self, mesh_graph):
|
|
"""Test updating an existing edge increments observation count."""
|
|
mesh_graph.add_edge('01', '7e')
|
|
mesh_graph.add_edge('01', '7e')
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['observation_count'] == 2
|
|
|
|
def test_update_edge_hop_position_average(self, mesh_graph):
|
|
"""Test that updating edge recalculates average hop position."""
|
|
mesh_graph.add_edge('01', '7e', hop_position=1)
|
|
mesh_graph.add_edge('01', '7e', hop_position=3)
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
# Average should be (1 + 3) / 2 = 2.0
|
|
assert edge['avg_hop_position'] == 2.0
|
|
|
|
# Add another observation
|
|
mesh_graph.add_edge('01', '7e', hop_position=2)
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
# Average should be (2.0 * 2 + 2) / 3 = 2.0
|
|
assert edge['avg_hop_position'] == 2.0
|
|
|
|
def test_update_edge_public_keys(self, mesh_graph):
|
|
"""Test that updating edge can add missing public keys."""
|
|
# Add edge without keys
|
|
mesh_graph.add_edge('01', '7e')
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['from_public_key'] is None
|
|
assert edge['to_public_key'] is None
|
|
|
|
# Update with keys
|
|
from_key = '0101010101010101010101010101010101010101010101010101010101010101'
|
|
to_key = '7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e'
|
|
mesh_graph.add_edge('01', '7e', from_public_key=from_key, to_public_key=to_key)
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['from_public_key'] == from_key
|
|
assert edge['to_public_key'] == to_key
|
|
|
|
def test_update_edge_geographic_distance(self, mesh_graph):
|
|
"""Test that updating edge can update geographic distance."""
|
|
mesh_graph.add_edge('01', '7e', geographic_distance=10.0)
|
|
mesh_graph.add_edge('01', '7e', geographic_distance=15.5)
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['geographic_distance'] == 15.5
|
|
|
|
def test_get_edge_nonexistent(self, mesh_graph):
|
|
"""Test getting a non-existent edge returns None."""
|
|
edge = mesh_graph.get_edge('01', '99')
|
|
assert edge is None
|
|
|
|
def test_has_edge(self, mesh_graph):
|
|
"""Test has_edge method."""
|
|
assert mesh_graph.has_edge('01', '7e') is False
|
|
|
|
mesh_graph.add_edge('01', '7e')
|
|
assert mesh_graph.has_edge('01', '7e') is True
|
|
assert mesh_graph.has_edge('7e', '01') is False # Direction matters
|
|
|
|
def test_prefix_normalization(self, mesh_graph):
|
|
"""Test that prefixes are normalized to lowercase and truncated."""
|
|
mesh_graph.add_edge('01AB', '7EFF')
|
|
|
|
# Should normalize to '01' and '7e'
|
|
assert mesh_graph.has_edge('01', '7e') is True
|
|
assert mesh_graph.has_edge('01ab', '7eff') is True
|
|
assert mesh_graph.has_edge('01AB', '7EFF') is True
|
|
|
|
def test_get_outgoing_edges(self, mesh_graph):
|
|
"""Test getting all outgoing edges from a node."""
|
|
mesh_graph.add_edge('01', '7e')
|
|
mesh_graph.add_edge('01', '86')
|
|
mesh_graph.add_edge('7e', '01') # Reverse direction
|
|
|
|
outgoing = mesh_graph.get_outgoing_edges('01')
|
|
assert len(outgoing) == 2
|
|
prefixes = {edge['to_prefix'] for edge in outgoing}
|
|
assert '7e' in prefixes
|
|
assert '86' in prefixes
|
|
|
|
def test_get_incoming_edges(self, mesh_graph):
|
|
"""Test getting all incoming edges to a node."""
|
|
mesh_graph.add_edge('01', '7e')
|
|
mesh_graph.add_edge('86', '7e')
|
|
mesh_graph.add_edge('7e', '01') # Reverse direction
|
|
|
|
incoming = mesh_graph.get_incoming_edges('7e')
|
|
assert len(incoming) == 2
|
|
prefixes = {edge['from_prefix'] for edge in incoming}
|
|
assert '01' in prefixes
|
|
assert '86' in prefixes
|
|
|
|
def test_empty_prefix_handling(self, mesh_graph):
|
|
"""Test that empty prefixes are ignored."""
|
|
initial_count = len(mesh_graph.edges)
|
|
|
|
mesh_graph.add_edge('', '7e')
|
|
mesh_graph.add_edge('01', '')
|
|
mesh_graph.add_edge('', '')
|
|
|
|
assert len(mesh_graph.edges) == initial_count
|
|
|
|
def test_edge_last_seen_updates(self, mesh_graph):
|
|
"""Test that last_seen timestamp updates on edge updates."""
|
|
first_time = datetime.now() - timedelta(seconds=5)
|
|
|
|
# Manually set first_seen to past time
|
|
mesh_graph.add_edge('01', '7e')
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
edge['first_seen'] = first_time
|
|
edge['last_seen'] = first_time
|
|
|
|
# Wait a moment and update
|
|
import time
|
|
time.sleep(0.1)
|
|
mesh_graph.add_edge('01', '7e')
|
|
|
|
edge = mesh_graph.get_edge('01', '7e')
|
|
assert edge['last_seen'] > first_time
|
|
assert edge['first_seen'] == first_time # First seen shouldn't change
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestMeshGraphMultiByteMerge:
|
|
"""Test MeshGraph 1-byte uniqueness merge and 2/3-byte merge/promote (multi-byte node identity)."""
|
|
|
|
def test_one_byte_merge_when_unique(self, mesh_graph):
|
|
"""1-byte observation merges into the single matching 2-byte edge (unique link)."""
|
|
mesh_graph.add_edge('0c01', '0d42') # 2-byte edge first
|
|
assert len(mesh_graph.edges) == 1
|
|
mesh_graph.add_edge('0c', '0d') # 1-byte: only one match -> merge
|
|
assert len(mesh_graph.edges) == 1
|
|
edge = mesh_graph.get_edge('0c', '0d')
|
|
assert edge is not None
|
|
assert edge['from_prefix'] == '0c01'
|
|
assert edge['to_prefix'] == '0d42'
|
|
assert edge['observation_count'] == 2
|
|
|
|
def test_one_byte_no_merge_when_multiple(self, mesh_graph):
|
|
"""1-byte observation does not merge when multiple edges prefix-match (ambiguous)."""
|
|
mesh_graph.add_edge('0c01', '0d42')
|
|
mesh_graph.add_edge('0c99', '0dee')
|
|
assert len(mesh_graph.edges) == 2
|
|
mesh_graph.add_edge('0c', '0d') # 1-byte: two matches -> create separate 1-byte edge
|
|
assert len(mesh_graph.edges) == 3
|
|
# The 1-byte edge (0c, 0d) should exist with count 1
|
|
one_byte_edge = mesh_graph.edges.get(('0c', '0d'))
|
|
assert one_byte_edge is not None
|
|
assert one_byte_edge['observation_count'] == 1
|
|
# 2-byte edges unchanged
|
|
assert mesh_graph.get_edge('0c01', '0d42')['observation_count'] == 1
|
|
assert mesh_graph.get_edge('0c99', '0dee')['observation_count'] == 1
|
|
|
|
def test_two_byte_merges_into_three_byte(self, mesh_graph):
|
|
"""2-byte observation merges into existing 3-byte edge (more specific)."""
|
|
mesh_graph.add_edge('0101c1', '8611ab') # 3-byte; to_prefix must start with 8611 for prefix_match
|
|
mesh_graph.add_edge('0101', '8611') # 2-byte: best match is 3-byte -> update that
|
|
assert len(mesh_graph.edges) == 1
|
|
edge = mesh_graph.get_edge('0101', '8611')
|
|
assert edge['from_prefix'] == '0101c1'
|
|
assert edge['to_prefix'] == '8611ab'
|
|
assert edge['observation_count'] == 2
|
|
|
|
def test_promote_one_byte_to_three_byte(self, mesh_graph):
|
|
"""When 3-byte observation follows 1-byte and 1-byte has no public_key, do NOT promote: keep 1-byte edge and merge count (preserves other e0 nodes)."""
|
|
mesh_graph.add_edge('01', '86') # 1-byte, no public_key
|
|
assert len(mesh_graph.edges) == 1
|
|
assert ('01', '86') in mesh_graph.edges
|
|
mesh_graph.add_edge('0101c1', '86ab12') # 3-byte: would promote, but 1-byte has no public_key -> merge into 1-byte
|
|
assert len(mesh_graph.edges) == 1
|
|
assert ('01', '86') in mesh_graph.edges
|
|
edge = mesh_graph.get_edge('01', '86')
|
|
assert edge is not None
|
|
assert edge['from_prefix'] == '01'
|
|
assert edge['to_prefix'] == '86'
|
|
assert edge['observation_count'] == 2
|
|
|
|
def test_promote_one_byte_to_three_byte_when_1byte_has_public_key(self, mesh_graph):
|
|
"""When 3-byte observation follows 1-byte and 1-byte edge has public_key, promote: remove 1-byte, add 3-byte with merged count."""
|
|
mesh_graph.add_edge('01', '86', from_public_key='01' * 32, to_public_key='86' * 32) # 1-byte with keys
|
|
assert len(mesh_graph.edges) == 1
|
|
assert ('01', '86') in mesh_graph.edges
|
|
mesh_graph.add_edge('0101c1', '86ab12') # 3-byte: best match has public_key -> promote
|
|
assert len(mesh_graph.edges) == 1
|
|
assert ('01', '86') not in mesh_graph.edges
|
|
edge = mesh_graph.get_edge('01', '86')
|
|
assert edge is not None
|
|
assert edge['from_prefix'] == '0101c1'
|
|
assert edge['to_prefix'] == '86ab12'
|
|
assert edge['observation_count'] == 2
|
|
|
|
def test_get_edge_returns_three_byte_when_present(self, mesh_graph):
|
|
"""get_edge(01, 86) returns the 3-byte edge when it exists (prefix match prefers best)."""
|
|
mesh_graph.add_edge('0101c1', '86ab12')
|
|
edge = mesh_graph.get_edge('01', '86')
|
|
assert edge is not None
|
|
assert edge['from_prefix'] == '0101c1'
|
|
assert edge['to_prefix'] == '86ab12'
|
|
assert edge['observation_count'] == 1
|