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_diff(self):
testdb = "test-%s" % self.filename
now = int(time.time())
self.addCleanup(self._remove, testdb)
whisper.create(testdb, self.retention)
whisper.create(self.filename, self.retention)
whisper.update(testdb, 1.0, now)
whisper.update(self.filename, 2.0, now)
results = whisper.diff(testdb, self.filename)
expected = [(0, [(int(now), 1.0, 2.0)], 1), (1, [], 0)]
self.assertEqual(results, expected)
def test_file_diff(self):
testdb = "test-%s" % self.filename
now = time.time()
self.addCleanup(self._remove, testdb)
whisper.create(testdb, self.retention)
whisper.create(self.filename, self.retention)
whisper.update(testdb, 1.0, now)
whisper.update(self.filename, 2.0, now)
# Merging 2 archives with different retentions should fail
with open(testdb, 'rb') as fh_1:
with open(self.filename, 'rb+') as fh_2:
results = whisper.file_diff(fh_1, fh_2)
expected = [(0, [(int(now), 1.0, 2.0)], 1), (1, [], 0)]
self.assertEqual(results, expected)
for i, (timestamp, value) in enumerate(data):
# is value in the fetched data?
self.assertEqual(value, fetch_data[i])
# check TimestampNotCovered
with AssertRaisesException(
whisper.TimestampNotCovered(
'Timestamp not covered by any archives in this database.')):
# in the futur
whisper.update(self.filename, 1.337, time.time() + 1)
with AssertRaisesException(
whisper.TimestampNotCovered(
'Timestamp not covered by any archives in this database.')):
# before the past
whisper.update(self.filename, 1.337, time.time() - retention_schema[0][1] - 1)
# When no timestamp is passed in, it should use the current time
original_lock = whisper.LOCK
whisper.LOCK = True
whisper.update(self.filename, 3.7337, None)
fetched = whisper.fetch(self.filename, 0)[1]
self.assertEqual(fetched[-1], 3.7337)
whisper.LOCK = original_lock
def test_diff(self):
testdb = "test-%s" % self.filename
now = int(time.time())
self.addCleanup(self._remove, testdb)
whisper.create(testdb, self.retention)
whisper.create(self.filename, self.retention)
whisper.update(testdb, 1.0, now)
whisper.update(self.filename, 2.0, now)
results = whisper.diff(testdb, self.filename)
expected = [(0, [(int(now), 1.0, 2.0)], 1), (1, [], 0)]
self.assertEqual(results, expected)
with AssertRaisesException(
whisper.TimestampNotCovered(
'Timestamp not covered by any archives in this database.')):
# in the futur
whisper.update(self.filename, 1.337, time.time() + 1)
with AssertRaisesException(
whisper.TimestampNotCovered(
'Timestamp not covered by any archives in this database.')):
# before the past
whisper.update(self.filename, 1.337, time.time() - retention_schema[0][1] - 1)
# When no timestamp is passed in, it should use the current time
original_lock = whisper.LOCK
whisper.LOCK = True
whisper.update(self.filename, 3.7337, None)
fetched = whisper.fetch(self.filename, 0)[1]
self.assertEqual(fetched[-1], 3.7337)
whisper.LOCK = original_lock
wsp = wsp or self.filename
schema = schema or [(1, 20)]
num_data_points = 20
# create sample data
self.addCleanup(self._remove, wsp)
whisper.create(wsp, schema, sparse=sparse, useFallocate=useFallocate)
tn = int(time.time()) - num_data_points
data = []
for i in range(num_data_points):
data.append((tn + 1 + i, random.random() * 10))
# test single update
whisper.update(wsp, data[0][1], data[0][0])
# test multi update
whisper.update_many(wsp, data[1:])
return data
def _update(self, datapoints):
"""
This method store in the datapoints in the current database.
:datapoints: is a list of tupple with the epoch timestamp and value
[(1368977629,10)]
"""
if len(datapoints) == 1:
timestamp, value = datapoints[0]
whisper.update(self.path, value, timestamp)
else:
whisper.update_many(self.path, datapoints)