def enums_of_primitives(draw):
"""Generate enum classes with primitive values."""
if is_py2:
names = draw(st.sets(st.text(alphabet=string.ascii_letters,
min_size=1),
min_size=1))
else:
names = draw(st.sets(st.text(min_size=1), min_size=1))
n = len(names)
vals = draw(st.one_of(st.sets(st.one_of(
st.integers(),
st.floats(allow_nan=False),
st.text(min_size=1)),
min_size=n, max_size=n)))
return Enum('HypEnum', list(zip(names, vals)))
python类floats()的实例源码
def float_attrs(draw, defaults=None):
"""
Generate a tuple of an attribute and a strategy that yields floats for that
attribute.
"""
default = NOTHING
if defaults is True or (defaults is None and draw(st.booleans())):
default = draw(st.floats())
return ((attr.ib(default=default), st.floats()))
def create_dummy_rate_file(rate_file):
rates = lists(floats(min_value=0.00001, allow_nan=False, allow_infinity=False), min_size=0, max_size=100).example()
max_year = datetime.datetime.now().year
date_times = lists(datetimes(min_year=2016, max_year=max_year), min_size=len(rates),
max_size=len(rates)).map(sorted).example()
with open(rate_file, 'a') as f:
for date_time, rate in zip(date_times, rates):
writer = csv.writer(f, lineterminator='\n')
market_data = [date_time.strftime("%Y-%m-%d %H:%M:%S"), rate]
writer.writerow(market_data)
return rates, date_times
def reals(min_value=None,
max_value=None,
exclude_zero=None,
shape=None,
dtype=np.float):
"""Real number strategy that excludes nan and inf.
Args:
min_value (Number, optional):
max_value (Number, optional):
exclude_zero (str, optional):
Choices from: (None, 'exact', 'near')
shape (int|tuple, optional):
None for scalar output and int or tuple of int for array output.
dtype (numpy.float):
Numpy float type
"""
# TODO: size as strategy
assert dtype is None or np.dtype(dtype).kind == u'f'
if min_value is not None and max_value is not None:
assert max_value > min_value
elements = st.floats(min_value, max_value, False, False)
# Filter values
if exclude_zero == 'exact':
elements = elements.filter(lambda x: x != 0.0)
elif exclude_zero == 'near':
elements = elements.filter(lambda x: not np.isclose(x, 0.0))
# Strategy
if shape is None:
return elements
else:
return arrays(dtype, shape, elements)
def unit_vectors(draw, min_angle=0, max_angle=2 * np.pi):
phi = draw(st.floats(min_angle, max_angle, False, False))
return np.array((np.cos(phi), np.sin(phi)), dtype=np.float64)
def test_both_multiplications_with_floats(self):
""" Left- and right- multiplications by floats should be identical """
results = []
for q in self.quats:
x = random.random()
results.append(numpy.allclose((x * q).as_array, (q * x).as_array))
at = all(results)
self.assertTrue(at)
def test_hna_start_end_floats(self, start, end):
"""Test the from_start_and_end class method of the SingleStrandedHelix using floats."""
sequence = 'GAGATATACACA'
if allclose(start, end):
with self.assertRaises(ValueError):
isambard.specifications.NucleicAcidStrand.from_start_and_end(start, end, sequence)
else:
ssnh = isambard.specifications.NucleicAcidStrand.from_start_and_end(start, end, sequence)
self.assertEqual(len(ssnh), 12)
def test_play_with_clock_publish_freq(self):
self._run_argument_numeric("clock_publish_freq", "hz", hst.floats)
def test_play_with_delay(self):
self._run_argument_numeric("delay", "delay", hst.floats)
def test_play_with_publish_rate_multiplier(self):
self._run_argument_numeric("publish_rate_multiplier", "rate",
hst.floats)
def test_play_with_start_time(self):
self._run_argument_numeric("start_time", "start", hst.floats)
def test_play_with_duration(self):
self._run_argument_numeric("duration", "duration", hst.floats)
def test_to_long_fail(self, fail_val):
"""
to_long() with incompatible params:
:param fail_val: random data of known incompatible types (floats, text)
"""
self.force_fail(fail_val, to_long, TypeError)
def test_to_bool_fail(self, fail_val):
"""
to_bool() with incompatible param:
:param fail_val: data of known incompatible type (floats, text)
"""
self.force_fail(fail_val, to_bool, TypeError)
def applications():
"""Mock of the YARN cluster apps REST resource."""
if 'last' in request.args:
return jsonify(redis.get(request.base_url))
d = st.fixed_dictionaries({
'allocatedMB': st.integers(-1),
'allocatedVCores': st.integers(-1),
'amContainerLogs': st.text(),
'amHostHttpAddress': st.text(),
'applicationTags': st.text(),
'applicationType': st.sampled_from(['MAPREDUCE', 'SPARK']),
'clusterId': st.integers(0),
'diagnostics': st.text(),
'elapsedTime': st.integers(0),
'finalStatus': st.sampled_from(['UNDEFINED', 'SUCCEEDED', 'FAILED', 'KILLED']),
'finishedTime': st.integers(0),
'id': st.text(string.ascii_letters, min_size=5, max_size=25),
'memorySeconds': st.integers(0),
'name': st.text(min_size=5),
'numAMContainerPreempted': st.integers(0),
'numNonAMContainerPreempted': st.integers(0),
'preemptedResourceMB': st.integers(0),
'preemptedResourceVCores': st.integers(0),
'progress': st.floats(0, 100),
'queue': st.text(),
'runningContainers': st.integers(-1),
'startedTime': st.integers(0),
'state': st.sampled_from(['NEW', 'NEW_SAVING', 'SUBMITTED', 'ACCEPTED', 'RUNNING', 'FINISHED', 'FAILED', 'KILLED']),
'trackingUI': st.text(),
'trackingUrl': st.just(os.environ['YARN_ENDPOINT']),
'user': st.text(),
'vcoreSeconds': st.integers(0)
})
result = json.dumps({
'apps': {
'app': st.lists(d, min_size=4, average_size=10).example()
}
})
redis.set(request.base_url, result)
return jsonify(result)
def mapreduce_application():
"""Mock of the mapreduce jobs REST resource."""
if 'last' in request.args:
return jsonify(redis.get(request.base_url))
d = st.fixed_dictionaries({
'startTime': st.integers(0),
'finishTime': st.integers(0),
'elapsedTime': st.integers(0),
'id': st.integers(0),
'name': st.text(),
'user': st.text(),
'state': st.sampled_from(['NEW', 'SUCCEEDED', 'RUNNING', 'FAILED', 'KILLED']),
'mapsTotal': st.integers(0),
'mapsCompleted': st.integers(0),
'reducesTotal': st.integers(0),
'reducesCompleted': st.integers(0),
'mapProgress': st.floats(0, 100),
'reduceProgress': st.floats(0, 100),
'mapsPending': st.integers(0),
'mapsRunning': st.integers(0),
'reducesPending': st.integers(0),
'reducesRunning': st.integers(0),
'uberized': st.booleans(),
'diagnostics': st.text(),
'newReduceAttempts': st.integers(0),
'runningReduceAttempts': st.integers(0),
'failedReduceAttempts': st.integers(0),
'killedReduceAttempts': st.integers(0),
'successfulReduceAttempts': st.integers(0),
'newMapAttempts': st.integers(0),
'runningMapAttempts': st.integers(0),
'failedMapAttempts': st.integers(0),
'killedMapAttempts': st.integers(0),
'successfulMapAttempts': st.integers(0)
})
result = json.dumps({
'jobs': {
'job': st.lists(d, average_size=3).example()
}
})
redis.set(request.base_url, result)
return jsonify(result)
