Files
continuum-proxy/tests/uplink_protocol.rs
T
octoturgeandClaude Opus 5 bda5e95b66 Generate go2rtc's config from what the printers report
go2rtc has been supervised since before this commit, with nothing to
serve: `CONTINUUM_GO2RTC_CONFIG` pointed at `./go2rtc.yaml`, and that
file did not exist in the repository or anywhere else. A watchdog was
restarting a process that had no cameras configured.

`camera.rs` writes it, from the `rtsp_url` a printer publishes in its own
`print.push_status` `ipcam` block rather than from `printers.toml`. The
URL is already something the machine tells us, so asking an operator to
keep a second copy of it - and to update it by hand when a printer's
address changes - would be asking for drift.

Two tasks, not one: the builder decides what should be served, the
watchdog keeps the thing serving it alive. They meet at one `Notify`,
because go2rtc reads its configuration only at startup, so a changed
config needs a restart. The watchdog now selects the child's exit against
that notification and skips its backoff on a deliberate change.

What this does *not* cover is most printers, and that is the honest
limit. Only the X1/H2 class publishes an RTSP URL - confirmed against two
real machines: the H2C reports
`rtsps://192.168.0.21:322/streaming/live/1`, the P1P reports an `ipcam`
block with `ipcam_dev: "1"` and no URL at all. "A camera exists" and "a
camera you can connect to" are different facts. A P1/A1 needs Bambu's
proprietary port-6000 JPEG stream, which this crate does not speak, so
those printers report no `camera_stream` and a UI should offer a player
only when one is present.

Details worth keeping:

- The config is rendered sorted so an unchanged fleet is byte-identical,
  and on the first pass it is compared against the file on disk - without
  that, every daemon restart rewrites an identical file and drops every
  live viewer.
- Written through a `.tmp` and renamed. The watchdog can restart go2rtc at
  any moment, and a half-written config fails to parse, which takes every
  working stream down rather than one.
- Never logged, and gitignored: the credentials go in the URL because
  that is the only place go2rtc takes them for an RTSP source, so the file
  holds printer access codes.
- A reported URL with no scheme is refused rather than repaired. One bad
  entry makes go2rtc fail to load the whole file.
- No `rtsp:` or `webrtc:` block. The API listener is what serves WebRTC to
  a browser, which is the only consumer here; opening go2rtc's own RTSP
  server would publish every printer's camera on the LAN unauthenticated.
- `stream_base_url` is reported in `hello`, not derived from the listen
  address: `:1984` is a bind address and not something a browser can open.
  Unset means "advertise nothing", which is the right default, because a
  wrong URL makes the UI offer a player that cannot connect. It is a LAN
  address either way - there is no relay, so this works for an operator in
  the workshop and not for anyone outside it.

63 tests, clippy unchanged at the two deliberate variant-size warnings,
`cargo deny check licenses` ok. go2rtc itself stays operator-installed and
is not distributed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-10-15 23:03:02 +00:00

288 lines
13 KiB
Rust

//! The other half of the `/ws/edge/v1` contract test.
//!
//! `continuum-backend/tests/edge_protocol.test.ts` checks that what this
//! gateway sends is what the backend accepts. This checks the reverse: that
//! every message the backend sends is one this gateway can actually read, and
//! that what this gateway writes matches the schema it claims to follow.
//!
//! Worth having on both sides because the two ends had drifted into entirely
//! different protocols - a handshake the gateway never sent, telemetry shaped
//! one printer per message instead of a batch, printer ids a gateway has no
//! way to know - and nothing failed anywhere. Both test files are hand-kept
//! mirrors; the canonical shape is
//! `continuum-schemas/schemas/continuum/v1/edge/`.
use continuum_proxy::printer::PrintControl;
use continuum_proxy::uplink::{ClientMessage, PrinterCommand, PrinterTelemetry, ServerMessage};
/// Verbatim from the backend's `src/routes/ws/edge.ts`. If the backend
/// changes one of these, this test is where it should break.
const BACKEND_MESSAGES: &[&str] = &[
r#"{"type":"hello_ack","session_id":"3f2b1c5e-0000-4000-8000-000000000000"}"#,
r#"{"type":"hello_error","message":"invalid edge token"}"#,
r#"{"type":"heartbeat_ack","seq":7}"#,
r#"{"type":"command","command_id":"9a7d2f10-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"set_fan_speed","percent":40}}"#,
r#"{"type":"command","command_id":"9a7d2f11-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"send_gcode_line","line":"M106 S0"}}"#,
r#"{"type":"command","command_id":"9a7d2f12-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"control_print","control":"pause"}}"#,
r#"{"type":"command","command_id":"9a7d2f13-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"control_print","control":"resume"}}"#,
r#"{"type":"command","command_id":"9a7d2f14-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"control_print","control":"stop"}}"#,
];
#[test]
fn every_backend_message_deserializes() {
for raw in BACKEND_MESSAGES {
let parsed: Result<ServerMessage, _> = serde_json::from_str(raw);
assert!(parsed.is_ok(), "backend sends this and the gateway cannot read it:\n {raw}\n {:?}", parsed.err());
}
}
#[test]
fn a_command_arrives_with_its_payload_intact() {
let raw = r#"{"type":"command","command_id":"cmd-1","printer_id":"p1p","command":{"action":"set_fan_speed","percent":40}}"#;
match serde_json::from_str::<ServerMessage>(raw).expect("should parse") {
ServerMessage::Command { command_id, printer_id, command } => {
assert_eq!(command_id, "cmd-1");
// The gateway's own id from printers.toml, not a cloud UUID - the
// backend translates on the way out.
assert_eq!(printer_id, "p1p");
match command {
PrinterCommand::SetFanSpeed { percent } => assert_eq!(percent, 40),
other => panic!("wrong command variant: {other:?}"),
}
}
other => panic!("wrong message variant: {other:?}"),
}
}
/// Each control verb has to arrive as the right `PrintControl`, not merely
/// parse. A silent mis-mapping here would send a stop where a pause was
/// asked for, which on a real machine is the difference between a print that
/// can be resumed and one that is over.
#[test]
fn each_print_control_verb_maps_to_its_own_action() {
for (verb, expected) in [
("pause", PrintControl::Pause),
("resume", PrintControl::Resume),
("stop", PrintControl::Stop),
] {
let raw = format!(
r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"control_print","control":"{verb}"}}}}"#
);
match serde_json::from_str::<ServerMessage>(&raw).expect("should parse") {
ServerMessage::Command { command: PrinterCommand::ControlPrint { control }, .. } => {
assert_eq!(control, expected, "'{verb}' mapped to the wrong action");
// And the wire word the vendor layer sends has to match the
// wire word that arrived - `as_str` is what builds the Bambu
// command name.
assert_eq!(control.as_str(), verb);
}
other => panic!("wrong variant for '{verb}': {other:?}"),
}
}
}
/// "cancel" is the trap: it is Moonraker's word, and the Klipper
/// implementation behind `control_print` really does send a cancel - but the
/// wire word is "stop". A verb only one vendor recognises must be rejected
/// here rather than reaching a gateway that cannot map it.
#[test]
fn a_vendor_specific_control_verb_is_rejected() {
for verb in ["cancel", "abort", "Pause", ""] {
let raw = format!(
r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"control_print","control":"{verb}"}}}}"#
);
assert!(
serde_json::from_str::<ServerMessage>(&raw).is_err(),
"'{verb}' parsed as a print control verb and should not have"
);
}
}
/// The schema's `StartPrint` variant, in the shape the backend dispatches.
/// Every field has to arrive intact: `url` is what gets downloaded and
/// `file_name` is what the printer is told to call it, so a swapped or
/// dropped one prints the wrong thing or nothing.
#[test]
fn a_start_print_command_arrives_with_its_fields_intact() {
let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"5b0e7c1a-0000-4000-8000-000000000000","file_name":"plate_1.3mf","url":"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"}}"#;
match serde_json::from_str::<ServerMessage>(raw).expect("should parse") {
ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url }, .. } => {
assert_eq!(job_id, "5b0e7c1a-0000-4000-8000-000000000000");
assert_eq!(file_name, "plate_1.3mf");
assert_eq!(
url,
"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"
);
}
other => panic!("wrong variant: {other:?}"),
}
}
/// The URL is a bearer credential until it expires, and `dispatch` logs
/// every command with `{:?}`.
#[test]
fn a_start_print_url_does_not_appear_in_debug_output() {
let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"j","file_name":"plate_1.3mf","url":"https://objects.example.com/x?X-Amz-Signature=deadbeef"}}"#;
let parsed = serde_json::from_str::<ServerMessage>(raw).expect("should parse");
let shown = format!("{parsed:?}");
assert!(!shown.contains("deadbeef"), "signature leaked into a log line: {shown}");
assert!(shown.contains("plate_1.3mf"), "the file name should still be logged: {shown}");
}
/// Each field is load-bearing and none has a sensible default: a command
/// missing its URL must be rejected here, not discovered at the printer.
#[test]
fn a_start_print_missing_any_required_field_is_rejected() {
let full = [
("job_id", r#""job_id":"5b0e7c1a-0000-4000-8000-000000000000""#),
("file_name", r#""file_name":"plate_1.3mf""#),
("url", r#""url":"https://objects.example.com/jobs/abc.3mf""#),
];
for missing in 0..full.len() {
let fields: Vec<&str> =
full.iter().enumerate().filter(|(i, _)| *i != missing).map(|(_, (_, f))| *f).collect();
let raw = format!(
r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"start_print",{}}}}}"#,
fields.join(",")
);
assert!(
serde_json::from_str::<ServerMessage>(&raw).is_err(),
"start_print without '{}' parsed and should not have:\n {raw}",
full[missing].0
);
}
}
#[test]
fn unknown_messages_are_rejected_rather_than_silently_accepted() {
// The old protocol. A gateway that accepted this would ack a handshake
// that never happened.
assert!(serde_json::from_str::<ServerMessage>(r#"{"type":"auth_ok","farmId":"f"}"#).is_err());
}
/// The bug this test exists for: serde emits `"job_name": null` for an empty
/// `Option` unless told not to, and the backend's validator reads a missing
/// key and an explicit null as different things. An idle printer has no job
/// name, so every batch containing one was rejected whole.
#[test]
fn empty_telemetry_fields_are_omitted_not_null() {
let msg = ClientMessage::Telemetry {
printers: vec![PrinterTelemetry {
printer_id: "p1p".into(),
name: "P1P".into(),
state: Some("IDLE".into()),
percent: Some(0),
// An idle printer is in no stage, so this is the second field
// after `job_name` that must be omitted rather than nulled.
stage: None,
layer: Some(0),
total_layers: Some(0),
remaining_minutes: Some(0),
job_name: None,
nozzle_temp: Some(23.375),
nozzle_target: Some(0.0),
bed_temp: Some(22.40625),
bed_target: Some(0.0),
print_error: Some(0),
active_alerts: Some(0),
// The common case, and the third field that must be omitted: a
// P1P reports that it has a camera without giving a URL for it.
camera_stream: None,
}],
};
let json = serde_json::to_string(&msg).expect("should serialize");
assert!(!json.contains("null"), "telemetry should omit empty fields, got:\n {json}");
assert!(!json.contains("job_name"), "an absent job name should not appear at all");
// The fields that *do* have values must still be there - an assertion on
// the absence of "null" alone would pass on an empty object.
assert!(json.contains(r#""state":"IDLE""#));
assert!(json.contains(r#""bed_temp":22.40625"#));
assert!(!json.contains("stage"), "an absent stage should not appear at all");
}
/// A stage of 0 means "printing" and must survive the wire, because
/// `skip_serializing_if = "Option::is_none"` on an `Option<i32>` is one
/// `unwrap_or_default` away from turning "printing" into "no stage". The
/// distinction matters: a printer at 14% and layer 0 is mid-bed-levelling if
/// the stage is 54 and genuinely stuck if it is 0.
#[test]
fn a_zero_stage_is_sent_rather_than_treated_as_absent() {
let msg = ClientMessage::Telemetry {
printers: vec![PrinterTelemetry {
printer_id: "p1p".into(),
name: "P1P".into(),
state: Some("RUNNING".into()),
percent: Some(14),
stage: Some(0),
layer: Some(1),
total_layers: Some(150),
remaining_minutes: Some(39),
job_name: Some("cube-test".into()),
nozzle_temp: Some(255.2),
nozzle_target: Some(255.0),
bed_temp: Some(79.3),
bed_target: Some(80.0),
print_error: Some(0),
active_alerts: Some(0),
camera_stream: Some("p1p".into()),
}],
};
let json = serde_json::to_string(&msg).expect("should serialize");
assert!(json.contains(r#""stage":0"#), "a zero stage must be on the wire, got:\n {json}");
}
#[test]
fn a_successful_command_result_omits_the_error() {
let json = serde_json::to_string(&ClientMessage::CommandResult {
command_id: "cmd-1".into(),
ok: true,
error: None,
})
.expect("should serialize");
assert_eq!(json, r#"{"type":"command_result","command_id":"cmd-1","ok":true}"#);
}
#[test]
fn a_failed_command_result_carries_the_reason() {
let json = serde_json::to_string(&ClientMessage::CommandResult {
command_id: "cmd-1".into(),
ok: false,
error: Some("no printer with id 'nope' in this gateway's fleet".into()),
})
.expect("should serialize");
assert!(json.contains(r#""ok":false"#));
assert!(json.contains("no printer with id 'nope'"));
}
/// A gateway with no configured stream base URL must omit the field, for the
/// same reason the telemetry fields are omitted: the backend reads a missing
/// key and an explicit null differently, and an unconfigured gateway is the
/// default case, not an edge case.
#[test]
fn a_hello_without_a_stream_base_url_omits_it() {
let json = serde_json::to_string(&ClientMessage::Hello {
gateway_id: "gw-1".into(),
farm_id: "farm-1".into(),
token: "t".into(),
stream_base_url: None,
})
.expect("should serialize");
assert!(!json.contains("stream_base_url"), "got:\n {json}");
// Without this the assertion above would also pass on an empty object.
assert!(json.contains(r#""gateway_id":"gw-1""#));
}
#[test]
fn a_hello_carries_the_stream_base_url_when_one_is_configured() {
let json = serde_json::to_string(&ClientMessage::Hello {
gateway_id: "gw-1".into(),
farm_id: "farm-1".into(),
token: "t".into(),
stream_base_url: Some("http://gateway.local:1984".into()),
})
.expect("should serialize");
assert!(json.contains(r#""stream_base_url":"http://gateway.local:1984""#), "got:\n {json}");
}