serai_full_stack_tests/
lib.rs

1/*
2#![allow(clippy::needless_pass_by_ref_mut)] // False positives
3
4use std::{
5  sync::{OnceLock, Mutex},
6  time::Duration,
7  fs,
8};
9
10use zeroize::Zeroizing;
11
12use ciphersuite::{group::ff::PrimeField, Ciphersuite, Ristretto};
13
14use serai_client::primitives::NetworkId;
15
16use messages::{CoordinatorMessage, ProcessorMessage};
17use serai_message_queue::{Service, Metadata, client::MessageQueue};
18
19use serai_client::Serai;
20
21use dockertest::{
22  PullPolicy, Image, LogAction, LogPolicy, LogSource, LogOptions, StartPolicy, Composition,
23  DockerOperations,
24};
25
26#[cfg(test)]
27mod tests;
28
29static UNIQUE_ID: OnceLock<Mutex<u16>> = OnceLock::new();
30
31pub fn coordinator_instance(
32  name: &str,
33  message_queue_key: <Ristretto as Ciphersuite>::F,
34) -> Composition {
35  serai_docker_tests::build("coordinator".to_string());
36
37  Composition::with_image(
38    Image::with_repository("serai-dev-coordinator").pull_policy(PullPolicy::Never),
39  )
40  .with_env(
41    [
42      ("MESSAGE_QUEUE_KEY".to_string(), hex::encode(message_queue_key.to_repr())),
43      ("DB_PATH".to_string(), "./coordinator-db".to_string()),
44      ("SERAI_KEY".to_string(), {
45        use serai_client::primitives::insecure_pair_from_name;
46        hex::encode(&insecure_pair_from_name(name).as_ref().secret.to_bytes()[.. 32])
47      }),
48      (
49        "RUST_LOG".to_string(),
50        "serai_coordinator=trace,".to_string() + "tributary_chain=trace," + "tendermint=trace",
51      ),
52    ]
53    .into(),
54  )
55}
56
57pub fn serai_composition(name: &str) -> Composition {
58  serai_docker_tests::build("serai".to_string());
59
60  let mut composition = Composition::with_image(
61    Image::with_repository("serai-dev-serai").pull_policy(PullPolicy::Never),
62  )
63  .with_cmd(vec![
64    "serai-node".to_string(),
65    "--unsafe-rpc-external".to_string(),
66    "--rpc-cors".to_string(),
67    "all".to_string(),
68    "--chain".to_string(),
69    "local".to_string(),
70    format!("--{}", name.to_lowercase()),
71  ]);
72  composition.publish_all_ports();
73  composition
74}
75
76pub type Handles = (String, String, String);
77pub fn coordinator_stack(name: &str) -> (Handles, <Ristretto as Ciphersuite>::F, Vec<Composition>) {
78  let serai_composition = serai_composition(name);
79
80  let (coord_key, message_queue_keys, message_queue_composition) =
81    serai_message_queue_tests::instance();
82
83  let coordinator_composition = coordinator_instance(name, coord_key);
84
85  // Give every item in this stack a unique ID
86  // Uses a Mutex as we can't generate a 8-byte random ID without hitting hostname length limits
87  let (first, unique_id) = {
88    let unique_id_mutex = UNIQUE_ID.get_or_init(|| Mutex::new(0));
89    let mut unique_id_lock = unique_id_mutex.lock().unwrap();
90    let first = *unique_id_lock == 0;
91    let unique_id = hex::encode(unique_id_lock.to_be_bytes());
92    *unique_id_lock += 1;
93    (first, unique_id)
94  };
95
96  let logs_path = [std::env::current_dir().unwrap().to_str().unwrap(), ".test-logs", "coordinator"]
97    .iter()
98    .collect::<std::path::PathBuf>();
99  if first {
100    let _ = fs::remove_dir_all(&logs_path);
101    fs::create_dir_all(&logs_path).expect("couldn't create logs directory");
102    assert!(
103      fs::read_dir(&logs_path).expect("couldn't read the logs folder").next().is_none(),
104      "logs folder wasn't empty, despite removing it at the start of the run",
105    );
106  }
107  let logs_path = logs_path.to_str().unwrap().to_string();
108
109  let mut compositions = vec![];
110  let mut handles = vec![];
111  for composition in [serai_composition, message_queue_composition, coordinator_composition] {
112    let name = format!("{}-{}", composition.handle(), &unique_id);
113
114    compositions.push(
115      composition
116        .with_start_policy(StartPolicy::Strict)
117        .with_container_name(name.clone())
118        .with_log_options(Some(LogOptions {
119          action: LogAction::ForwardToFile { path: logs_path.clone() },
120          policy: LogPolicy::Always,
121          source: LogSource::Both,
122        })),
123    );
124
125    handles.push(compositions.last().unwrap().handle());
126  }
127
128  let coordinator_composition = compositions.last_mut().unwrap();
129  coordinator_composition.inject_container_name(handles.remove(0), "SERAI_HOSTNAME");
130  coordinator_composition.inject_container_name(handles.remove(0), "MESSAGE_QUEUE_RPC");
131
132  (
133    (compositions[0].handle(), compositions[1].handle(), compositions[2].handle()),
134    message_queue_keys[&NetworkId::Bitcoin],
135    compositions,
136  )
137}
138
139pub struct Processor {
140  network: NetworkId,
141
142  serai_rpc: String,
143  #[allow(unused)]
144  message_queue_handle: String,
145  #[allow(unused)]
146  coordinator_handle: String,
147
148  next_send_id: u64,
149  next_recv_id: u64,
150  queue: MessageQueue,
151}
152
153impl Processor {
154  pub async fn new(
155    network: NetworkId,
156    ops: &DockerOperations,
157    handles: (String, String, String),
158    processor_key: <Ristretto as Ciphersuite>::F,
159  ) -> Processor {
160    let message_queue_rpc = ops.handle(&handles.1).host_port(2287).unwrap();
161    let message_queue_rpc = format!("{}:{}", message_queue_rpc.0, message_queue_rpc.1);
162
163    // Sleep until the Substrate RPC starts
164    let serai_rpc = ops.handle(&handles.0).host_port(9944).unwrap();
165    let serai_rpc = format!("ws://{}:{}", serai_rpc.0, serai_rpc.1);
166    // Bound execution to 60 seconds
167    for _ in 0 .. 60 {
168      tokio::time::sleep(Duration::from_secs(1)).await;
169      let Ok(client) = Serai::new(&serai_rpc).await else { continue };
170      if client.get_latest_block_hash().await.is_err() {
171        continue;
172      }
173      break;
174    }
175
176    // The Serai RPC may or may not be started
177    // Assume it is and continue, so if it's a few seconds late, it's still within tolerance
178
179    Processor {
180      network,
181
182      serai_rpc,
183      message_queue_handle: handles.1,
184      coordinator_handle: handles.2,
185
186      next_send_id: 0,
187      next_recv_id: 0,
188      queue: MessageQueue::new(
189        Service::Processor(network),
190        message_queue_rpc,
191        Zeroizing::new(processor_key),
192      ),
193    }
194  }
195
196  pub async fn serai(&self) -> Serai {
197    Serai::new(&self.serai_rpc).await.unwrap()
198  }
199
200  /// Send a message to a processor as its coordinator.
201  pub async fn send_message(&mut self, msg: impl Into<ProcessorMessage>) {
202    let msg: ProcessorMessage = msg.into();
203    self
204      .queue
205      .queue(
206        Metadata {
207          from: Service::Processor(self.network),
208          to: Service::Coordinator,
209          intent: msg.intent(),
210        },
211        serde_json::to_string(&msg).unwrap().into_bytes(),
212      )
213      .await;
214    self.next_send_id += 1;
215  }
216
217  /// Receive a message from a processor as its coordinator.
218  pub async fn recv_message(&mut self) -> CoordinatorMessage {
219    let msg = tokio::time::timeout(Duration::from_secs(10), self.queue.next(self.next_recv_id))
220      .await
221      .unwrap();
222    assert_eq!(msg.from, Service::Coordinator);
223    assert_eq!(msg.id, self.next_recv_id);
224    self.queue.ack(self.next_recv_id).await;
225    self.next_recv_id += 1;
226    serde_json::from_slice(&msg.msg).unwrap()
227  }
228}
229*/
serai_full_stack_tests/lib.rs

serai_full_stack_tests/
lib.rs
