tawasuyu/brahman

1//! Tests del `run_sync_async` sobre canales async in-memory.
2//!
3//! Equivalentes a los del harness síncrono pero ejecutados sobre
4//! `tokio::io::duplex` — la misma lógica protocolar viajando sobre
5//! bytes serializados con postcard, encuadrados con length-prefix, y
6//! transportados por una pipa async. Si esto pasa, lo único que falta
7//! para el sync sobre TCP/QUIC/libp2p es enchufar el transporte real.
8
9use minga_core::{parse, ContentHash, Keypair, MemStore, Mst, NodeStore};
10use minga_p2p::{run_sync_async, SyncSession};
11
12fn kp(seed: u8) -> Keypair {
13    Keypair::from_seed(&[seed; 32])
14}
15
16fn build_repo(sources: &[&str]) -> (Mst, MemStore, Vec<ContentHash>) {
17    let mut mst = Mst::new();
18    let mut store = MemStore::new();
19    let mut roots = Vec::new();
20    for src in sources {
21        let n = parse::rust(src).unwrap();
22        let h = store.put(&n);
23        mst.insert(h);
24        roots.push(h);
25    }
26    (mst, store, roots)
27}
28
29#[tokio::test]
30async fn async_sync_identical_repos() {
31    let sources = &["fn add(x: i32, y: i32) -> i32 { x + y }"];
32    let (mst_a, store_a, _) = build_repo(sources);
33    let (mst_b, store_b, _) = build_repo(sources);
34
35    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
36    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
37
38    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
39
40    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
41    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
42
43    let a = task_a.await.unwrap().unwrap();
44    let b = task_b.await.unwrap().unwrap();
45
46    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
47}
48
49#[tokio::test]
50async fn async_sync_one_empty_pulls_everything() {
51    let sources = &["fn complex(x: i32) -> i32 { let y = x * 2; y + 1 }"];
52    let (mst_a, store_a, _) = build_repo(sources);
53    let (mst_b, store_b, _) = build_repo(&[]);
54    let store_a_size = store_a.len();
55
56    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
57    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
58
59    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
60
61    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
62    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
63
64    let a = task_a.await.unwrap().unwrap();
65    let b = task_b.await.unwrap().unwrap();
66
67    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
68    assert_eq!(a.store().len(), b.store().len());
69    assert_eq!(b.store().len(), store_a_size);
70}
71
72#[tokio::test]
73async fn async_sync_disjoint_sets_merge() {
74    let only_a = &[
75        "fn alpha() -> i32 { 1 }",
76        "fn beta(x: i32) -> i32 { x + 1 }",
77    ];
78    let only_b = &[
79        "fn gamma(y: i32) -> bool { y > 0 }",
80        "fn delta() -> &'static str { \"hello\" }",
81    ];
82
83    let (mst_a, store_a, _) = build_repo(only_a);
84    let (mst_b, store_b, _) = build_repo(only_b);
85
86    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
87    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
88
89    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
90
91    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
92    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
93
94    let a = task_a.await.unwrap().unwrap();
95    let b = task_b.await.unwrap().unwrap();
96
97    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
98    assert_eq!(a.mst().len(), 4);
99}
100
101#[tokio::test]
102async fn async_sync_propagates_authenticated_identity() {
103    // Cada peer debe acabar conociendo el DID verificado del otro,
104    // exactamente como en el harness síncrono.
105    let kp_a = kp(10);
106    let kp_b = kp(20);
107    let did_a = kp_a.did();
108    let did_b = kp_b.did();
109
110    let session_a = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp_a);
111    let session_b = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp_b);
112
113    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
114
115    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
116    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
117
118    let a = task_a.await.unwrap().unwrap();
119    let b = task_b.await.unwrap().unwrap();
120
121    assert_eq!(a.peer_did(), Some(did_b));
122    assert_eq!(b.peer_did(), Some(did_a));
123}
124
125#[tokio::test]
126async fn async_sync_propagates_attestations() {
127    use minga_core::{Attestation, AttestationStore};
128
129    let kp_a = kp(30);
130    let kp_b = kp(40);
131
132    let (mst_a, store_a, roots_a) = build_repo(&["fn from_a() -> i32 { 1 }"]);
133    let (mst_b, store_b, roots_b) = build_repo(&["fn from_b() -> i32 { 2 }"]);
134
135    let mut atts_a = AttestationStore::new();
136    atts_a
137        .add(Attestation::create(&kp_a, roots_a[0]))
138        .unwrap();
139
140    let mut atts_b = AttestationStore::new();
141    atts_b
142        .add(Attestation::create(&kp_b, roots_b[0]))
143        .unwrap();
144
145    let session_a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
146    let session_b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
147
148    let (a_stream, b_stream) = tokio::io::duplex(128 * 1024);
149
150    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
151    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
152
153    let a = task_a.await.unwrap().unwrap();
154    let b = task_b.await.unwrap().unwrap();
155
156    // Los DIDs y atestaciones cruzaron correctamente sobre el wire.
157    assert_eq!(a.attestations().authors_of(&roots_a[0]), vec![kp_a.did()]);
158    assert_eq!(a.attestations().authors_of(&roots_b[0]), vec![kp_b.did()]);
159    assert_eq!(b.attestations().authors_of(&roots_a[0]), vec![kp_a.did()]);
160    assert_eq!(b.attestations().authors_of(&roots_b[0]), vec![kp_b.did()]);
161}