mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-07-21 09:02:54 +02:00
feat: implement traceroute app packet handling across the stack (#463)
* feat: implement traceroute app packet handling across the stack * run linter * tests: fix * Spec: add more unit tests
This commit is contained in:
l5y
@@ -777,6 +777,44 @@ module PotatoMesh
|
||||
|
||||
private :resolve_numeric_metric
|
||||
|
||||
# Normalise a traceroute hop entry to a numeric node identifier.
|
||||
#
|
||||
# @param hop [Object] raw hop entry from the payload.
|
||||
# @return [Integer, nil] coerced node ID or nil when the value is unusable.
|
||||
def coerce_trace_node_id(hop)
|
||||
case hop
|
||||
when Integer
|
||||
return hop
|
||||
when Numeric
|
||||
return hop.to_i
|
||||
when String
|
||||
trimmed = hop.strip
|
||||
return nil if trimmed.empty?
|
||||
return Integer(trimmed, 10) if trimmed.match?(/\A-?\d+\z/)
|
||||
|
||||
parts = canonical_node_parts(trimmed)
|
||||
return parts[1] if parts
|
||||
when Hash
|
||||
candidate = hop["node_id"] || hop[:node_id] || hop["id"] || hop[:id] || hop["num"] || hop[:num]
|
||||
return coerce_trace_node_id(candidate)
|
||||
end
|
||||
|
||||
nil
|
||||
rescue ArgumentError
|
||||
nil
|
||||
end
|
||||
|
||||
# Extract hop identifiers from a traceroute payload preserving order.
|
||||
#
|
||||
# @param hops_value [Object] raw hops array or path collection.
|
||||
# @return [Array<Integer>] ordered list of coerced hop identifiers.
|
||||
def normalize_trace_hops(hops_value)
|
||||
return [] if hops_value.nil?
|
||||
|
||||
hop_entries = hops_value.is_a?(Array) ? hops_value : [hops_value]
|
||||
hop_entries.filter_map { |entry| coerce_trace_node_id(entry) }
|
||||
end
|
||||
|
||||
def insert_telemetry(db, payload)
|
||||
return unless payload.is_a?(Hash)
|
||||
|
||||
@@ -1206,6 +1244,74 @@ module PotatoMesh
|
||||
})
|
||||
end
|
||||
|
||||
# Persist a traceroute observation and its hop path.
|
||||
#
|
||||
# @param db [SQLite3::Database] open database handle.
|
||||
# @param payload [Hash] traceroute payload as produced by the ingestor.
|
||||
# @return [void]
|
||||
def insert_trace(db, payload)
|
||||
return unless payload.is_a?(Hash)
|
||||
|
||||
trace_identifier = coerce_integer(payload["id"] || payload["packet_id"] || payload["packetId"])
|
||||
trace_identifier ||= coerce_integer(payload["trace_id"])
|
||||
request_id = coerce_integer(payload["request_id"] || payload["req"])
|
||||
trace_identifier ||= request_id
|
||||
|
||||
now = Time.now.to_i
|
||||
rx_time = coerce_integer(payload["rx_time"])
|
||||
rx_time = now if rx_time.nil? || rx_time > now
|
||||
rx_iso = string_or_nil(payload["rx_iso"]) || Time.at(rx_time).utc.iso8601
|
||||
|
||||
metrics = normalize_json_object(payload["metrics"])
|
||||
src = coerce_integer(payload["src"] || payload["source"] || payload["from"])
|
||||
dest = coerce_integer(payload["dest"] || payload["destination"] || payload["to"])
|
||||
rssi = coerce_integer(payload["rssi"]) || coerce_integer(metrics["rssi"])
|
||||
snr = coerce_float(payload["snr"]) || coerce_float(metrics["snr"])
|
||||
elapsed_ms = coerce_integer(
|
||||
payload["elapsed_ms"] ||
|
||||
payload["latency_ms"] ||
|
||||
metrics&.[]("elapsed_ms") ||
|
||||
metrics&.[]("latency_ms") ||
|
||||
metrics&.[]("latencyMs"),
|
||||
)
|
||||
|
||||
hops_value = payload.key?("hops") ? payload["hops"] : payload["path"]
|
||||
hops = normalize_trace_hops(hops_value)
|
||||
|
||||
all_nodes = [src, dest, *hops].compact.uniq
|
||||
all_nodes.each do |node|
|
||||
ensure_unknown_node(db, node, node, heard_time: rx_time)
|
||||
touch_node_last_seen(db, node, node, rx_time: rx_time, source: :trace)
|
||||
end
|
||||
|
||||
with_busy_retry do
|
||||
db.execute <<~SQL, [trace_identifier, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms]
|
||||
INSERT INTO traces(id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms)
|
||||
VALUES(?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
request_id=COALESCE(excluded.request_id,traces.request_id),
|
||||
src=COALESCE(excluded.src,traces.src),
|
||||
dest=COALESCE(excluded.dest,traces.dest),
|
||||
rx_time=excluded.rx_time,
|
||||
rx_iso=excluded.rx_iso,
|
||||
rssi=COALESCE(excluded.rssi,traces.rssi),
|
||||
snr=COALESCE(excluded.snr,traces.snr),
|
||||
elapsed_ms=COALESCE(excluded.elapsed_ms,traces.elapsed_ms)
|
||||
SQL
|
||||
|
||||
trace_id = trace_identifier || db.last_insert_row_id
|
||||
return unless trace_id
|
||||
|
||||
db.execute("DELETE FROM trace_hops WHERE trace_id = ?", [trace_id])
|
||||
hops.each_with_index do |hop_id, index|
|
||||
db.execute(
|
||||
"INSERT INTO trace_hops(trace_id, hop_index, node_id) VALUES(?,?,?)",
|
||||
[trace_id, index, hop_id],
|
||||
)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def insert_message(db, message)
|
||||
return unless message.is_a?(Hash)
|
||||
|
||||
|
||||
@@ -81,10 +81,10 @@ module PotatoMesh
|
||||
return false unless File.exist?(PotatoMesh::Config.db_path)
|
||||
|
||||
db = open_database(readonly: true)
|
||||
required = %w[nodes messages positions telemetry neighbors instances]
|
||||
required = %w[nodes messages positions telemetry neighbors instances traces trace_hops]
|
||||
tables =
|
||||
db.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions','telemetry','neighbors','instances')",
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions','telemetry','neighbors','instances','traces','trace_hops')",
|
||||
).flatten
|
||||
(required - tables).empty?
|
||||
rescue SQLite3::Exception
|
||||
@@ -99,7 +99,7 @@ module PotatoMesh
|
||||
def init_db
|
||||
FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path))
|
||||
db = open_database
|
||||
%w[nodes messages positions telemetry neighbors instances].each do |schema|
|
||||
%w[nodes messages positions telemetry neighbors instances traces].each do |schema|
|
||||
sql_file = File.expand_path("../../../../data/#{schema}.sql", __dir__)
|
||||
db.execute_batch(File.read(sql_file))
|
||||
end
|
||||
@@ -178,6 +178,15 @@ module PotatoMesh
|
||||
db.execute("ALTER TABLE telemetry ADD COLUMN #{name} #{type}")
|
||||
telemetry_columns << name
|
||||
end
|
||||
|
||||
trace_tables =
|
||||
db.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('traces','trace_hops')",
|
||||
).flatten
|
||||
unless trace_tables.include?("traces") && trace_tables.include?("trace_hops")
|
||||
traces_schema = File.expand_path("../../../../data/traces.sql", __dir__)
|
||||
db.execute_batch(File.read(traces_schema))
|
||||
end
|
||||
rescue SQLite3::SQLException, Errno::ENOENT => e
|
||||
warn_log(
|
||||
"Failed to apply schema upgrade",
|
||||
|
||||
@@ -110,7 +110,7 @@ module PotatoMesh
|
||||
cleaned_strings = string_values.compact.map(&:to_s).map(&:strip).reject(&:empty?).uniq
|
||||
cleaned_numbers = numeric_values.compact.map do |value|
|
||||
begin
|
||||
Integer(value, 10)
|
||||
value.is_a?(String) ? Integer(value, 10) : Integer(value)
|
||||
rescue ArgumentError, TypeError
|
||||
nil
|
||||
end
|
||||
@@ -464,6 +464,79 @@ module PotatoMesh
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
|
||||
def query_traces(limit, node_ref: nil)
|
||||
limit = coerce_query_limit(limit)
|
||||
db = open_database(readonly: true)
|
||||
db.results_as_hash = true
|
||||
params = []
|
||||
where_clauses = []
|
||||
|
||||
if node_ref
|
||||
tokens = node_reference_tokens(node_ref)
|
||||
numeric_values = tokens[:numeric_values]
|
||||
if numeric_values.empty?
|
||||
return []
|
||||
end
|
||||
placeholders = Array.new(numeric_values.length, "?").join(", ")
|
||||
candidate_clauses = []
|
||||
candidate_clauses << "src IN (#{placeholders})"
|
||||
candidate_clauses << "dest IN (#{placeholders})"
|
||||
candidate_clauses << "id IN (SELECT trace_id FROM trace_hops WHERE node_id IN (#{placeholders}))"
|
||||
where_clauses << "(#{candidate_clauses.join(" OR ")})"
|
||||
3.times { params.concat(numeric_values) }
|
||||
end
|
||||
|
||||
sql = <<~SQL
|
||||
SELECT id, request_id, src, dest, rx_time, rx_iso, rssi, snr, elapsed_ms
|
||||
FROM traces
|
||||
SQL
|
||||
sql += " WHERE #{where_clauses.join(" AND ")}\n" if where_clauses.any?
|
||||
sql += <<~SQL
|
||||
ORDER BY rx_time DESC
|
||||
LIMIT ?
|
||||
SQL
|
||||
params << limit
|
||||
rows = db.execute(sql, params)
|
||||
|
||||
trace_ids = rows.map { |row| coerce_integer(row["id"]) }.compact
|
||||
hops_by_trace = Hash.new { |hash, key| hash[key] = [] }
|
||||
unless trace_ids.empty?
|
||||
placeholders = Array.new(trace_ids.length, "?").join(", ")
|
||||
hop_rows =
|
||||
db.execute(
|
||||
"SELECT trace_id, hop_index, node_id FROM trace_hops WHERE trace_id IN (#{placeholders}) ORDER BY trace_id, hop_index",
|
||||
trace_ids,
|
||||
)
|
||||
hop_rows.each do |hop|
|
||||
trace_id = coerce_integer(hop["trace_id"])
|
||||
node_id = coerce_integer(hop["node_id"])
|
||||
next unless trace_id && node_id
|
||||
|
||||
hops_by_trace[trace_id] << node_id
|
||||
end
|
||||
end
|
||||
|
||||
rows.each do |r|
|
||||
rx_time = coerce_integer(r["rx_time"])
|
||||
r["rx_time"] = rx_time if rx_time
|
||||
r["rx_iso"] = Time.at(rx_time).utc.iso8601 if rx_time && string_or_nil(r["rx_iso"]).nil?
|
||||
r["request_id"] = coerce_integer(r["request_id"])
|
||||
r["src"] = coerce_integer(r["src"])
|
||||
r["dest"] = coerce_integer(r["dest"])
|
||||
r["rssi"] = coerce_integer(r["rssi"])
|
||||
r["snr"] = coerce_float(r["snr"])
|
||||
r["elapsed_ms"] = coerce_integer(r["elapsed_ms"])
|
||||
|
||||
trace_id = coerce_integer(r["id"])
|
||||
if trace_id && hops_by_trace.key?(trace_id)
|
||||
r["hops"] = hops_by_trace[trace_id]
|
||||
end
|
||||
end
|
||||
rows.map { |row| compact_api_row(row) }
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@@ -135,6 +135,20 @@ module PotatoMesh
|
||||
query_telemetry(limit, node_ref: node_ref).to_json
|
||||
end
|
||||
|
||||
app.get "/api/traces" do
|
||||
content_type :json
|
||||
limit = [params["limit"]&.to_i || 200, 1000].min
|
||||
query_traces(limit).to_json
|
||||
end
|
||||
|
||||
app.get "/api/traces/:id" do
|
||||
content_type :json
|
||||
node_ref = string_or_nil(params["id"])
|
||||
halt 400, { error: "missing node id" }.to_json unless node_ref
|
||||
limit = [params["limit"]&.to_i || 200, 1000].min
|
||||
query_traces(limit, node_ref: node_ref).to_json
|
||||
end
|
||||
|
||||
app.get "/api/instances" do
|
||||
# Prevent the federation catalog from being exposed when federation is disabled.
|
||||
halt 404 unless federation_enabled?
|
||||
|
||||
@@ -321,6 +321,25 @@ module PotatoMesh
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
|
||||
app.post "/api/traces" do
|
||||
require_token!
|
||||
content_type :json
|
||||
begin
|
||||
data = JSON.parse(read_json_body)
|
||||
rescue JSON::ParserError
|
||||
halt 400, { error: "invalid JSON" }.to_json
|
||||
end
|
||||
trace_packets = data.is_a?(Array) ? data : [data]
|
||||
halt 400, { error: "too many traces" }.to_json if trace_packets.size > 1000
|
||||
db = open_database
|
||||
trace_packets.each do |packet|
|
||||
insert_trace(db, packet)
|
||||
end
|
||||
{ status: "ok" }.to_json
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@@ -96,6 +96,8 @@ RSpec.describe "Potato Mesh Sinatra app" do
|
||||
def clear_database
|
||||
with_db do |db|
|
||||
db.execute("DELETE FROM instances")
|
||||
db.execute("DELETE FROM trace_hops")
|
||||
db.execute("DELETE FROM traces")
|
||||
db.execute("DELETE FROM neighbors")
|
||||
db.execute("DELETE FROM messages")
|
||||
db.execute("DELETE FROM nodes")
|
||||
@@ -270,6 +272,30 @@ RSpec.describe "Potato Mesh Sinatra app" do
|
||||
let(:nodes_fixture) { JSON.parse(File.read(fixture_path("nodes.json"))) }
|
||||
let(:messages_fixture) { JSON.parse(File.read(fixture_path("messages.json"))) }
|
||||
let(:telemetry_fixture) { JSON.parse(File.read(fixture_path("telemetry.json"))) }
|
||||
let(:trace_fixture) do
|
||||
[
|
||||
{
|
||||
"id" => 9_001,
|
||||
"request_id" => 17,
|
||||
"src" => 2_658_361_180,
|
||||
"dest" => 4_242_424_242,
|
||||
"rx_time" => reference_time.to_i - 2,
|
||||
"hops" => [2_658_361_180, 19_088_743, 4_242_424_242],
|
||||
"rssi" => -83,
|
||||
"snr" => 5.0,
|
||||
"elapsed_ms" => 842,
|
||||
},
|
||||
{
|
||||
"packet_id" => 9_002,
|
||||
"req" => 21,
|
||||
"from" => 19_088_743,
|
||||
"destination" => 2_658_361_180,
|
||||
"rx_time" => reference_time.to_i - 5,
|
||||
"path" => [{ "node_id" => "0xbeadf00d" }, { "node_id" => 19_088_743 }],
|
||||
"metrics" => { "snr" => 3.5, "latency_ms" => 1_020 },
|
||||
},
|
||||
]
|
||||
end
|
||||
let(:reference_time) do
|
||||
latest = nodes_fixture.map { |node| node["last_heard"] }.compact.max
|
||||
Time.at((latest || Time.now.to_i) + 1000)
|
||||
@@ -3186,6 +3212,88 @@ RSpec.describe "Potato Mesh Sinatra app" do
|
||||
end
|
||||
end
|
||||
|
||||
describe "POST /api/traces" do
|
||||
it "stores traces with hop paths and updates last heard timestamps" do
|
||||
payload = trace_fixture
|
||||
|
||||
post "/api/traces", payload.to_json, auth_headers
|
||||
|
||||
expect(last_response).to be_ok
|
||||
expect(JSON.parse(last_response.body)).to eq("status" => "ok")
|
||||
|
||||
with_db(readonly: true) do |db|
|
||||
db.results_as_hash = true
|
||||
|
||||
traces = db.execute("SELECT * FROM traces ORDER BY rx_time DESC")
|
||||
expect(traces.size).to eq(payload.size)
|
||||
|
||||
primary = traces.find { |row| row["id"] == payload.first["id"] }
|
||||
expect(primary["request_id"]).to eq(payload.first["request_id"])
|
||||
expect(primary["src"]).to eq(payload.first["src"])
|
||||
expect(primary["dest"]).to eq(payload.first["dest"])
|
||||
expect(primary["rx_time"]).to eq(payload.first["rx_time"])
|
||||
expect(primary["rx_iso"]).to eq(Time.at(payload.first["rx_time"]).utc.iso8601)
|
||||
expect(primary["rssi"]).to eq(payload.first["rssi"])
|
||||
expect(primary["snr"]).to eq(payload.first["snr"])
|
||||
expect(primary["elapsed_ms"]).to eq(payload.first["elapsed_ms"])
|
||||
|
||||
primary_hops = db.execute(
|
||||
"SELECT hop_index, node_id FROM trace_hops WHERE trace_id = ? ORDER BY hop_index",
|
||||
[primary["id"]],
|
||||
)
|
||||
expect(primary_hops.map { |row| row["node_id"] }).to eq(payload.first["hops"])
|
||||
|
||||
secondary = traces.find { |row| row["id"] == payload.last["packet_id"] }
|
||||
expect(secondary["request_id"]).to eq(payload.last["req"])
|
||||
expect(secondary["src"]).to eq(payload.last["from"])
|
||||
expect(secondary["dest"]).to eq(payload.last["destination"])
|
||||
expect(secondary["rssi"]).to be_nil
|
||||
expect(secondary["snr"]).to eq(payload.last.dig("metrics", "snr"))
|
||||
expect(secondary["elapsed_ms"]).to eq(payload.last.dig("metrics", "latency_ms"))
|
||||
|
||||
secondary_hops = db.execute(
|
||||
"SELECT hop_index, node_id FROM trace_hops WHERE trace_id = ? ORDER BY hop_index",
|
||||
[secondary["id"]],
|
||||
)
|
||||
expect(secondary_hops.map { |row| row["node_id"] }).to eq([0xBEADF00D, 19_088_743])
|
||||
|
||||
node_ids = [
|
||||
payload.first["src"],
|
||||
payload.first["dest"],
|
||||
payload.first["hops"][1],
|
||||
0xBEADF00D,
|
||||
].map { |num| format("!%08x", num & 0xFFFFFFFF) }
|
||||
|
||||
placeholders = node_ids.map { "?" }.join(",")
|
||||
rows = db.execute("SELECT node_id, last_heard FROM nodes WHERE node_id IN (#{placeholders})", node_ids)
|
||||
expect(rows.size).to eq(node_ids.size)
|
||||
latest_last_heard = rows.map { |row| row["last_heard"] }.max
|
||||
expect(latest_last_heard).to eq(payload.first["rx_time"])
|
||||
end
|
||||
end
|
||||
|
||||
it "returns 400 when the payload is not valid JSON" do
|
||||
post "/api/traces", "{", auth_headers
|
||||
|
||||
expect(last_response.status).to eq(400)
|
||||
expect(JSON.parse(last_response.body)).to eq("error" => "invalid JSON")
|
||||
end
|
||||
|
||||
it "returns 400 when more than 1000 traces are provided" do
|
||||
payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } }
|
||||
|
||||
post "/api/traces", payload.to_json, auth_headers
|
||||
|
||||
expect(last_response.status).to eq(400)
|
||||
expect(JSON.parse(last_response.body)).to eq("error" => "too many traces")
|
||||
|
||||
with_db(readonly: true) do |db|
|
||||
count = db.get_first_value("SELECT COUNT(*) FROM traces")
|
||||
expect(count).to eq(0)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
it "returns 400 when more than 1000 messages are provided" do
|
||||
payload = Array.new(1001) { |i| { "packet_id" => i + 1 } }
|
||||
|
||||
@@ -4242,6 +4350,60 @@ RSpec.describe "Potato Mesh Sinatra app" do
|
||||
end
|
||||
end
|
||||
|
||||
describe "GET /api/traces" do
|
||||
it "returns stored traces ordered by receive time" do
|
||||
clear_database
|
||||
post "/api/traces", trace_fixture.to_json, auth_headers
|
||||
expect(last_response).to be_ok
|
||||
|
||||
get "/api/traces"
|
||||
|
||||
expect(last_response).to be_ok
|
||||
payload = JSON.parse(last_response.body)
|
||||
expect(payload.length).to eq(trace_fixture.length)
|
||||
expect(payload.map { |row| row["id"] }).to eq([trace_fixture.first["id"], trace_fixture.last["packet_id"]])
|
||||
|
||||
latest = payload.first
|
||||
expect(latest["request_id"]).to eq(trace_fixture.first["request_id"])
|
||||
expect(latest["src"]).to eq(trace_fixture.first["src"])
|
||||
expect(latest["dest"]).to eq(trace_fixture.first["dest"])
|
||||
expect(latest["hops"]).to eq(trace_fixture.first["hops"])
|
||||
expect(latest["rx_iso"]).to eq(Time.at(trace_fixture.first["rx_time"]).utc.iso8601)
|
||||
|
||||
earlier = payload.last
|
||||
expect(earlier["request_id"]).to eq(trace_fixture.last["req"])
|
||||
expect(earlier["hops"]).to eq([0xBEADF00D, 19_088_743])
|
||||
expect(earlier["elapsed_ms"]).to eq(trace_fixture.last.dig("metrics", "latency_ms"))
|
||||
end
|
||||
|
||||
it "filters traces by node reference across sources" do
|
||||
clear_database
|
||||
post "/api/traces", trace_fixture.to_json, auth_headers
|
||||
expect(last_response).to be_ok
|
||||
|
||||
get "/api/traces/#{trace_fixture.first["src"]}"
|
||||
|
||||
expect(last_response).to be_ok
|
||||
filtered = JSON.parse(last_response.body)
|
||||
expect(filtered.map { |row| row["id"] }).to include(trace_fixture.first["id"], trace_fixture.last["packet_id"])
|
||||
|
||||
get "/api/traces/!beadf00d"
|
||||
|
||||
expect(last_response).to be_ok
|
||||
bead_filtered = JSON.parse(last_response.body)
|
||||
expect(bead_filtered.map { |row| row["id"] }).to eq([trace_fixture.last["packet_id"]])
|
||||
expect(bead_filtered.first["hops"]).to eq([0xBEADF00D, 19_088_743])
|
||||
end
|
||||
|
||||
it "returns an empty list when no traces are stored" do
|
||||
clear_database
|
||||
get "/api/traces"
|
||||
|
||||
expect(last_response).to be_ok
|
||||
expect(JSON.parse(last_response.body)).to eq([])
|
||||
end
|
||||
end
|
||||
|
||||
describe "GET /nodes/:id" do
|
||||
before do
|
||||
import_nodes_fixture
|
||||
|
||||
@@ -165,4 +165,22 @@ RSpec.describe PotatoMesh::App::Database do
|
||||
expect(telemetry_columns).to include("soil_temperature", "lux", "iaq")
|
||||
expect(telemetry_columns).to include("rx_time", "battery_level")
|
||||
end
|
||||
|
||||
it "creates trace tables when absent" do
|
||||
SQLite3::Database.new(PotatoMesh::Config.db_path) do |db|
|
||||
db.execute("CREATE TABLE nodes(node_id TEXT)")
|
||||
db.execute("CREATE TABLE messages(id INTEGER PRIMARY KEY)")
|
||||
end
|
||||
|
||||
expect(column_names_for("traces")).to be_empty
|
||||
expect(column_names_for("trace_hops")).to be_empty
|
||||
|
||||
harness_class.ensure_schema_upgrades
|
||||
|
||||
traces_columns = column_names_for("traces")
|
||||
expect(traces_columns).to include("request_id", "src", "dest", "rx_time", "rx_iso", "elapsed_ms")
|
||||
|
||||
hop_columns = column_names_for("trace_hops")
|
||||
expect(hop_columns).to include("trace_id", "hop_index", "node_id")
|
||||
end
|
||||
end
|
||||
|
||||
Reference in New Issue
Block a user