Function test_parallel_get_set_config

mne/utils/tests/test_config.py:279–345

Test that uses parallel workers to get and set config. All the workers update the same configuration file concurrently. In a correct implementation with proper path file locking, the final config file remains valid JSON and includes all expected updates.

(tmp_path: Path)

@pytest.mark.slowtest
def test_parallel_get_set_config(tmp_path: Path):
    """Test that uses parallel workers to get and set config.

    All the workers update the same configuration file concurrently. In a
    correct implementation with proper path file locking, the final
    config file remains valid JSON and includes all expected updates.

    """
    pytest.importorskip("joblib")
    pytest.importorskip("filelock")
    from joblib import Parallel, delayed

    # Use the temporary directory as our home directory.
    home_dir = str(tmp_path)
    # get_config_path will return home_dir/.mne/mne-python.json
    config_file = get_config_path(home_dir=home_dir)

    # if the config file already exists, remove it
    if os.path.exists(config_file):
        os.remove(config_file)

    # Ensure that the .mne directory exists.
    config_dir = tmp_path / ".mne"
    config_dir.mkdir(exist_ok=True)

    # Write an initial (valid) config file.
    initial_config = {"initial": "True"}
    with open(config_file, "w") as f:
        json.dump(initial_config, f)

    n_workers = 50
    iterations = 10

    # Launch multiple workers concurrently using joblib.
    Parallel(n_jobs=10)(
        delayed(_worker_update_config_loop)(home_dir, worker_id, iterations)
        for worker_id in range(n_workers)
    )

    # Now, read back the config file.
    final_config = get_config(home_dir=home_dir)
    expected_keys = set()
    expected_values = set()
    # For each worker and iteration, check that the expected key/value pair is present.
    for worker_id in range(n_workers):
        for i in range(iterations):
            expected_key = f"worker_{worker_id}_{i}"
            expected_value = f"value_{worker_id}_{i}"

            assert final_config.get(expected_key) == expected_value, (
                f"Missing or incorrect value for key {expected_key}"
            )
            expected_keys.add(expected_key)
            expected_values.add(expected_value)

    # include the initial key/value pair
    # that was written before the workers started