volli-manager 0.1.12

Manager for volli
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
95
96
97
98
99
100
101
102
103
104
105

use super::{HealthContext, ManagerContext, SecurityContext, StateContext};
use crate::{
    ManagerPeerEntry,
    connection::{CommunicationContext, manager::process_peer_update},
};
use std::collections::HashMap;
use std::sync::{
    Arc,
    atomic::{AtomicU32, AtomicU64},
};
use tempfile::TempDir;
use tokio::{
    runtime::Builder,
    sync::{Mutex, RwLock, broadcast},
};
use volli_core::{EnvironmentConfig, Role, WorkerConfig, override_config_dir, override_env_config};

#[test]
fn self_not_added_from_peer_list() {
    let base = TempDir::new().unwrap();
    let _cfg = override_config_dir(Some(base.path()));
    let _env = override_env_config(EnvironmentConfig {
        interval_ms: Some(0),
        backoff_ms: Some(0),
        ..Default::default()
    });

    let rt = Builder::new_current_thread().enable_all().build().unwrap();
    rt.block_on(async {
        let self_meta = ManagerPeerEntry {
            manager_id: "self".into(),
            manager_name: "self".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            host: "127.0.0.1".into(),
            tcp_port: 4242,
            quic_port: 4242,
            pub_fp: "0".repeat(64),
            csk_ver: 0,
            tls_cert: String::new(),
            tls_fp: String::new(),
            health: None,
        };
        let peers = Arc::new(Mutex::new(HashMap::new()));
        let (alive_tx, _) = broadcast::channel(8);
        let workers = Arc::new(Mutex::new(HashMap::new()));
        let csk = Arc::new(RwLock::new([0u8; 32]));
        let csk_ver = Arc::new(AtomicU32::new(0));
        let peer_version = Arc::new(AtomicU64::new(0));
        let (dial_tx, _dial_rx) = tokio::sync::mpsc::unbounded_channel();
        let ctx = ManagerContext::new_peer(
            SecurityContext {
                signing: None,
                csk,
                csk_ver,
            },
            StateContext {
                peers: peers.clone(),
                workers,
                self_meta: self_meta.clone(),
                peer_version,
                command_distributor: None,
            },
            CommunicationContext {
                alive_tx,
                dial_tx,
                profile: "profile".into(),
            },
            WorkerConfig {
                host: "127.0.0.1".into(),
                quic_port: 4242,
                tcp_port: 0,
                protocol: None,
                token: "t".into(),
                fingerprint: String::new(),
                cert: Vec::new(),
                role: Role::Manager,
            },
            HealthContext::default(),
        );

        let remote_meta = ManagerPeerEntry {
            manager_id: "remote".into(),
            manager_name: "remote".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            host: "127.0.0.2".into(),
            tcp_port: 0,
            quic_port: 0,
            pub_fp: "0".repeat(64),
            csk_ver: 0,
            tls_cert: String::new(),
            tls_fp: String::new(),
            health: None,
        };
        assert!(
            process_peer_update(&ctx, remote_meta.clone()).await,
            "remote peer should be persisted"
        );
        assert!(
            !process_peer_update(&ctx, self_meta.clone()).await,
            "self entry should be ignored"
        );
    });
}