Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_roundtrip_pickle(self):
ser = get_serializer(Format.PICKLE)
deser = get_deserializer(Format.PICKLE)
fp = BytesIO()
ser(date_sample, fp)
fp.seek(0)
obj = deser(fp)
self.assertEqual(obj, date_sample)
def test_get_serializer_json(self):
ser = get_serializer(Format.JSON)
self.assertEqual(ser, _serialize_json)
def test_get_serializer_pickle(self):
ser = get_serializer(Format.PICKLE)
self.assertEqual(ser, _serialize_pickle)
def __init__(self, prefix=None, debug=False, record_format=Format.JSON):
if debug:
self._set_logger(__name__, logging.DEBUG)
record_format = record_format.lower()
if record_format not in Format.ALLOWED:
LOG.warning("Record format not allowed. Falling back to default.")
record_format = Format.DEFAULT
self._serializer = get_serializer(record_format)
self._filename_re = re.compile(r'.*\..*_(?P\d+).{0}'.format(
record_format))
self._record_format = record_format
self.prefix = prefix
self._uuid = str(uuid.uuid4())
self._data_path = None
self._mode = None
self._session = None
self._index = {}
self.events = []
self.clients = []