# -*- coding: utf-8 -*-
# Author: Alexandre Gramfort <alexandre.gramfort@inria.fr>
#         Denis Engemann <denis.engemann@gmail.com>
#
# License: BSD-3-Clause

from copy import deepcopy
from pathlib import Path
from functools import partial
from io import BytesIO
import os
import os.path as op
import pathlib
import pickle
import shutil
import sys

import numpy as np
from numpy.testing import (assert_array_almost_equal, assert_array_equal,
                           assert_allclose)
import pytest

from mne.datasets import testing
from mne.filter import filter_data
from mne.io.constants import FIFF
from mne.io import RawArray, concatenate_raws, read_raw_fif, base
from mne.io.open import read_tag, read_tag_info
from mne.io.tag import _read_tag_header
from mne.io.tests.test_raw import _test_concat, _test_raw_reader
from mne import (concatenate_events, find_events, equalize_channels,
                 compute_proj_raw, pick_types, pick_channels, create_info,
                 pick_info)
from mne.utils import (requires_pandas, assert_object_equal, _dt_to_stamp,
                       requires_mne, run_subprocess, _record_warnings,
                       assert_and_remove_boundary_annot)
from mne.annotations import Annotations

testing_path = testing.data_path(download=False)
data_dir = op.join(testing_path, 'MEG', 'sample')
fif_fname = op.join(data_dir, 'sample_audvis_trunc_raw.fif')
ms_fname = op.join(testing_path, 'SSS', 'test_move_anon_raw.fif')
skip_fname = op.join(testing_path, 'misc', 'intervalrecording_raw.fif')

base_dir = op.join(op.dirname(__file__), '..', '..', 'tests', 'data')
test_fif_fname = op.join(base_dir, 'test_raw.fif')
test_fif_gz_fname = op.join(base_dir, 'test_raw.fif.gz')
ctf_fname = op.join(base_dir, 'test_ctf_raw.fif')
ctf_comp_fname = op.join(base_dir, 'test_ctf_comp_raw.fif')
fif_bad_marked_fname = op.join(base_dir, 'test_withbads_raw.fif')
bad_file_works = op.join(base_dir, 'test_bads.txt')
bad_file_wrong = op.join(base_dir, 'test_wrong_bads.txt')
hp_fname = op.join(base_dir, 'test_chpi_raw_hp.txt')
hp_fif_fname = op.join(base_dir, 'test_chpi_raw_sss.fif')


@testing.requires_testing_data
def test_acq_skip(tmp_path):
    """Test treatment of acquisition skips."""
    raw = read_raw_fif(skip_fname, preload=True)
    picks = [1, 2, 10]
    assert len(raw.times) == 17000
    annotations = raw.annotations
    assert len(annotations) == 3  # there are 3 skips
    assert_allclose(annotations.onset, [14, 19, 23])
    assert_allclose(annotations.duration, [2., 2., 3.])  # inclusive!
    data, times = raw.get_data(
        picks, reject_by_annotation='omit', return_times=True)
    expected_data, expected_times = zip(raw[picks, :2000],
                                        raw[picks, 4000:7000],
                                        raw[picks, 9000:11000],
                                        raw[picks, 14000:17000])
    expected_times = np.concatenate(list(expected_times), axis=-1)
    assert_allclose(times, expected_times)
    expected_data = list(expected_data)
    assert_allclose(data, np.concatenate(expected_data, axis=-1), atol=1e-22)

    # Check that acquisition skips are handled properly in filtering
    kwargs = dict(l_freq=None, h_freq=50., fir_design='firwin')
    raw_filt = raw.copy().filter(picks=picks, **kwargs)
    for data in expected_data:
        filter_data(data, raw.info['sfreq'], copy=False, **kwargs)
    data = raw_filt.get_data(picks, reject_by_annotation='omit')
    assert_allclose(data, np.concatenate(expected_data, axis=-1), atol=1e-22)

    # Check that acquisition skips are handled properly during I/O
    fname = tmp_path / 'test_raw.fif'
    raw.save(fname, fmt=raw.orig_format)
    # first: file size should not increase much (orig data is missing
    # 7 of 17 buffers, so if we write them out it should increase the file
    # size quite a bit.
    orig_size = op.getsize(skip_fname)
    new_size = op.getsize(fname)
    max_size = int(1.05 * orig_size)  # almost the same + annotations
    assert new_size < max_size, (new_size, max_size)
    raw_read = read_raw_fif(fname)
    assert raw_read.annotations is not None
    assert_allclose(raw.times, raw_read.times)
    assert_allclose(raw_read[:][0], raw[:][0], atol=1e-17)
    # Saving with a bad buffer length emits warning
    raw.pick_channels(raw.ch_names[:2])
    with _record_warnings() as w:
        raw.save(fname, buffer_size_sec=0.5, overwrite=True)
    assert len(w) == 0
    with pytest.warns(RuntimeWarning, match='did not fit evenly'):
        raw.save(fname, buffer_size_sec=2., overwrite=True)


def test_fix_types():
    """Test fixing of channel types."""
    for fname, change, bads in (
        (hp_fif_fname, True, ["MEG0111"]),
        (test_fif_fname, False, []),
        (ctf_fname, False, [])
    ):
        raw = read_raw_fif(fname)
        raw.info["bads"] = bads
        mag_picks = pick_types(raw.info, meg='mag', exclude=[])
        other_picks = np.setdiff1d(np.arange(len(raw.ch_names)), mag_picks)
        # we don't actually have any files suffering from this problem, so
        # fake it
        if change:
            for ii in mag_picks:
                raw.info['chs'][ii]['coil_type'] = FIFF.FIFFV_COIL_VV_MAG_T2
        orig_types = np.array([ch['coil_type'] for ch in raw.info['chs']])
        raw.fix_mag_coil_types()
        new_types = np.array([ch['coil_type'] for ch in raw.info['chs']])
        if not change:
            assert_array_equal(orig_types, new_types)
        else:
            assert_array_equal(orig_types[other_picks], new_types[other_picks])
            assert ((orig_types[mag_picks] != new_types[mag_picks]).all())
            assert ((new_types[mag_picks] ==
                     FIFF.FIFFV_COIL_VV_MAG_T3).all())


def test_concat(tmp_path):
    """Test RawFIF concatenation."""
    # we trim the file to save lots of memory and some time
    raw = read_raw_fif(test_fif_fname)
    raw.crop(0, 2.)
    test_name = tmp_path / 'test_raw.fif'
    raw.save(test_name)
    # now run the standard test
    _test_concat(partial(read_raw_fif), test_name)


@testing.requires_testing_data
def test_hash_raw():
    """Test hashing raw objects."""
    raw = read_raw_fif(fif_fname)
    pytest.raises(RuntimeError, raw.__hash__)
    raw = read_raw_fif(fif_fname).crop(0, 0.5)
    raw_size = raw._size
    raw.load_data()
    raw_load_size = raw._size
    assert (raw_size < raw_load_size)
    raw_2 = read_raw_fif(fif_fname).crop(0, 0.5)
    raw_2.load_data()
    assert hash(raw) == hash(raw_2)
    # do NOT use assert_equal here, failing output is terrible
    assert pickle.dumps(raw) == pickle.dumps(raw_2)

    raw_2._data[0, 0] -= 1
    assert hash(raw) != hash(raw_2)


@testing.requires_testing_data
def test_maxshield():
    """Test maxshield warning."""
    with pytest.warns(RuntimeWarning, match='Internal Active Shielding') as w:
        read_raw_fif(ms_fname, allow_maxshield=True)
    assert ('test_raw_fiff.py' in w[0].filename)


@testing.requires_testing_data
def test_subject_info(tmp_path):
    """Test reading subject information."""
    raw = read_raw_fif(fif_fname).crop(0, 1)
    assert (raw.info['subject_info'] is None)
    # fake some subject data
    keys = ['id', 'his_id', 'last_name', 'first_name', 'birthday', 'sex',
            'hand']
    vals = [1, 'foobar', 'bar', 'foo', (1901, 2, 3), 0, 1]
    subject_info = dict()
    for key, val in zip(keys, vals):
        subject_info[key] = val
    raw.info['subject_info'] = subject_info
    out_fname = tmp_path / 'test_subj_info_raw.fif'
    raw.save(out_fname, overwrite=True)
    raw_read = read_raw_fif(out_fname)
    for key in keys:
        assert subject_info[key] == raw_read.info['subject_info'][key]
    assert raw.info['meas_date'] == raw_read.info['meas_date']

    for key in ['secs', 'usecs', 'version']:
        assert raw.info['meas_id'][key] == raw_read.info['meas_id'][key]
    assert_array_equal(raw.info['meas_id']['machid'],
                       raw_read.info['meas_id']['machid'])


@testing.requires_testing_data
def test_copy_append():
    """Test raw copying and appending combinations."""
    raw = read_raw_fif(fif_fname, preload=True).copy()
    raw_full = read_raw_fif(fif_fname)
    raw_full.append(raw)
    data = raw_full[:, :][0]
    assert data.shape[1] == 2 * raw._data.shape[1]


@testing.requires_testing_data
def test_output_formats(tmp_path):
    """Test saving and loading raw data using multiple formats."""
    formats = ['short', 'int', 'single', 'double']
    tols = [1e-4, 1e-7, 1e-7, 1e-15]

    # let's fake a raw file with different formats
    raw = read_raw_fif(test_fif_fname).crop(0, 1)

    temp_file = tmp_path / 'raw.fif'
    for ii, (fmt, tol) in enumerate(zip(formats, tols)):
        # Let's test the overwriting error throwing while we're at it
        if ii > 0:
            pytest.raises(IOError, raw.save, temp_file, fmt=fmt)
        raw.save(temp_file, fmt=fmt, overwrite=True)
        raw2 = read_raw_fif(temp_file)
        raw2_data = raw2[:, :][0]
        assert_allclose(raw2_data, raw[:, :][0], rtol=tol, atol=1e-25)
        assert raw2.orig_format == fmt


def _compare_combo(raw, new, times, n_times):
    """Compare data."""
    for ti in times:  # let's do a subset of points for speed
        orig = raw[:, ti % n_times][0]
        # these are almost_equals because of possible dtype differences
        assert_allclose(orig, new[:, ti][0])


@pytest.mark.slowtest
@testing.requires_testing_data
def test_multiple_files(tmp_path):
    """Test loading multiple files simultaneously."""
    # split file
    raw = read_raw_fif(fif_fname).crop(0, 10)
    raw.load_data()
    raw.load_data()  # test no operation
    split_size = 3.  # in seconds
    sfreq = raw.info['sfreq']
    nsamp = (raw.last_samp - raw.first_samp)
    tmins = np.round(np.arange(0., nsamp, split_size * sfreq))
    tmaxs = np.concatenate((tmins[1:] - 1, [nsamp]))
    tmaxs /= sfreq
    tmins /= sfreq
    assert raw.n_times == len(raw.times)

    # going in reverse order so the last fname is the first file (need later)
    raws = [None] * len(tmins)
    for ri in range(len(tmins) - 1, -1, -1):
        fname = tmp_path / ('test_raw_split-%d_raw.fif' % ri)
        raw.save(fname, tmin=tmins[ri], tmax=tmaxs[ri])
        raws[ri] = read_raw_fif(fname)
        assert (len(raws[ri].times) ==
                int(round((tmaxs[ri] - tmins[ri]) *
                          raw.info['sfreq'])) + 1)  # + 1 b/c inclusive
    events = [find_events(r, stim_channel='STI 014') for r in raws]
    last_samps = [r.last_samp for r in raws]
    first_samps = [r.first_samp for r in raws]

    # test concatenation of split file
    pytest.raises(ValueError, concatenate_raws, raws, True, events[1:])
    all_raw_1, events1 = concatenate_raws(raws, preload=False,
                                          events_list=events)
    assert_allclose(all_raw_1.times, raw.times)
    assert raw.first_samp == all_raw_1.first_samp
    assert raw.last_samp == all_raw_1.last_samp
    assert_allclose(raw[:, :][0], all_raw_1[:, :][0])
    raws[0] = read_raw_fif(fname)
    all_raw_2 = concatenate_raws(raws, preload=True)
    assert_allclose(raw[:, :][0], all_raw_2[:, :][0])

    # test proper event treatment for split files
    events2 = concatenate_events(events, first_samps, last_samps)
    events3 = find_events(all_raw_2, stim_channel='STI 014')
    assert_array_equal(events1, events2)
    assert_array_equal(events1, events3)

    # test various methods of combining files
    raw = read_raw_fif(fif_fname, preload=True)
    n_times = raw.n_times
    # make sure that all our data match
    times = list(range(0, 2 * n_times, 999))
    # add potentially problematic points
    times.extend([n_times - 1, n_times, 2 * n_times - 1])

    raw_combo0 = concatenate_raws([read_raw_fif(f)
                                   for f in [fif_fname, fif_fname]],
                                  preload=True)
    _compare_combo(raw, raw_combo0, times, n_times)
    raw_combo = concatenate_raws([read_raw_fif(f)
                                  for f in [fif_fname, fif_fname]],
                                 preload=False)
    _compare_combo(raw, raw_combo, times, n_times)
    raw_combo = concatenate_raws([read_raw_fif(f)
                                  for f in [fif_fname, fif_fname]],
                                 preload='memmap8.dat')
    _compare_combo(raw, raw_combo, times, n_times)
    assert raw[:, :][0].shape[1] * 2 == raw_combo0[:, :][0].shape[1]
    assert raw_combo0[:, :][0].shape[1] == raw_combo0.n_times

    # with all data preloaded, result should be preloaded
    raw_combo = read_raw_fif(fif_fname, preload=True)
    raw_combo.append(read_raw_fif(fif_fname, preload=True))
    assert (raw_combo.preload is True)
    assert raw_combo.n_times == raw_combo._data.shape[1]
    _compare_combo(raw, raw_combo, times, n_times)

    # with any data not preloaded, don't set result as preloaded
    raw_combo = concatenate_raws([read_raw_fif(fif_fname, preload=True),
                                  read_raw_fif(fif_fname, preload=False)])
    assert (raw_combo.preload is False)
    assert_array_equal(find_events(raw_combo, stim_channel='STI 014'),
                       find_events(raw_combo0, stim_channel='STI 014'))
    _compare_combo(raw, raw_combo, times, n_times)

    # user should be able to force data to be preloaded upon concat
    raw_combo = concatenate_raws([read_raw_fif(fif_fname, preload=False),
                                  read_raw_fif(fif_fname, preload=True)],
                                 preload=True)
    assert (raw_combo.preload is True)
    _compare_combo(raw, raw_combo, times, n_times)

    raw_combo = concatenate_raws([read_raw_fif(fif_fname, preload=False),
                                  read_raw_fif(fif_fname, preload=True)],
                                 preload='memmap3.dat')
    _compare_combo(raw, raw_combo, times, n_times)

    raw_combo = concatenate_raws([
        read_raw_fif(fif_fname, preload=True),
        read_raw_fif(fif_fname, preload=True)], preload='memmap4.dat')
    _compare_combo(raw, raw_combo, times, n_times)

    raw_combo = concatenate_raws([
        read_raw_fif(fif_fname, preload=False),
        read_raw_fif(fif_fname, preload=False)], preload='memmap5.dat')
    _compare_combo(raw, raw_combo, times, n_times)

    # verify that combining raws with different projectors throws an exception
    raw.add_proj([], remove_existing=True)
    pytest.raises(ValueError, raw.append,
                  read_raw_fif(fif_fname, preload=True))

    # now test event treatment for concatenated raw files
    events = [find_events(raw, stim_channel='STI 014'),
              find_events(raw, stim_channel='STI 014')]
    last_samps = [raw.last_samp, raw.last_samp]
    first_samps = [raw.first_samp, raw.first_samp]
    events = concatenate_events(events, first_samps, last_samps)
    events2 = find_events(raw_combo0, stim_channel='STI 014')
    assert_array_equal(events, events2)

    # check out the len method
    assert len(raw) == raw.n_times
    assert len(raw) == raw.last_samp - raw.first_samp + 1


@testing.requires_testing_data
@pytest.mark.parametrize('on_mismatch', ('ignore', 'warn', 'raise'))
def test_concatenate_raws(on_mismatch):
    """Test error handling during raw concatenation."""
    raw = read_raw_fif(fif_fname).crop(0, 10)
    raws = [raw, raw.copy()]
    raws[1].info['dev_head_t']['trans'] += 0.1
    kws = dict(raws=raws, on_mismatch=on_mismatch)

    if on_mismatch == 'ignore':
        concatenate_raws(**kws)
    elif on_mismatch == 'warn':
        with pytest.warns(RuntimeWarning, match='different head positions'):
            concatenate_raws(**kws)
    elif on_mismatch == 'raise':
        with pytest.raises(ValueError, match='different head positions'):
            concatenate_raws(**kws)


@testing.requires_testing_data
@pytest.mark.parametrize('mod', (
    'meg',
    pytest.param('raw', marks=[
        pytest.mark.filterwarnings(
            'ignore:.*naming conventions.*:RuntimeWarning'),
        pytest.mark.slowtest]),
))
def test_split_files(tmp_path, mod, monkeypatch):
    """Test writing and reading of split raw files."""
    raw_1 = read_raw_fif(fif_fname, preload=True)
    # Test a very close corner case

    assert_allclose(raw_1.buffer_size_sec, 10., atol=1e-2)  # samp rate
    split_fname = tmp_path / f'split_raw_{mod}.fif'
    # intended filenames
    split_fname_elekta_part2 = tmp_path / f'split_raw_{mod}-1.fif'
    split_fname_bids_part1 = tmp_path / f'split_raw_split-01_{mod}.fif'
    split_fname_bids_part2 = tmp_path / f'split_raw_split-02_{mod}.fif'
    raw_1.set_annotations(Annotations([2.], [5.5], 'test'))

    # Check that if BIDS is used and no split is needed it defaults to
    # simple writing without _split- entity.
    raw_1.save(split_fname, split_naming='bids', verbose=True)
    assert op.isfile(split_fname)
    assert not op.isfile(split_fname_bids_part1)
    for split_naming in ('neuromag', 'bids'):
        with pytest.raises(FileExistsError, match='Destination file'):
            raw_1.save(split_fname, split_naming=split_naming, verbose=True)
    os.remove(split_fname)
    with open(split_fname_bids_part1, 'w'):
        pass
    with pytest.raises(FileExistsError, match='Destination file'):
        raw_1.save(split_fname, split_naming='bids', verbose=True)
    assert not op.isfile(split_fname)
    raw_1.save(split_fname, split_naming='neuromag', verbose=True)  # okay
    os.remove(split_fname)
    os.remove(split_fname_bids_part1)

    raw_1.save(split_fname, buffer_size_sec=1.0, split_size='10MB',
               verbose=True)

    # check that the filenames match the intended pattern
    assert op.isfile(split_fname)
    assert op.isfile(split_fname_elekta_part2)
    # check that filenames are being formatted correctly for BIDS
    raw_1.save(split_fname, buffer_size_sec=1.0, split_size='10MB',
               split_naming='bids', overwrite=True, verbose=True)
    assert op.isfile(split_fname_bids_part1)
    assert op.isfile(split_fname_bids_part2)

    annot = Annotations(np.arange(20), np.ones((20,)), 'test')
    raw_1.set_annotations(annot)
    split_fname = op.join(tmp_path, 'split_raw.fif')
    raw_1.save(split_fname, buffer_size_sec=1.0, split_size='10MB')
    raw_2 = read_raw_fif(split_fname)
    assert_allclose(raw_2.buffer_size_sec, 1., atol=1e-2)  # samp rate
    assert_allclose(raw_1.annotations.onset, raw_2.annotations.onset)
    assert_allclose(raw_1.annotations.duration, raw_2.annotations.duration,
                    rtol=0.001 / raw_2.info['sfreq'])
    assert_array_equal(raw_1.annotations.description,
                       raw_2.annotations.description)

    data_1, times_1 = raw_1[:, :]
    data_2, times_2 = raw_2[:, :]
    assert_array_equal(data_1, data_2)
    assert_array_equal(times_1, times_2)

    raw_bids = read_raw_fif(split_fname_bids_part1)
    data_bids, times_bids = raw_bids[:, :]
    assert_array_equal(data_1, data_bids)
    assert_array_equal(times_1, times_bids)
    del raw_bids
    # split missing behaviors
    os.remove(split_fname_bids_part2)
    with pytest.raises(ValueError, match='manually renamed'):
        read_raw_fif(split_fname_bids_part1, on_split_missing='raise')
    with pytest.warns(RuntimeWarning, match='Split raw file detected'):
        read_raw_fif(split_fname_bids_part1, on_split_missing='warn')
    read_raw_fif(split_fname_bids_part1, on_split_missing='ignore')

    # test the case where we only end up with one buffer to write
    # (GH#3210). These tests rely on writing meas info and annotations
    # taking up a certain number of bytes, so if we change those functions
    # somehow, the numbers below for e.g. split_size might need to be
    # adjusted.
    raw_crop = raw_1.copy().crop(0, 5)
    raw_crop.set_annotations(Annotations([2.], [5.5], 'test'),
                             emit_warning=False)
    with pytest.raises(ValueError,
                       match='after writing measurement information'):
        raw_crop.save(split_fname, split_size='1MB',  # too small a size
                      buffer_size_sec=1., overwrite=True)
    with pytest.raises(ValueError,
                       match='too large for the given split size'):
        raw_crop.save(split_fname,
                      split_size=3003000,  # still too small, now after Info
                      buffer_size_sec=1., overwrite=True)
    # just barely big enough here; the right size to write exactly one buffer
    # at a time so we hit GH#3210 if we aren't careful
    raw_crop.save(split_fname, split_size='4.5MB',
                  buffer_size_sec=1., overwrite=True)
    raw_read = read_raw_fif(split_fname)
    assert_allclose(raw_crop[:][0], raw_read[:][0], atol=1e-20)

    # Check our buffer arithmetic

    # 1 buffer required
    raw_crop = raw_1.copy().crop(0, 1)
    raw_crop.save(split_fname, buffer_size_sec=1., overwrite=True)
    raw_read = read_raw_fif(split_fname)
    assert_array_equal(np.diff(raw_read._raw_extras[0]['bounds']), (301,))
    assert_allclose(raw_crop[:][0], raw_read[:][0])
    # 2 buffers required
    raw_crop.save(split_fname, buffer_size_sec=0.5, overwrite=True)
    raw_read = read_raw_fif(split_fname)
    assert_array_equal(np.diff(raw_read._raw_extras[0]['bounds']), (151, 150))
    assert_allclose(raw_crop[:][0], raw_read[:][0])
    # 2 buffers required
    raw_crop.save(split_fname,
                  buffer_size_sec=1. - 1.01 / raw_crop.info['sfreq'],
                  overwrite=True)
    raw_read = read_raw_fif(split_fname)
    assert_array_equal(np.diff(raw_read._raw_extras[0]['bounds']), (300, 1))
    assert_allclose(raw_crop[:][0], raw_read[:][0])
    raw_crop.save(split_fname,
                  buffer_size_sec=1. - 2.01 / raw_crop.info['sfreq'],
                  overwrite=True)
    raw_read = read_raw_fif(split_fname)
    assert_array_equal(np.diff(raw_read._raw_extras[0]['bounds']), (299, 2))
    assert_allclose(raw_crop[:][0], raw_read[:][0])

    # proper ending
    assert op.isdir(tmp_path)
    with pytest.raises(ValueError, match='must end with an underscore'):
        raw_crop.save(
            tmp_path / 'test.fif', split_naming='bids', verbose='error')

    # reserved file is deleted
    fname = tmp_path / 'test_raw.fif'
    monkeypatch.setattr(base, '_write_raw_fid', _err)
    with pytest.raises(RuntimeError, match='Killed mid-write'):
        raw_1.save(fname, split_size='10MB', split_naming='bids')
    assert op.isfile(fname)
    assert not op.isfile(tmp_path / 'test_split-01_raw.fif')


def _err(*args, **kwargs):
    raise RuntimeError('Killed mid-write')


def _no_write_file_name(fid, kind, data):
    assert kind == FIFF.FIFF_REF_FILE_NAME  # the only string we actually write
    return


def test_split_numbers(tmp_path, monkeypatch):
    """Test handling of split files using numbers instead of names."""
    monkeypatch.setattr(base, 'write_string', _no_write_file_name)
    raw = read_raw_fif(test_fif_fname).pick('eeg')
    # gh-8339
    dashes_fname = tmp_path / 'sub-1_ses-2_task-3_raw.fif'
    raw.save(dashes_fname, split_size='5MB',
             buffer_size_sec=1.)
    assert op.isfile(dashes_fname)
    next_fname = str(dashes_fname)[:-4] + '-1.fif'
    assert op.isfile(next_fname)
    raw_read = read_raw_fif(dashes_fname)
    assert_allclose(raw.times, raw_read.times)
    assert_allclose(raw.get_data(), raw_read.get_data(), atol=1e-16)


def test_load_bad_channels(tmp_path):
    """Test reading/writing of bad channels."""
    # Load correctly marked file (manually done in mne_process_raw)
    raw_marked = read_raw_fif(fif_bad_marked_fname)
    correct_bads = raw_marked.info['bads']
    raw = read_raw_fif(test_fif_fname)
    # Make sure it starts clean
    assert_array_equal(raw.info['bads'], [])

    # Test normal case
    raw.load_bad_channels(bad_file_works)
    # Write it out, read it in, and check
    raw.save(tmp_path / 'foo_raw.fif')
    raw_new = read_raw_fif(tmp_path / 'foo_raw.fif')
    assert correct_bads == raw_new.info['bads']
    # Reset it
    raw.info['bads'] = []

    # Test bad case
    pytest.raises(ValueError, raw.load_bad_channels, bad_file_wrong)

    # Test forcing the bad case
    with pytest.warns(RuntimeWarning, match='1 bad channel'):
        raw.load_bad_channels(bad_file_wrong, force=True)

    # write it out, read it in, and check
    raw.save(tmp_path / 'foo_raw.fif', overwrite=True)
    raw_new = read_raw_fif(tmp_path / 'foo_raw.fif')
    assert correct_bads == raw_new.info['bads']

    # Check that bad channels are cleared
    raw.load_bad_channels(None)
    raw.save(tmp_path / 'foo_raw.fif', overwrite=True)
    raw_new = read_raw_fif(tmp_path / 'foo_raw.fif')
    assert raw_new.info['bads'] == []


@pytest.mark.slowtest
@testing.requires_testing_data
def test_io_raw(tmp_path):
    """Test IO for raw data (Neuromag)."""
    rng = np.random.RandomState(0)
    # test unicode io
    for chars in [u'äöé', 'a']:
        with read_raw_fif(fif_fname) as r:
            assert ('Raw' in repr(r))
            assert (op.basename(fif_fname) in repr(r))
            r.info['description'] = chars
            temp_file = tmp_path / 'raw.fif'
            r.save(temp_file, overwrite=True)
            with read_raw_fif(temp_file) as r2:
                desc2 = r2.info['description']
            assert desc2 == chars

    # Let's construct a simple test for IO first
    raw = read_raw_fif(fif_fname).crop(0, 3.5)
    raw.load_data()
    # put in some data that we know the values of
    data = rng.randn(raw._data.shape[0], raw._data.shape[1])
    raw._data[:, :] = data
    # save it somewhere
    fname = tmp_path / 'test_copy_raw.fif'
    raw.save(fname, buffer_size_sec=1.0)
    # read it in, make sure the whole thing matches
    raw = read_raw_fif(fname)
    assert_allclose(data, raw[:, :][0], rtol=1e-6, atol=1e-20)
    # let's read portions across the 1-sec tag boundary, too
    inds = raw.time_as_index([1.75, 2.25])
    sl = slice(inds[0], inds[1])
    assert_allclose(data[:, sl], raw[:, sl][0], rtol=1e-6, atol=1e-20)


@pytest.mark.parametrize('fname_in, fname_out', [
    (test_fif_fname, 'raw.fif'),
    pytest.param(test_fif_gz_fname, 'raw.fif.gz', marks=pytest.mark.slowtest),
    (ctf_fname, 'raw.fif')])
def test_io_raw_additional(fname_in, fname_out, tmp_path):
    """Test IO for raw data (Neuromag + CTF + gz)."""
    fname_out = tmp_path / fname_out
    raw = read_raw_fif(fname_in).crop(0, 2)

    nchan = raw.info['nchan']
    ch_names = raw.info['ch_names']
    meg_channels_idx = [k for k in range(nchan)
                        if ch_names[k][0] == 'M']
    n_channels = 100
    meg_channels_idx = meg_channels_idx[:n_channels]
    start, stop = raw.time_as_index([0, 5], use_rounding=True)
    data, times = raw[meg_channels_idx, start:(stop + 1)]
    meg_ch_names = [ch_names[k] for k in meg_channels_idx]

    # Set up pick list: MEG + STI 014 - bad channels
    include = ['STI 014']
    include += meg_ch_names
    picks = pick_types(raw.info, meg=True, eeg=False, stim=True,
                       misc=True, ref_meg=True, include=include,
                       exclude='bads')

    # Writing with drop_small_buffer True
    raw.save(fname_out, picks, tmin=0, tmax=4, buffer_size_sec=3,
             drop_small_buffer=True, overwrite=True)
    raw2 = read_raw_fif(fname_out)

    sel = pick_channels(raw2.ch_names, meg_ch_names)
    data2, times2 = raw2[sel, :]
    assert (times2.max() <= 3)

    # Writing
    raw.save(fname_out, picks, tmin=0, tmax=5, overwrite=True)

    if fname_in in (fif_fname, fif_fname + '.gz'):
        assert len(raw.info['dig']) == 146

    raw2 = read_raw_fif(fname_out)

    sel = pick_channels(raw2.ch_names, meg_ch_names)
    data2, times2 = raw2[sel, :]

    assert_allclose(data, data2, rtol=1e-6, atol=1e-20)
    assert_allclose(times, times2)
    assert_allclose(raw.info['sfreq'], raw2.info['sfreq'], rtol=1e-5)

    # check transformations
    for trans in ['dev_head_t', 'dev_ctf_t', 'ctf_head_t']:
        if raw.info[trans] is None:
            assert (raw2.info[trans] is None)
        else:
            assert_array_equal(raw.info[trans]['trans'],
                               raw2.info[trans]['trans'])

            # check transformation 'from' and 'to'
            if trans.startswith('dev'):
                from_id = FIFF.FIFFV_COORD_DEVICE
            else:
                from_id = FIFF.FIFFV_MNE_COORD_CTF_HEAD
            if trans[4:8] == 'head':
                to_id = FIFF.FIFFV_COORD_HEAD
            else:
                to_id = FIFF.FIFFV_MNE_COORD_CTF_HEAD
            for raw_ in [raw, raw2]:
                assert raw_.info[trans]['from'] == from_id
                assert raw_.info[trans]['to'] == to_id

    if fname_in == fif_fname or fname_in == fif_fname + '.gz':
        assert_allclose(raw.info['dig'][0]['r'], raw2.info['dig'][0]['r'])

    # test warnings on bad filenames
    raw_badname = tmp_path / 'test-bad-name.fif.gz'
    with pytest.warns(RuntimeWarning, match='raw.fif'):
        raw.save(raw_badname)
    with pytest.warns(RuntimeWarning, match='raw.fif'):
        read_raw_fif(raw_badname)


@testing.requires_testing_data
@pytest.mark.parametrize('dtype', ('complex128', 'complex64'))
def test_io_complex(tmp_path, dtype):
    """Test IO with complex data types."""
    rng = np.random.RandomState(0)
    n_ch = 5
    raw = read_raw_fif(fif_fname).crop(0, 1).pick(np.arange(n_ch)).load_data()
    data_orig = raw.get_data()
    imag_rand = np.array(1j * rng.randn(n_ch, len(raw.times)), dtype=dtype)
    raw_cp = raw.copy()
    raw_cp._data = np.array(raw_cp._data, dtype)
    raw_cp._data += imag_rand
    with pytest.warns(RuntimeWarning, match='Saving .* complex data.'):
        raw_cp.save(tmp_path / 'raw.fif', overwrite=True)

    raw2 = read_raw_fif(tmp_path / 'raw.fif')
    raw2_data, _ = raw2[:]
    assert_allclose(raw2_data, raw_cp._data)
    # with preloading
    raw2 = read_raw_fif(tmp_path / 'raw.fif', preload=True)
    raw2_data, _ = raw2[:]
    assert_allclose(raw2_data, raw_cp._data)
    assert_allclose(data_orig, raw_cp._data.real)


@testing.requires_testing_data
def test_getitem():
    """Test getitem/indexing of Raw."""
    for preload in [False, True, 'memmap.dat']:
        raw = read_raw_fif(fif_fname, preload=preload)
        data, times = raw[0, :]
        data1, times1 = raw[0]
        assert_array_equal(data, data1)
        assert_array_equal(times, times1)
        data, times = raw[0:2, :]
        data1, times1 = raw[0:2]
        assert_array_equal(data, data1)
        assert_array_equal(times, times1)
        data1, times1 = raw[[0, 1]]
        assert_array_equal(data, data1)
        assert_array_equal(times, times1)
        assert_array_equal(raw[raw.ch_names[0]][0][0], raw[0][0][0])
        assert_array_equal(
            raw[-10:-1, :][0],
            raw[len(raw.ch_names) - 10:len(raw.ch_names) - 1, :][0])
        with pytest.raises(ValueError, match='No appropriate channels'):
            raw[slice(-len(raw.ch_names) - 1), slice(None)]
        with pytest.raises(ValueError, match='must be'):
            raw[-1000]


@testing.requires_testing_data
def test_proj(tmp_path):
    """Test SSP proj operations."""
    for proj in [True, False]:
        raw = read_raw_fif(fif_fname, preload=False)
        if proj:
            raw.apply_proj()
        assert (all(p['active'] == proj for p in raw.info['projs']))

        data, times = raw[0:2, :]
        data1, times1 = raw[0:2]
        assert_array_equal(data, data1)
        assert_array_equal(times, times1)

        # test adding / deleting proj
        if proj:
            pytest.raises(ValueError, raw.add_proj, [],
                          {'remove_existing': True})
            pytest.raises(ValueError, raw.del_proj, 0)
        else:
            projs = deepcopy(raw.info['projs'])
            n_proj = len(raw.info['projs'])
            raw.del_proj(0)
            assert len(raw.info['projs']) == n_proj - 1
            raw.add_proj(projs, remove_existing=False)
            # Test that already existing projections are not added.
            assert len(raw.info['projs']) == n_proj
            raw.add_proj(projs[:-1], remove_existing=True)
            assert len(raw.info['projs']) == n_proj - 1

    # test apply_proj() with and without preload
    for preload in [True, False]:
        raw = read_raw_fif(fif_fname, preload=preload)
        data, times = raw[:, 0:2]
        raw.apply_proj()
        data_proj_1 = np.dot(raw._projector, data)

        # load the file again without proj
        raw = read_raw_fif(fif_fname, preload=preload)

        # write the file with proj. activated, make sure proj has been applied
        raw.save(tmp_path / 'raw.fif', proj=True, overwrite=True)
        raw2 = read_raw_fif(tmp_path / 'raw.fif')
        data_proj_2, _ = raw2[:, 0:2]
        assert_allclose(data_proj_1, data_proj_2)
        assert (all(p['active'] for p in raw2.info['projs']))

        # read orig file with proj. active
        raw2 = read_raw_fif(fif_fname, preload=preload)
        raw2.apply_proj()
        data_proj_2, _ = raw2[:, 0:2]
        assert_allclose(data_proj_1, data_proj_2)
        assert (all(p['active'] for p in raw2.info['projs']))

        # test that apply_proj works
        raw.apply_proj()
        data_proj_2, _ = raw[:, 0:2]
        assert_allclose(data_proj_1, data_proj_2)
        assert_allclose(data_proj_2, np.dot(raw._projector, data_proj_2))

    # Test that picking removes projectors ...
    raw = read_raw_fif(fif_fname)
    n_projs = len(raw.info['projs'])
    raw.pick_types(meg=False, eeg=True)
    assert len(raw.info['projs']) == n_projs - 3

    # ... but only if it doesn't apply to any channels in the dataset anymore.
    raw = read_raw_fif(fif_fname)
    n_projs = len(raw.info['projs'])
    raw.pick_types(meg='mag', eeg=True)
    assert len(raw.info['projs']) == n_projs

    # I/O roundtrip of an MEG projector with a Raw that only contains EEG
    # data.
    out_fname = tmp_path / 'test_raw.fif'
    raw = read_raw_fif(test_fif_fname, preload=True).crop(0, 0.002)
    proj = raw.info['projs'][-1]
    raw.pick_types(meg=False, eeg=True)
    raw.add_proj(proj)  # Restore, because picking removed it!
    raw._data.fill(0)
    raw._data[-1] = 1.
    raw.save(out_fname)
    raw = read_raw_fif(out_fname, preload=False)
    raw.apply_proj()
    assert_allclose(raw[:, :][0][:1], raw[0, :][0])


@testing.requires_testing_data
@pytest.mark.parametrize('preload', [False, True, 'memmap.dat'])
def test_preload_modify(preload, tmp_path):
    """Test preloading and modifying data."""
    rng = np.random.RandomState(0)
    raw = read_raw_fif(fif_fname, preload=preload)

    nsamp = raw.last_samp - raw.first_samp + 1
    picks = pick_types(raw.info, meg='grad', exclude='bads')

    data = rng.randn(len(picks), nsamp // 2)

    try:
        raw[picks, :nsamp // 2] = data
    except RuntimeError:
        if not preload:
            return
        else:
            raise

    tmp_fname = tmp_path / 'raw.fif'
    raw.save(tmp_fname, overwrite=True)

    raw_new = read_raw_fif(tmp_fname)
    data_new, _ = raw_new[picks, :nsamp // 2]

    assert_allclose(data, data_new)


@pytest.mark.slowtest
@testing.requires_testing_data
def test_filter():
    """Test filtering (FIR and IIR) and Raw.apply_function interface."""
    raw = read_raw_fif(fif_fname).crop(0, 7)
    raw.load_data()
    sig_dec_notch = 12
    sig_dec_notch_fit = 12
    picks_meg = pick_types(raw.info, meg=True, exclude='bads')
    picks = picks_meg[:4]

    trans = 2.0
    filter_params = dict(picks=picks, filter_length='auto',
                         h_trans_bandwidth=trans, l_trans_bandwidth=trans,
                         fir_design='firwin')
    raw_lp = raw.copy().filter(None, 8.0, **filter_params)
    raw_hp = raw.copy().filter(16.0, None, **filter_params)
    raw_bp = raw.copy().filter(8.0 + trans, 16.0 - trans, **filter_params)
    raw_bs = raw.copy().filter(16.0, 8.0, **filter_params)

    data, _ = raw[picks, :]

    lp_data, _ = raw_lp[picks, :]
    hp_data, _ = raw_hp[picks, :]
    bp_data, _ = raw_bp[picks, :]
    bs_data, _ = raw_bs[picks, :]

    tols = dict(atol=1e-20, rtol=1e-5)
    assert_allclose(bs_data, lp_data + hp_data, **tols)
    assert_allclose(data, lp_data + bp_data + hp_data, **tols)
    assert_allclose(data, bp_data + bs_data, **tols)

    filter_params_iir = dict(picks=picks, n_jobs=2, method='iir',
                             iir_params=dict(output='ba'))
    raw_lp_iir = raw.copy().filter(None, 4.0, **filter_params_iir)
    raw_hp_iir = raw.copy().filter(8.0, None, **filter_params_iir)
    raw_bp_iir = raw.copy().filter(4.0, 8.0, **filter_params_iir)
    del filter_params_iir
    lp_data_iir, _ = raw_lp_iir[picks, :]
    hp_data_iir, _ = raw_hp_iir[picks, :]
    bp_data_iir, _ = raw_bp_iir[picks, :]
    summation = lp_data_iir + hp_data_iir + bp_data_iir
    assert_array_almost_equal(data[:, 100:-100], summation[:, 100:-100], 11)

    # make sure we didn't touch other channels
    data, _ = raw[picks_meg[4:], :]
    bp_data, _ = raw_bp[picks_meg[4:], :]
    assert_array_equal(data, bp_data)
    bp_data_iir, _ = raw_bp_iir[picks_meg[4:], :]
    assert_array_equal(data, bp_data_iir)

    # ... and that inplace changes are inplace
    raw_copy = raw.copy()
    assert np.may_share_memory(raw._data, raw._data)
    assert not np.may_share_memory(raw_copy._data, raw._data)
    # this could be assert_array_equal but we do this to mirror the call below
    assert (raw._data[0] == raw_copy._data[0]).all()
    raw_copy.filter(None, 20., n_jobs=2, **filter_params)
    assert not (raw._data[0] == raw_copy._data[0]).all()
    assert_array_equal(raw.copy().filter(None, 20., **filter_params)._data,
                       raw_copy._data)

    # do a very simple check on line filtering
    raw_bs = raw.copy().filter(60.0 + trans, 60.0 - trans, **filter_params)
    data_bs, _ = raw_bs[picks, :]
    raw_notch = raw.copy().notch_filter(
        60.0, picks=picks, n_jobs=2, method='fir',
        trans_bandwidth=2 * trans)
    data_notch, _ = raw_notch[picks, :]
    assert_array_almost_equal(data_bs, data_notch, sig_dec_notch)

    # now use the sinusoidal fitting
    assert raw.times[-1] < 10  # catch error with filter_length > n_times
    raw_notch = raw.copy().notch_filter(
        None, picks=picks, n_jobs=2, method='spectrum_fit',
        filter_length='10s')
    data_notch, _ = raw_notch[picks, :]
    data, _ = raw[picks, :]
    assert_array_almost_equal(data, data_notch, sig_dec_notch_fit)

    # filter should set the "lowpass" and "highpass" parameters
    raw = RawArray(np.random.randn(3, 1000),
                   create_info(3, 1000., ['eeg'] * 2 + ['stim']))
    with raw.info._unlock():
        raw.info['lowpass'] = raw.info['highpass'] = None
    for kind in ('none', 'lowpass', 'highpass', 'bandpass', 'bandstop'):
        print(kind)
        h_freq = l_freq = None
        if kind in ('lowpass', 'bandpass'):
            h_freq = 70
        if kind in ('highpass', 'bandpass'):
            l_freq = 30
        if kind == 'bandstop':
            l_freq, h_freq = 70, 30
        assert (raw.info['lowpass'] is None)
        assert (raw.info['highpass'] is None)
        kwargs = dict(l_trans_bandwidth=20, h_trans_bandwidth=20,
                      filter_length='auto', phase='zero', fir_design='firwin')
        raw_filt = raw.copy().filter(l_freq, h_freq, picks=np.arange(1),
                                     **kwargs)
        assert (raw.info['lowpass'] is None)
        assert (raw.info['highpass'] is None)
        raw_filt = raw.copy().filter(l_freq, h_freq, **kwargs)
        wanted_h = h_freq if kind != 'bandstop' else None
        wanted_l = l_freq if kind != 'bandstop' else None
        assert raw_filt.info['lowpass'] == wanted_h
        assert raw_filt.info['highpass'] == wanted_l
        # Using all data channels should still set the params (GH#3259)
        raw_filt = raw.copy().filter(l_freq, h_freq, picks=np.arange(2),
                                     **kwargs)
        assert raw_filt.info['lowpass'] == wanted_h
        assert raw_filt.info['highpass'] == wanted_l


def test_filter_picks():
    """Test filtering default channel picks."""
    ch_types = ['mag', 'grad', 'eeg', 'seeg', 'dbs', 'misc', 'stim', 'ecog',
                'hbo', 'hbr']
    info = create_info(ch_names=ch_types, ch_types=ch_types, sfreq=256)
    raw = RawArray(data=np.zeros((len(ch_types), 1000)), info=info)

    # -- Deal with meg mag grad and fnirs exceptions
    ch_types = ('misc', 'stim', 'meg', 'eeg', 'seeg', 'dbs', 'ecog')

    # -- Filter data channels
    for ch_type in ('mag', 'grad', 'eeg', 'seeg', 'dbs', 'ecog', 'hbo', 'hbr'):
        picks = {ch: ch == ch_type for ch in ch_types}
        picks['meg'] = ch_type if ch_type in ('mag', 'grad') else False
        picks['fnirs'] = ch_type if ch_type in ('hbo', 'hbr') else False
        raw_ = raw.copy().pick_types(**picks)
        raw_.filter(10, 30, fir_design='firwin')

    # -- Error if no data channel
    for ch_type in ('misc', 'stim'):
        picks = {ch: ch == ch_type for ch in ch_types}
        raw_ = raw.copy().pick_types(**picks)
        pytest.raises(ValueError, raw_.filter, 10, 30)


@testing.requires_testing_data
def test_crop():
    """Test cropping raw files."""
    # split a concatenated file to test a difficult case
    raw = concatenate_raws([read_raw_fif(f)
                            for f in [fif_fname, fif_fname]])
    split_size = 10.  # in seconds
    sfreq = raw.info['sfreq']
    nsamp = (raw.last_samp - raw.first_samp + 1)

    # do an annoying case (off-by-one splitting)
    tmins = np.r_[1., np.round(np.arange(0., nsamp - 1, split_size * sfreq))]
    tmins = np.sort(tmins)
    tmaxs = np.concatenate((tmins[1:] - 1, [nsamp - 1]))
    tmaxs /= sfreq
    tmins /= sfreq
    raws = [None] * len(tmins)
    for ri, (tmin, tmax) in enumerate(zip(tmins, tmaxs)):
        raws[ri] = raw.copy().crop(tmin, tmax)
        if ri < len(tmins) - 1:
            assert_allclose(
                raws[ri].times,
                raw.copy().crop(tmin, tmins[ri + 1], include_tmax=False).times)
        assert raws[ri]
    all_raw_2 = concatenate_raws(raws, preload=False)
    assert raw.first_samp == all_raw_2.first_samp
    assert raw.last_samp == all_raw_2.last_samp
    assert_array_equal(raw[:, :][0], all_raw_2[:, :][0])

    tmins = np.round(np.arange(0., nsamp - 1, split_size * sfreq))
    tmaxs = np.concatenate((tmins[1:] - 1, [nsamp - 1]))
    tmaxs /= sfreq
    tmins /= sfreq

    # going in reverse order so the last fname is the first file (need it
    # later)
    raws = [None] * len(tmins)
    for ri, (tmin, tmax) in enumerate(zip(tmins, tmaxs)):
        raws[ri] = raw.copy().crop(tmin, tmax)
    # test concatenation of split file
    all_raw_1 = concatenate_raws(raws, preload=False)

    all_raw_2 = raw.copy().crop(0, None)
    for ar in [all_raw_1, all_raw_2]:
        assert raw.first_samp == ar.first_samp
        assert raw.last_samp == ar.last_samp
        assert_array_equal(raw[:, :][0], ar[:, :][0])

    # test shape consistency of cropped raw
    data = np.zeros((1, 1002001))
    info = create_info(1, 1000)
    raw = RawArray(data, info)
    for tmin in range(0, 1001, 100):
        raw1 = raw.copy().crop(tmin=tmin, tmax=tmin + 2)
        assert raw1[:][0].shape == (1, 2001)

    # degenerate
    with pytest.raises(ValueError, match='No samples.*when include_tmax=Fals'):
        raw.crop(0, 0, include_tmax=False)

    # edge cases cropping to exact duration +/- 1 sample
    data = np.zeros((1, 100))
    info = create_info(1, 100)
    raw = RawArray(data, info)
    with pytest.raises(ValueError, match='tmax \\(1\\) must be less than or '):
        raw.copy().crop(tmax=1, include_tmax=True)
    raw1 = raw.copy().crop(tmax=1 - 1 / raw.info['sfreq'], include_tmax=True)
    assert raw.n_times == raw1.n_times
    raw2 = raw.copy().crop(tmax=1, include_tmax=False)
    assert raw.n_times == raw2.n_times
    raw3 = raw.copy().crop(tmax=1 - 1 / raw.info['sfreq'], include_tmax=False)
    assert raw.n_times - 1 == raw3.n_times


@testing.requires_testing_data
def test_resample_equiv():
    """Test resample (with I/O and multiple files)."""
    raw = read_raw_fif(fif_fname).crop(0, 1)
    raw_preload = raw.copy().load_data()
    for r in (raw, raw_preload):
        r.resample(r.info['sfreq'] / 4.)
    assert_allclose(raw._data, raw_preload._data)


@pytest.mark.slowtest
@testing.requires_testing_data
@pytest.mark.parametrize('preload, n, npad', [
    (True, 512, 'auto'),
    (False, 512, 0),
])
def test_resample(tmp_path, preload, n, npad):
    """Test resample (with I/O and multiple files)."""
    raw = read_raw_fif(fif_fname)
    raw.crop(0, raw.times[n - 1])
    assert len(raw.times) == n
    if preload:
        raw.load_data()
    raw_resamp = raw.copy()
    sfreq = raw.info['sfreq']
    # test parallel on upsample
    raw_resamp.resample(sfreq * 2, n_jobs=2, npad=npad)
    assert raw_resamp.n_times == len(raw_resamp.times)
    raw_resamp.save(tmp_path / 'raw_resamp-raw.fif')
    raw_resamp = read_raw_fif(tmp_path / 'raw_resamp-raw.fif', preload=True)
    assert sfreq == raw_resamp.info['sfreq'] / 2
    assert raw.n_times == raw_resamp.n_times // 2
    assert raw_resamp.get_data().shape[1] == raw_resamp.n_times
    assert raw.get_data().shape[0] == raw_resamp._data.shape[0]
    # test non-parallel on downsample
    raw_resamp.resample(sfreq, n_jobs=None, npad=npad)
    assert raw_resamp.info['sfreq'] == sfreq
    assert raw.get_data().shape == raw_resamp._data.shape
    assert raw.first_samp == raw_resamp.first_samp
    assert raw.last_samp == raw.last_samp
    # upsampling then downsampling doubles resampling error, but this still
    # works (hooray). Note that the stim channels had to be sub-sampled
    # without filtering to be accurately preserved
    # note we have to treat MEG and EEG+STIM channels differently (tols)
    assert_allclose(raw.get_data()[:306, 200:-200],
                    raw_resamp._data[:306, 200:-200],
                    rtol=1e-2, atol=1e-12)
    assert_allclose(raw.get_data()[306:, 200:-200],
                    raw_resamp._data[306:, 200:-200],
                    rtol=1e-2, atol=1e-7)

    # now check multiple file support w/resampling, as order of operations
    # (concat, resample) should not affect our data
    raw1 = raw.copy()
    raw2 = raw.copy()
    raw3 = raw.copy()
    raw4 = raw.copy()
    raw1 = concatenate_raws([raw1, raw2])
    raw1.resample(10., npad=npad)
    raw3.resample(10., npad=npad)
    raw4.resample(10., npad=npad)
    raw3 = concatenate_raws([raw3, raw4])
    assert_array_equal(raw1._data, raw3._data)
    assert_array_equal(raw1._first_samps, raw3._first_samps)
    assert_array_equal(raw1._last_samps, raw3._last_samps)
    assert_array_equal(raw1._raw_lengths, raw3._raw_lengths)
    assert raw1.first_samp == raw3.first_samp
    assert raw1.last_samp == raw3.last_samp
    assert raw1.info['sfreq'] == raw3.info['sfreq']

    # smoke test crop after resample
    raw4.crop(tmin=raw4.times[1], tmax=raw4.times[-1])

    # test resampling of stim channel

    # basic decimation
    stim = [1, 1, 1, 1, 0, 0, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0]
    raw = RawArray([stim], create_info(1, len(stim), ['stim']))
    assert_allclose(raw.resample(8., npad=npad)._data,
                    [[1, 1, 0, 0, 1, 1, 0, 0]])

    # decimation of multiple stim channels
    raw = RawArray(2 * [stim], create_info(2, len(stim), 2 * ['stim']))
    assert_allclose(raw.resample(8., npad=npad, verbose='error')._data,
                    [[1, 1, 0, 0, 1, 1, 0, 0],
                     [1, 1, 0, 0, 1, 1, 0, 0]])

    # decimation that could potentially drop events if the decimation is
    # done naively
    stim = [0, 0, 0, 1, 1, 0, 0, 0]
    raw = RawArray([stim], create_info(1, len(stim), ['stim']))
    assert_allclose(raw.resample(4., npad=npad)._data,
                    [[0, 1, 1, 0]])

    # two events are merged in this case (warning)
    stim = [0, 0, 1, 1, 1, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0, 0]
    raw = RawArray([stim], create_info(1, len(stim), ['stim']))
    with pytest.warns(RuntimeWarning, match='become unreliable'):
        raw.resample(8., npad=npad)

    # events are dropped in this case (warning)
    stim = [0, 1, 1, 0, 0, 1, 1, 0]
    raw = RawArray([stim], create_info(1, len(stim), ['stim']))
    with pytest.warns(RuntimeWarning, match='become unreliable'):
        raw.resample(4., npad=npad)

    # test resampling events: this should no longer give a warning
    # we often have first_samp != 0, include it here too
    stim = [0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 1]  # an event at end
    # test is on half the sfreq, but should work with trickier ones too
    o_sfreq, sfreq_ratio = len(stim), 0.5
    n_sfreq = o_sfreq * sfreq_ratio
    first_samp = len(stim) // 2
    raw = RawArray([stim], create_info(1, o_sfreq, ['stim']),
                   first_samp=first_samp)
    events = find_events(raw)
    raw, events = raw.resample(n_sfreq, events=events, npad=npad)
    # Try index into raw.times with resampled events:
    raw.times[events[:, 0] - raw.first_samp]
    n_fsamp = int(first_samp * sfreq_ratio)  # how it's calc'd in base.py
    # NB np.round used for rounding event times, which has 0.5 as corner case:
    # https://docs.scipy.org/doc/numpy/reference/generated/numpy.around.html
    assert_array_equal(
        events,
        np.array([[np.round(1 * sfreq_ratio) + n_fsamp, 0, 1],
                  [np.round(10 * sfreq_ratio) + n_fsamp, 0, 1],
                  [np.minimum(np.round(15 * sfreq_ratio),
                              raw._data.shape[1] - 1) + n_fsamp, 0, 1]]))

    # test copy flag
    stim = [1, 1, 1, 1, 0, 0, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0]
    raw = RawArray([stim], create_info(1, len(stim), ['stim']))
    raw_resampled = raw.copy().resample(4., npad=npad)
    assert (raw_resampled is not raw)
    raw_resampled = raw.resample(4., npad=npad)
    assert (raw_resampled is raw)

    # resample should still work even when no stim channel is present
    raw = RawArray(np.random.randn(1, 100), create_info(1, 100, ['eeg']))
    with raw.info._unlock():
        raw.info['lowpass'] = 50.
    raw.resample(10, npad=npad)
    assert raw.info['lowpass'] == 5.
    assert len(raw) == 10


def test_resample_stim():
    """Test stim_picks argument."""
    data = np.ones((2, 1000))
    info = create_info(2, 1000., ('eeg', 'misc'))
    raw = RawArray(data, info)
    raw.resample(500., stim_picks='misc')


@testing.requires_testing_data
def test_hilbert():
    """Test computation of analytic signal using hilbert."""
    raw = read_raw_fif(fif_fname, preload=True)
    picks_meg = pick_types(raw.info, meg=True, exclude='bads')
    picks = picks_meg[:4]

    raw_filt = raw.copy()
    raw_filt.filter(10, 20, picks=picks, l_trans_bandwidth='auto',
                    h_trans_bandwidth='auto', filter_length='auto',
                    phase='zero', fir_window='blackman', fir_design='firwin')
    raw_filt_2 = raw_filt.copy()

    raw2 = raw.copy()
    raw3 = raw.copy()
    raw.apply_hilbert(picks, n_fft='auto')
    raw2.apply_hilbert(picks, n_fft='auto', envelope=True)

    # Test custom n_fft
    raw_filt.apply_hilbert(picks, n_fft='auto')
    n_fft = 2 ** int(np.ceil(np.log2(raw_filt_2.n_times + 1000)))
    raw_filt_2.apply_hilbert(picks, n_fft=n_fft)
    assert raw_filt._data.shape == raw_filt_2._data.shape
    assert_allclose(raw_filt._data[:, 50:-50], raw_filt_2._data[:, 50:-50],
                    atol=1e-13, rtol=1e-2)
    with pytest.raises(ValueError, match='n_fft.*must be at least the number'):
        raw3.apply_hilbert(picks, n_fft=raw3.n_times - 100)

    env = np.abs(raw._data[picks, :])
    assert_allclose(env, raw2._data[picks, :], rtol=1e-2, atol=1e-13)


@testing.requires_testing_data
def test_raw_copy():
    """Test Raw copy."""
    raw = read_raw_fif(fif_fname, preload=True)
    data, _ = raw[:, :]
    copied = raw.copy()
    copied_data, _ = copied[:, :]
    assert_array_equal(data, copied_data)
    assert sorted(raw.__dict__.keys()) == sorted(copied.__dict__.keys())

    raw = read_raw_fif(fif_fname, preload=False)
    data, _ = raw[:, :]
    copied = raw.copy()
    copied_data, _ = copied[:, :]
    assert_array_equal(data, copied_data)
    assert sorted(raw.__dict__.keys()) == sorted(copied.__dict__.keys())


@requires_pandas
def test_to_data_frame():
    """Test raw Pandas exporter."""
    from pandas import Timedelta
    raw = read_raw_fif(test_fif_fname).crop(0, 1).load_data()
    df = raw.to_data_frame(index='time')
    assert ((df.columns == raw.ch_names).all())
    df = raw.to_data_frame(index=None)
    assert ('time' in df.columns)
    assert_array_equal(df.values[:, 1], raw._data[0] * 1e13)
    assert_array_equal(df.values[:, 3], raw._data[2] * 1e15)
    # test long format
    df_long = raw.to_data_frame(long_format=True)
    assert len(df_long) == raw.get_data().size
    expected = ('time', 'channel', 'ch_type', 'value')
    assert set(expected) == set(df_long.columns)
    # test bad time format
    with pytest.raises(ValueError, match='not a valid time format. Valid'):
        raw.to_data_frame(time_format='foo')
    # test time format error handling
    raw.set_meas_date(None)
    with pytest.warns(RuntimeWarning, match='Cannot convert to Datetime when'):
        df = raw.to_data_frame(time_format='datetime')
    assert isinstance(df['time'].iloc[0], Timedelta)


@requires_pandas
@pytest.mark.parametrize('time_format', (None, 'ms', 'timedelta', 'datetime'))
def test_to_data_frame_time_format(time_format):
    """Test time conversion in epochs Pandas exporter."""
    from pandas import Timedelta, Timestamp, to_timedelta
    raw = read_raw_fif(test_fif_fname, preload=True)
    # test time_format
    df = raw.to_data_frame(time_format=time_format)
    dtypes = {None: np.float64, 'ms': np.int64, 'timedelta': Timedelta,
              'datetime': Timestamp}
    assert isinstance(df['time'].iloc[0], dtypes[time_format])
    # test values
    _, times = raw[0, :10]
    offset = 0.
    if time_format == 'datetime':
        times += raw.first_time
        offset = raw.info['meas_date']
    elif time_format == 'timedelta':
        offset = Timedelta(0.)
    funcs = {None: lambda x: x,
             'ms': lambda x: np.rint(x * 1e3).astype(int),  # s → ms
             'timedelta': partial(to_timedelta, unit='s'),
             'datetime': partial(to_timedelta, unit='s')
             }
    assert_array_equal(funcs[time_format](times) + offset, df['time'][:10])


def test_add_channels():
    """Test raw splitting / re-appending channel types."""
    rng = np.random.RandomState(0)
    raw = read_raw_fif(test_fif_fname).crop(0, 1).load_data()
    assert raw._orig_units == {}
    raw_nopre = read_raw_fif(test_fif_fname, preload=False)
    raw_eeg_meg = raw.copy().pick_types(meg=True, eeg=True)
    raw_eeg = raw.copy().pick_types(eeg=True)
    raw_meg = raw.copy().pick_types(meg=True)
    raw_stim = raw.copy().pick_types(stim=True)
    raw_new = raw_meg.copy().add_channels([raw_eeg, raw_stim])
    assert (
        all(ch in raw_new.ch_names
            for ch in list(raw_stim.ch_names) + list(raw_meg.ch_names))
    )
    raw_new = raw_meg.copy().add_channels([raw_eeg])

    assert (ch in raw_new.ch_names for ch in raw.ch_names)
    assert_array_equal(raw_new[:, :][0], raw_eeg_meg[:, :][0])
    assert_array_equal(raw_new[:, :][1], raw[:, :][1])
    assert (all(ch not in raw_new.ch_names for ch in raw_stim.ch_names))

    # Testing force updates
    raw_arr_info = create_info(['1', '2'], raw_meg.info['sfreq'], 'eeg')
    orig_head_t = raw_arr_info['dev_head_t']
    raw_arr = rng.randn(2, raw_eeg.n_times)
    raw_arr = RawArray(raw_arr, raw_arr_info)
    # This should error because of conflicts in Info
    raw_arr.info['dev_head_t'] = orig_head_t
    with pytest.raises(ValueError, match='mutually inconsistent dev_head_t'):
        raw_meg.copy().add_channels([raw_arr])
    raw_meg.copy().add_channels([raw_arr], force_update_info=True)
    # Make sure that values didn't get overwritten
    assert_object_equal(raw_arr.info['dev_head_t'], orig_head_t)
    # Make sure all variants work
    for simult in (False, True):  # simultaneous adding or not
        raw_new = raw_meg.copy()
        if simult:
            raw_new.add_channels([raw_eeg, raw_stim])
        else:
            raw_new.add_channels([raw_eeg])
            raw_new.add_channels([raw_stim])
        for other in (raw_meg, raw_stim, raw_eeg):
            assert_allclose(
                raw_new.copy().pick_channels(other.ch_names).get_data(),
                other.get_data())

    # Now test errors
    raw_badsf = raw_eeg.copy()
    with raw_badsf.info._unlock():
        raw_badsf.info['sfreq'] = 3.1415927
    raw_eeg.crop(.5)

    pytest.raises(RuntimeError, raw_meg.add_channels, [raw_nopre])
    pytest.raises(RuntimeError, raw_meg.add_channels, [raw_badsf])
    pytest.raises(ValueError, raw_meg.add_channels, [raw_eeg])
    pytest.raises(ValueError, raw_meg.add_channels, [raw_meg])
    pytest.raises(TypeError, raw_meg.add_channels, raw_badsf)


@testing.requires_testing_data
def test_save(tmp_path):
    """Test saving raw."""
    temp_fname = tmp_path / 'test_raw.fif'
    shutil.copyfile(fif_fname, temp_fname)
    raw = read_raw_fif(temp_fname, preload=False)
    # can't write over file being read
    with pytest.raises(ValueError, match='to the same file'):
        raw.save(temp_fname)
    raw.load_data()
    # can't overwrite file without overwrite=True
    with pytest.raises(IOError, match='file exists'):
        raw.save(fif_fname)

    # test abspath support and annotations
    orig_time = _dt_to_stamp(raw.info['meas_date'])[0] + raw._first_time
    annot = Annotations([10], [5], ['test'], orig_time=orig_time)
    raw.set_annotations(annot)
    annot = raw.annotations
    new_fname = tmp_path / 'break_raw.fif'
    raw.save(new_fname, overwrite=True)
    new_raw = read_raw_fif(new_fname, preload=False)
    pytest.raises(ValueError, new_raw.save, new_fname)
    assert_array_almost_equal(annot.onset, new_raw.annotations.onset)
    assert_array_equal(annot.duration, new_raw.annotations.duration)
    assert_array_equal(annot.description, new_raw.annotations.description)
    assert annot.orig_time == new_raw.annotations.orig_time

    # test set_meas_date(None)
    raw.set_meas_date(None)
    raw.save(new_fname, overwrite=True)
    new_raw = read_raw_fif(new_fname, preload=False)
    assert new_raw.info['meas_date'] is None


@testing.requires_testing_data
def test_annotation_crop(tmp_path):
    """Test annotation sync after cropping and concatenating."""
    annot = Annotations([5., 11., 15.], [2., 1., 3.], ['test', 'test', 'test'])
    raw = read_raw_fif(fif_fname, preload=False)
    raw.set_annotations(annot)
    r1 = raw.copy().crop(2.5, 7.5)
    r2 = raw.copy().crop(12.5, 17.5)
    r3 = raw.copy().crop(10., 12.)
    raw = concatenate_raws([r1, r2, r3])  # segments reordered
    assert_and_remove_boundary_annot(raw, 2)
    onsets = raw.annotations.onset
    durations = raw.annotations.duration
    # 2*5s clips combined with annotations at 2.5s + 2s clip, annotation at 1s
    assert_array_almost_equal(onsets[:3], [47.95, 52.95, 56.46], decimal=2)
    assert_array_almost_equal([2., 2.5, 1.], durations[:3], decimal=2)

    # test annotation clipping
    orig_time = _dt_to_stamp(raw.info['meas_date'])
    orig_time = orig_time[0] + orig_time[1] * 1e-6 + raw._first_time - 1.
    annot = Annotations([0., raw.times[-1]], [2., 2.], 'test', orig_time)
    with pytest.warns(RuntimeWarning, match='Limited .* expanding outside'):
        raw.set_annotations(annot)
    assert_allclose(raw.annotations.duration,
                    [1., 1. + 1. / raw.info['sfreq']], atol=1e-3)

    # make sure we can overwrite the file we loaded when preload=True
    new_fname = tmp_path / 'break_raw.fif'
    raw.save(new_fname)
    new_raw = read_raw_fif(new_fname, preload=True)
    new_raw.save(new_fname, overwrite=True)


@testing.requires_testing_data
def test_with_statement():
    """Test with statement."""
    for preload in [True, False]:
        with read_raw_fif(fif_fname, preload=preload) as raw_:
            print(raw_)


def test_compensation_raw(tmp_path):
    """Test Raw compensation."""
    raw_3 = read_raw_fif(ctf_comp_fname)
    assert raw_3.compensation_grade == 3
    data_3, times = raw_3[:, :]

    # data come with grade 3
    for ii in range(2):
        raw_3_new = raw_3.copy()
        if ii == 0:
            raw_3_new.load_data()
        raw_3_new.apply_gradient_compensation(3)
        assert raw_3_new.compensation_grade == 3
        data_new, times_new = raw_3_new[:, :]
        assert_array_equal(times, times_new)
        assert_array_equal(data_3, data_new)

    # change to grade 0
    raw_0 = raw_3.copy().apply_gradient_compensation(0)
    assert raw_0.compensation_grade == 0
    data_0, times_new = raw_0[:, :]
    assert_array_equal(times, times_new)
    assert (np.mean(np.abs(data_0 - data_3)) > 1e-12)
    # change to grade 1
    raw_1 = raw_0.copy().apply_gradient_compensation(1)
    assert raw_1.compensation_grade == 1
    data_1, times_new = raw_1[:, :]
    assert_array_equal(times, times_new)
    assert (np.mean(np.abs(data_1 - data_3)) > 1e-12)
    pytest.raises(ValueError, raw_1.apply_gradient_compensation, 33)
    raw_bad = raw_0.copy()
    raw_bad.add_proj(compute_proj_raw(raw_0, duration=0.5, verbose='error'))
    raw_bad.apply_proj()
    pytest.raises(RuntimeError, raw_bad.apply_gradient_compensation, 1)
    # with preload
    tols = dict(rtol=1e-12, atol=1e-25)
    raw_1_new = raw_3.copy().load_data().apply_gradient_compensation(1)
    assert raw_1_new.compensation_grade == 1
    data_1_new, times_new = raw_1_new[:, :]
    assert_array_equal(times, times_new)
    assert (np.mean(np.abs(data_1_new - data_3)) > 1e-12)
    assert_allclose(data_1, data_1_new, **tols)
    # change back
    raw_3_new = raw_1.copy().apply_gradient_compensation(3)
    data_3_new, times_new = raw_3_new[:, :]
    assert_allclose(data_3, data_3_new, **tols)
    raw_3_new = raw_1.copy().load_data().apply_gradient_compensation(3)
    data_3_new, times_new = raw_3_new[:, :]
    assert_allclose(data_3, data_3_new, **tols)

    for load in (False, True):
        for raw in (raw_0, raw_1):
            raw_3_new = raw.copy()
            if load:
                raw_3_new.load_data()
            raw_3_new.apply_gradient_compensation(3)
            assert raw_3_new.compensation_grade == 3
            data_3_new, times_new = raw_3_new[:, :]
            assert_array_equal(times, times_new)
            assert (np.mean(np.abs(data_3_new - data_1)) > 1e-12)
            assert_allclose(data_3, data_3_new, **tols)

    # Try IO with compensation
    temp_file = tmp_path / 'raw.fif'
    raw_3.save(temp_file, overwrite=True)
    for preload in (True, False):
        raw_read = read_raw_fif(temp_file, preload=preload)
        assert raw_read.compensation_grade == 3
        data_read, times_new = raw_read[:, :]
        assert_array_equal(times, times_new)
        assert_allclose(data_3, data_read, **tols)
        raw_read.apply_gradient_compensation(1)
        data_read, times_new = raw_read[:, :]
        assert_array_equal(times, times_new)
        assert_allclose(data_1, data_read, **tols)

    # Now save the file that has modified compensation
    # and make sure the compensation is the same as it was,
    # but that we can undo it

    # These channels have norm 1e-11/1e-12, so atol=1e-18 isn't awesome,
    # but it's due to the single precision of the info['comps'] leading
    # to inexact inversions with saving/loading (casting back to single)
    # in between (e.g., 1->3->1 will degrade like this)
    looser_tols = dict(rtol=1e-6, atol=1e-18)
    raw_1.save(temp_file, overwrite=True)
    for preload in (True, False):
        raw_read = read_raw_fif(temp_file, preload=preload, verbose=True)
        assert raw_read.compensation_grade == 1
        data_read, times_new = raw_read[:, :]
        assert_array_equal(times, times_new)
        assert_allclose(data_1, data_read, **looser_tols)
        raw_read.apply_gradient_compensation(3, verbose=True)
        data_read, times_new = raw_read[:, :]
        assert_array_equal(times, times_new)
        assert_allclose(data_3, data_read, **looser_tols)


@requires_mne
def test_compensation_raw_mne(tmp_path):
    """Test Raw compensation by comparing with MNE-C."""
    def compensate_mne(fname, grad):
        tmp_fname = tmp_path / 'mne_ctf_test_raw.fif'
        cmd = ['mne_process_raw', '--raw', fname, '--save', tmp_fname,
               '--grad', str(grad), '--projoff', '--filteroff']
        run_subprocess(cmd)
        return read_raw_fif(tmp_fname, preload=True)

    for grad in [0, 2, 3]:
        raw_py = read_raw_fif(ctf_comp_fname, preload=True)
        raw_py.apply_gradient_compensation(grad)
        raw_c = compensate_mne(ctf_comp_fname, grad)
        assert_allclose(raw_py._data, raw_c._data, rtol=1e-6, atol=1e-17)
        assert raw_py.info['nchan'] == raw_c.info['nchan']
        for ch_py, ch_c in zip(raw_py.info['chs'], raw_c.info['chs']):
            for key in ('ch_name', 'coil_type', 'scanno', 'logno', 'unit',
                        'coord_frame', 'kind'):
                assert ch_py[key] == ch_c[key]
            for key in ('loc', 'unit_mul', 'range', 'cal'):
                assert_allclose(ch_py[key], ch_c[key])


@testing.requires_testing_data
def test_drop_channels_mixin():
    """Test channels-dropping functionality."""
    raw = read_raw_fif(fif_fname, preload=True)
    drop_ch = raw.ch_names[:3]
    ch_names = raw.ch_names[3:]

    ch_names_orig = raw.ch_names
    dummy = raw.copy().drop_channels(drop_ch)
    assert ch_names == dummy.ch_names
    assert ch_names_orig == raw.ch_names
    assert len(ch_names_orig) == raw._data.shape[0]

    raw.drop_channels(drop_ch)
    assert ch_names == raw.ch_names
    assert len(ch_names) == len(raw._cals)
    assert len(ch_names) == raw._data.shape[0]

    # Test that dropping all channels a projector applies to will lead to the
    # removal of said projector.
    raw = read_raw_fif(fif_fname).crop(0, 1)
    n_projs = len(raw.info['projs'])
    eeg_names = raw.info['projs'][-1]['data']['col_names']
    with pytest.raises(RuntimeError, match='loaded'):
        raw.copy().apply_proj().drop_channels(eeg_names)
    raw.load_data().drop_channels(eeg_names)  # EEG proj
    assert len(raw.info['projs']) == n_projs - 1

    # Dropping EEG channels with custom ref removes info['custom_ref_applied']
    raw = read_raw_fif(fif_fname).crop(0, 1).load_data()
    raw.set_eeg_reference()
    assert raw.info['custom_ref_applied']
    raw.drop_channels(eeg_names)
    assert not raw.info['custom_ref_applied']


@testing.requires_testing_data
@pytest.mark.parametrize('preload', (True, False))
def test_pick_channels_mixin(preload):
    """Test channel-picking functionality."""
    raw = read_raw_fif(fif_fname, preload=preload)
    raw_orig = raw.copy()
    ch_names = raw.ch_names[:3]

    ch_names_orig = raw.ch_names
    dummy = raw.copy().pick_channels(ch_names)
    assert ch_names == dummy.ch_names
    assert ch_names_orig == raw.ch_names
    assert len(ch_names_orig) == raw.get_data().shape[0]

    raw.pick_channels(ch_names)  # copy is False
    assert ch_names == raw.ch_names
    assert len(ch_names) == len(raw._cals)
    assert len(ch_names) == raw.get_data().shape[0]
    with pytest.raises(ValueError, match='must be'):
        raw.pick_channels(ch_names[0])

    assert_allclose(raw[:][0], raw_orig[:3][0])


@testing.requires_testing_data
def test_equalize_channels():
    """Test equalization of channels."""
    raw1 = read_raw_fif(fif_fname, preload=True)

    raw2 = raw1.copy()
    ch_names = raw1.ch_names[2:]
    raw1.drop_channels(raw1.ch_names[:1])
    raw2.drop_channels(raw2.ch_names[1:2])
    my_comparison = [raw1, raw2]
    my_comparison = equalize_channels(my_comparison)
    for e in my_comparison:
        assert ch_names == e.ch_names


def test_memmap(tmp_path):
    """Test some interesting memmapping cases."""
    # concatenate_raw
    memmaps = [str(tmp_path / str(ii)) for ii in range(3)]
    raw_0 = read_raw_fif(test_fif_fname, preload=memmaps[0])
    assert raw_0._data.filename == memmaps[0]
    raw_1 = read_raw_fif(test_fif_fname, preload=memmaps[1])
    assert raw_1._data.filename == memmaps[1]
    raw_0.append(raw_1, preload=memmaps[2])
    assert raw_0._data.filename == memmaps[2]
    # add_channels
    orig_data = raw_0[:][0]
    new_ch_info = pick_info(raw_0.info, [0])
    new_ch_info['chs'][0]['ch_name'] = 'foo'
    new_ch_info._update_redundant()
    new_data = np.linspace(0, 1, len(raw_0.times))[np.newaxis]
    ch = RawArray(new_data, new_ch_info)
    raw_0.add_channels([ch])
    if sys.platform == 'darwin':
        assert not hasattr(raw_0._data, 'filename')
    else:
        assert raw_0._data.filename == memmaps[2]
    assert_allclose(orig_data, raw_0[:-1][0], atol=1e-7)
    assert_allclose(new_data, raw_0[-1][0], atol=1e-7)

    # now let's see if .copy() actually works; it does, but eventually
    # we should make it optionally memmap to a new filename rather than
    # create an in-memory version (filename=None)
    raw_0 = read_raw_fif(test_fif_fname, preload=memmaps[0])
    assert raw_0._data.filename == memmaps[0]
    assert raw_0._data[:1, 3:5].all()
    raw_1 = raw_0.copy()
    assert isinstance(raw_1._data, np.memmap)
    assert raw_1._data.filename is None
    raw_0._data[:] = 0.
    assert not raw_0._data.any()
    assert raw_1._data[:1, 3:5].all()
    # other things like drop_channels and crop work but do not use memmapping,
    # eventually we might want to add support for some of these as users
    # require them.


# These are slow on Azure Windows so let's do a subset
@pytest.mark.parametrize('kind', [
    'file',
    pytest.param('bytes', marks=pytest.mark.slowtest),
])
@pytest.mark.parametrize('preload', [
    True,
    pytest.param(str, marks=pytest.mark.slowtest),
])
@pytest.mark.parametrize('split', [
    False,
    pytest.param(True, marks=pytest.mark.slowtest),
])
def test_file_like(kind, preload, split, tmp_path):
    """Test handling with file-like objects."""
    if split:
        fname = tmp_path / 'test_raw.fif'
        read_raw_fif(test_fif_fname).save(fname, split_size='5MB')
        assert op.isfile(fname)
        assert op.isfile(str(fname)[:-4] + '-1.fif')
    else:
        fname = test_fif_fname
    if preload is str:
        preload = str(tmp_path / 'memmap')
    with open(str(fname), 'rb') as file_fid:
        fid = BytesIO(file_fid.read()) if kind == 'bytes' else file_fid
        assert not fid.closed
        assert not file_fid.closed
        with pytest.raises(ValueError, match='preload must be used with file'):
            read_raw_fif(fid)
        assert not fid.closed
        assert not file_fid.closed
        # Use test_preloading=False but explicitly pass the preload type
        # so that we don't bother testing preload=False
        kwargs = dict(fname=fid, preload=preload, on_split_missing='ignore',
                      test_preloading=False, test_kwargs=False)
        _test_raw_reader(read_raw_fif, **kwargs)
        assert not fid.closed
        assert not file_fid.closed
    assert file_fid.closed


def test_str_like():
    """Test handling with str-like objects."""
    fname = pathlib.Path(test_fif_fname)
    raw_path = read_raw_fif(fname, preload=True)
    raw_str = read_raw_fif(test_fif_fname, preload=True)
    assert_allclose(raw_path._data, raw_str._data)


@pytest.mark.parametrize('fname', [
    test_fif_fname,
    testing._pytest_param(fif_fname),
    testing._pytest_param(ms_fname),
])
def test_bad_acq(fname):
    """Test handling of acquisition errors."""
    # see gh-7844
    raw = read_raw_fif(fname, allow_maxshield='yes').load_data()
    with open(fname, 'rb') as fid:
        for ent in raw._raw_extras[0]['ent']:
            fid.seek(ent.pos, 0)
            tag = _read_tag_header(fid)
            # hack these, others (kind, type) should be correct
            tag.pos, tag.next = ent.pos, ent.next
            assert tag == ent


@testing.requires_testing_data
@pytest.mark.skipif(sys.platform not in ('darwin', 'linux'),
                    reason='Needs proper symlinking')
def test_split_symlink(tmp_path):
    """Test split files with symlinks."""
    # regression test for gh-9221
    (tmp_path / 'first').mkdir()
    first = tmp_path / 'first' / 'test_raw.fif'
    raw = read_raw_fif(fif_fname).pick('meg').load_data()
    raw.save(first, buffer_size_sec=1, split_size='10MB', verbose=True)
    second = str(first)[:-4] + '-1.fif'
    assert op.isfile(second)
    assert not op.isfile(str(first)[:-4] + '-2.fif')
    (tmp_path / 'a').mkdir()
    (tmp_path / 'b').mkdir()
    new_first = tmp_path / 'a' / 'test_raw.fif'
    new_second = tmp_path / 'b' / 'test_raw-1.fif'
    shutil.move(first, new_first)
    shutil.move(second, new_second)
    os.symlink(new_first, first)
    os.symlink(new_second, second)
    raw_new = read_raw_fif(first)
    assert_allclose(raw_new.get_data(), raw.get_data())


@testing.requires_testing_data
def test_corrupted(tmp_path):
    """Test that a corrupted file can still be read."""
    # Must be a file written by Neuromag, not us, since we don't write the dir
    # at the end, so use the skip one (straight from acq).
    raw = read_raw_fif(skip_fname)
    with open(skip_fname, 'rb') as fid:
        tag = read_tag_info(fid)
        tag = read_tag(fid)
        dirpos = int(tag.data)
        assert dirpos == 12641532
        fid.seek(0)
        data = fid.read(dirpos)
    bad_fname = tmp_path / 'test_raw.fif'
    with open(bad_fname, 'wb') as fid:
        fid.write(data)
    with pytest.warns(RuntimeWarning, match='.*tag directory.*corrupt.*'):
        raw_bad = read_raw_fif(bad_fname)
    assert_allclose(raw.get_data(), raw_bad.get_data())


@testing.requires_testing_data
def test_expand_user(tmp_path, monkeypatch):
    """Test that we're expanding `~` before reading and writing."""
    monkeypatch.setenv('HOME', str(tmp_path))
    monkeypatch.setenv('USERPROFILE', str(tmp_path))  # Windows

    path_in = Path(fif_fname)
    path_out = tmp_path / path_in.name
    path_home = Path('~') / path_in.name

    shutil.copyfile(
        src=path_in,
        dst=path_out
    )

    raw = read_raw_fif(fname=path_home, preload=True)
    raw.save(fname=path_home, overwrite=True)