sofie_testing/
messages.rs

1use std::time::Duration;
2
3use sofie_core::{SqlValue, Timestamp};
4use sofie_db::Event;
5use sofie_db::worker::{Drained, Failure};
6
7use crate::TestApp;
8
9pub(crate) async fn advance(app: &TestApp, by: Duration) -> usize {
10    let until = now(app)
11        .checked_add(by)
12        .expect("advancing within the calendar");
13    let mut completed = app.drain().await;
14    while let Some(due) = next_due(app, until).await {
15        move_to(app, due);
16        completed += app.drain().await;
17    }
18    move_to(app, until);
19    completed + app.drain().await
20}
21
22pub(crate) async fn drain(app: &TestApp) -> usize {
23    let drained = drain_reporting(app).await;
24    assert!(
25        drained.failures.is_empty(),
26        "jobs failed while draining: {:?}",
27        drained.failures
28    );
29    drained.completed
30}
31
32/// Each failed job is retried once its backoff is over, as `advance` reaches it.
33pub async fn drain_failing(app: &TestApp) -> Vec<Failure> {
34    drain_reporting(app).await.failures
35}
36
37async fn drain_reporting(app: &TestApp) -> Drained {
38    app.worker
39        .drain_reporting()
40        .await
41        .expect("draining the worker")
42}
43
44/// Every `E` committed, delivered or not; for a test that waits until one is.
45#[must_use]
46pub fn raised<E: Event>(app: &TestApp) -> Vec<E> {
47    sofie_db::raised(app.db(), E::NAME)
48        .into_iter()
49        .map(|payload| E::from_json(payload).expect("reading a raised event"))
50        .collect()
51}
52
53pub fn assert_raised<E: Event>(app: &TestApp) -> Vec<E> {
54    let raised = raised::<E>(app);
55    assert!(!raised.is_empty(), "no `{}` was raised", E::NAME);
56    raised
57}
58
59pub(crate) fn now(app: &TestApp) -> Timestamp {
60    app.clock.clock().now().timestamp()
61}
62
63async fn next_due(app: &TestApp, until: Timestamp) -> Option<Timestamp> {
64    let rows = app
65        .query(
66            "SELECT min(due) FROM (
67                SELECT next_at AS due FROM sofie.schedules WHERE next_at > $1 AND next_at <= $2
68                UNION ALL
69                SELECT available_at FROM sofie.messages
70                WHERE dead_at IS NULL AND available_at > $1 AND available_at <= $2
71            ) AS due",
72            &[SqlValue::Timestamp(now(app)), SqlValue::Timestamp(until)],
73        )
74        .await;
75    match rows.first().and_then(|row| row.first()) {
76        Some(SqlValue::Timestamp(due)) => Some(*due),
77        _ => None,
78    }
79}
80
81fn move_to(app: &TestApp, moment: Timestamp) {
82    let micros = moment.as_microsecond() - now(app).as_microsecond();
83    let forward = u64::try_from(micros).expect("the clock only moves forward");
84    app.clock
85        .advance(Duration::from_micros(forward))
86        .expect("moving the test clock");
87}
sofie_testing/messages.rs

sofie_testing/
messages.rs
