def test_warning_raised(self):
ret_val = "lololol"
for test in [{"args": {}, # No args just means deprecated
"warning": deprecation.DeprecatedWarning},
{"args": {"deprecated_in": "1.0",
"current_version": "2.0"},
"warning": deprecation.DeprecatedWarning},
{"args": {"deprecated_in": "1.0",
"removed_in": "2.0",
"current_version": "2.0"},
"warning": deprecation.UnsupportedWarning}]:
with self.subTest(**test):
class Test(object):
@deprecation.deprecated(**test["args"])
def method(self):
return ret_val
with warnings.catch_warnings(record=True) as caught_warnings:
warnings.simplefilter("always")
sot = Test()
self.assertEqual(ret_val, sot.method())
self.assertEqual(len(caught_warnings), 1)
self.assertEqual(caught_warnings[0].category, test["warning"])
python类catch_warnings()的实例源码
def main(argv):
parser = build_cli_parser()
opts, args = parser.parse_args(argv)
if not opts.url or not opts.token or not opts.sensor:
print "Missing required param; run with --help for usage"
sys.exit(-1)
cb = cbapi.CbApi(opts.url, token=opts.token, ssl_verify=opts.ssl_verify, ignore_system_proxy=True)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
sensor = cb.sensor(opts.sensor)
pprint.pprint(sensor)
def __init__(self, period):
# Quiet the warning about GTK Tooltip deprecation
import warnings
with warnings.catch_warnings():
warnings.filterwarnings('ignore', category=DeprecationWarning)
self.figure = pyplot.figure()
self.figure.canvas.set_window_title('REDHAWK Speedometer')
# Create a line graph of throughput over time
self.axes = self.figure.add_subplot(111)
self.axes.set_xlabel('Time')
self.axes.set_ylabel('Throughput (Bps)')
self.x = []
self.y = []
self.line, = self.axes.plot(self.x, self.y)
self.axes.set_xlim(xmax=period)
self.figure.subplots_adjust(bottom=0.2)
sliderax = self.figure.add_axes([0.2, 0.05, 0.6, 0.05])
min_size = 1024
max_size = 8*1024*1024
default_size = 1*1024*1024
self.slider = matplotlib.widgets.Slider(sliderax, 'Transfer size', min_size, max_size,
default_size, '%.0f')
self.figure.show()
def test_propertyInitialization(self):
"""
Tests for the correct initialization of 'property' kind properties
based on whether command line is set, and overrides via launch().
"""
# First, test with defaults
comp = sb.launch('sdr/dom/components/property_init/property_init.spd.xml')
self.assertFalse('initial' in comp.cmdline_args)
self.assertFalse('cmdline' in comp.initialize_props)
comp.releaseObject()
# Test with overrides
comp = sb.launch('sdr/dom/components/property_init/property_init.spd.xml',
properties={'cmdline':'override', 'initial':'override'})
self.assertFalse('initial' in comp.cmdline_args)
self.assertFalse('cmdline' in comp.initialize_props)
self.assertEquals('override', comp.cmdline)
self.assertEquals('override', comp.initial)
comp.releaseObject()
# Test with overrides in deprecated 'execparams' and 'configure' arguments
with warnings.catch_warnings():
warnings.simplefilter('ignore', DeprecationWarning)
comp = sb.launch('sdr/dom/components/property_init/property_init.spd.xml',
execparams={'cmdline':'override'}, configure={'initial':'override'})
self.assertFalse('initial' in comp.cmdline_args)
self.assertFalse('cmdline' in comp.initialize_props)
self.assertEquals('override', comp.cmdline)
self.assertEquals('override', comp.initial)
comp.releaseObject()
# Test with misplaced command line property in deprecated 'configure' argument
comp = sb.launch('sdr/dom/components/property_init/property_init.spd.xml',
configure={'cmdline':'override'})
self.assertFalse('initial' in comp.cmdline_args)
self.assertFalse('cmdline' in comp.initialize_props)
self.assertEquals('override', comp.cmdline)
comp.releaseObject()
def test_ignore_nan(self):
""" Test that NaNs are handled correctly """
stream = [np.random.random(size = (16,12)) for _ in range(5)]
for s in stream:
s[randint(0, 15), randint(0,11)] = np.nan
with catch_warnings():
simplefilter('ignore')
from_iaverage = last(iaverage(stream, ignore_nan = True))
from_numpy = np.nanmean(np.dstack(stream), axis = 2)
self.assertTrue(np.allclose(from_iaverage, from_numpy))
def test_ddof(self):
""" Test that the ddof parameter is equivalent to numpy's """
stream = [np.random.random((16, 7, 3)) for _ in range(10)]
stack = np.stack(stream, axis = -1)
with catch_warnings():
simplefilter('ignore')
for axis in (0, 1, 2, None):
for ddof in range(4):
with self.subTest('axis = {}, ddof = {}'.format(axis, ddof)):
from_numpy = np.var(stack, axis = axis, ddof = ddof)
from_ivar = last(ivar(stream, axis = axis, ddof = ddof))
self.assertSequenceEqual(from_numpy.shape, from_ivar.shape)
self.assertTrue(np.allclose(from_ivar, from_numpy))
def test_against_numpy_std(self):
stream = [np.random.random((16, 7, 3)) for _ in range(10)]
stack = np.stack(stream, axis = -1)
with catch_warnings():
simplefilter('ignore')
for axis in (0, 1, 2, None):
for ddof in range(4):
with self.subTest('axis = {}, ddof = {}'.format(axis, ddof)):
from_numpy = np.std(stack, axis = axis, ddof = ddof)
from_ivar = last(istd(stream, axis = axis, ddof = ddof))
self.assertSequenceEqual(from_numpy.shape, from_ivar.shape)
self.assertTrue(np.allclose(from_ivar, from_numpy))
def test_slider_slice_warning(self):
index = 3
kwargs = dict(grid_longitude=index)
coords = ('time', 'grid_latitude')
plot = Plot2D(self.cube, self.axes, coords=coords)
plot.alias(wibble=2)
wmsg = ("expected to be called with alias 'wibble' for dimension 2, "
"rather than with 'grid_longitude'")
with warnings.catch_warnings():
# Cause all warnings to raise an exception.
warnings.simplefilter('error')
with self.assertRaisesRegexp(UserWarning, wmsg):
plot(**kwargs)
def test_auto_deltas_fail_warn(self):
with warnings.catch_warnings(record=True) as ws:
warnings.simplefilter('always')
loader = BlazeLoader()
expr = bz.data(self.df, dshape=self.dshape)
from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.warn,
missing_values=self.missing_values,
)
self.assertEqual(len(ws), 1)
w = ws[0].message
self.assertIsInstance(w, NoDeltasWarning)
self.assertIn(str(expr), str(w))
def test_warning_from_explicit_topic(self, valid_message_data):
data_with_topic = self._make_message_data(
valid_message_data,
topic=str('explicit_topic')
)
with warnings.catch_warnings(record=True) as w:
self.message_class(**data_with_topic)
assert len(w) == 1
assert "Passing in topics explicitly is deprecated." in w[0].message
def test_specify_contains_pii_triggers_warnings(self, valid_message_data):
message_data = self._make_message_data(
valid_message_data,
contains_pii=False
)
with warnings.catch_warnings(record=True) as warns:
self.message_class(**message_data)
assert len(warns) == 1
contains_pii_warning = warns[0]
assert issubclass(contains_pii_warning.category, DeprecationWarning)
assert ('contains_pii is deprecated. Please stop passing it in.'
in contains_pii_warning.message)
def test_specify_keys_triggers_warnings(self, valid_message_data):
message_data = self._make_message_data(
valid_message_data,
keys=(1, 2)
)
with warnings.catch_warnings(record=True) as warns:
self.message_class(**message_data)
assert len(warns) == 1
keys_warning = warns[0]
assert issubclass(keys_warning.category, DeprecationWarning)
assert ('Passing in keys explicitly is deprecated.'
in keys_warning.message)
def assertItemsEqual(self, expected_seq, actual_seq, msg=None):
"""An unordered sequence specific comparison. It asserts that
actual_seq and expected_seq have the same element counts.
Equivalent to::
self.assertEqual(Counter(iter(actual_seq)),
Counter(iter(expected_seq)))
Asserts that each element has the same count in both sequences.
Example:
- [0, 1, 1] and [1, 0, 1] compare equal.
- [0, 0, 1] and [0, 1] compare unequal.
"""
first_seq, second_seq = list(actual_seq), list(expected_seq)
with warnings.catch_warnings():
if sys.py3kwarning:
# Silence Py3k warning raised during the sorting
for _msg in ["(code|dict|type) inequality comparisons",
"builtin_function_or_method order comparisons",
"comparing unequal types"]:
warnings.filterwarnings("ignore", _msg, DeprecationWarning)
try:
first = collections.Counter(first_seq)
second = collections.Counter(second_seq)
except TypeError:
# Handle case with unhashable elements
differences = _count_diff_all_purpose(first_seq, second_seq)
else:
if first == second:
return
differences = _count_diff_hashable(first_seq, second_seq)
if differences:
standardMsg = 'Element counts were not equal:\n'
lines = ['First has %d, Second has %d: %r' % diff for diff in differences]
diffMsg = '\n'.join(lines)
standardMsg = self._truncateMessage(standardMsg, diffMsg)
msg = self._formatMessage(msg, standardMsg)
self.fail(msg)
def initiate_send(self):
while self.producer_fifo and self.connected:
first = self.producer_fifo[0]
# handle empty string/buffer or None entry
if not first:
del self.producer_fifo[0]
if first is None:
self.handle_close()
return
# handle classic producer behavior
obs = self.ac_out_buffer_size
try:
with catch_warnings():
if py3kwarning:
filterwarnings("ignore", ".*buffer", DeprecationWarning)
data = buffer(first, 0, obs)
except TypeError:
data = first.more()
if data:
self.producer_fifo.appendleft(data)
else:
del self.producer_fifo[0]
continue
# send the data
try:
num_sent = self.send(data)
except socket.error:
self.handle_error()
return
if num_sent:
if num_sent < len(data) or obs < len(first):
self.producer_fifo[0] = first[num_sent:]
else:
del self.producer_fifo[0]
# we tried to send some actual data
return
def _sorted(iterable):
with warnings.catch_warnings():
if _sys.py3kwarning:
warnings.filterwarnings("ignore", "comparing unequal types "
"not supported", DeprecationWarning)
return sorted(iterable)
def index_worker(self, queue, size=200):
actions = []
indexed = 0
while True:
item = queue.get()
if item is None:
break
id_submission, analysis = item
doc = {
'_index': 'fcc-comments',
'_type': 'document',
'_op_type': 'update',
'_id': id_submission,
'doc': {'analysis': analysis},
}
actions.append(doc)
if len(actions) == size:
with warnings.catch_warnings():
warnings.simplefilter('ignore')
try:
response = bulk(self.es, actions)
indexed += response[0]
print('\tanalyzed %s/%s\t%s%%' % (indexed, self.limit,
int(indexed / self.limit * 100)))
actions = []
except ConnectionTimeout:
print('error indexing: connection timeout')
with warnings.catch_warnings():
warnings.simplefilter('ignore')
response = bulk(self.es, actions)
indexed += response[0]
print('indexed %s' % (indexed))
def bulk_index(self, queue, size=20):
actions = []
indexed = 0
ids = set()
while True:
item = queue.get()
if item is None:
break
doc_id = item
doc = {
'_index': 'fcc-comments',
'_type': 'document',
'_op_type': 'update',
'_id': doc_id,
'doc': {'analysis.sentiment_sig_terms_ordered': True},
}
actions.append(doc)
ids.add(doc_id)
if len(actions) == size:
with warnings.catch_warnings():
warnings.simplefilter('ignore')
try:
response = bulk(self.es, actions)
indexed += response[0]
if not indexed % 200:
print('\tindexed %s/%s\t%s%%' % (indexed, self.limit,
int(indexed / self.limit * 100)))
actions = []
except ConnectionTimeout:
print('error indexing: connection timeout')
with warnings.catch_warnings():
warnings.simplefilter('ignore')
response = bulk(self.es, actions)
indexed += response[0]
print('indexed %s' % (indexed))
ids = list(ids)
#print('%s\n%s' % (len(ids), ' '.join(ids))
def make_table(self, *args, **kwargs):
# type: (*Any, **Any) -> str
with warnings.catch_warnings():
# PendingDeprecationWarning acknowledged.
warnings.simplefilter("ignore")
table = super(HtmlMultiDiff, self).make_table(*args, **kwargs)
def unescape_zeroonetwo(m):
return unichr(ord(m.group(1)) - ord('0'))
table = re.sub('\x02([012])', unescape_zeroonetwo, table)
return table
def get_random_giphy(phrase):
"""Return the URL of a random GIF related to the phrase, if possible"""
with warnings.catch_warnings():
warnings.simplefilter('ignore')
giphy = giphypop.Giphy()
results = giphy.search_list(phrase=phrase, limit=100)
if not results:
raise ValueError('There were no results for that phrase')
return random.choice(results).media_url
def test():
"""
little test, NOT FOR REAL USE, put target filename as first and only argument
will plot the pitch and strength v. time to screen
safety is not guaranteed.
"""
# imports specific to this test
import sys
import warnings
from scipy.io import wavfile
import matplotlib.pyplot as plt
from psola.experiment_config import ExperimentConfig
# get the data and do the estimation
filename = sys.argv[1] # filename is first command line arg
cfg = ExperimentConfig() # use default settings
with warnings.catch_warnings():
warnings.simplefilter("ignore") # ignore annoying WavFileWarning
fs, data = wavfile.read(filename)
pitch, t, strength = pitch_estimation(data, fs, cfg)
# Plot estimated pitch and strength of pitch values
f, (ax1, ax2) = plt.subplots(2, 1, sharex=True, figsize=(16,9))
ax1.plot(t, pitch); ax1.set_title('Pitch v. Time'); ax1.set_ylabel('Freq (Hz)')
ax2.plot(t, strength); ax2.set_title('Strength v. Time'); ax1.set_ylabel('Strength')
plt.show()
