it.rs

//Integration test for webhdfs-rs

use webhdfs::{config::HttpsConfig, sync_client::*, *};
//use ReadHdfsFile;

use std::convert::TryInto;
use std::fs::{read, File};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::Path;
use std::time::Duration;

#[test]
fn webhdfs_test() {
    println!("Integration test -- start");
    //env_logger::init();
    let _ = env_logger::builder().is_test(true).try_init();

    //let do_itt = if let Ok(..) = std::env::var("WEBHDFS_BYPASS_ITT") { false } else { true };
    //fn run_shell(cmdline: &str, msg: &'static str) {
    //    use std::process::Command;
    //    assert!(Command::new("bash").arg("-c").arg(cmdline).status().expect("could not run bash").success(), msg)
    //}
    //if do_itt { run_shell("./itt.sh --prepare", "Could not prepare"); }

    fn file_as_string(path: &str) -> String {
        String::from_utf8_lossy(&read(path).expect("cannot file-as-stirng"))
            .to_owned()
            .to_string()
    }

    fn file_as_string_opt(path: &str) -> Option<String> {
        read(path)
            .map(|s| String::from_utf8_lossy(&s).to_owned().to_string())
            .ok()
    }

    let entrypoint = file_as_string("./test-data/entrypoint");
    let alt_entrypoint = file_as_string_opt("./test-data/alt-entrypoint");
    let has_alt_entrypoint = alt_entrypoint.is_some();

    let natmap = crate::config::read_kv_file("./test-data/natmap").expect("cannot read natmap");
    let user = file_as_string_opt("./test-data/user");
    let dtoken = file_as_string_opt("./test-data/dtoken");
    let scheme = file_as_string("./test-data/scheme");
    println!(
        "
entrypoint='{e}'
alt_entrypoint='{ae:?}'
natmap={n:?}
user={u:?}
dtoken={d:?}",
        e = entrypoint,
        ae = alt_entrypoint,
        n = natmap,
        u = user,
        d = dtoken
    );
    let nm = NatMap::new(natmap.into_iter()).expect("cannot build natmap");
    let mut https_config = HttpsConfig::new();
    https_config.danger_accept_invalid_certs = Some(true);
    https_config.danger_accept_invalid_hostnames = Some(true);
    let entrypoint_uri = format!("{}://{}", scheme, entrypoint);
    let b = SyncHdfsClientBuilder::new(entrypoint_uri.parse().expect("Cannot parse entrypoint"))
        .default_timeout(Duration::from_secs(180))
        .natmap(nm)
        .https_settings(https_config.into());
    let b = if let Some(w) = alt_entrypoint {
        let alt_entrypoint_uri = format!("{}://{}", scheme, w);
        b.alt_entrypoint(
            alt_entrypoint_uri
                .parse()
                .expect("Cannot parse alt_entrypoint"),
        )
    } else {
        b
    };
    let b = if let Some(w) = dtoken {
        b.delegation_token(w)
    } else {
        b
    };
    let b = if let Some(w) = user {
        b.user_name(w)
    } else {
        b
    };
    let mut cx = b.build().expect("cannot HdfsContext::new");

    let readscript = file_as_string("./test-data/readscript");
    let writescript = file_as_string("./test-data/writescript");
    let source = file_as_string("./test-data/source");
    let target = file_as_string("./test-data/target");
    let size = file_as_string("./test-data/size").parse::<i64>().unwrap();
    let dirsource = file_as_string_opt("./test-data/dirsource");
    println!(
        "
source='{s}'
readscript='{r}'
target='{t}'
writescript='{w}'
size={z}",
        s = source,
        r = readscript,
        t = target,
        w = writescript,
        z = size
    );
    let (source_dir, source_sfn) =
        source.split_at(source.rfind('/').expect("source does not contain '/'"));
    let (_, source_fn) = source_sfn.split_at(1);

    //------------------------------------------------
    //Test directory listing
    println!("Test dir and stat");

    //Ok(ListStatusResponse { file_statuses: FileStatuses { file_status: [FileStatus {
    // access_time: 1564409836087, block_size: 134217728, group: "hadoop", length: 423941508,
    // modification_time: 1564409849727, owner: "root", path_suffix: "soc-pokec-relationships.txt",
    // permission: "644", replication: 3, type_: "FILE"
    //}] } })
    let dir_resp = cx.dir(source_dir);
    println!("Dir: {:?}", dir_resp);
    //assert_eq!(source_fn, dir_resp.unwrap().file_statuses.file_status[0].path_suffix);
    dir_resp
        .unwrap()
        .file_statuses
        .file_status
        .into_iter()
        .find(|fs| fs.path_suffix == source_fn)
        .ok_or("cannot find sourcefile in hdfs")
        .unwrap();

    if let Some(dirsource_dir) = dirsource {
        let dir_resp = cx.dir(&dirsource_dir);
        println!("Dir: {:?}", dir_resp);
        let _ = dir_resp.unwrap();
    }

    let stat_resp = cx.stat(&source);
    println!("Stat: {:?}", stat_resp);
    assert_eq!(size, stat_resp.unwrap().file_status.length);

    println!("Read test");

    //Parse program
    #[derive(Debug)]
    enum Op {
        Seek(i64),
        Read(i64, String),
    }

    fn parse_size(s: &str) -> i64 {
        if s.ends_with('k') {
            &s[0..s.len() - 1].parse::<i64>().unwrap() * 1024
        } else if s.ends_with('m') {
            &s[0..s.len() - 1].parse::<i64>().unwrap() * 1024 * 1024
        } else {
            s.parse().unwrap()
        }
    }

    let p = readscript
        .split(' ')
        .filter(|e| !e.is_empty())
        .map(|s| {
            let mut i = s.split(':');
            let optype = i.next().unwrap();
            let arg = parse_size(i.next().unwrap());
            match optype {
                "s" => Op::Seek(arg),
                "r" => Op::Read(arg, i.next().unwrap().to_owned()),
                _ => panic!("invalid optype '{}'", s),
            }
        })
        .collect::<Vec<Op>>();

    //allocate and initialize a large master buffer
    let master_buffer_size = p
        .iter()
        .map(|w| if let Op::Read(len, _) = w { *len } else { 0 })
        .max()
        .unwrap()
        .try_into()
        .unwrap();
    print!("alloc_mb(len={})...", master_buffer_size);
    let mut b = vec![0; master_buffer_size];
    println!("done");

    let mut file = ReadHdfsFile::open(cx, source.clone()).unwrap();

    for op in p {
        println!("{:?}...", op);
        match op {
            Op::Seek(o) => {
                file.seek(SeekFrom::Start(o.try_into().unwrap())).unwrap();
            }
            Op::Read(l, f) => {
                let length: usize = l.try_into().unwrap();
                let readcount = file.read(&mut b[0..length]).unwrap();
                assert_eq!(length, readcount);
                let writecount = File::create(&Path::new(&f))
                    .unwrap()
                    .write(&b[0..length])
                    .unwrap();
                assert_eq!(length, writecount);
            }
        }
    }

    let (cx, _, _) = file.into_parts();

    println!("Write test");
    let files = writescript
        .split(' ')
        .filter(|e| !e.is_empty())
        .collect::<Vec<&str>>();
    let mut file =
        WriteHdfsFile::create(cx, target, CreateOptions::new(), AppendOptions::new()).unwrap();
    let mut count = 0usize;

    for file_name in files {
        //std::thread::sleep(std::time::Duration::from_secs(3));
        println!("{}", file_name);
        let fb = read(Path::new(&file_name)).expect("couldn't read wseg");
        count += file.write(&fb).unwrap();
    }

    assert_eq!(count, size as usize);

    let (mut cx, _) = file.into_parts();

    //MKDIRS/DELETE(dir) test
    let dir_to_make = file_as_string("./test-data/dir-to-make");
    cx.mkdirs(&dir_to_make, MkdirsOptions::new())
        .expect("mkdirs");
    let mkdirs_stat_resp = cx.stat(&dir_to_make);
    //println!("Mkdirs Stat: {:?}", mkdirs_stat_resp);
    assert_eq!(
        dirent_type::DIRECTORY,
        mkdirs_stat_resp.unwrap().file_status.type_
    );

    let dir_to_remove = file_as_string("./test-data/dir-to-remove");
    let rmdir_stat_resp = cx.stat(&dir_to_remove);
    //println!("Stat: {:?}", rmdir_stat_resp);
    assert_eq!(
        dirent_type::DIRECTORY,
        rmdir_stat_resp.unwrap().file_status.type_
    );
    cx.delete(&dir_to_remove, DeleteOptions::new())
        .expect("delete (dir)");
    let x = cx.stat(&dir_to_remove).expect_err("delete(dir) failed");
    println!("{}", x);

    //failover test
    if has_alt_entrypoint {
        println!("Failover test");
        let standby_state = cx.fostate().next();
        let mut cx = cx.with_fostate(standby_state);
        let dir_resp = cx.dir(source_dir);
        println!("Dir: {:?}", dir_resp);
        //assert_eq!(source_fn, dir_resp.unwrap().file_statuses.file_status[0].path_suffix);
        dir_resp
            .unwrap()
            .file_statuses
            .file_status
            .into_iter()
            .find(|fs| fs.path_suffix == source_fn)
            .ok_or("cannot find sourcefile in hdfs")
            .unwrap();

        //do the same again - to make sure the correct state was memoized on the previous step
        //the 1st request should go directly to the active node (consult the logs)
        let dir_resp = cx.dir(source_dir);
        println!("Dir(2): {:?}", dir_resp);
        //assert_eq!(source_fn, dir_resp.unwrap().file_statuses.file_status[0].path_suffix);
        dir_resp
            .unwrap()
            .file_statuses
            .file_status
            .into_iter()
            .find(|fs| fs.path_suffix == source_fn)
            .ok_or("cannot find sourcefile in hdfs")
            .unwrap();

        //TODO: test all other control paths (get/put binary, op_json, op_empty)starting in standby nn state
        //at least read and write (small chunks/files)

        //get binary
        let cx = cx.with_fostate(standby_state);
        let mut file = ReadHdfsFile::open(cx, source.clone()).unwrap();
        let mut b = vec![0; b.capacity()];
        file.read_exact(&mut b).unwrap();
    } else {
        println!("No alt_entrypoint specified -- skip failover test");
    }

    //if do_itt { run_shell("./itt.sh --validate", "Validation failed"); }
}