def test_read_on_missing(self, instance):
        with h5py.File('test', driver='core', backing_store=False) as h5f:
            instance.write(h5f)
            names = ['randomname']

            def _read(**kwargs):
                return self.TEST_CLASS.read(h5f, names=names, format='hdf5',
                                            **kwargs)

            # check on_missing='error' (default) raises ValueError
            with pytest.raises(ValueError) as exc:
                _read()
            assert str(exc.value) == ('\'randomname\' not found in any input '
                                      'file')

            # check on_missing='warn' prints warning
            with pytest.warns(UserWarning):
                _read(on_missing='warn')

            # check on_missing='ignore' does nothing
            with pytest.warns(None) as record:
                _read(on_missing='ignore')
            assert not record.list

            # check on_missing=<anything else> raises exception
            with pytest.raises(ValueError) as exc:
                _read(on_missing='blah')

def test_warnings():
    a = Input(shape=(3,), name='input_a')
    b = Input(shape=(3,), name='input_b')

    a_2 = Dense(4, name='dense_1')(a)
    dp = Dropout(0.5, name='dropout')
    b_2 = dp(b)

    model = Model([a, b], [a_2, b_2])

    optimizer = 'rmsprop'
    loss = 'mse'
    loss_weights = [1., 0.5]
    model.compile(optimizer, loss, metrics=[], loss_weights=loss_weights,
                  sample_weight_mode=None)

    def gen_data(batch_sz):
        while True:
            yield ([np.random.random((batch_sz, 3)), np.random.random((batch_sz, 3))],
                   [np.random.random((batch_sz, 4)), np.random.random((batch_sz, 3))])

    with pytest.warns(Warning) as w:
        out = model.fit_generator(gen_data(4), steps_per_epoch=10, use_multiprocessing=True, workers=2)
    warning_raised = any(['Sequence' in str(w_.message) for w_ in w])
    assert warning_raised, 'No warning raised when using generator with processes.'

    with pytest.warns(None) as w:
        out = model.fit_generator(RandomSequence(3), steps_per_epoch=4, use_multiprocessing=True, workers=2)
    assert all(['Sequence' not in str(w_.message) for w_ in w]), 'A warning was raised for Sequence.'

def test_PR_424():
    """Ensure deprecation and user warnings are triggered."""
    import warnings
    warnings.simplefilter('always') # Alert us of deprecation warnings.

    # Recommended use
    ColorClip([1000, 600], color=(60, 60, 60), duration=10).close()

    with pytest.warns(DeprecationWarning):
        # Uses `col` so should work the same as above, but give warning.
        ColorClip([1000, 600], col=(60, 60, 60), duration=10).close()

    # Catch all warnings as record.
    with pytest.warns(None) as record:
        # Should give 2 warnings and use `color`, not `col`
        ColorClip([1000, 600], color=(60, 60, 60), duration=10, col=(2,2,2)).close()

    message1 = 'The `ColorClip` parameter `col` has been deprecated. ' + \
               'Please use `color` instead.'
    message2 = 'The arguments `color` and `col` have both been passed to ' + \
               '`ColorClip` so `col` has been ignored.'

    # Assert that two warnings popped and validate the message text.
    assert len(record) == 2
    assert str(record[0].message) == message1
    assert str(record[1].message) == message2

def test_docstring_parameters():
    """Test module docstring formatting."""
    from numpydoc import docscrape
    incorrect = []
    for name in public_modules:
        with pytest.warns(None):  # traits warnings
            module = __import__(name, globals())
        for submod in name.split('.')[1:]:
            module = getattr(module, submod)
        classes = inspect.getmembers(module, inspect.isclass)
        for cname, cls in classes:
            if cname.startswith('_') and cname not in _doc_special_members:
                continue
            with pytest.warns(None) as w:
                cdoc = docscrape.ClassDoc(cls)
            for ww in w:
                if 'Using or importing the ABCs' not in str(ww.message):
                    raise RuntimeError('Error for __init__ of %s in %s:\n%s'
                                       % (cls, name, ww))
            if hasattr(cls, '__init__'):
                incorrect += check_parameters_match(cls.__init__, cdoc, cls)
            for method_name in cdoc.methods:
                method = getattr(cls, method_name)
                incorrect += check_parameters_match(method, cls=cls)
            if hasattr(cls, '__call__'):
                incorrect += check_parameters_match(cls.__call__, cls=cls)
        functions = inspect.getmembers(module, inspect.isfunction)
        for fname, func in functions:
            if fname.startswith('_'):
                continue
            incorrect += check_parameters_match(func)
    msg = '\n' + '\n'.join(sorted(list(set(incorrect))))
    if len(incorrect) > 0:
        raise AssertionError(msg)

 def test_xindex(self):
     x = numpy.linspace(0, 100, num=self.data.shape[0])
     # test simple
     series = self.create(xindex=x)
     self.assertQuantityEqual(
         series.xindex, units.Quantity(x, self.TEST_CLASS._default_xunit))
     # test deleter
     del series.xindex
     del series.xindex
     x1 = series.x0.value + series.shape[0] * series.dx.value
     x_default = numpy.linspace(series.x0.value, x1, num=series.shape[0],
                                endpoint=False)
     self.assertQuantityEqual(
         series.xindex,
         units.Quantity(x_default, self.TEST_CLASS._default_xunit))
     # test setting of x0 and dx
     series = self.create(xindex=units.Quantity(x, 'Farad'))
     self.assertEqual(series.x0, units.Quantity(x[0], 'Farad'))
     self.assertEqual(series.dx, units.Quantity(x[1] - x[0], 'Farad'))
     self.assertEqual(series.xunit, units.Farad)
     self.assertEqual(series.xspan, (x[0], x[-1] + x[1] - x[0]))
     # test that setting xindex warns about ignoring dx or x0
     with pytest.warns(UserWarning):
         series = self.create(xindex=units.Quantity(x, 'Farad'), dx=1)
     with pytest.warns(UserWarning):
         series = self.create(xindex=units.Quantity(x, 'Farad'), x0=0)
     # test non-regular xindex
     x = numpy.logspace(0, 2, num=self.data.shape[0])
     series = self.create(xindex=units.Quantity(x, 'Mpc'))
     def _get_dx():
         series.dx
     self.assertRaises(AttributeError, _get_dx)
     self.assertEqual(series.x0, units.Quantity(1, 'Mpc'))
     self.assertEqual(series.xspan, (x[0], x[-1] + x[-1] - x[-2]))

def check_min_samples_leaf(name):
    X, y = hastie_X, hastie_y

    # Test if leaves contain more than leaf_count training examples
    ForestEstimator = FOREST_ESTIMATORS[name]

    # test boundary value
    with pytest.warns(DeprecationWarning, match='min_samples_leaf'):
        assert_raises(ValueError,
                      ForestEstimator(min_samples_leaf=-1).fit, X, y)
    with pytest.warns(DeprecationWarning, match='min_samples_leaf'):
        assert_raises(ValueError,
                      ForestEstimator(min_samples_leaf=0).fit, X, y)

    est = ForestEstimator(min_samples_leaf=5, n_estimators=1, random_state=0)
    with pytest.warns(DeprecationWarning, match='min_samples_leaf'):
        est.fit(X, y)
    out = est.estimators_[0].tree_.apply(X)
    node_counts = np.bincount(out)
    # drop inner nodes
    leaf_count = node_counts[node_counts != 0]
    assert_greater(np.min(leaf_count), 4,
                   "Failed with {0}".format(name))

    est = ForestEstimator(min_samples_leaf=0.25, n_estimators=1,
                          random_state=0)
    with pytest.warns(DeprecationWarning, match='min_samples_leaf'):
        est.fit(X, y)
    out = est.estimators_[0].tree_.apply(X)
    node_counts = np.bincount(out)
    # drop inner nodes
    leaf_count = node_counts[node_counts != 0]
    assert_greater(np.min(leaf_count), len(X) * 0.25 - 1,
                   "Failed with {0}".format(name))

def test_ica_ctf():
    """Test run ICA computation on ctf data with/without compensation."""
    method = 'fastica'
    raw = read_raw_ctf(ctf_fname, preload=True)
    events = make_fixed_length_events(raw, 99999)
    for comp in [0, 1]:
        raw.apply_gradient_compensation(comp)
        epochs = Epochs(raw, events, None, -0.2, 0.2, preload=True)
        evoked = epochs.average()

        # test fit
        for inst in [raw, epochs]:
            ica = ICA(n_components=2, random_state=0, max_iter=2,
                      method=method)
            with pytest.warns(UserWarning, match='did not converge'):
                ica.fit(inst)

        # test apply and get_sources
        for inst in [raw, epochs, evoked]:
            ica.apply(inst)
            ica.get_sources(inst)

    # test mixed compensation case
    raw.apply_gradient_compensation(0)
    ica = ICA(n_components=2, random_state=0, max_iter=2, method=method)
    with pytest.warns(UserWarning, match='did not converge'):
        ica.fit(raw)
    raw.apply_gradient_compensation(1)
    epochs = Epochs(raw, events, None, -0.2, 0.2, preload=True)
    evoked = epochs.average()
    for inst in [raw, epochs, evoked]:
        with pytest.raises(RuntimeError, match='Compensation grade of ICA'):
            ica.apply(inst)
        with pytest.raises(RuntimeError, match='Compensation grade of ICA'):
            ica.get_sources(inst)

def test_version_mismatch_file():

    testfile = str(get_test_data_path('version_mismatch.fits'))

    with pytest.warns(None) as w:
        with asdf.open(testfile,
                ignore_version_mismatch=False) as fits_handle:
            assert fits_handle.tree['a'] == complex(0j)
    # This is the warning that we expect from opening the FITS file
    assert len(w) == 1, display_warnings(w)
    assert str(w[0].message) == (
        "'tag:stsci.edu:asdf/core/complex' with version 7.0.0 found in file "
        "'{}', but latest supported version is 1.0.0".format(testfile))

    # Make sure warning does not occur when warning is ignored (default)
    with pytest.warns(None) as w:
        with asdf.open(testfile) as fits_handle:
            assert fits_handle.tree['a'] == complex(0j)
    assert len(w) == 0, display_warnings(w)

    with pytest.warns(None) as w:
        with fits_embed.AsdfInFits.open(testfile,
                ignore_version_mismatch=False) as fits_handle:
            assert fits_handle.tree['a'] == complex(0j)
    assert len(w) == 1
    assert str(w[0].message) == (
        "'tag:stsci.edu:asdf/core/complex' with version 7.0.0 found in file "
        "'{}', but latest supported version is 1.0.0".format(testfile))

    # Make sure warning does not occur when warning is ignored (default)
    with pytest.warns(None) as w:
        with fits_embed.AsdfInFits.open(testfile) as fits_handle:
            assert fits_handle.tree['a'] == complex(0j)
    assert len(w) == 0, display_warnings(w)

    def test_order_after_filter_does_not_warn_on_absent_keys(self):
        with pytest.warns(None) as warnings:
            sources = self.flourish.sources.order_by('type')
            assert type(sources) == Flourish
            assert len(sources) == 6
            assert len(warnings) == 1
            assert (
                str(warnings[0].message) ==
                'sorting sources by "type" failed: '
                'not all sources have that attribute'
            )

        with pytest.warns(None) as warnings:
            sources = self.flourish.sources.filter(type='post')
            assert len(sources) == 5
            sources = sources.order_by('type', 'published')
            assert type(sources) == Flourish
            assert len(sources) == 5
            assert len(warnings) == 0
            assert [
                    'series/part-one',
                    'series/part-two',
                    'thing-one',
                    'thing-two',
                    'series/part-three',
                ] == [source.slug for source in sources]

def test_compression_args():

    z = create(100, compression='zlib', compression_opts=9)
    assert isinstance(z, Array)
    assert 'zlib' == z.compressor.codec_id
    assert 9 == z.compressor.level

    # 'compressor' overrides 'compression'
    z = create(100, compressor=Zlib(9), compression='bz2', compression_opts=1)
    assert isinstance(z, Array)
    assert 'zlib' == z.compressor.codec_id
    assert 9 == z.compressor.level

    # 'compressor' ignores 'compression_opts'
    z = create(100, compressor=Zlib(9), compression_opts=1)
    assert isinstance(z, Array)
    assert 'zlib' == z.compressor.codec_id
    assert 9 == z.compressor.level

    with pytest.warns(UserWarning):
        # 'compressor' overrides 'compression'
        create(100, compressor=Zlib(9), compression='bz2', compression_opts=1)
    with pytest.warns(UserWarning):
        # 'compressor' ignores 'compression_opts'
        create(100, compressor=Zlib(9), compression_opts=1)

def test_deprecated_rfc6979_signature():
    with pytest.warns(CryptographyDeprecationWarning):
        sig = encode_rfc6979_signature(1, 1)
    assert sig == b"0\x06\x02\x01\x01\x02\x01\x01"
    with pytest.warns(CryptographyDeprecationWarning):
        decoded = decode_rfc6979_signature(sig)
    assert decoded == (1, 1)

def test_old_input_deprecation_warning():
    with nengo.Network():
        c = nengo.networks.CircularConvolution(n_neurons=10, dimensions=1)
        with pytest.warns(DeprecationWarning):
            assert c.A is c.input_a
        with pytest.warns(DeprecationWarning):
            assert c.B is c.input_b

def test_pairwise_boolean_distance(metric):
    # test that we convert to boolean arrays for boolean distances
    rng = np.random.RandomState(0)
    X = rng.randn(5, 4)
    Y = X.copy()
    Y[0, 0] = 1 - Y[0, 0]

    # ignore conversion to boolean in pairwise_distances
    with ignore_warnings(category=DataConversionWarning):
        for Z in [Y, None]:
            res = pairwise_distances(X, Z, metric=metric)
            res[np.isnan(res)] = 0
            assert np.sum(res != 0) == 0

    # non-boolean arrays are converted to boolean for boolean
    # distance metrics with a data conversion warning
    msg = "Data was converted to boolean for metric %s" % metric
    with pytest.warns(DataConversionWarning, match=msg):
        pairwise_distances(X, metric=metric)

    # Check that the warning is raised if X is boolean by Y is not boolean:
    with pytest.warns(DataConversionWarning, match=msg):
        pairwise_distances(X.astype(bool), Y=Y, metric=metric)

    # Check that no warning is raised if X is already boolean and Y is None:
    with pytest.warns(None) as records:
        pairwise_distances(X.astype(bool), metric=metric)
    assert len(records) == 0

def test_tokenize_dense_sparse_array(cls_name):
    rng = np.random.RandomState(1234)

    with pytest.warns(None):
        # ignore scipy.sparse.SparseEfficiencyWarning
        a = sp.rand(10, 10000, random_state=rng).asformat(cls_name)
    b = a.copy()

    assert tokenize(a) == tokenize(b)

    # modifying the data values
    if hasattr(b, 'data'):
        b.data[:10] = 1
    elif cls_name == 'dok':
        b[3, 3] = 1
    else:
        raise ValueError

    assert tokenize(a) != tokenize(b)

    # modifying the data indices
    with pytest.warns(None):
        b = a.copy().asformat('coo')
        b.row[:10] = np.arange(10)
        b = b.asformat(cls_name)
    assert tokenize(a) != tokenize(b)

def test_deprecation():
    """Test our various warnings."""
    with pytest.warns(deprecation.MetpyDeprecationWarning):
        FakeyMcFakeface.dontuse()
        assert FakeyMcFakeface.dontuse.__doc__ == "Don't use."
    with pytest.warns(deprecation.MetpyDeprecationWarning):
        FakeyMcFakeface.really_dontuse()

def test_io_inverse_operator():
    """Test IO of inverse_operator."""
    tempdir = _TempDir()
    inverse_operator = read_inverse_operator(fname_inv)
    x = repr(inverse_operator)
    assert (x)
    assert (isinstance(inverse_operator['noise_cov'], Covariance))
    # just do one example for .gz, as it should generalize
    _compare_io(inverse_operator, '.gz')

    # test warnings on bad filenames
    inv_badname = op.join(tempdir, 'test-bad-name.fif.gz')
    with pytest.warns(RuntimeWarning, match='-inv.fif'):
        write_inverse_operator(inv_badname, inverse_operator)
    with pytest.warns(RuntimeWarning, match='-inv.fif'):
        read_inverse_operator(inv_badname)

    # make sure we can write and read
    inv_fname = op.join(tempdir, 'test-inv.fif')
    args = (10, 1. / 9., 'dSPM')
    inv_prep = prepare_inverse_operator(inverse_operator, *args)
    write_inverse_operator(inv_fname, inv_prep)
    inv_read = read_inverse_operator(inv_fname)
    _compare(inverse_operator, inv_read)
    inv_read_prep = prepare_inverse_operator(inv_read, *args)
    _compare(inv_prep, inv_read_prep)
    inv_prep_prep = prepare_inverse_operator(inv_prep, *args)
    _compare(inv_prep, inv_prep_prep)

    def test_default(self):
        """ Default semantics in the presence or absence of a file """
        fname = self.mktemp()

        # No existing file; create a new file and open RW
        with pytest.warns(H5pyDeprecationWarning):
            with File(fname) as f:
                self.assertTrue(f)
                self.assertEqual(f.mode, 'r+')

        # Existing readonly file; open read-only
        os.chmod(fname, stat.S_IREAD)
        # Running as root (e.g. in a docker container) gives 'r+' as the file
        # mode, even for a read-only file.  See
        # https://github.com/h5py/h5py/issues/696
        exp_mode = 'r+' if os.stat(fname).st_uid == 0 and platform != "win32" else 'r'
        try:
            with pytest.warns(H5pyDeprecationWarning):
                with File(fname) as f:
                    self.assertTrue(f)
                    self.assertEqual(f.mode, exp_mode)
        finally:
            os.chmod(fname, stat.S_IWRITE)

        # File exists but is not HDF5; raise IOError
        with open(fname, 'wb') as f:
            f.write(b'\x00')
        with pytest.warns(H5pyDeprecationWarning):
            with self.assertRaises(IOError):
                File(fname)

    def test_deprecations(self):
        with pytest.warns(PendingDeprecationWarning) as record:
            @api_view(["GET"], exclude_from_schema=True)
            def view(request):
                pass

        assert len(record) == 1
        assert str(record[0].message) == (
            "The `exclude_from_schema` argument to `api_view` is pending "
            "deprecation. Use the `schema` decorator instead, passing `None`."
        )

        class OldFashionedExcludedView(APIView):
            exclude_from_schema = True

            def get(self, request, *args, **kwargs):
                pass

        patterns = [
            url('^excluded-old-fashioned/$', OldFashionedExcludedView.as_view()),
        ]

        inspector = EndpointEnumerator(patterns)
        with pytest.warns(PendingDeprecationWarning) as record:
            inspector.get_api_endpoints()

        assert len(record) == 1
        assert str(record[0].message) == (
            "The `OldFashionedExcludedView.exclude_from_schema` attribute is "
            "pending deprecation. Set `schema = None` instead."
        )

async def test_app_ctor() -> None:
    loop = asyncio.get_event_loop()
    with pytest.warns(DeprecationWarning):
        app = web.Application(loop=loop)
    with pytest.warns(DeprecationWarning):
        assert loop is app.loop
    assert app.logger is log.web_logger

def test_from_networkx_with_bad_attributes():
    G = nx.Graph()
    G.add_nodes_from([(0, {"index": "a", "attr_1": 10}),
                      (1, {"index": "b", "attr_1": 20})])
    G.add_edges_from([[0, 1]])

    with pytest.warns(UserWarning):
        renderer = from_networkx(G, nx.circular_layout)
        assert renderer.node_renderer.data_source.data["index"] == [0, 1]
        assert renderer.node_renderer.data_source.data["attr_1"] == [10, 20]

    G = nx.Graph()
    G.add_nodes_from([0, 1])
    G.add_edges_from([(0, 1, {"start": "A", "attr_1": 10})])

    with pytest.warns(UserWarning):
        renderer = from_networkx(G, nx.circular_layout)
        assert renderer.edge_renderer.data_source.data["start"] == [0]
        assert renderer.edge_renderer.data_source.data["end"] == [1]
        assert renderer.edge_renderer.data_source.data["attr_1"] == [10]

    G = nx.Graph()
    G.add_nodes_from([0, 1])
    G.add_edges_from([(0, 1, {"end": "A", "attr_1": 10})])

    with pytest.warns(UserWarning):
        renderer = from_networkx(G, nx.circular_layout)
        assert renderer.edge_renderer.data_source.data["start"] == [0]
        assert renderer.edge_renderer.data_source.data["end"] == [1]
        assert renderer.edge_renderer.data_source.data["attr_1"] == [10]