Repository: allenai/dolma (language: python)

Issue:
Add rust unit tests
I'm starting with bloom and for the simple functions but I hope to expand coverage.

This will help with refactoring while ensuring correctness


Discussion / hints:

Fix patch:
diff --git a/src/bloom_filter.rs b/src/bloom_filter.rs
index 9ede63bd..f7805e4c 100644
--- a/src/bloom_filter.rs
+++ b/src/bloom_filter.rs
@@ -11,6 +11,7 @@ use std::mem::size_of;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicU32, Ordering};
 
+mod bloom_test;
 // A thread-safe bloom filter.
 pub struct BloomFilter {
     bits: Vec<AtomicU32>,
@@ -85,13 +86,10 @@ impl BloomFilter {
             hash_builder_seeds.push(seeds);
         }
 
-        let mut bits = Vec::new();
         let number_of_u32 = size_in_bytes / size_of::<AtomicU32>();
-        bits.reserve_exact(number_of_u32);
-        for _ in 0..number_of_u32 {
-            bits.push(AtomicU32::new(0));
-        }
-
+        let bits: Vec<AtomicU32> = std::iter::repeat_with(|| AtomicU32::new(0))
+            .take(number_of_u32)
+            .collect();
         Self {
             bits,
             hash_builder_seeds,
@@ -138,8 +136,7 @@ impl BloomFilter {
         }
 
         let number_of_elements = stream.read_u64::<LittleEndian>()?;
-        let mut bits = Vec::new();
-        bits.reserve_exact(number_of_elements as usize);
+        let mut bits = Vec::with_capacity(number_of_elements as usize);
         for _ in 0..number_of_elements {
             bits.push(AtomicU32::new(stream.read_u32::<NativeEndian>()?));
         }
@@ -220,8 +217,7 @@ impl BloomFilter {
                 return false;
             }
         }
-
-        return true;
+        true
     }
 
     pub fn contains(&self, s: &VecDeque<&str>) -> bool {
@@ -237,7 +233,7 @@ impl BloomFilter {
         } else {
             log::info!("Creating new bloom filter...");
             let mut bloom_filter_size: usize = config.size_in_bytes;
-            if bloom_filter_size <= 0 {
+            if bloom_filter_size == 0 {
                 bloom_filter_size = BloomFilter::suggest_size_in_bytes(
                     config.estimated_doc_count,
                     config.desired_false_positive_rate,
diff --git a/src/deduper.rs b/src/deduper.rs
index 2fcb7b28..98e6694b 100644
--- a/src/deduper.rs
+++ b/src/deduper.rs
@@ -43,12 +43,9 @@ pub fn run(config: DeduperConfig) -> Result<u32, u32> {
         let failed_shard_count_ref = failed_shard_count_ref.clone();
         threadpool.execute(move || {
             let result = write_attributes(path, work_dirs, dedupe, bloom_filter);
-            match result {
-                Ok(_) => {}
-                Err(e) => {
-                    log::error!("Failed to process {:?}: {}", p, e);
-                    failed_shard_count_ref.fetch_add(1, Ordering::Relaxed);
-                }
+            if let Err(e) = result {
+                log::error!("Failed to process {:?}: {}", p, e);
+                failed_shard_count_ref.fetch_add(1, Ordering::Relaxed);
             }
         });
     }
@@ -56,20 +53,22 @@ pub fn run(config: DeduperConfig) -> Result<u32, u32> {
 
     let bloom_filter_file = PathBuf::from(&config.bloom_filter.file);
     log::info!("Writing bloom filter to {:?}...", config.bloom_filter.file);
-    bloom_filter.write_to_file(&bloom_filter_file).unwrap();
-    log::info!("Bloom filter written.");
-
-    let failure_count = failed_shard_count_ref.fetch_add(0, Ordering::Relaxed);
-    match failure_count {
-        0 => {
-            log::info!("Done!");
-            return Ok(failure_count);
-        }
-        _ => {
-            log::error!("{} shards failed to process.", failure_count);
-            return Err(failure_count);
+    match bloom_filter.write_to_file(&bloom_filter_file) {
+        Ok(_) => log::info!("Bloom filter written."),
+        Err(e) => {
+            log::error!("Write failed: {}", e);
+            panic!("Failed to write bloom filter");
         }
     }
+
+    let failure_count = failed_shard_count_ref.load(Ordering::Relaxed);
+    if failure_count == 0 {
+        log::info!("Done!");
+        Ok(failure_count)
+    } else {
+        log::error!("{} shards failed to process.", failure_count);
+        Err(failure_count)
+    }
 }
 
 // Write attributes for the documents in the given file:
@@ -87,12 +86,8 @@ fn write_attributes(
     };
 
     let attrs_location = {
-        let mut attr_prefix = "/attributes/".to_owned();
-        attr_prefix.push_str(&dedupe_config.name);
-        attr_prefix.push_str("/");
-        docs_location
-            .to_owned()
-            .replace("/documents/", &attr_prefix)
+        let attr_prefix = format!("/attributes/{}/", &dedupe_config.name);
+        docs_location.replace("/documents/", &attr_prefix)
     };
     let local_output = cache.prepare_output(&attrs_location)?;
     if local_output.exists() {
@@ -133,10 +128,9 @@ fn write_attributes(
             GzEncoder::new(tmp_output, Compression::default()),
         );
 
-        let mut line_number = 0;
-        for line in reader.lines() {
-            match line {
-                Ok(_) => {}
+        for (line_number, line) in reader.lines().enumerate() {
+            let line = match line {
+                Ok(line) => line,
                 Err(e) => {
                     log::error!(
                         "Error reading line {} of {}: {}",
@@ -146,45 +140,39 @@ fn write_attributes(
                     );
                     break;
                 }
-            }
-            line_number += 1;
-            let line = line?;
+            };
             let data: Value = serde_json::from_str(&line)?;
             let mut attributes = json!({});
 
-            match dedupe_config.documents {
-                Some(ref cfg) => {
-                    let document_key = {
-                        let mut finder = jsonpath_rust::JsonPathFinder::from_str("{}", &cfg.key)
-                            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
-                            .unwrap();
-                        finder.set_json(Box::new(data.clone()));
-                        finder
-                            .find()
-                            .as_array()
-                            .unwrap()
-                            .get(0)
-                            .unwrap()
-                            .as_str()
-                            .unwrap()
-                            .to_string()
-                    };
+            if let Some(ref cfg) = dedupe_config.documents {
+                let document_key = {
+                    let mut finder = jsonpath_rust::JsonPathFinder::from_str("{}", &cfg.key)
+                        .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
+                        .unwrap();
+                    finder.set_json(Box::new(data.clone()));
+                    finder
+                        .find()
+                        .as_array()
+                        .unwrap()
+                        .get(0)
+                        .unwrap()
+                        .as_str()
+                        .unwrap()
+                        .to_string()
+                };
 
-                    if dedupe_config.skip_empty.unwrap_or(false) && document_key.trim().is_empty() {
-                        // skip empty documents if dedupe_config.skip_empty is true
-                        // and the document key is empty after trimming (i.e., removing whitespace)
-                        continue;
-                    } else {
-                        let mut dedupe_key = VecDeque::with_capacity(1);
-                        dedupe_key.push_back(document_key.as_str());
-                        if bloom_filter.contains(&dedupe_key) {
-                            attributes[&cfg.attribute_name] = Value::Bool(true);
-                        } else if !bloom_filter.read_only {
-                            bloom_filter.insert(&dedupe_key);
-                        }
+                if dedupe_config.skip_empty.unwrap_or(false) && document_key.trim().is_empty() {
+                    // skip empty documents if dedupe_config.skip_empty is true
+                    // and the document key is empty after trimming (i.e., removing whitespace)
+                    continue;
+                } else {
+                    let dedupe_key = VecDeque::from([document_key.as_str()]);
+                    if bloom_filter.contains(&dedupe_key) {
+                        attributes[&cfg.attribute_name] = Value::Bool(true);
+                    } else if !bloom_filter.read_only {
+                        bloom_filter.insert(&dedupe_key);
                     }
                 }
-                None => {}
             }
             match dedupe_config.paragraphs {
                 None => {}
@@ -193,7 +181,7 @@ fn write_attributes(
                     let text = data["text"].as_str().unwrap();
                     let text_length = text.len();
                     let mut offset = 0;
-                    let paragraphs = text.split("\n");
+                    let paragraphs = text.split('\n');
                     let mut duplicate_paragraph_spans = Vec::new();
                     for p in paragraphs {
                         let par_start = offset;
@@ -208,13 +196,12 @@ fn write_attributes(
                             // and the paragraph is empty after trimming (i.e., removing whitespace)
                             continue;
                         } else {
-                            let mut dedupe_key = VecDeque::with_capacity(1);
-                            dedupe_key.push_back(p);
+                            let dedupe_key = VecDeque::from([p]);
                             if bloom_filter.contains(&dedupe_key) {
                                 let span = vec![
                                     Value::Number(par_start.into()),
                                     Value::Number(par_end.into()),
-                                    Value::Number(1.into()),
+                                    Value::from(1),
                                 ];
                                 // add span to duplicate_paragraph_spans
                                 duplicate_paragraph_spans.push(Value::Array(span));
diff --git a/src/lib.rs b/src/lib.rs
index ade5dcc6..3add8972 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -15,26 +15,26 @@ use std::env;
 fn deduper_entrypoint(config_str: &str) -> PyResult<()> {
     let config: DeduperConfig = DeduperConfig::parse_from_string(config_str).unwrap();
 
-    match deduper::run(config) {
-        Ok(_) => Ok(()),
-        Err(cnt) => Err(exceptions::PyRuntimeError::new_err(format!(
+    if let Err(cnt) = deduper::run(config) {
+        return Err(exceptions::PyRuntimeError::new_err(format!(
             "Failed with {} errors",
             cnt
-        ))),
+        )));
     }
+    Ok(())
 }
 
 #[pyfunction]
 fn mixer_entrypoint(config_str: &str) -> PyResult<()> {
     //Result<u32, PyErr> {
     let config: MixerConfig = MixerConfig::parse_from_string(config_str).unwrap();
-    match mixer::run(config) {
-        Ok(_) => Ok(()),
-        Err(cnt) => Err(exceptions::PyRuntimeError::new_err(format!(
+    if let Err(cnt) = mixer::run(config) {
+        return Err(exceptions::PyRuntimeError::new_err(format!(
             "Failed with {} errors",
             cnt
-        ))),
+        )));
     }
+    Ok(())
 }
 
 // A Python module implemented in Rust. The name of this function must match
diff --git a/src/mixer.rs b/src/mixer.rs
index 0a140618..d89ab939 100644
--- a/src/mixer.rs
+++ b/src/mixer.rs
@@ -12,8 +12,7 @@ pub fn run(config: MixerConfig) -> Result<u32, u32> {
     let shards = Shard::split_streams(&config.streams).unwrap();
 
     let threadpool = ThreadPool::new(config.processes);
-    let failed_shard_count = AtomicU32::new(0);
-    let failed_shard_count_ref = Arc::new(failed_shard_count);
+    let failed_shard_count_ref = Arc::new(AtomicU32::new(0));
     for shard in shards {
         let output_path = Path::new(&config.work_dir.output.clone()).join(&shard.output);
         if output_path.exists() {
@@ -26,27 +25,21 @@ pub fn run(config: MixerConfig) -> Result<u32, u32> {
 
         threadpool.execute(move || {
             log::info!("Building output {:?}...", shard.output);
-            match shard.clone().process(work_dirs) {
-                Ok(_) => {}
-                Err(e) => {
-                    log::error!("Error processing {:?}: {}", shard.output, e);
-                    failed_shard_count_ref.fetch_add(1, Ordering::Relaxed);
-                }
+            if let Err(e) = shard.clone().process(work_dirs) {
+                log::error!("Error processing {:?}: {}", shard.output, e);
+                failed_shard_count_ref.fetch_add(1, Ordering::Relaxed);
             }
         });
     }
     threadpool.join();
 
-    let failure_count = failed_shard_count_ref.fetch_add(0, Ordering::Relaxed);
-    match failure_count {
-        0 => {
-            log::info!("Done!");
-            return Ok(failure_count);
-        }
-        _ => {
-            log::error!("{} shards failed to process.", failure_count);
-            return Err(failure_count);
-        }
+    let failure_count = failed_shard_count_ref.load(Ordering::Relaxed);
+    if failure_count == 0 {
+        log::info!("Done!");
+        Ok(failure_count)
+    } else {
+        log::error!("{} shards failed to process.", failure_count);
+        Err(failure_count)
     }
 }
 
diff --git a/src/s3_util.rs b/src/s3_util.rs
index 6e329813..75ba1036 100644
--- a/src/s3_util.rs
+++ b/src/s3_util.rs
@@ -100,10 +100,7 @@ pub async fn object_size(
         .send()
         .await
         .map_err(|e| io::Error::new(io::ErrorKind::Other, e));
-    match resp {
-        Ok(resp) => Ok(resp.content_length as usize),
-        Err(e) => Err(e),
-    }
+    Ok(resp?.content_length as usize)
 }
 
 // Expand wildcard patterns into a list of object paths
@@ -112,7 +109,7 @@ pub async fn object_size(
 // or:   a/*/b.txt -> a/1/b.txt, a/2/b.txt, a/3/b.txt
 pub fn find_objects_matching_patterns(
     s3_client: &S3Client,
-    patterns: &Vec<String>,
+    patterns: &[String],
 ) -> Result<Vec<String>, io::Error> {
     let rt = tokio::runtime::Builder::new_current_thread()
         .enable_all()
@@ -141,7 +138,7 @@ pub fn find_objects_matching_patterns(
                     return Err(io::Error::new(io::ErrorKind::Other, e));
                 }
             };
-            let resp = if token.is_some() {
+            let resp = if let Some(token_value) = token {
                 log::info!("Listing objects in bucket={}, prefix={}", bucket, key);
                 rt.block_on(
                     s3_client
@@ -149,7 +146,7 @@ pub fn find_objects_matching_patterns(
                         .bucket(bucket)
                         .prefix(key)
                         .delimiter("/")
-                        .continuation_token(token.unwrap())
+                        .continuation_token(token_value)
                         .send(),
                 )
                 .unwrap()
@@ -234,10 +231,19 @@ mod test {
     use std::io;
     use std::io::{BufRead, BufReader};
     use std::path::Path;
-    use std::str::FromStr;
 
     use flate2::read::MultiGzDecoder;
 
+    fn skip_dolma_aws_tests() -> bool {
+        if std::env::var_os("DOLMA_TESTS_SKIP_AWS")
+            .is_some_and(|var| var.eq_ignore_ascii_case("true"))
+        {
+            println!("Skipping test_download_file because DOLMA_TESTS_SKIP_AWS=True");
+            return true;
+        }
+        false
+    }
+
     fn compare_contents(expected: &str, actual: &str) {
         let expected_lines = BufReader::new(MultiGzDecoder::new(
             OpenOptions::new()
@@ -291,12 +297,10 @@ mod test {
 
     #[test]
     fn test_object_size() -> Result<(), io::Error> {
-        if std::env::var_os("DOLMA_TESTS_SKIP_AWS")
-            .is_some_and(|var| var.eq_ignore_ascii_case("true"))
-        {
-            println!("Skipping test_download_file because DOLMA_TESTS_SKIP_AWS=True");
+        if skip_dolma_aws_tests() {
             return Ok(());
         }
+
         let rt = tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()
@@ -313,10 +317,7 @@ mod test {
 
     #[test]
     fn test_download_file() -> Result<(), io::Error> {
-        if std::env::var_os("DOLMA_TESTS_SKIP_AWS")
-            .is_some_and(|var| var.eq_ignore_ascii_case("true"))
-        {
-            println!("Skipping test_download_file because DOLMA_TESTS_SKIP_AWS=True");
+        if skip_dolma_aws_tests() {
             return Ok(());
         }
         let rt = tokio::runtime::Builder::new_current_thread()
@@ -342,45 +343,25 @@ mod test {
 
     #[test]
     fn test_find_objects_matching_patterns() -> Result<(), io::Error> {
-        if std::env::var_os("DOLMA_TESTS_SKIP_AWS")
-            .is_some_and(|var| var.eq_ignore_ascii_case("true"))
-        {
-            println!("Skipping test_download_file because DOLMA_TESTS_SKIP_AWS=True");
+        if skip_dolma_aws_tests() {
             return Ok(());
         }
         let s3_client = new_client(None)?;
 
         let patterns =
-            vec![
-                String::from_str("s3://ai2-llm/pretraining-data/tests/mixer/expected/*.json.gz")
-                    .unwrap(),
-            ];
+            vec!["s3://ai2-llm/pretraining-data/tests/mixer/expected/*.json.gz".to_string()];
 
         let resp = find_objects_matching_patterns(&s3_client, &patterns).unwrap();
         let mut matches: HashSet<String> = HashSet::from_iter(resp.iter().map(|s| s.to_owned()));
 
         // list the contents of `tests/data/expected` and check that they match
-        match read_dir("tests/data/expected") {
-            Ok(entries) => {
-                for entry in entries {
-                    match entry {
-                        Ok(entry) => {
-                            let mut remote_path = String::from_str(
-                                "s3://ai2-llm/pretraining-data/tests/mixer/expected/",
-                            )
-                            .unwrap();
-                            remote_path.push_str(entry.file_name().to_str().unwrap());
-                            matches.remove(&remote_path);
-                        }
-                        Err(err) => {
-                            return Err(err);
-                        }
-                    }
-                }
-            }
-            Err(err) => {
-                return Err(err);
-            }
+        let entries = read_dir("tests/data/expected")?;
+        for entry in entries {
+            let remote_path = format!(
+                "s3://ai2-llm/pretraining-data/tests/mixer/expected/{}",
+                entry?.file_name().to_str().unwrap()
+            );
+            matches.remove(&remote_path);
         }
 
         assert_eq!(matches.len(), 0);
diff --git a/src/shard.rs b/src/shard.rs
index 5a57cfdb..d4d16efd 100644
--- a/src/shard.rs
+++ b/src/shard.rs
@@ -49,9 +49,7 @@ impl Shard {
                 .map(|(input, size)| {
                     let mut attr_paths = Vec::new();
                     for prefix in stream_config.attributes.iter() {
-                        let mut attr_prefix = "/attributes/".to_owned();
-                        attr_prefix.push_str(prefix);
-                        attr_prefix.push_str("/");
+                        let attr_prefix = format!("/attributes/{}/", prefix);
                         let attr_path = input.replace("/documents/", &attr_prefix);
                         attr_paths.push(attr_path);
                     }
@@ -65,8 +63,7 @@ impl Shard {
                 })
                 .collect::<Vec<(DocumentPaths, usize)>>();
             let mut shard_size = inputs_with_sizes[0].1;
-            let mut shard_inputs: Vec<DocumentPaths> = Vec::new();
-            shard_inputs.push(inputs_with_sizes[0].0.clone());
+            let mut shard_inputs: Vec<DocumentPaths> = vec![inputs_with_sizes[0].0.clone()];
             for (input, size) in inputs_with_sizes[1..].iter() {
                 if *size == 0 {
                     log::warn!(
@@ -95,7 +92,7 @@ impl Shard {
                 }
                 shard_inputs.push(input.clone());
             }
-            if shard_inputs.len() > 0 {
+            if !shard_inputs.is_empty() {
                 let output = format!(
                     "{}/{}-{:04}.json.gz",
                     stream_config.output.path, stream_config.name, stream_shard_count
@@ -153,7 +150,7 @@ impl Shard {
                 let mut local_attr_readers = Vec::new();
                 let mut attr_reader_failure_counts = Vec::new();
                 for attr in &input_path.attribute_paths {
-                    let local_attr_file = cache.prepare_input(&attr)?;
+                    let local_attr_file = cache.prepare_input(attr)?;
                     let f = OpenOptions::new()
                         .read(true)
                         .write(false)
@@ -189,8 +186,9 @@ impl Shard {
                     let line = line?;
                     let mut data: Value = serde_json::from_str(&line)?;
                     let mut attrs = serde_json::Map::new();
-                    let mut attr_reader_index = 0;
-                    for (_, attr_reader) in local_attr_readers.iter_mut() {
+                    for (attr_reader_index, (_, attr_reader)) in
+                        local_attr_readers.iter_mut().enumerate()
+                    {
                         match attr_reader.next() {
                             Some(Ok(line)) => {
                                 let attr_data: Value = serde_json::from_str(&line)?;
@@ -249,7 +247,6 @@ impl Shard {
                                 break;
                             }
                         }
-                        attr_reader_index += 1;
                     }
 
                     if !attrs.is_empty() {
@@ -302,7 +299,8 @@ impl Shard {
                                         }
                                         if !is_inside_span {
                                             if i == replacements[span_index].end {
-                                                if replacements[span_index].replacement.len() > 0 {
+                                                if !replacements[span_index].replacement.is_empty()
+                                                {
                                                     let replacement_text = replacements[span_index]
                                                         .replacement
                                                         .to_owned()
@@ -335,19 +333,15 @@ impl Shard {
                                     i += 1;
                                     byte_index_with_char = chars.next();
                                 }
-                                if span_index < replacements.len() {
-                                    if replacements[span_index].replacement.len() > 0 {
-                                        let replacement_text = replacements[span_index]
-                                            .replacement
-                                            .to_owned()
-                                            .replace(
-                                                "{}",
-                                                old_text[span_start_byte_index..]
-                                                    .to_owned()
-                                                    .as_str(),
-                                            );
-                                        new_text.push_str(&replacement_text);
-                                    }
+                                if span_index < replacements.len()
+                                    && !replacements[span_index].replacement.is_empty()
+                                {
+                                    let replacement_text =
+                                        replacements[span_index].replacement.to_owned().replace(
+                                            "{}",
+                                            old_text[span_start_byte_index..].to_owned().as_str(),
+                                        );
+                                    new_text.push_str(&replacement_text);
                                 }
                                 data["text"] = Value::String(new_text);
                             }
@@ -361,15 +355,16 @@ impl Shard {
                     }
                 }
                 cache.finalize_input(&input_path.doc_path)?;
-                for i in 0..input_path.attribute_paths.len() {
-                    if attr_reader_failure_counts[i] > 0 {
+                for (index, attribute_path) in input_path.attribute_paths.iter().enumerate() {
+                    let failure_count = attr_reader_failure_counts[index];
+                    if failure_count > 0 {
                         log::warn!(
                             "Failed to read {} attributes from {}",
-                            input_path.attribute_paths[i],
-                            attr_reader_failure_counts[i]
+                            attribute_path,
+                            failure_count
                         );
                     }
-                    cache.finalize_input(&input_path.attribute_paths[i])?;
+                    cache.finalize_input(attribute_path)?;
                 }
                 log::info!(
                     "Dropped {} of {} documents from {}",
@@ -476,7 +471,7 @@ pub mod shard_config {
         // Check the json for the existence of any element matching the configured include/exclude patterns
         // Determine whether to keep the document based on the include/exclude matches
         pub fn should_keep(&self, json: &Value) -> Result<bool, String> {
-            let mut keep = self.include.len() == 0;
+            let mut keep = self.include.is_empty();
             for pattern in self.include.iter() {
                 let mut finder = JsonPathFinder::from_str("{}", pattern)?;
                 finder.set_json(Box::new(json.clone()));
@@ -527,7 +522,7 @@ impl FileCache {
             rt.block_on(s3_util::download_to_file(
                 &self.s3_client,
                 bucket,
-                &key,
+                key,
                 &path,
             ))?;
             Ok(path.clone())
@@ -549,7 +544,7 @@ impl FileCache {
     pub fn finalize_input(&self, location: &str) -> Result<(), io::Error> {
         if location.starts_with("s3://") {
             let (_, _, path) = cached_s3_location!(location, &self.work.input);
-            std::fs::remove_file(&path)?;
+            std::fs::remove_file(path)?;
             Ok(())
         } else {
             Ok(())
@@ -581,7 +576,7 @@ impl FileCache {
                 .enable_all()
                 .build()
                 .unwrap();
-            rt.block_on(s3_util::upload_file(&self.s3_client, &path, &bucket, &key))?;
+            rt.block_on(s3_util::upload_file(&self.s3_client, &path, bucket, key))?;
             std::fs::remove_file(&path)?;
             {
                 // Create empty file to indicate that the shard is done.
@@ -589,9 +584,8 @@ impl FileCache {
             }
             Ok(())
         } else {
-            let tmp_path = location.to_owned() + ".tmp";
-            let tmp_path = Path::new(tmp_path.as_str());
-            std::fs::rename(&tmp_path, &location)?;
+            std::fs::rename(Path::new(&(location.to_owned() + ".tmp")), location)?;
+
             Ok(())
         }
     }
@@ -602,8 +596,8 @@ pub fn find_objects_matching_patterns(patterns: &Vec<String>) -> Result<Vec<Stri
     if s3_url_count == 0 {
         let mut matches = Vec::new();
         for pattern in patterns.iter() {
-            for entry in
-                glob(pattern).expect(format! {"Invalid file pattern: {}", pattern.clone()}.as_str())
+            for entry in glob(pattern)
+                .unwrap_or_else(|_| panic!("Invalid file pattern: {}", pattern.clone()))
             {
                 matches.push(entry.unwrap().to_str().unwrap().to_owned());
             }
@@ -644,11 +638,8 @@ pub fn get_object_sizes(locations: &Vec<String>) -> Result<Vec<usize>, io::Error
             .par_iter()
             .map(|location| {
                 let (bucket, key) = s3_util::split_url(location).unwrap();
-                let resp = rt.block_on(s3_util::object_size(&s3_client, &bucket, &key));
-                match resp {
-                    Ok(size) => size,
-                    Err(_) => 0,
-                }
+                rt.block_on(s3_util::object_size(&s3_client, bucket, key))
+                    .unwrap_or(0)
             })
             .collect();
         Ok(sizes)


Test patch:
diff --git a/src/bloom_filter/bloom_test.rs b/src/bloom_filter/bloom_test.rs
new file mode 100644
index 00000000..15f375d8
--- /dev/null
+++ b/src/bloom_filter/bloom_test.rs
@@ -0,0 +1,56 @@
+#[cfg(test)]
+mod tests {
+    use super::super::BloomFilter;
+
+    // n: number of items in filter. p: false positive rate
+    // m: number of bits in filter. k: number of hashers
+    // n = ceil(m / (-k / log(1 - exp(log(p) / k))))
+    // p = pow(1 - exp(-k / (m / n)), k)
+    // m = ceil((n * log(p)) / log(1 / pow(2, log(2))));
+    // k = round((m / n) * log(2));
+
+    #[test]
+    fn bloom_optimal_hasher_number() {
+        let size_in_bytes = 1_000_000_000;
+        let expected_elements = 1_000_000_000;
+        assert_eq!(
+            BloomFilter::optimal_number_of_hashers(size_in_bytes, expected_elements),
+            6
+        );
+        assert_eq!(
+            BloomFilter::optimal_number_of_hashers(1_000_000, 500_000),
+            12
+        )
+    }
+    #[test]
+    fn bloom_test_prob_of_false_positive() {
+        // calculated from https://hur.st/bloomfilter/
+        let size_in_bytes = 1_000_000_000;
+        let expected_elements = 1_000_000_000;
+        let num_hashers = 8;
+        assert_eq!(
+            BloomFilter::prob_of_false_positive(size_in_bytes, expected_elements, num_hashers),
+            0.025_491_740_593_406_025 as f64
+        );
+        assert_eq!(
+            BloomFilter::prob_of_false_positive(1_048_576, 524288, 2),
+            0.013_806_979_447_406_826 as f64
+        )
+    }
+
+    #[test]
+    fn bloom_suggest_size() {
+        // it's hard to derive this exactly since the algorithm is doing closest power of 2
+        // instead of exact theoretical optimum
+        let expected_elements = 1_000_000;
+        let desired_false_positive_rate = 0.0001 as f64;
+        let theoretical_optimum = ((expected_elements as f64 * desired_false_positive_rate.ln())
+            / f64::ln(1.0 / 2.0f64.powf(2.0f64.ln())))
+        .ceil()
+        .div_euclid(8f64) as usize;
+        let suggested_size =
+            BloomFilter::suggest_size_in_bytes(expected_elements, desired_false_positive_rate);
+        assert_eq!(suggested_size, 4_194_304);
+        assert_eq!(suggested_size, theoretical_optimum.next_power_of_two())
+    }
+}