⎇ 72758e75ce515f355f27615671c95ecaf8c2a8bc
crates/core/brahman-handshake/tests/handshake.rs
History
Who touched what
raw
feat(brahman-handshake): ListSessions endpoint + cliente + UI broker-explorer
1//! Tests de integración: levanta server + client en el mismo proceso,
2//! ejercita el round-trip completo del protocolo.
3
4use std::collections::BTreeSet;
5use std::sync::Arc;
6use std::time::Duration;
7
8use brahman_broker::{Broker, BrokerConfig};
9use brahman_card::{
10 Card, CgroupSpec, Flow, Flows, NamespaceSet, Payload, ResourceLimits, SomaSpec, Supervision,
11 TypeRef, CARD_SCHEMA_VERSION,
12};
13use brahman_handshake::{
14 client::{Client, ClientError},
15 codec::{read_frame, write_frame},
16 messages::{Frame, HandshakeError, Hello, Ping},
17 server::{Server, ServerConfig},
18};
19use tokio::net::UnixStream;
20use tokio::sync::Mutex;
21use ulid::Ulid;
22
23fn sample_card(label: &str) -> Card {
24 Card {
25 schema_version: CARD_SCHEMA_VERSION,
26 id: Ulid::new(),
27 lineage: None,
28 label: label.into(),
29 provides: BTreeSet::new(),
30 requires: BTreeSet::new(),
31 soma: SomaSpec {
32 cgroup: CgroupSpec {
33 path: "ente.slice/test".into(),
34 cpu_weight: None,
35 io_weight: None,
36 },
37 namespaces: NamespaceSet::default(),
38 rlimits: ResourceLimits::default(),
39 cpu_affinity: None,
40 },
41 payload: Payload::Virtual,
42 supervision: Supervision::OneShot,
43 ..Default::default()
44 }
45}
46
47/// Genera una ruta de socket única bajo TMPDIR. No la creamos —
48/// el server la creará al hacer bind.
49fn sock_path(name: &str) -> std::path::PathBuf {
50 std::env::temp_dir().join(format!(
51 "brahman-test-{}-{}-{}.sock",
52 name,
53 std::process::id(),
54 Ulid::new()
55 ))
56}
57
58#[tokio::test]
59async fn full_handshake_roundtrip() {
60 let path = sock_path("happy");
61 let server = Server::bind(&path, ServerConfig { init_attached: true, broker: None, net: None, policy: None }).unwrap();
62
63 let session_handle = tokio::spawn({
64 async move {
65 let session = server.accept_one().await.unwrap();
66 session.handle().await.unwrap();
67 }
68 });
69
70 let mut client = Client::connect(&path, sample_card("alpha")).await.unwrap();
71 assert!(client.server_info().init_attached);
72 assert_eq!(
73 client.server_info().protocol_version,
74 brahman_card::PROTOCOL_VERSION
75 );
76
77 let mut last = 0u64;
78 for _ in 0..3 {
79 let ts = client.ping().await.unwrap();
80 assert!(ts >= last);
81 last = ts;
82 tokio::time::sleep(Duration::from_millis(2)).await;
83 }
84 client.farewell().await.unwrap();
85
86 tokio::time::timeout(Duration::from_secs(2), session_handle)
87 .await
88 .expect("server hung after farewell")
89 .unwrap();
90}
91
92#[tokio::test]
93async fn list_sessions_returns_currently_registered() {
94 // Levantamos un server con broker (requerido para que el registro
95 // pase por el path real) y conectamos 3 clientes. El último pide
96 // ListSessions y debe ver a los 2 anteriores + a sí mismo.
97 let path = sock_path("listsess");
98 let broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
99 let server = Server::bind(
100 &path,
101 ServerConfig {
102 init_attached: true,
103 broker: Some(broker),
104 net: None,
105 policy: None,
106 },
107 )
108 .unwrap();
109
110 // Una task accept loop genérica para los 3 clientes.
111 let server_handle = tokio::spawn(async move {
112 for _ in 0..3 {
113 let session = server.accept_one().await.unwrap();
114 tokio::spawn(async move {
115 let _ = session.handle().await;
116 });
117 }
118 // Mantener el server vivo para que las sesiones puedan
119 // mantenerse abiertas mientras el observer pregunta.
120 std::future::pending::<()>().await;
121 });
122
123 let mut alpha = Client::connect(&path, sample_card("producer-alpha"))
124 .await
125 .unwrap();
126 let mut beta = Client::connect(&path, sample_card("producer-beta"))
127 .await
128 .unwrap();
129 // observer es el que va a preguntar.
130 let mut observer = Client::connect(&path, sample_card("observer"))
131 .await
132 .unwrap();
133
134 let list = observer.list_sessions().await.unwrap();
135 assert_eq!(list.entries.len(), 3, "deberían verse 3 sesiones activas");
136
137 let labels: BTreeSet<&str> = list.entries.iter().map(|e| e.label.as_str()).collect();
138 assert!(labels.contains("producer-alpha"));
139 assert!(labels.contains("producer-beta"));
140 assert!(labels.contains("observer"));
141
142 // schema_version + conscious sanity en la propia entry del observer.
143 let me = list
144 .entries
145 .iter()
146 .find(|e| e.label == "observer")
147 .unwrap();
148 assert_eq!(me.schema_version, brahman_card::CARD_SCHEMA_VERSION);
149 assert!(!me.conscious, "observer no envió WIT — debería ser agnostic");
150
151 alpha.farewell().await.unwrap();
152 beta.farewell().await.unwrap();
153 observer.farewell().await.unwrap();
154 server_handle.abort();
155}
156
157#[tokio::test]
158async fn rejects_invalid_card_client_side() {
159 let path = sock_path("invalid");
160 let server = Server::bind(&path, ServerConfig::default()).unwrap();
161 let session_handle = tokio::spawn(async move {
162 // No esperamos que el server complete: el cliente corta antes.
163 let _ = tokio::time::timeout(Duration::from_secs(1), async move {
164 let session = server.accept_one().await.unwrap();
165 session.handle().await.unwrap();
166 })
167 .await;
168 });
169
170 let mut bad = sample_card("placeholder");
171 bad.label = String::new();
172 let err = Client::connect(&path, bad).await.unwrap_err();
173 assert!(matches!(err, ClientError::InvalidCard(_)));
174
175 session_handle.abort();
176}
177
178#[tokio::test]
179async fn server_rejects_protocol_mismatch() {
180 let path = sock_path("mismatch");
181 let server = Server::bind(&path, ServerConfig::default()).unwrap();
182 let session_handle = tokio::spawn(async move {
183 let session = server.accept_one().await.unwrap();
184 session.handle().await.unwrap();
185 });
186
187 let mut stream = UnixStream::connect(&path).await.unwrap();
188 let hello = Hello {
189 schema_version: CARD_SCHEMA_VERSION,
190 protocol_version: "999.0.0".into(),
191 card: sample_card("future-module").into(),
192 wit: None,
193 signature: None,
194 identity_cert: None,
195 };
196 write_frame(&mut stream, &Frame::Hello(hello)).await.unwrap();
197
198 match read_frame(&mut stream).await.unwrap() {
199 Frame::Error(HandshakeError::ProtocolMismatch(_)) => {}
200 other => panic!("esperado ProtocolMismatch, got {other:?}"),
201 }
202
203 tokio::time::timeout(Duration::from_secs(2), session_handle)
204 .await
205 .expect("server hung after rejecting")
206 .unwrap();
207}
208
209// =====================================================================
210// Integración handshake ↔ broker
211// =====================================================================
212
213fn card_with_flows(label: &str, input: Vec<Flow>, output: Vec<Flow>) -> Card {
214 Card {
215 schema_version: CARD_SCHEMA_VERSION,
216 id: Ulid::new(),
217 label: label.into(),
218 soma: SomaSpec {
219 cgroup: CgroupSpec {
220 path: "ente.slice/test".into(),
221 cpu_weight: None,
222 io_weight: None,
223 },
224 namespaces: NamespaceSet::default(),
225 rlimits: ResourceLimits::default(),
226 cpu_affinity: None,
227 },
228 payload: Payload::Virtual,
229 supervision: Supervision::OneShot,
230 flow: Flows { input, output },
231 ..Default::default()
232 }
233}
234
235fn flow(name: &str, ty: TypeRef) -> Flow {
236 Flow {
237 name: name.into(),
238 ty,
239 pin_to: None,
240 }
241}
242
243/// Espera hasta que `broker.len() >= n` o timeout.
244async fn wait_for_broker_len(broker: &Arc<Mutex<Broker>>, n: usize) {
245 for _ in 0..50 {
246 if broker.lock().await.len() >= n {
247 return;
248 }
249 tokio::time::sleep(Duration::from_millis(10)).await;
250 }
251 panic!("broker no alcanzó {n} entradas en 500ms");
252}
253
254#[tokio::test]
255async fn broker_registers_and_unregisters_with_session() {
256 let path = sock_path("broker-lifecycle");
257 let broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
258 let server = Server::bind(
259 &path,
260 ServerConfig {
261 init_attached: false,
262 broker: Some(broker.clone()),
263 net: None,
264 policy: None,
265 },
266 )
267 .unwrap();
268
269 let session_handle = tokio::spawn(async move {
270 let session = server.accept_one().await.unwrap();
271 session.handle().await.unwrap();
272 });
273
274 let mut client = Client::connect(&path, sample_card("alpha")).await.unwrap();
275 let session_id = client.session();
276
277 // Tras el handshake, la Card debe estar registrada en el broker.
278 wait_for_broker_len(&broker, 1).await;
279 {
280 let b = broker.lock().await;
281 assert_eq!(b.len(), 1);
282 assert!(b.sessions().any(|s| s == session_id));
283 }
284
285 client.farewell().await.unwrap();
286 tokio::time::timeout(Duration::from_secs(2), session_handle)
287 .await
288 .expect("server colgó tras farewell")
289 .unwrap();
290
291 // Tras el cleanup, el broker queda vacío.
292 {
293 let b = broker.lock().await;
294 assert_eq!(b.len(), 0);
295 }
296}
297
298#[tokio::test]
299async fn broker_matches_two_live_modules() {
300 let path = sock_path("broker-match");
301 let broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
302 let server = Server::bind(
303 &path,
304 ServerConfig {
305 init_attached: false,
306 broker: Some(broker.clone()),
307 net: None,
308 policy: None,
309 },
310 )
311 .unwrap();
312
313 // Server loop: usa la API run() para manejar accept+spawn.
314 let server_handle = tokio::spawn(async move {
315 let _ = server.run().await;
316 });
317
318 // Productor: emite "out" tipo string.
319 let producer_card = card_with_flows(
320 "dht",
321 vec![],
322 vec![flow(
323 "out",
324 TypeRef::Primitive {
325 name: "string".into(),
326 },
327 )],
328 );
329 let mut producer = Client::connect(&path, producer_card).await.unwrap();
330 wait_for_broker_len(&broker, 1).await;
331
332 // Consumidor: pide "in" tipo string.
333 let consumer_card = card_with_flows(
334 "ui",
335 vec![flow(
336 "in",
337 TypeRef::Primitive {
338 name: "string".into(),
339 },
340 )],
341 vec![],
342 );
343 let mut consumer = Client::connect(&path, consumer_card).await.unwrap();
344 wait_for_broker_len(&broker, 2).await;
345
346 // El broker debe encontrar el match consumer.in ← producer.out.
347 let m = {
348 let b = broker.lock().await;
349 b.find_producer_for(consumer.session(), "in")
350 }
351 .expect("broker no encontró match");
352 assert_eq!(m.consumer_label, "ui");
353 assert_eq!(m.producer_label, "dht");
354 assert_eq!(m.producer.flow_name, "out");
355
356 // Cuando el productor se va, el match desaparece.
357 producer.farewell().await.unwrap();
358 for _ in 0..50 {
359 if broker.lock().await.len() < 2 {
360 break;
361 }
362 tokio::time::sleep(Duration::from_millis(10)).await;
363 }
364 {
365 let b = broker.lock().await;
366 assert!(b.find_producer_for(consumer.session(), "in").is_none());
367 }
368
369 consumer.farewell().await.unwrap();
370 server_handle.abort();
371}
372
373#[tokio::test]
374async fn match_event_pushed_on_producer_arrival() {
375 use brahman_handshake::messages::MatchEventKind;
376
377 let path = sock_path("push-match");
378 let broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
379 let server = Server::bind(
380 &path,
381 ServerConfig {
382 init_attached: false,
383 broker: Some(broker.clone()),
384 net: None,
385 policy: None,
386 },
387 )
388 .unwrap();
389
390 let server_handle = tokio::spawn(async move {
391 let _ = server.run().await;
392 });
393
394 // El consumidor llega primero — sin productor, no hay match aún.
395 let consumer_card = card_with_flows(
396 "ui",
397 vec![flow(
398 "in",
399 TypeRef::Primitive {
400 name: "json".into(),
401 },
402 )],
403 vec![],
404 );
405 let mut consumer = Client::connect(&path, consumer_card).await.unwrap();
406
407 // No debería haber evento todavía.
408 let no_event = consumer
409 .await_event(Duration::from_millis(100))
410 .await
411 .unwrap();
412 assert!(no_event.is_none(), "evento inesperado: {no_event:?}");
413
414 // Llega el productor → consumer debe recibir Available.
415 let producer_card = card_with_flows(
416 "dht",
417 vec![],
418 vec![flow(
419 "out",
420 TypeRef::Primitive {
421 name: "json".into(),
422 },
423 )],
424 );
425 let mut producer = Client::connect(&path, producer_card).await.unwrap();
426
427 let ev = consumer
428 .await_event(Duration::from_secs(2))
429 .await
430 .unwrap()
431 .expect("Available no llegó");
432 assert_eq!(ev.kind, MatchEventKind::Available);
433 assert_eq!(ev.consumer_flow, "in");
434 assert_eq!(ev.producer_label, "dht");
435 assert_eq!(ev.producer_flow, "out");
436
437 // El productor se va → consumer debe recibir Lost.
438 producer.farewell().await.unwrap();
439 let ev = consumer
440 .await_event(Duration::from_secs(2))
441 .await
442 .unwrap()
443 .expect("Lost no llegó");
444 assert_eq!(ev.kind, MatchEventKind::Lost);
445 assert_eq!(ev.consumer_flow, "in");
446
447 consumer.farewell().await.unwrap();
448 server_handle.abort();
449}
450
451#[tokio::test]
452async fn ping_before_hello_rejected() {
453 let path = sock_path("ping-no-hello");
454 let server = Server::bind(&path, ServerConfig::default()).unwrap();
455 let session_handle = tokio::spawn(async move {
456 let session = server.accept_one().await.unwrap();
457 session.handle().await.unwrap();
458 });
459
460 // Conectamos y mandamos un Ping sin haber saludado.
461 let mut stream = UnixStream::connect(&path).await.unwrap();
462 write_frame(
463 &mut stream,
464 &Frame::Ping(Ping {
465 session: Ulid::new(),
466 }),
467 )
468 .await
469 .unwrap();
470
471 match read_frame(&mut stream).await.unwrap() {
472 Frame::Error(HandshakeError::Rejected(_)) => {}
473 other => panic!("esperado Rejected, got {other:?}"),
474 }
475
476 tokio::time::timeout(Duration::from_secs(2), session_handle)
477 .await
478 .expect("server hung after rejecting")
479 .unwrap();
480}