def _validate_dict_data(self, expected, actual):
"""Validate dictionary data.
Compare expected dictionary data vs actual dictionary data.
The values in the 'expected' dictionary can be strings, bools, ints,
longs, or can be a function that evaluates a variable and returns a
bool.
"""
self.log.debug('actual: {}'.format(repr(actual)))
self.log.debug('expected: {}'.format(repr(expected)))
for k, v in six.iteritems(expected):
if k in actual:
if (isinstance(v, six.string_types) or
isinstance(v, bool) or
isinstance(v, six.integer_types)):
# handle explicit values
if v != actual[k]:
return "{}:{}".format(k, actual[k])
# handle function pointers, such as not_null or valid_ip
elif not v(actual[k]):
return "{}:{}".format(k, actual[k])
else:
return "key '{}' does not exist".format(k)
return None
python类integer_types()的实例源码
def _validate_dict_data(self, expected, actual):
"""Validate dictionary data.
Compare expected dictionary data vs actual dictionary data.
The values in the 'expected' dictionary can be strings, bools, ints,
longs, or can be a function that evaluates a variable and returns a
bool.
"""
self.log.debug('actual: {}'.format(repr(actual)))
self.log.debug('expected: {}'.format(repr(expected)))
for k, v in six.iteritems(expected):
if k in actual:
if (isinstance(v, six.string_types) or
isinstance(v, bool) or
isinstance(v, six.integer_types)):
# handle explicit values
if v != actual[k]:
return "{}:{}".format(k, actual[k])
# handle function pointers, such as not_null or valid_ip
elif not v(actual[k]):
return "{}:{}".format(k, actual[k])
else:
return "key '{}' does not exist".format(k)
return None
def test_yahoo_bars_to_panel_source(self):
env = TradingEnvironment()
finder = AssetFinder(env.engine)
stocks = ['AAPL', 'GE']
env.write_data(equities_identifiers=stocks)
start = pd.datetime(1993, 1, 1, 0, 0, 0, 0, pytz.utc)
end = pd.datetime(2002, 1, 1, 0, 0, 0, 0, pytz.utc)
data = factory.load_bars_from_yahoo(stocks=stocks,
indexes={},
start=start,
end=end)
check_fields = ['sid', 'open', 'high', 'low', 'close',
'volume', 'price']
copy_panel = data.copy()
sids = finder.map_identifier_index_to_sids(
data.items, data.major_axis[0]
)
copy_panel.items = sids
source = DataPanelSource(copy_panel)
for event in source:
for check_field in check_fields:
self.assertIn(check_field, event)
self.assertTrue(isinstance(event['volume'], (integer_types)))
self.assertTrue(event['sid'] in sids)
def fetch_size(self, v):
"""
Sets the number of rows that are fetched at a time.
*Note that driver's default fetch size is 5000.*
.. code-block:: python
for user in User.objects().fetch_size(500):
print(user)
"""
if not isinstance(v, six.integer_types):
raise TypeError
if v == self._fetch_size:
return self
if v < 1:
raise QueryException("fetch size less than 1 is not allowed")
clone = copy.deepcopy(self)
clone._fetch_size = v
return clone
def __init__(self, value):
"""
Initializer value can be:
- integer_type: absolute nanoseconds in the day
- datetime.time: built-in time
- string_type: a string time of the form "HH:MM:SS[.mmmuuunnn]"
"""
if isinstance(value, six.integer_types):
self._from_timestamp(value)
elif isinstance(value, datetime.time):
self._from_time(value)
elif isinstance(value, six.string_types):
self._from_timestring(value)
else:
raise TypeError('Time arguments must be a whole number, datetime.time, or string')
def __init__(self, value):
"""
Initializer value can be:
- integer_type: absolute days from epoch (1970, 1, 1). Can be negative.
- datetime.date: built-in date
- string_type: a string time of the form "yyyy-mm-dd"
"""
if isinstance(value, six.integer_types):
self.days_from_epoch = value
elif isinstance(value, (datetime.date, datetime.datetime)):
self._from_timetuple(value.timetuple())
elif isinstance(value, six.string_types):
self._from_datestring(value)
else:
raise TypeError('Date arguments must be a whole number, datetime.date, or string')
def _validate_dict_data(self, expected, actual):
"""Validate dictionary data.
Compare expected dictionary data vs actual dictionary data.
The values in the 'expected' dictionary can be strings, bools, ints,
longs, or can be a function that evaluates a variable and returns a
bool.
"""
self.log.debug('actual: {}'.format(repr(actual)))
self.log.debug('expected: {}'.format(repr(expected)))
for k, v in six.iteritems(expected):
if k in actual:
if (isinstance(v, six.string_types) or
isinstance(v, bool) or
isinstance(v, six.integer_types)):
# handle explicit values
if v != actual[k]:
return "{}:{}".format(k, actual[k])
# handle function pointers, such as not_null or valid_ip
elif not v(actual[k]):
return "{}:{}".format(k, actual[k])
else:
return "key '{}' does not exist".format(k)
return None
def _validate_dict_data(self, expected, actual):
"""Validate dictionary data.
Compare expected dictionary data vs actual dictionary data.
The values in the 'expected' dictionary can be strings, bools, ints,
longs, or can be a function that evaluates a variable and returns a
bool.
"""
self.log.debug('actual: {}'.format(repr(actual)))
self.log.debug('expected: {}'.format(repr(expected)))
for k, v in six.iteritems(expected):
if k in actual:
if (isinstance(v, six.string_types) or
isinstance(v, bool) or
isinstance(v, six.integer_types)):
# handle explicit values
if v != actual[k]:
return "{}:{}".format(k, actual[k])
# handle function pointers, such as not_null or valid_ip
elif not v(actual[k]):
return "{}:{}".format(k, actual[k])
else:
return "key '{}' does not exist".format(k)
return None
def _asFileDescriptor(obj):
fd = None
if not isinstance(obj, integer_types):
meth = getattr(obj, "fileno", None)
if meth is not None:
obj = meth()
if isinstance(obj, integer_types):
fd = obj
if not isinstance(fd, integer_types):
raise TypeError("argument must be an int, or have a fileno() method.")
elif fd < 0:
raise ValueError(
"file descriptor cannot be a negative integer (%i)" % (fd,))
return fd
def use_certificate_file(self, certfile, filetype=FILETYPE_PEM):
"""
Load a certificate from a file
:param certfile: The name of the certificate file (``bytes`` or
``unicode``).
:param filetype: (optional) The encoding of the file, default is PEM
:return: None
"""
certfile = _path_string(certfile)
if not isinstance(filetype, integer_types):
raise TypeError("filetype must be an integer")
use_result = _lib.SSL_CTX_use_certificate_file(
self._context, certfile, filetype
)
if not use_result:
_raise_current_error()
def use_privatekey_file(self, keyfile, filetype=_UNSPECIFIED):
"""
Load a private key from a file
:param keyfile: The name of the key file (``bytes`` or ``unicode``)
:param filetype: (optional) The encoding of the file, default is PEM
:return: None
"""
keyfile = _path_string(keyfile)
if filetype is _UNSPECIFIED:
filetype = FILETYPE_PEM
elif not isinstance(filetype, integer_types):
raise TypeError("filetype must be an integer")
use_result = _lib.SSL_CTX_use_PrivateKey_file(
self._context, keyfile, filetype)
if not use_result:
self._raise_passphrase_exception()
def bio_read(self, bufsiz):
"""
When using non-socket connections this function reads the "dirty" data
that would have traveled away on the network.
:param bufsiz: The maximum number of bytes to read
:return: The string read.
"""
if self._from_ssl is None:
raise TypeError("Connection sock was not None")
if not isinstance(bufsiz, integer_types):
raise TypeError("bufsiz must be an integer")
buf = _ffi.new("char[]", bufsiz)
result = _lib.BIO_read(self._from_ssl, buf, bufsiz)
if result <= 0:
self._handle_bio_errors(self._from_ssl, result)
return _ffi.buffer(buf, result)[:]
def bytes(num_bytes):
"""
Get some random bytes from the PRNG as a string.
This is a wrapper for the C function ``RAND_bytes``.
:param num_bytes: The number of bytes to fetch.
:return: A string of random bytes.
"""
if not isinstance(num_bytes, _integer_types):
raise TypeError("num_bytes must be an integer")
if num_bytes < 0:
raise ValueError("num_bytes must not be negative")
result_buffer = _ffi.new("char[]", num_bytes)
result_code = _lib.RAND_bytes(result_buffer, num_bytes)
if result_code == -1:
# TODO: No tests for this code path. Triggering a RAND_bytes failure
# might involve supplying a custom ENGINE? That's hard.
_raise_current_error()
return _ffi.buffer(result_buffer)[:]
def __init__(self, key, length, algorithm, backend):
if not isinstance(backend, HMACBackend):
raise UnsupportedAlgorithm(
"Backend object does not implement HMACBackend.",
_Reasons.BACKEND_MISSING_INTERFACE
)
if len(key) < 16:
raise ValueError("Key length has to be at least 128 bits.")
if not isinstance(length, six.integer_types):
raise TypeError("Length parameter must be an integer type.")
if length < 6 or length > 8:
raise ValueError("Length of HOTP has to be between 6 to 8.")
if not isinstance(algorithm, (SHA1, SHA256, SHA512)):
raise TypeError("Algorithm must be SHA1, SHA256 or SHA512.")
self._key = key
self._length = length
self._algorithm = algorithm
self._backend = backend
def __init__(self, ca, path_length):
if not isinstance(ca, bool):
raise TypeError("ca must be a boolean value")
if path_length is not None and not ca:
raise ValueError("path_length must be None when ca is False")
if (
path_length is not None and
(not isinstance(path_length, six.integer_types) or path_length < 0)
):
raise TypeError(
"path_length must be a non-negative integer or None"
)
self._ca = ca
self._path_length = path_length
def serial_number(self, number):
"""
Sets the certificate serial number.
"""
if not isinstance(number, six.integer_types):
raise TypeError('Serial number must be of integral type.')
if self._serial_number is not None:
raise ValueError('The serial number may only be set once.')
if number < 0:
raise ValueError('The serial number should be non-negative.')
if utils.bit_length(number) > 160: # As defined in RFC 5280
raise ValueError('The serial number should not be more than 160 '
'bits.')
return CertificateBuilder(
self._issuer_name, self._subject_name,
self._public_key, number, self._not_valid_before,
self._not_valid_after, self._extensions
)
def _asFileDescriptor(obj):
fd = None
if not isinstance(obj, integer_types):
meth = getattr(obj, "fileno", None)
if meth is not None:
obj = meth()
if isinstance(obj, integer_types):
fd = obj
if not isinstance(fd, integer_types):
raise TypeError("argument must be an int, or have a fileno() method.")
elif fd < 0:
raise ValueError(
"file descriptor cannot be a negative integer (%i)" % (fd,))
return fd
def use_privatekey_file(self, keyfile, filetype=_UNSPECIFIED):
"""
Load a private key from a file
:param keyfile: The name of the key file (``bytes`` or ``unicode``)
:param filetype: (optional) The encoding of the file, default is PEM
:return: None
"""
keyfile = _path_string(keyfile)
if filetype is _UNSPECIFIED:
filetype = FILETYPE_PEM
elif not isinstance(filetype, integer_types):
raise TypeError("filetype must be an integer")
use_result = _lib.SSL_CTX_use_PrivateKey_file(
self._context, keyfile, filetype)
if not use_result:
self._raise_passphrase_exception()
def set_verify(self, mode, callback):
"""
Set the verify mode and verify callback
:param mode: The verify mode, this is either VERIFY_NONE or
VERIFY_PEER combined with possible other flags
:param callback: The Python callback to use
:return: None
See SSL_CTX_set_verify(3SSL) for further details.
"""
if not isinstance(mode, integer_types):
raise TypeError("mode must be an integer")
if not callable(callback):
raise TypeError("callback must be callable")
self._verify_helper = _VerifyHelper(callback)
self._verify_callback = self._verify_helper.callback
_lib.SSL_CTX_set_verify(self._context, mode, self._verify_callback)
def bio_read(self, bufsiz):
"""
When using non-socket connections this function reads the "dirty" data
that would have traveled away on the network.
:param bufsiz: The maximum number of bytes to read
:return: The string read.
"""
if self._from_ssl is None:
raise TypeError("Connection sock was not None")
if not isinstance(bufsiz, integer_types):
raise TypeError("bufsiz must be an integer")
buf = _ffi.new("char[]", bufsiz)
result = _lib.BIO_read(self._from_ssl, buf, bufsiz)
if result <= 0:
self._handle_bio_errors(self._from_ssl, result)
return _ffi.buffer(buf, result)[:]
def bytes(num_bytes):
"""
Get some random bytes from the PRNG as a string.
This is a wrapper for the C function ``RAND_bytes``.
:param num_bytes: The number of bytes to fetch.
:return: A string of random bytes.
"""
if not isinstance(num_bytes, _integer_types):
raise TypeError("num_bytes must be an integer")
if num_bytes < 0:
raise ValueError("num_bytes must not be negative")
result_buffer = _ffi.new("char[]", num_bytes)
result_code = _lib.RAND_bytes(result_buffer, num_bytes)
if result_code == -1:
# TODO: No tests for this code path. Triggering a RAND_bytes failure
# might involve supplying a custom ENGINE? That's hard.
_raise_current_error()
return _ffi.buffer(result_buffer)[:]
def __init__(self, key, length, algorithm, backend):
if not isinstance(backend, HMACBackend):
raise UnsupportedAlgorithm(
"Backend object does not implement HMACBackend.",
_Reasons.BACKEND_MISSING_INTERFACE
)
if len(key) < 16:
raise ValueError("Key length has to be at least 128 bits.")
if not isinstance(length, six.integer_types):
raise TypeError("Length parameter must be an integer type.")
if length < 6 or length > 8:
raise ValueError("Length of HOTP has to be between 6 to 8.")
if not isinstance(algorithm, (SHA1, SHA256, SHA512)):
raise TypeError("Algorithm must be SHA1, SHA256 or SHA512.")
self._key = key
self._length = length
self._algorithm = algorithm
self._backend = backend
def __init__(self, ca, path_length):
if not isinstance(ca, bool):
raise TypeError("ca must be a boolean value")
if path_length is not None and not ca:
raise ValueError("path_length must be None when ca is False")
if (
path_length is not None and
(not isinstance(path_length, six.integer_types) or path_length < 0)
):
raise TypeError(
"path_length must be a non-negative integer or None"
)
self._ca = ca
self._path_length = path_length
def __init__(self, require_explicit_policy, inhibit_policy_mapping):
if require_explicit_policy is not None and not isinstance(
require_explicit_policy, six.integer_types
):
raise TypeError(
"require_explicit_policy must be a non-negative integer or "
"None"
)
if inhibit_policy_mapping is not None and not isinstance(
inhibit_policy_mapping, six.integer_types
):
raise TypeError(
"inhibit_policy_mapping must be a non-negative integer or None"
)
if inhibit_policy_mapping is None and require_explicit_policy is None:
raise ValueError(
"At least one of require_explicit_policy and "
"inhibit_policy_mapping must not be None"
)
self._require_explicit_policy = require_explicit_policy
self._inhibit_policy_mapping = inhibit_policy_mapping
def serial_number(self, number):
"""
Sets the certificate serial number.
"""
if not isinstance(number, six.integer_types):
raise TypeError('Serial number must be of integral type.')
if self._serial_number is not None:
raise ValueError('The serial number may only be set once.')
if number < 0:
raise ValueError('The serial number should be non-negative.')
if utils.bit_length(number) > 160: # As defined in RFC 5280
raise ValueError('The serial number should not be more than 160 '
'bits.')
return CertificateBuilder(
self._issuer_name, self._subject_name,
self._public_key, number, self._not_valid_before,
self._not_valid_after, self._extensions
)
def __getitem__(self, s):
if isinstance(s, six.integer_types):
# Single index
assert s >= 0, 'negative indexes are not supported'
qs = copy(self)
qs._limits = (s, 1)
return six.next(iter(qs))
else:
# Slice
assert s.step in (None, 1), 'step is not supported in slices'
start = s.start or 0
stop = s.stop or 2**63 - 1
assert start >= 0 and stop >= 0, 'negative indexes are not supported'
assert start <= stop, 'start of slice cannot be smaller than its end'
qs = copy(self)
qs._limits = (start, stop - start)
return qs
def fetch_size(self, v):
"""Sets the number of rows that are fetched at a time.
*Note that driver's default fetch size is 5000.*
.. code-block:: python
for user in User.objects().fetch_size(500):
print(user)
"""
if not isinstance(v, six.integer_types):
raise TypeError
if v == self._fetch_size:
return self
if v < 1:
raise QueryException("fetch size less than 1 is not allowed")
clone = copy.deepcopy(self)
clone._fetch_size = v
return clone
def timestamp_normalized(self):
"""
we're expecting self.timestamp to be either a long, int, a datetime, or a timedelta
:return:
"""
if not self.timestamp:
return None
if isinstance(self.timestamp, six.integer_types):
return self.timestamp
if isinstance(self.timestamp, timedelta):
tmp = datetime.now() + self.timestamp
else:
tmp = self.timestamp
return int(time.mktime(tmp.timetuple()) * 1e+6 + tmp.microsecond)
def validate_emscripten_memory_size(memory_size):
if isinstance(memory_size, six.integer_types):
if memory_size <= 0:
raise ValueError("Invalid memory size value {size}: a positive number expected".format(size=memory_size))
return ["-s", "TOTAL_MEMORY=" + str(memory_size)]
elif isinstance(memory_size, str):
import re
match = re.match(r"^(\d+)([MK]?)$", memory_size)
if not match:
raise ValueError("Invalid memory size value {size}: "
"an integer expected with an optional M(ega) or K(ilo) suffix "
"(e.g. 256M)".format(size=memory_size))
number = int(match.group(1))
suffix = match.group(2)
memory_size = {"": number, "K": number * 1024, "M": number * 1048576}[suffix]
return validate_emscripten_memory_size(memory_size)
elif memory_size is all:
return ["-s", "ALLOW_MEMORY_GROWTH=1"]
else:
raise TypeError("Invalid memory size type: an integer, string, or all expected")
def validate(cls, value, minimum=None, maximum=None):
if value is None:
return None
if not isinstance(value, six.integer_types):
try:
value = int(value)
except Exception:
LOG.exception('Failed to convert value to int')
raise exception.InvalidValue(value=value, type=cls.type_name)
if minimum is not None and value < minimum:
message = _("Integer '%(value)s' is smaller than "
"'%(min)d'.") % {'value': value, 'min': minimum}
raise exception.InvalidValue(message=message)
if maximum is not None and value > maximum:
message = _("Integer '%(value)s' is large than "
"'%(max)d'.") % {'value': value, 'max': maximum}
raise exception.InvalidValue(message=message)
return value
