From 89651b01dbe137d7ea5eff67f987c5bfbb6c004a Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 12 Jul 2026 07:52:51 -0700 Subject: [PATCH] feat(web-viewer): multi-byte evidence mode for the mesh graph Single-byte repeater identity guessing has been persistently unreliable, and mesh_connections flattens away evidence provenance (confirmed_2byte never persists). observed_paths retains it: bytes_per_hop marks which paths carry unambiguous multi-byte hops. - /api/mesh/edges?evidence=multibyte derives edges purely from unique multi-byte path observations (consecutive hop pairs, aggregated with a distinct-path count); 2-byte edges coalesce into a 3-byte edge only on a unique prefix match, mirroring MeshGraph.add_edge - default mode tags each edge evidence=multibyte|singlebyte by key length - Evidence filter on the mesh page (persisted with the other filters); single-byte edges render dashed in map and graph views, with an evidence line in tooltips and a legend entry - client-side haversine fallback for edges without a stored distance - prefix-compatible edge/node matching so degree sizing and node details work when edge and node prefixes differ in resolution - /api/mesh/stats reports multibyte_edges; shown on the Edges stat card Co-Authored-By: Claude Fable 5 --- modules/web_viewer/app.py | 131 ++++++++++++++++++++++++- modules/web_viewer/templates/mesh.html | 126 ++++++++++++++++++++---- tests/test_web_viewer_app.py | 124 +++++++++++++++++++++++ 3 files changed, 361 insertions(+), 20 deletions(-) diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 35dbe09..4b1b26b 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -464,6 +464,105 @@ class BotDataViewer: finally: conn.close() + def _derive_multibyte_evidence_edges( + self, days: int | None = None, min_observations: int | None = None + ) -> list[dict[str, Any]]: + """Derive mesh edges purely from multi-byte path evidence. + + Splits each observed_paths row with bytes_per_hop >= 2 into consecutive + hop pairs and aggregates per directed pair. Unlike mesh_connections, this + never mixes in single-byte observations, so edge identity is unambiguous + (up to 2/3-byte prefix collisions, which are rare). + + Edges observed at 2-byte resolution are coalesced into a 3-byte edge when + exactly one 3-byte edge prefix-matches both endpoints — the same + unique-match rule MeshGraph.add_edge applies at write time. + + Returns edge dicts matching the /api/mesh/edges schema, plus: + path_count — number of distinct observed paths crossing the edge + evidence — always 'multibyte' + """ + query = ''' + SELECT path_hex, bytes_per_hop, observation_count, first_seen, last_seen + FROM observed_paths + WHERE bytes_per_hop >= 2 + ''' + params: list[Any] = [] + if days is not None: + query += ' AND last_seen >= datetime("now", "-" || ? || " days")' + params.append(days) + + with self._with_db_connection() as conn: + rows = conn.execute(query, params).fetchall() + + edges: dict[tuple[str, str], dict[str, Any]] = {} + for row in rows: + step = (row['bytes_per_hop'] or 0) * 2 + path_hex = (row['path_hex'] or '').lower() + if step < 4 or not path_hex or len(path_hex) % step != 0: + continue + hops = [path_hex[i:i + step] for i in range(0, len(path_hex), step)] + obs = row['observation_count'] or 1 + for i in range(len(hops) - 1): + key = (hops[i], hops[i + 1]) + agg = edges.get(key) + if agg is None: + edges[key] = agg = { + 'observation_count': 0, + 'path_count': 0, + 'first_seen': row['first_seen'], + 'last_seen': row['last_seen'], + 'hop_position_sum': 0.0, + } + agg['observation_count'] += obs + agg['path_count'] += 1 + if row['first_seen'] and (agg['first_seen'] is None or row['first_seen'] < agg['first_seen']): + agg['first_seen'] = row['first_seen'] + if row['last_seen'] and (agg['last_seen'] is None or row['last_seen'] > agg['last_seen']): + agg['last_seen'] = row['last_seen'] + # hop_position is the 1-based index of the receiving hop (matches + # graph_trace_helper semantics); weighted by observation count. + agg['hop_position_sum'] += (i + 1) * obs + + # Coalesce 2-byte edges into a 3-byte edge when exactly one matches. + # (Hops within a path share one resolution, so keys are homogeneous.) + by_truncated_key: dict[tuple[str, str], list[tuple[str, str]]] = {} + for key in edges: + if len(key[0]) == 6: + by_truncated_key.setdefault((key[0][:4], key[1][:4]), []).append(key) + for key in [k for k in edges if len(k[0]) == 4]: + candidates = by_truncated_key.get(key, []) + if len(candidates) == 1: + target = edges[candidates[0]] + source = edges.pop(key) + target['observation_count'] += source['observation_count'] + target['path_count'] += source['path_count'] + target['hop_position_sum'] += source['hop_position_sum'] + if source['first_seen'] and (target['first_seen'] is None or source['first_seen'] < target['first_seen']): + target['first_seen'] = source['first_seen'] + if source['last_seen'] and (target['last_seen'] is None or source['last_seen'] > target['last_seen']): + target['last_seen'] = source['last_seen'] + + result = [] + for (from_prefix, to_prefix), agg in edges.items(): + if min_observations is not None and agg['observation_count'] < min_observations: + continue + result.append({ + 'from_prefix': from_prefix, + 'to_prefix': to_prefix, + 'from_public_key': None, + 'to_public_key': None, + 'observation_count': agg['observation_count'], + 'path_count': agg['path_count'], + 'first_seen': agg['first_seen'], + 'last_seen': agg['last_seen'], + 'avg_hop_position': agg['hop_position_sum'] / agg['observation_count'], + 'geographic_distance': None, + 'evidence': 'multibyte', + }) + result.sort(key=lambda e: e['last_seen'] or '', reverse=True) + return result + def _resolve_path(self, path_input: str) -> dict[str, Any]: """Resolve a hex path to repeater names/locations for the mesh map. @@ -2018,7 +2117,12 @@ class BotDataViewer: @self.app.route('/api/mesh/edges') def api_mesh_edges(): - """Get all graph edges with metadata""" + """Get all graph edges with metadata. + + evidence=multibyte derives edges purely from unique multi-byte path + observations (observed_paths, bytes_per_hop >= 2), bypassing the + mesh_connections merge heuristics that single-byte evidence feeds into. + """ conn = None try: # Get optional query parameters @@ -2026,6 +2130,20 @@ class BotDataViewer: days = request.args.get('days', type=int) min_distance = request.args.get('min_distance', type=float) max_distance = request.args.get('max_distance', type=float) + evidence = request.args.get('evidence', 'all') + + if evidence == 'multibyte': + edges = self._derive_multibyte_evidence_edges( + days=days, min_observations=min_observations + ) + prefix_hex_chars = max( + (len(e['from_prefix']) for e in edges), default=2 + ) + return jsonify({ + 'edges': edges, + 'prefix_hex_chars': max(2, prefix_hex_chars), + 'evidence': 'multibyte', + }) conn = self._get_db_connection() cursor = conn.cursor() @@ -2072,6 +2190,10 @@ class BotDataViewer: for row in rows: fp, tp = row['from_prefix'], row['to_prefix'] prefix_hex_chars = max(prefix_hex_chars, len(fp) if fp else 0, len(tp) if tp else 0) + # Edges keyed at 4+ hex chars were necessarily created (or promoted) + # by a multi-byte path observation; 2-char keys carry only ambiguous + # single-byte evidence. + is_multibyte = bool(fp) and bool(tp) and len(fp) >= 4 and len(tp) >= 4 edges.append({ 'from_prefix': fp.lower() if fp else '', 'to_prefix': tp.lower() if tp else '', @@ -2081,7 +2203,8 @@ class BotDataViewer: 'first_seen': row['first_seen'], 'last_seen': row['last_seen'], 'avg_hop_position': row['avg_hop_position'], - 'geographic_distance': row['geographic_distance'] + 'geographic_distance': row['geographic_distance'], + 'evidence': 'multibyte' if is_multibyte else 'singlebyte' }) return jsonify({'edges': edges, 'prefix_hex_chars': prefix_hex_chars or 2}) @@ -2123,7 +2246,8 @@ class BotDataViewer: MAX(geographic_distance) as max_distance, COUNT(CASE WHEN from_public_key IS NOT NULL THEN 1 END) as edges_with_from_key, COUNT(CASE WHEN to_public_key IS NOT NULL THEN 1 END) as edges_with_to_key, - COUNT(CASE WHEN from_public_key IS NOT NULL AND to_public_key IS NOT NULL THEN 1 END) as edges_with_both_keys + COUNT(CASE WHEN from_public_key IS NOT NULL AND to_public_key IS NOT NULL THEN 1 END) as edges_with_both_keys, + COUNT(CASE WHEN LENGTH(from_prefix) >= 4 AND LENGTH(to_prefix) >= 4 THEN 1 END) as multibyte_edges FROM mesh_connections ''') edge_stats = cursor.fetchone() @@ -2169,6 +2293,7 @@ class BotDataViewer: 'edges_with_from_key': edge_stats['edges_with_from_key'] or 0, 'edges_with_to_key': edge_stats['edges_with_to_key'] or 0, 'edges_with_both_keys': edge_stats['edges_with_both_keys'] or 0, + 'multibyte_edges': edge_stats['multibyte_edges'] or 0, 'top_connected': [{'prefix': prefix, 'count': count} for prefix, count in top_connected], 'recent_edges_24h': recent_edges } diff --git a/modules/web_viewer/templates/mesh.html b/modules/web_viewer/templates/mesh.html index f779974..b453eb9 100644 --- a/modules/web_viewer/templates/mesh.html +++ b/modules/web_viewer/templates/mesh.html @@ -74,7 +74,7 @@
Edges
0
-
Total Connections
+
Total Connections
@@ -182,6 +182,13 @@ +
+ + +
@@ -191,9 +198,9 @@
-
+
- +
@@ -227,6 +234,15 @@ Bidirectional +
+
+ + Multi-byte evidence +
+
+ + Single-byte (uncertain) +
@@ -503,6 +519,55 @@ return filteredNodeMap[prefix] || filteredNodes.find(n => n.prefix.startsWith(prefix) || prefix.startsWith(n.prefix)) || null; } + // Prefix-compatible equality across resolutions (2, 4, or 6 hex chars): + // equal, or one is a prefix of the other. Needed wherever edges and nodes + // may carry different resolutions (e.g. 4-char edge vs 6-char node prefix). + function prefixesCompatible(a, b) { + if (!a || !b) return false; + return a === b || a.startsWith(b) || b.startsWith(a); + } + + function edgeTouchesNode(edge, node) { + return prefixesCompatible(edge.from_prefix, node.prefix) || prefixesCompatible(edge.to_prefix, node.prefix); + } + + // Great-circle distance fallback for edges without a stored distance + // (multi-byte derived edges carry none; compute from resolved node coords) + function haversineKm(lat1, lon1, lat2, lon2) { + const toRad = deg => deg * Math.PI / 180; + const dLat = toRad(lat2 - lat1); + const dLon = toRad(lon2 - lon1); + const a = Math.sin(dLat / 2) ** 2 + + Math.cos(toRad(lat1)) * Math.cos(toRad(lat2)) * Math.sin(dLon / 2) ** 2; + return 6371 * 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a)); + } + + function getEdgeDistance(edge, fromNode, toNode) { + if (edge.geographic_distance !== null && edge.geographic_distance !== undefined) { + return edge.geographic_distance; + } + if (fromNode && toNode) { + return haversineKm(fromNode.latitude, fromNode.longitude, toNode.latitude, toNode.longitude); + } + return null; + } + + function isSingleByteEdge(edge) { + return edge.evidence === 'singlebyte'; + } + + function getEvidenceLabel(edge) { + if (edge.evidence === 'multibyte') { + return edge.path_count + ? `Multi-byte (${edge.path_count} unique path${edge.path_count === 1 ? '' : 's'})` + : 'Multi-byte confirmed'; + } + if (edge.evidence === 'singlebyte') { + return 'Single-byte (identity uncertain)'; + } + return null; + } + // Detect if input is a hex path (2+ hex values) function detectPathInput(input) { if (!input || input.trim().length === 0) return false; @@ -914,6 +979,10 @@ document.getElementById('stat-node-count').textContent = stats.node_count || 0; document.getElementById('stat-edge-count').textContent = stats.total_edges || 0; + if (stats.multibyte_edges !== undefined) { + document.getElementById('stat-edge-detail').textContent = + `${stats.multibyte_edges} multi-byte confirmed`; + } document.getElementById('stat-avg-obs').textContent = stats.avg_observations || 0; document.getElementById('stat-avg-dist').textContent = stats.avg_distance ? stats.avg_distance + ' km' : 'N/A'; @@ -945,7 +1014,9 @@ try { // Load edges first so we can pass prefix_hex_chars to nodes (keeps node prefix in sync with edge prefixes) - const edgesResponse = await fetch('/api/mesh/edges'); + const evidence = document.getElementById('filter-evidence').value; + const edgesUrl = evidence === 'multibyte' ? '/api/mesh/edges?evidence=multibyte' : '/api/mesh/edges'; + const edgesResponse = await fetch(edgesUrl); const edgesData = await edgesResponse.json(); allEdges = edgesData.edges || []; graphPrefixHexChars = (edgesData.prefix_hex_chars && [2, 4, 6].includes(edgesData.prefix_hex_chars)) @@ -992,7 +1063,8 @@ minObs: document.getElementById('filter-min-obs').value, edgeDays: document.getElementById('filter-edge-days').value, nodeDays: document.getElementById('filter-node-days').value, - starredOnly: document.getElementById('filter-starred').checked + starredOnly: document.getElementById('filter-starred').checked, + evidence: document.getElementById('filter-evidence').value // Note: searchTerm is NOT saved - it should not persist across reloads }; localStorage.setItem('meshGraphFilters', JSON.stringify(filters)); @@ -1020,6 +1092,9 @@ if (filters.starredOnly !== undefined) { document.getElementById('filter-starred').checked = filters.starredOnly; } + if (filters.evidence === 'all' || filters.evidence === 'multibyte') { + document.getElementById('filter-evidence').value = filters.evidence; + } // Note: searchTerm is NOT loaded - it should reset on page reload } } catch (error) { @@ -1101,6 +1176,13 @@ } } + // Evidence mode is a server-side parameter, so changing it re-fetches data + // (loadData applies the current filters and re-renders when done) + function onEvidenceChange() { + saveFilters(); + loadData().catch(error => console.error('Error reloading for evidence mode:', error)); + } + // Update filter label function updateFilterLabel(inputId, labelId) { const value = document.getElementById(inputId).value; @@ -1331,22 +1413,26 @@ const color = getEdgeColor(edge); const weight = Math.max(1, Math.log10(edge.observation_count) * 2); - + const edgeDistance = getEdgeDistance(edge, fromNode, toNode); + const evidenceLabel = getEvidenceLabel(edge); + const line = L.polyline( [[fromNode.latitude, fromNode.longitude], [toNode.latitude, toNode.longitude]], { color: color, weight: weight, opacity: getEdgeOpacity(edge), + dashArray: isSingleByteEdge(edge) ? '6 6' : null, // Dash ambiguous single-byte evidence pane: 'overlayPane' // Use overlayPane for edges (lower z-index) } ); - + // Simple hover tooltip for quick info line.bindTooltip(` ${formatPrefixWithBytes(edge.from_prefix)} → ${formatPrefixWithBytes(edge.to_prefix)}
Observations: ${edge.observation_count}
- Distance: ${edge.geographic_distance ? edge.geographic_distance.toFixed(1) + ' km' : 'N/A'}
+ ${evidenceLabel ? 'Evidence: ' + evidenceLabel + '
' : ''} + Distance: ${edgeDistance ? edgeDistance.toFixed(1) + ' km' : 'N/A'}
Last Seen: ${edge.last_seen ? new Date(edge.last_seen).toLocaleString() : 'Never'} `); @@ -1370,9 +1456,13 @@ Observations: ${edge.observation_count} + ${evidenceLabel ? ` + Evidence: + ${evidenceLabel} + ` : ''} Distance: - ${edge.geographic_distance ? edge.geographic_distance.toFixed(2) + ' km' : 'N/A'} + ${edgeDistance ? edgeDistance.toFixed(2) + ' km' : 'N/A'} Avg Hop Position: @@ -1657,6 +1747,7 @@ // Only add if we haven't seen this edge ID before if (!edgeMapById.has(edgeId)) { + const evidenceLabel = getEvidenceLabel(edge); edgeMapById.set(edgeId, { id: edgeId, // Use unique IDs for edge ID from: fromNodeId, // Use unique node ID @@ -1666,8 +1757,10 @@ color: getEdgeColor(edge), opacity: getEdgeOpacity(edge) }, + dashes: isSingleByteEdge(edge), // Dash ambiguous single-byte evidence arrows: 'to', title: `${edge.from_prefix} → ${edge.to_prefix} (${edge.observation_count} obs)` + + (evidenceLabel ? ` — ${evidenceLabel}` : '') }); } }); @@ -2353,9 +2446,9 @@ function getNodeSize(node) { // Calculate degree (number of connections) using current filteredEdges // This ensures node size updates when filters change - const degree = filteredEdges.filter(e => - e.from_prefix === node.prefix || e.to_prefix === node.prefix - ).length; + // Prefix-compatible matching so degree counts work when edges and nodes + // carry different resolutions (e.g. 4-char edge vs 6-char node prefix) + const degree = filteredEdges.filter(e => edgeTouchesNode(e, node)).length; // Base size 6, scale with degree, cap at 16 return Math.max(6, Math.min(16, 6 + degree * 0.5)); } @@ -2406,11 +2499,10 @@ } function showNodeDetails(node) { - const connections = filteredEdges.filter(e => - e.from_prefix === node.prefix || e.to_prefix === node.prefix - ); - const incoming = connections.filter(e => e.to_prefix === node.prefix).length; - const outgoing = connections.filter(e => e.from_prefix === node.prefix).length; + // Prefix-compatible matching across resolutions (see edgeTouchesNode) + const connections = filteredEdges.filter(e => edgeTouchesNode(e, node)); + const incoming = connections.filter(e => prefixesCompatible(e.to_prefix, node.prefix)).length; + const outgoing = connections.filter(e => prefixesCompatible(e.from_prefix, node.prefix)).length; document.getElementById('nodeDetailsTitle').textContent = node.name; document.getElementById('nodeDetailsBody').innerHTML = ` diff --git a/tests/test_web_viewer_app.py b/tests/test_web_viewer_app.py index b64be7b..90dc1c1 100644 --- a/tests/test_web_viewer_app.py +++ b/tests/test_web_viewer_app.py @@ -373,6 +373,130 @@ class TestApiExportPaths: assert response.status_code == 200 +# --------------------------------------------------------------------------- +# api_mesh_edges (evidence modes) +# --------------------------------------------------------------------------- + + +def _seed_observed_path(db_path, path_hex, bytes_per_hop, observation_count=1, + last_seen=None, packet_type='advert'): + """Insert an observed_paths row; from/to prefixes derived from the path.""" + hex_chars = bytes_per_hop * 2 + hops = [path_hex[i:i + hex_chars] for i in range(0, len(path_hex), hex_chars)] + ts = last_seen or time.strftime('%Y-%m-%dT%H:%M:%S') + with sqlite3.connect(db_path, timeout=60) as conn: + conn.execute(""" + INSERT INTO observed_paths + (from_prefix, to_prefix, path_hex, path_length, bytes_per_hop, + packet_type, observation_count, first_seen, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, (hops[0], hops[-1], path_hex, len(path_hex) // 2, bytes_per_hop, + packet_type, observation_count, ts, ts)) + conn.commit() + + +class TestApiMeshEdgesEvidence: + def test_default_mode_tags_evidence_by_key_length(self, viewer_with_db): + with sqlite3.connect(viewer_with_db.db_path, timeout=60) as conn: + conn.execute(""" + INSERT INTO mesh_connections (from_prefix, to_prefix, observation_count) + VALUES ('aa', 'bb', 5), ('aabb', 'ccdd', 3) + """) + conn.commit() + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/edges') + assert response.status_code == 200 + edges = {(e['from_prefix'], e['to_prefix']): e + for e in json.loads(response.data)['edges']} + assert edges[('aa', 'bb')]['evidence'] == 'singlebyte' + assert edges[('aabb', 'ccdd')]['evidence'] == 'multibyte' + + def test_multibyte_mode_derives_consecutive_pairs(self, viewer_with_db): + # 2-byte path with 3 hops -> two directed edges; 1-byte row excluded + _seed_observed_path(viewer_with_db.db_path, 'aaaabbbbcccc', 2, observation_count=2) + _seed_observed_path(viewer_with_db.db_path, 'ddee', 1, observation_count=9) + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/edges?evidence=multibyte') + assert response.status_code == 200 + data = json.loads(response.data) + assert data['evidence'] == 'multibyte' + edges = {(e['from_prefix'], e['to_prefix']): e for e in data['edges']} + assert set(edges) == {('aaaa', 'bbbb'), ('bbbb', 'cccc')} + edge = edges[('aaaa', 'bbbb')] + assert edge['observation_count'] == 2 + assert edge['path_count'] == 1 + assert edge['evidence'] == 'multibyte' + assert edge['avg_hop_position'] == 1 + assert edges[('bbbb', 'cccc')]['avg_hop_position'] == 2 + + def test_multibyte_mode_coalesces_unique_lower_resolution(self, viewer_with_db): + # One 3-byte edge plus a 2-byte observation of the same link: + # unique prefix match, so the 2-byte counts merge into the 3-byte edge + _seed_observed_path(viewer_with_db.db_path, 'aaaa11bbbb22', 3, observation_count=4) + _seed_observed_path(viewer_with_db.db_path, 'aaaabbbb', 2, observation_count=3) + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/edges?evidence=multibyte') + data = json.loads(response.data) + edges = {(e['from_prefix'], e['to_prefix']): e for e in data['edges']} + assert set(edges) == {('aaaa11', 'bbbb22')} + assert edges[('aaaa11', 'bbbb22')]['observation_count'] == 7 + assert edges[('aaaa11', 'bbbb22')]['path_count'] == 2 + + def test_multibyte_mode_keeps_ambiguous_resolutions_separate(self, viewer_with_db): + # Two distinct 3-byte edges share the same 4-char truncation: + # the 2-byte observation is ambiguous and must stay its own edge + _seed_observed_path(viewer_with_db.db_path, 'aaaa11bbbb22', 3) + _seed_observed_path(viewer_with_db.db_path, 'aaaa33bbbb44', 3) + _seed_observed_path(viewer_with_db.db_path, 'aaaabbbb', 2) + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/edges?evidence=multibyte') + data = json.loads(response.data) + keys = {(e['from_prefix'], e['to_prefix']) for e in data['edges']} + assert keys == {('aaaa11', 'bbbb22'), ('aaaa33', 'bbbb44'), ('aaaa', 'bbbb')} + + def test_multibyte_mode_min_observations_applied_after_merge(self, viewer_with_db): + _seed_observed_path(viewer_with_db.db_path, 'aaaa11bbbb22', 3, observation_count=2) + _seed_observed_path(viewer_with_db.db_path, 'aaaabbbb', 2, observation_count=2) + _seed_observed_path(viewer_with_db.db_path, 'cccc55dddd66', 3, observation_count=1) + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/edges?evidence=multibyte&min_observations=4') + data = json.loads(response.data) + keys = {(e['from_prefix'], e['to_prefix']) for e in data['edges']} + # merged edge has 4 observations and survives; the other has 1 + assert keys == {('aaaa11', 'bbbb22')} + + def test_multibyte_mode_days_filter(self, viewer_with_db): + _seed_observed_path(viewer_with_db.db_path, 'aaaa11bbbb22', 3, + last_seen='2020-01-01T00:00:00') + _seed_observed_path(viewer_with_db.db_path, 'cccc55dddd66', 3) + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/edges?evidence=multibyte&days=7') + data = json.loads(response.data) + keys = {(e['from_prefix'], e['to_prefix']) for e in data['edges']} + assert keys == {('cccc55', 'dddd66')} + + def test_stats_include_multibyte_edge_count(self, viewer_with_db): + with sqlite3.connect(viewer_with_db.db_path, timeout=60) as conn: + conn.execute(""" + INSERT INTO mesh_connections (from_prefix, to_prefix, observation_count) + VALUES ('aa', 'bb', 5), ('aabb', 'ccdd', 3), ('aabb11', 'ccdd22', 1) + """) + conn.commit() + + with viewer_with_db.app.test_client() as client: + response = client.get('/api/mesh/stats') + assert response.status_code == 200 + stats = json.loads(response.data) + assert stats['total_edges'] == 3 + assert stats['multibyte_edges'] == 2 + + # --------------------------------------------------------------------------- # api_geocode_contact # ---------------------------------------------------------------------------