lean_lmdb

use super::*;

#[test]
fn reports_the_crate_version() {
    assert_eq!(version(), ("lean_lmdb", "0.3.0"));
}

#[test]
fn validates_environment_option_bounds() {
    for (map_size, max_dbs, max_readers) in [(MIN_MAP_SIZE, 1, 1), (MAX_MAP_SIZE, 1024, 4096)] {
        assert!(
            validate_options(map_size, max_dbs, max_readers, false, true, 0, false, true).is_ok()
        );
    }

    for (map_size, max_dbs, max_readers) in [
        (MIN_MAP_SIZE - 4096, 8, 16),
        (MIN_MAP_SIZE + 1, 8, 16),
        (MAX_MAP_SIZE + 4096, 8, 16),
        (MIN_MAP_SIZE, 0, 16),
        (MIN_MAP_SIZE, 1025, 16),
        (MIN_MAP_SIZE, 8, 0),
        (MIN_MAP_SIZE, 8, 4097),
    ] {
        assert_eq!(
            validate_options(map_size, max_dbs, max_readers, false, false, 0, false, true),
            Err(LifecycleError::InvalidEnvironmentOptions)
        );
    }

    assert_eq!(
        validate_options(MIN_MAP_SIZE, 8, 16, true, true, 0, false, true),
        Err(LifecycleError::InvalidEnvironmentOptions)
    );
    assert_eq!(
        validate_options(MIN_MAP_SIZE, 8, 16, false, true, 3, false, true),
        Err(LifecycleError::InvalidEnvironmentOptions)
    );
    assert_eq!(
        validate_options(MIN_MAP_SIZE, 8, 16, true, false, 2, false, true),
        Err(LifecycleError::InvalidEnvironmentOptions)
    );
    assert_eq!(
        validate_options(MIN_MAP_SIZE, 8, 16, false, true, 1, true, false),
        Ok(EnvironmentOptions {
            map_size: MIN_MAP_SIZE as usize,
            max_dbs: 8,
            max_readers: 16,
            read_only: false,
            durability: Durability::NoMetaSync,
            fixed_map: true,
            read_ahead: false,
        })
    );
}

#[test]
fn validates_database_names() {
    assert_eq!(validate_database_name(b"named-db"), Ok("named-db"));
    assert!(validate_database_name(b"").is_err());
    assert!(validate_database_name(&vec![b'a'; 512]).is_err());
    assert!(validate_database_name(b"has\0nul").is_err());
    assert!(validate_database_name(&[0xff]).is_err());
    assert!(validate_database_name(&vec![b'a'; 511]).is_ok());
}

#[test]
fn canonical_paths_collapse_equivalent_spellings() {
    let dir = fresh_dir("canonical");
    fs::create_dir_all(&dir).expect("create isolated test directory");
    let direct = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), false)
        .expect("canonical path");
    let equivalent =
        canonical_environment_path(dir.join(".").to_str().expect("UTF-8 temp path"), false)
            .expect("equivalent canonical path");
    assert_eq!(direct, equivalent);
    fs::remove_dir_all(dir).expect("remove isolated test directory");
}

#[test]
fn shared_registry_reuses_compatible_state_and_rejects_incompatible_options() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("reuse");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let first = open_shared_environment(canonical.clone(), options(), true).expect("first open");
    let second = open_shared_environment(canonical.clone(), options(), true).expect("shared open");
    assert!(Arc::ptr_eq(&first, &second));
    assert_eq!(first.id, second.id);

    for incompatible in [
        EnvironmentOptions {
            max_readers: options().max_readers + 1,
            ..options()
        },
        EnvironmentOptions {
            read_ahead: false,
            ..options()
        },
    ] {
        assert_eq!(
            open_shared_environment(canonical.clone(), incompatible, true).err(),
            Some(LifecycleError::IncompatibleEnvironmentOptions)
        );
    }
    drop(second);
    drop(first);
    cleanup_environment(&canonical);
}

#[test]
fn concurrent_opens_create_exactly_one_shared_state() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("concurrent");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let barrier = Arc::new(Barrier::new(8));
    let mut workers = Vec::new();
    for _ in 0..8 {
        let barrier = Arc::clone(&barrier);
        let path = canonical.clone();
        workers.push(thread::spawn(move || {
            barrier.wait();
            open_shared_environment(path, options(), true).expect("concurrent open")
        }));
    }
    let states: Vec<_> = workers
        .into_iter()
        .map(|worker| worker.join().expect("worker did not panic"))
        .collect();
    assert!(states.iter().all(|state| state.id == states[0].id));
    drop(states);
    cleanup_environment(&canonical);
}

#[test]
fn named_databases_are_created_reopened_and_missing_is_stable() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("databases");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let state =
        open_shared_environment(canonical.clone(), options(), true).expect("environment open");

    create_named_database(&state, "first").expect("create first database");
    create_named_database(&state, "second").expect("create second database");
    open_named_database(&state, "first").expect("reopen first database");
    open_named_database(&state, "second").expect("reopen second database");
    assert_eq!(
        open_named_database(&state, "missing").err(),
        Some(LifecycleError::DatabaseNotFound)
    );
    drop(state);
    cleanup_environment(&canonical);
}

#[test]
fn read_only_create_false_requires_existing_data_file() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("read_only_missing");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let read_only = EnvironmentOptions {
        read_only: true,
        ..options()
    };

    assert_eq!(
        open_shared_environment(canonical.clone(), read_only, false).err(),
        Some(LifecycleError::PathNotFound)
    );
    cleanup_environment(&canonical);
}

#[test]
fn stale_registry_entries_are_replaced_with_new_monotonic_ids() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("stale");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let first_id = open_shared_environment(canonical.clone(), options(), true)
        .expect("first open")
        .id;
    assert_eq!(registry().get(&canonical).map(Weak::strong_count), Some(0));
    let second_id = open_shared_environment(canonical.clone(), options(), true)
        .expect("reopen")
        .id;
    assert!(second_id > first_id);
    cleanup_environment(&canonical);
}