olai_stack_runtime/
fake.rs

1//! [`FakeEngine`]: an in-memory [`ComposeEngine`] for tests (feature `testing`).
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::path::{Path, PathBuf};
5use std::sync::{Arc, Mutex, MutexGuard};
6
7use async_trait::async_trait;
8use futures::StreamExt;
9use tokio::sync::broadcast;
10
11use crate::engine::{ComposeEngine, DownOptions, EngineStream, LogOptions, UpOptions};
12use crate::error::{Error, Result};
13use crate::events::{EngineEvent, EventAction, EventFilter};
14use crate::logs::LogLine;
15use crate::status::{Health, ProjectStatus, ServiceStatus, State};
16
17/// One call the fake received, for assertions.
18#[derive(Clone, Debug, PartialEq, Eq)]
19pub enum FakeCall {
20    /// [`up`](ComposeEngine::up) of the project `project` in `dir`, with the `env` it was given.
21    Up {
22        /// The project directory.
23        dir: PathBuf,
24        /// The project name read from its head `compose.yaml`.
25        project: String,
26        /// The variables passed.
27        env: BTreeMap<String, String>,
28    },
29    /// [`down`](ComposeEngine::down) of `project`.
30    Down {
31        /// The project name.
32        project: String,
33        /// Whether volumes were removed.
34        volumes: bool,
35    },
36    /// [`network_rm`](ComposeEngine::network_rm) of `name`.
37    NetworkRm(String),
38}
39
40#[derive(Default)]
41struct Inner {
42    projects: BTreeMap<String, Vec<ServiceStatus>>,
43    /// The directory each project was started from, as Compose reports its config files.
44    dirs: BTreeMap<String, PathBuf>,
45    /// The services (and their health once up) a project gets, set by the test.
46    scripts: BTreeMap<String, Vec<(String, Health)>>,
47    fail_up: BTreeSet<String>,
48    networks: BTreeSet<String>,
49    unavailable: bool,
50    calls: Vec<FakeCall>,
51    clock: i64,
52}
53
54/// An in-memory Docker: `up` creates a project with scripted services, `down` removes it, and
55/// both raise the [events](ComposeEngine::events) Docker would. Clones share state, so a test
56/// can hand one to the code under test and inspect another.
57///
58/// A project's name is read from the `name:` line of the head `compose.yaml` in the directory
59/// `up` is given, as Compose does for the files trestle renders. Its services default to a
60/// single healthy `app`; [`script`](Self::script) sets others.
61#[derive(Clone)]
62pub struct FakeEngine {
63    inner: Arc<Mutex<Inner>>,
64    events: broadcast::Sender<EngineEvent>,
65}
66
67impl Default for FakeEngine {
68    fn default() -> Self {
69        FakeEngine {
70            inner: Arc::default(),
71            events: broadcast::channel(1024).0,
72        }
73    }
74}
75
76impl FakeEngine {
77    /// An empty Docker host.
78    pub fn new() -> Self {
79        Self::default()
80    }
81
82    fn lock(&self) -> MutexGuard<'_, Inner> {
83        self.inner.lock().unwrap_or_else(|e| e.into_inner())
84    }
85
86    /// Give `project` these services, each reaching `health` once up.
87    pub fn script(
88        &self,
89        project: &str,
90        services: impl IntoIterator<Item = (&'static str, Health)>,
91    ) {
92        self.lock().scripts.insert(
93            project.to_owned(),
94            services
95                .into_iter()
96                .map(|(s, h)| (s.to_owned(), h))
97                .collect(),
98        );
99    }
100
101    /// Make the next `up` of `project` fail, as an unhealthy `--wait` does.
102    pub fn fail_up(&self, project: &str) {
103        self.lock().fail_up.insert(project.to_owned());
104    }
105
106    /// Make every call fail as if the daemon were down (or back up with `false`).
107    pub fn set_unavailable(&self, unavailable: bool) {
108        self.lock().unavailable = unavailable;
109    }
110
111    /// Add a project as if something else had started it (e.g. a runtime that since died).
112    pub fn adopt(&self, project: &str) {
113        let services = self.services_for(project);
114        self.lock().projects.insert(project.to_owned(), services);
115    }
116
117    /// Declare the network `name`, as the first project joining it would.
118    pub fn add_network(&self, name: &str) {
119        self.lock().networks.insert(name.to_owned());
120    }
121
122    /// The networks that exist.
123    pub fn networks(&self) -> BTreeSet<String> {
124        self.lock().networks.clone()
125    }
126
127    /// The projects that exist.
128    pub fn projects(&self) -> BTreeSet<String> {
129        self.lock().projects.keys().cloned().collect()
130    }
131
132    /// Every call so far, in order.
133    pub fn calls(&self) -> Vec<FakeCall> {
134        self.lock().calls.clone()
135    }
136
137    /// Change the health of `service` in `project`, raising a health event.
138    pub fn set_health(&self, project: &str, service: &str, health: Health) {
139        let mut inner = self.lock();
140        let Some(container) = inner.projects.get_mut(project).and_then(|services| {
141            let s = services.iter_mut().find(|s| s.service == service)?;
142            s.health = health;
143            Some(s.container.clone())
144        }) else {
145            return;
146        };
147        let event = next_event(
148            &mut inner,
149            project,
150            service,
151            &container,
152            EventAction::Health(health),
153        );
154        drop(inner);
155        let _ = self.events.send(event);
156    }
157
158    fn services_for(&self, project: &str) -> Vec<ServiceStatus> {
159        let script = self
160            .lock()
161            .scripts
162            .get(project)
163            .cloned()
164            .unwrap_or_else(|| vec![("app".to_owned(), Health::Healthy)]);
165        script
166            .into_iter()
167            .map(|(service, health)| ServiceStatus {
168                container: format!("{project}-{service}-1"),
169                service,
170                state: State::Running,
171                health,
172                exit_code: 0,
173                status: "Up".into(),
174                published: Vec::new(),
175            })
176            .collect()
177    }
178
179    fn check_available(&self) -> Result<()> {
180        if self.lock().unavailable {
181            return Err(Error::Unavailable {
182                reason: "the fake Docker daemon is down".into(),
183            });
184        }
185        Ok(())
186    }
187
188    fn emit(&self, project: &str, services: &[ServiceStatus], actions: &[fn() -> EventAction]) {
189        let mut inner = self.lock();
190        let mut events = Vec::new();
191        for s in services {
192            for action in actions {
193                events.push(next_event(
194                    &mut inner,
195                    project,
196                    &s.service,
197                    &s.container,
198                    action(),
199                ));
200            }
201        }
202        drop(inner);
203        for e in events {
204            let _ = self.events.send(e);
205        }
206    }
207}
208
209fn next_event(
210    inner: &mut Inner,
211    project: &str,
212    service: &str,
213    container: &str,
214    action: EventAction,
215) -> EngineEvent {
216    inner.clock += 1;
217    EngineEvent {
218        project: project.to_owned(),
219        service: service.to_owned(),
220        container: container.to_owned(),
221        action,
222        time_nanos: inner.clock,
223    }
224}
225
226/// The `name:` of the head `compose.yaml` in `dir`.
227fn project_in(dir: &Path) -> Result<String> {
228    let path = dir.join("compose.yaml");
229    let text = std::fs::read_to_string(&path).map_err(|source| Error::Io {
230        command: format!("read {}", path.display()),
231        source,
232    })?;
233    text.lines()
234        .find_map(|l| l.strip_prefix("name:"))
235        .map(|n| n.trim().trim_matches('"').to_owned())
236        .ok_or_else(|| Error::Io {
237            command: format!("read {}", path.display()),
238            source: std::io::Error::other("no `name:` in the head compose.yaml"),
239        })
240}
241
242#[async_trait]
243impl ComposeEngine for FakeEngine {
244    async fn up(&self, dir: &Path, opts: &UpOptions) -> Result<()> {
245        self.check_available()?;
246        let project = project_in(dir)?;
247        self.lock().calls.push(FakeCall::Up {
248            dir: dir.to_owned(),
249            project: project.clone(),
250            env: opts.env.clone(),
251        });
252        let services = self.services_for(&project);
253        let failed = self.lock().fail_up.remove(&project);
254        self.lock()
255            .projects
256            .insert(project.clone(), services.clone());
257        self.lock().dirs.insert(project.clone(), dir.to_owned());
258        self.emit(
259            &project,
260            &services,
261            &[|| EventAction::Create, || EventAction::Start],
262        );
263        if failed {
264            return Err(Error::Failed {
265                command: "compose up -d --wait".into(),
266                status: exit_status(1),
267                stderr: format!("container {project}-app-1 is unhealthy"),
268            });
269        }
270        Ok(())
271    }
272
273    async fn down(&self, dir: &Path, opts: &DownOptions) -> Result<()> {
274        self.check_available()?;
275        let project = project_in(dir)?;
276        self.lock().calls.push(FakeCall::Down {
277            project: project.clone(),
278            volumes: opts.volumes,
279        });
280        let removed = self.lock().projects.remove(&project);
281        self.lock().dirs.remove(&project);
282        if let Some(services) = removed {
283            self.emit(
284                &project,
285                &services,
286                &[
287                    || EventAction::Stop,
288                    || EventAction::Die { exit_code: Some(0) },
289                    || EventAction::Destroy,
290                ],
291            );
292        }
293        Ok(())
294    }
295
296    async fn ls(&self) -> Result<Vec<ProjectStatus>> {
297        self.check_available()?;
298        let inner = self.lock();
299        Ok(inner
300            .projects
301            .iter()
302            .map(|(name, services)| {
303                let mut states = BTreeMap::new();
304                for s in services {
305                    let state = match &s.state {
306                        State::Running => "running",
307                        State::Exited => "exited",
308                        _ => "created",
309                    };
310                    *states.entry(state.to_owned()).or_insert(0) += 1;
311                }
312                let status = states
313                    .iter()
314                    .map(|(s, n)| format!("{s}({n})"))
315                    .collect::<Vec<_>>()
316                    .join(", ");
317                ProjectStatus {
318                    name: name.clone(),
319                    status,
320                    states,
321                    config_files: inner
322                        .dirs
323                        .get(name)
324                        .map(|d| d.join("compose.yaml").display().to_string())
325                        .unwrap_or_default(),
326                }
327            })
328            .collect())
329    }
330
331    async fn ps(&self, project: &str) -> Result<Vec<ServiceStatus>> {
332        self.check_available()?;
333        Ok(self
334            .lock()
335            .projects
336            .get(project)
337            .cloned()
338            .unwrap_or_default())
339    }
340
341    async fn logs(&self, project: &str, _opts: &LogOptions) -> Result<EngineStream<LogLine>> {
342        self.check_available()?;
343        let lines: Vec<Result<LogLine>> = self
344            .lock()
345            .projects
346            .get(project)
347            .into_iter()
348            .flatten()
349            .map(|s| {
350                Ok(LogLine {
351                    container: format!("{}-1", s.service),
352                    timestamp: None,
353                    text: format!("{} started", s.service),
354                })
355            })
356            .collect();
357        Ok(futures::stream::iter(lines).boxed())
358    }
359
360    async fn events(&self, filter: &EventFilter) -> Result<EngineStream<EngineEvent>> {
361        self.check_available()?;
362        let filter = filter.clone();
363        let rx = self.events.subscribe();
364        let stream = futures::stream::unfold(rx, |mut rx| async move {
365            loop {
366                match rx.recv().await {
367                    Ok(event) => return Some((event, rx)),
368                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
369                    Err(broadcast::error::RecvError::Closed) => return None,
370                }
371            }
372        })
373        .filter(move |e| futures::future::ready(filter.matches(e)))
374        .map(Ok);
375        Ok(stream.boxed())
376    }
377
378    async fn network_rm(&self, name: &str) -> Result<bool> {
379        self.check_available()?;
380        let mut inner = self.lock();
381        inner.calls.push(FakeCall::NetworkRm(name.to_owned()));
382        Ok(inner.networks.remove(name))
383    }
384}
385
386#[cfg(unix)]
387fn exit_status(code: i32) -> std::process::ExitStatus {
388    use std::os::unix::process::ExitStatusExt;
389    std::process::ExitStatus::from_raw(code << 8)
390}
391
392#[cfg(windows)]
393fn exit_status(code: i32) -> std::process::ExitStatus {
394    use std::os::windows::process::ExitStatusExt;
395    std::process::ExitStatus::from_raw(code as u32)
396}
olai_stack_runtime/fake.rs

olai_stack_runtime/
fake.rs
