liminal-server 0.25.4

Standalone server for the liminal messaging bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

//! SRV-041 R4: the startup draft pull behind the listening door.
use super::pull_at_start;
use crate::auth_lys::drafts::DraftPuller;
use crate::auth_lys::drafts::PullSignal;
use crate::auth_lys::drafts::admin::{AdminDrafts, AdminLimits};
use crate::auth_lys::live::drafts::tests::{Issuer, TargetStore, limits};
use crate::auth_lys::lys_pass_adapter::LysPassAdapter;
use std::sync::Arc;

type TestResult<T = ()> = Result<T, Box<dyn std::error::Error>>;

fn route(
    issuer: &Issuer,
    dir: &std::path::Path,
) -> TestResult<tokio::sync::Mutex<AdminDrafts<TargetStore>>> {
    let puller = DraftPuller::new(
        &issuer.api,
        TargetStore::new(dir.join("site.json")),
        limits()?,
    )?;
    let adapter = Arc::new(LysPassAdapter::from_keys_json(
        "{\"keys\":[]}",
        "https://issuer.example".to_owned(),
    )?);
    let limits = AdminLimits {
        credential_bytes: limits()?.metadata_bytes,
        body_bytes: limits()?.metadata_bytes,
    };
    Ok(tokio::sync::Mutex::new(AdminDrafts::new(
        puller,
        adapter,
        "network-a".to_owned(),
        limits,
    )?))
}

fn lys_runtime() -> TestResult<tokio::runtime::Runtime> {
    Ok(tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()?)
}

#[test]
fn a_startup_pull_lys_refuses_is_logged_and_the_next_signal_pulls() -> TestResult {
    let dir = tempfile::tempdir()?;
    let issuer = Issuer::new(vec![
        (
            503,
            serde_json::json!({"refusal":"fixture_issuer_unavailable",
                "reason":"The fixture issuer is unavailable."}),
        ),
        (200, serde_json::json!({"drafts":[],"next":null,"total":0})),
    ])?;
    let route = route(&issuer, dir.path())?;
    let runtime = lys_runtime()?;
    assert!(
        !pull_at_start(&route, "connector-fixture", runtime.handle()),
        "a pull Lys refused was reported complete"
    );
    let report = runtime.block_on(async {
        route
            .lock()
            .await
            .signal("connector-fixture", PullSignal::Console, 2)
            .await
    })?;
    assert_eq!(
        report.pages, 1,
        "the console pull after a refused start did not pull"
    );
    assert_eq!(issuer.calls()?, 2);
    issuer.finish()
}

#[test]
fn a_console_pull_after_the_startup_pull_is_answered_and_never_repeats_it() -> TestResult {
    let dir = tempfile::tempdir()?;
    let empty = serde_json::json!({"drafts":[],"next":null,"total":0});
    let issuer = Issuer::new(vec![(200, empty.clone()), (200, empty)])?;
    let route = route(&issuer, dir.path())?;
    let runtime = lys_runtime()?;
    assert!(pull_at_start(&route, "connector-fixture", runtime.handle()));
    let report = runtime.block_on(async {
        route
            .lock()
            .await
            .signal("connector-fixture", PullSignal::Console, 3)
            .await
    })?;
    assert_eq!(report.pages, 1);
    assert_eq!(issuer.calls()?, 2);
    issuer.finish()
}