Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_o3db_sync/tests/basic.rs

11.0 KiB, 93 runs

created by r1870400018:834, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1use oxedyne_fe2o3_core::{
2 prelude::*,
3 alt::Override,
4 rand::Rand,
5};
6use oxedyne_fe2o3_crypto::enc::EncryptionScheme;
7use oxedyne_fe2o3_hash::{
8 csum::ChecksumScheme,
9 hash::HashScheme,
10};
11use oxedyne_fe2o3_iop_db::api::{
12 Meta,
13 RestSchemesOverride,
14};
15use oxedyne_fe2o3_jdat::{
16 prelude::*,
17};
18use oxedyne_fe2o3_o3db_sync::{
19 base::{
20 constant,
21 index::ZoneInd,
22 },
23 comm::{
24 response::Wait,
25 },
26 data::core::RestSchemesInput,
27 file::zdir::ZoneDir,
28 test::{
29 dbapi,
30 file::{
31 delete_all_index_files,
32 corrupt_an_index_file,
33 },
34 setup,
35 },
36};
37use oxedyne_fe2o3_test::error::delayed_error;
38
39use std::{
40 collections::BTreeMap,
41 path::Path,
42 thread,
43 time::Duration,
44};
45
46
47const wait: Wait = constant::USER_REQUEST_WAIT;
48
49pub fn test_basic(_filter: &'static str) -> Outcome<()> {
50
51 res!(std::fs::create_dir_all("./test_db"));
52 let db_root = res!(Path::new("./test_db").canonicalize());
53 // + + + +
54 // 1234567890123456789012345678901234
55 let key = dat!("A long key exceeding 32 bytes");
56 let mut valvec = vec![0u8; 5_000];
57 Rand::fill_u8(&mut valvec[..]);
58 let val = dat!(valvec.clone());
59
60 let mut enckey = [0u8; 32];
61 Rand::fill_u8(&mut enckey);
62 let aes_gcm = res!(EncryptionScheme::new_aes_256_gcm_with_key(&enckey[..]));
63 let _sha3_256 = HashScheme::new_sha3_256();
64 let crc32 = ChecksumScheme::new_crc32();
65 let schms2: RestSchemesOverride<EncryptionScheme, HashScheme> =
66 RestSchemesOverride::default().set_encrypter(Override::Default(aes_gcm.clone()));
67 let schms2 = Some(&schms2);
68 let user = setup::Uid::default();
69 //let meta = Meta::<{ setup::UID_LEN }, setup::Uid>::new(setup::Uid::default());
70 let schms_input = RestSchemesInput::new(
71 Some(aes_gcm.clone()),
72 None::<HashScheme>,
73 None::<HashScheme>,
74 Some(crc32.clone()),
75 );
76 //let user1_name = fmt!("Alice83");
77 //let user1_email = fmt!("alice83@gmail.com");
78 //let user1_pass = fmt!("alice_pass");
79
80 let mut cfg = res!(setup::default_cfg());
81 cfg.cache_size_limit_bytes = 100_000;
82 cfg.rest_chunk_threshold = 700;
83 cfg.num_cbots_per_zone = 2;
84 cfg.num_zones = 3;
85 cfg.zone_overrides = mapdat!{
86 1u16 => mapdat!{
87 "dir" => "../test_db_zone_container",
88 "max_size" => 100u64,
89 },
90 3u16 => mapdat!{
91 "dir" => "",
92 "max_size" => 100u64,
93 },
94 }.get_map().unwrap();
95
96 let error_delay = 2;
97
98 {
99 test!(sync_log::stream(), "+---------------------------------------------+");
100 test!(sync_log::stream(), "| NEW OZONE SESSION |");
101 test!(sync_log::stream(), "| Wipe all traces of previous test. |");
102 test!(sync_log::stream(), "| Start database. |");
103 test!(sync_log::stream(), "| Store and fetch some simple data. |");
104 test!(sync_log::stream(), "| Store and fetch some chunked data: |");
105 test!(sync_log::stream(), "| * Including one cycle wiping the cache. |");
106 test!(sync_log::stream(), "| Gracefully shut down the database. |");
107 test!(sync_log::stream(), "+---------------------------------------------+");
108 // Wipe all traces of previous test.
109 // Start database.
110 let mut db = match setup::start_db(
111 db_root.clone(),
112 Some(cfg.clone()),
113 schms_input.clone(),
114 Some(fmt!("./test_db_zone_container")),
115 true,
116 true,
117 ) {
118 // These pauses on errors are needed to capture tardy messages from asynchronous
119 // logging.
120 Err(e) => return Err(delayed_error(e, error_delay)),
121 Ok(db) => db,
122 };
123
124 thread::sleep(Duration::from_secs(1));
125
126 // Store and fetch some simple data.
127 match dbapi::simple(&mut db, schms2, user) {
128 Err(e) => return Err(delayed_error(e, error_delay)),
129 _ => (),
130 }
131
132 // Store and fetch some simple data.
133 match dbapi::simple_api(&mut db, user) {
134 Err(e) => return Err(delayed_error(e, error_delay)),
135 _ => (),
136 }
137
138 // Store and fetch some chunked data:
139 // * Including one cycle wiping the cache.
140 match dbapi::store_chunked_data(
141 &mut db,
142 schms2,
143 user,
144 key.clone(),
145 val.clone(),
146 ) {
147 Err(e) => return Err(delayed_error(e, error_delay)),
148 _ => (),
149 };
150
151 //res!(db.dump_caches(constant::USER_REQUEST_WAIT));
152
153 // While debugging, extra work can mean we start reading data before bots have written it.
154 thread::sleep(Duration::from_secs(2));
155 match dbapi::fetch_chunked_data(
156 &mut db,
157 &key,
158 &valvec,
159 user,
160 schms2,
161 ) {
162 Err(e) => return Err(delayed_error(e, error_delay)),
163 _ => (),
164 }
165 test!(sync_log::stream(), "Listing files...");
166 res!(db.api().list_files(wait));
167 //res!(db.dump_caches(constant::USER_REQUEST_WAIT));
168 test!(sync_log::stream(), "Shutting db down...");
169 // Gracefully shut down the database.
170 res!(db.shutdown());
171 }
172
173 thread::sleep(Duration::from_secs(1));
174
175 let zdirs: BTreeMap<ZoneInd, ZoneDir>;
176
177 {
178 test!(sync_log::stream(), "+---------------------------------------------+");
179 test!(sync_log::stream(), "| NEW OZONE SESSION |");
180 test!(sync_log::stream(), "| Start database: |");
181 test!(sync_log::stream(), "| * Including caching index files. |");
182 test!(sync_log::stream(), "| Fetch chunked data from previous session. |");
183 test!(sync_log::stream(), "| Gracefully shut down the database. |");
184 test!(sync_log::stream(), "+---------------------------------------------+");
185 let mut db = match setup::start_db(
186 db_root.clone(),
187 Some(cfg.clone()),
188 schms_input.clone(),
189 Some(fmt!("./test_db_zone_container")),
190 true,
191 false,
192 ) {
193 Err(e) => return Err(delayed_error(e, error_delay)),
194 Ok(db) => db,
195 };
196
197 thread::sleep(Duration::from_secs(1));
198
199 zdirs = res!(db.api().get_zone_dirs());
200 //res!(db.dump_caches(constant::USER_REQUEST_WAIT));
201 match dbapi::fetch_chunked_data(
202 &mut db,
203 &key,
204 &valvec,
205 user,
206 schms2,
207 ) {
208 Err(e) => return Err(delayed_error(e, error_delay)),
209 _ => (),
210 }
211
212 test!(sync_log::stream(), "Demonstrating collecting the state of ozone resources, ");
213 test!(sync_log::stream(), "which is regularly reported by each zone to the supervisor.");
214 let zstats = res!(db.api().ozone_state(constant::USER_REQUEST_WAIT));
215 for (i, zstat) in zstats.iter().enumerate() {
216 test!(sync_log::stream(), "Zone {} {:?}", i+1, zstat);
217 }
218
219 test!(sync_log::stream(), "Shutting db down...");
220 res!(db.shutdown());
221 }
222
223 thread::sleep(Duration::from_secs(1));
224
225 res!(delete_all_index_files(&zdirs));
226
227 {
228 test!(sync_log::stream(), "+---------------------------------------------+");
229 test!(sync_log::stream(), "| NEW OZONE SESSION |");
230 test!(sync_log::stream(), "| Delete all index files. |");
231 test!(sync_log::stream(), "| Start database: |");
232 test!(sync_log::stream(), "| * Including caching data files. |");
233 test!(sync_log::stream(), "| Fetch chunked data from previous session. |");
234 test!(sync_log::stream(), "| Gracefully shut down the database. |");
235 test!(sync_log::stream(), "+---------------------------------------------+");
236 let mut db = match setup::start_db(
237 db_root.clone(),
238 Some(cfg.clone()),
239 schms_input.clone(),
240 Some(fmt!("./test_db_zone_container")),
241 true,
242 false,
243 ) {
244 Err(e) => return Err(delayed_error(e, error_delay)),
245 Ok(db) => db,
246 };
247
248 match dbapi::fetch_chunked_data(
249 &mut db,
250 &key,
251 &valvec,
252 user,
253 schms2,
254 ) {
255 Err(e) => return Err(delayed_error(e, error_delay)),
256 _ => (),
257 }
258
259 test!(sync_log::stream(), "Listing files...");
260 res!(db.api().list_files(wait));
261 test!(sync_log::stream(), "Shutting db down...");
262 res!(db.shutdown());
263 }
264
265 thread::sleep(Duration::from_secs(1));
266
267 res!(corrupt_an_index_file(&zdirs));
268
269 {
270 test!(sync_log::stream(), "+---------------------------------------------+");
271 test!(sync_log::stream(), "| NEW OZONE SESSION |");
272 test!(sync_log::stream(), "| Corrupt a single byte of one index file. |");
273 test!(sync_log::stream(), "| Start database: |");
274 test!(sync_log::stream(), "| * Including caching index files. |");
275 test!(sync_log::stream(), "| Fetch chunked data from previous session. |");
276 test!(sync_log::stream(), "| Gracefully shut down the database. |");
277 test!(sync_log::stream(), "| |");
278 test!(sync_log::stream(), "| Note: We are expecting one error when an |");
279 test!(sync_log::stream(), "| igcbot discovers the corruption in the |");
280 test!(sync_log::stream(), "| index file and switches to indexing the |");
281 test!(sync_log::stream(), "| data file. |");
282 test!(sync_log::stream(), "+---------------------------------------------+");
283
284 let mut db = match setup::start_db(
285 db_root.clone(),
286 Some(cfg.clone()),
287 schms_input.clone(),
288 Some(fmt!("./test_db_zone_container")),
289 true,
290 false,
291 ) {
292 Err(e) => return Err(delayed_error(e, error_delay)),
293 Ok(db) => db,
294 };
295
296 test!(sync_log::stream(), "We expected an error during initial caching of an index file.");
297 test!(sync_log::stream(), "This appears to have successfully resolved when the data file was instead cached.");
298
299 test!(sync_log::stream(), "Listing files...");
300 res!(db.api().list_files(wait));
301
302 match dbapi::fetch_chunked_data(
303 &mut db,
304 &key,
305 &valvec,
306 user,
307 schms2,
308 ) {
309 Err(e) => return Err(delayed_error(e, error_delay)),
310 _ => (),
311 }
312 test!(sync_log::stream(), "Shutting db down...");
313 res!(db.shutdown());
314 }
315
316 Ok(())
317}