Switch bracketed time comparisons to use from_now

Author: jagerman

tests/test_dm.py

@@ -5,7 +5,7 @@
 from sogs.model.user import SystemUser
 import nacl.bindings as salt
 from nacl.utils import random
-import time
+from util import from_now
 
 
 def test_dm_default_empty(client, blind_user):
@@ -71,21 +71,21 @@ def test_dm_send(client, blind_user, blind_user2):
     r = sogs_post(client, f'/inbox/{blind_user2.session_id}', post, blind_user)
     assert r.status_code == 201
     data = r.json
-    assert -1 < data.pop('posted_at') - time.time() < 1
-    assert -1 < data.pop('expires_at') - config.DM_EXPIRY - time.time() < 1
+    assert data.pop('posted_at') == from_now.seconds(0)
+    assert data.pop('expires_at') == from_now.seconds(config.DM_EXPIRY)
     assert data == {k: v for k, v in msg_expected.items() if k != 'message'}
 
     r = sogs_get(client, '/inbox', blind_user2)
     assert r.status_code == 200
     assert len(r.json) == 1
     data = r.json[0]
-    assert -1 < data.pop('posted_at') - time.time() < 1
-    assert -1 < data.pop('expires_at') - config.DM_EXPIRY - time.time() < 1
+    assert data.pop('posted_at') == from_now.seconds(0)
+    assert data.pop('expires_at') == from_now.seconds(config.DM_EXPIRY)
     assert data == msg_expected
 
     r = sogs_get(client, '/outbox', blind_user)
     assert len(r.json) == 1
     data = r.json[0]
-    assert -1 < data.pop('posted_at') - time.time() < 1
-    assert -1 < data.pop('expires_at') - config.DM_EXPIRY - time.time() < 1
+    assert data.pop('posted_at') == from_now.seconds(0)
+    assert data.pop('expires_at') == from_now.seconds(config.DM_EXPIRY)
     assert data == msg_expected

tests/test_room_routes.py

@@ -571,7 +571,7 @@ def test_fetch_since(client, room, user, no_rate_limit):
                 assert post['seqno'] == j
                 assert utils.decode_base64(post['data']) == f"fake data {j}".encode()
                 assert utils.decode_base64(post['signature']) == pad64(f"fake sig {j}")
-                assert -10 <= post['posted'] - time.time() <= 10
+                assert post['posted'] == from_now.seconds(0, 10)
 
                 top_fetched = post['seqno']
 
@@ -693,7 +693,7 @@ def room_json():
     ri = room_json()
     assert ri['info_updates'] == 2
     assert filter_timestamps(ri['pinned_messages']) == [{'id': 3, 'pinned_by': admin.session_id}]
-    assert -1 < ri['pinned_messages'][0]['pinned_at'] - time.time() < 1
+    assert ri['pinned_messages'][0]['pinned_at'] == from_now.now()
 
     url = "/room/test-room/pin/7"
     r = sogs_post(client, url, {}, admin)
@@ -752,7 +752,9 @@ def room_json():
         {'id': 3, 'pinned_by': admin.session_id},
         {'id': 7, 'pinned_by': admin.session_id},
     ]
-    assert time.time() - 1 < rpm[0]['pinned_at'] < rpm[1]['pinned_at'] < time.time() + 1
+    assert rpm[0]['pinned_at'] == from_now.now()
+    assert rpm[0]['pinned_at'] < rpm[1]['pinned_at']
+    assert rpm[1]['pinned_at'] == from_now.now()
 
     url = "/room/test-room/unpin/all"
     r = sogs_post(client, url, {}, admin)
@@ -776,7 +778,7 @@ def test_posting(client, room, user, user2, mod, global_mod):
         'data': d,
         'signature': s,
     }
-    assert -1 < p1['posted'] - time.time() < 1
+    assert p1['posted'] == from_now.now()
 
     url_get = "/room/test-room/messages/since/0"
     r = sogs_get(client, url_get, user)
@@ -806,7 +808,7 @@ def test_whisper_to(client, room, user, user2, mod, global_mod):
         'whisper_mods': False,
         'whisper_to': user2.session_id,
     }
-    assert -1 < msg['posted'] - time.time() < 1
+    assert msg['posted'] == from_now.now()
 
     url_get = "/room/test-room/messages/since/0"
     # user shouldn't get the whisper:
@@ -852,7 +854,7 @@ def test_whisper_mods(client, room, user, user2, mod, global_mod, admin):
         'whisper': True,
         'whisper_mods': True,
     }
-    assert -1 < msg['posted'] - time.time() < 1
+    assert msg['posted'] == from_now.now()
 
     url_get = "/room/test-room/messages/since/0"
 
@@ -972,7 +974,7 @@ def test_edits(client, room, user, user2, mod, global_admin):
         'data': d,
         'signature': s,
     }
-    assert -1 < p1['posted'] - time.time() < 1
+    assert p1['posted'] == from_now.now()
 
     url_get = "/room/test-room/messages/since/0"
     r = sogs_get(client, url_get, user)
@@ -1000,7 +1002,9 @@ def test_edits(client, room, user, user2, mod, global_admin):
 
     r = sogs_get(client, url_get, user)
     assert filter_timestamps(r.json) == filter_timestamps([p1])
-    assert time.time() - 1 < r.json[0]['posted'] < r.json[0]['edited'] < time.time() + 1
+    assert r.json[0]['posted'] == from_now.now()
+    assert r.json[0]['posted'] < r.json[0]['edited']
+    assert r.json[0]['edited'] == from_now.now()
     p1['edited'] = r.json[0]['edited']
 
     d, s = (utils.encode_base64(x) for x in (b"post 2", pad64("sig 2")))
@@ -1014,7 +1018,7 @@ def test_edits(client, room, user, user2, mod, global_admin):
         'data': d,
         'signature': s,
     }
-    assert -1 < p2['posted'] - time.time() < 1
+    assert p2['posted'] == from_now.now()
 
     d, s = (utils.encode_base64(x) for x in (b"post 1c", pad64("sig 1c")))
     time.sleep(0.001)

tests/test_rooms.py

@@ -4,7 +4,7 @@
 from sogs.model.room import Room, get_rooms
 from sogs.model.file import File
 from sogs import config
-from util import pad64
+from util import pad64, from_now
 
 
 def test_create(room, room2):
@@ -446,15 +446,14 @@ def test_upload(room, user):
 
     import os
 
-    now = time.time()
     file = File(id=room.upload_file(content=b'abc', uploader=user, filename="abc.txt", lifetime=30))
 
     assert file.id
     assert file.room.id and file.room.id == room.id
     assert file.uploader.id and file.uploader.id == user.id
     assert file.size == 3
-    assert now - 1 < file.uploaded < now + 1
-    assert now - 1 < file.expiry - 30 < now + 1
+    assert file.uploaded == from_now.now()
+    assert file.expiry == from_now.seconds(30)
     assert os.path.isfile(file.path)
     assert os.path.getsize(file.path) == file.size
     assert file.path == f'{config.UPLOAD_PATH}/{room.token}/{file.id}_abc.txt'
@@ -466,8 +465,8 @@ def test_upload(room, user):
     assert file.room.id and file.room.id == room.id
     assert file.uploader.id and file.uploader.id == user.id
     assert file.size == 4
-    assert now - 1 < file.uploaded < now + 1
-    assert now - 1 < file.expiry - 15 * 86400 < now + 1
+    assert file.uploaded == from_now.now()
+    assert file.expiry == from_now.days(15)
     assert os.path.isfile(file.path)
     assert os.path.getsize(file.path) == file.size
     assert file.path == f'{config.UPLOAD_PATH}/{room.token}/{file.id}_(unnamed)'
@@ -478,13 +477,11 @@ def test_upload_expiry(room, user):
 
     import os
 
-    now = time.time()
-    now = time.time()
     file = File(id=room.upload_file(content=b'abc', uploader=user, filename="abc.txt", lifetime=-1))
 
     assert file.id
-    assert now - 1 < file.uploaded < now + 1
-    assert now - 2 < file.expiry < now
+    assert file.uploaded == from_now.now()
+    assert file.expiry == from_now.seconds(-1)
     assert os.path.isfile(file.path)
     assert os.path.getsize(file.path) == file.size
     assert file.path == f'{config.UPLOAD_PATH}/{room.token}/{file.id}_abc.txt'
@@ -562,19 +559,15 @@ def test_pinning(room, room2, user, mod, admin, global_admin, no_rate_limit):
     room.pin(msgs[5]['id'], admin)
     assert msgs[5]['id'] == 6
 
-    assert -1 < room.pinned_messages[0]['pinned_at'] - time.time() < 1
-    del room.pinned_messages[0]['pinned_at']
+    assert room.pinned_messages[0].pop('pinned_at') == from_now.now()
     assert room.pinned_messages == [{"id": 6, "pinned_by": admin.session_id}]
 
     time.sleep(0.001)
 
     room.pin(7, global_admin)
-    assert (
-        time.time() - 1
-        < room.pinned_messages[0]['pinned_at']
-        < room.pinned_messages[1]['pinned_at']
-        < time.time() + 1
-    )
+    assert room.pinned_messages[0]['pinned_at'] == from_now.now()
+    assert room.pinned_messages[0]['pinned_at'] < room.pinned_messages[1]['pinned_at']
+    assert room.pinned_messages[1]['pinned_at'] == from_now.now()
     old_ts_t = room.pinned_messages[1]['pinned_at']
     rpm = room.pinned_messages.copy()
     for pm in rpm:
@@ -588,12 +581,9 @@ def test_pinning(room, room2, user, mod, admin, global_admin, no_rate_limit):
     # Re-pin (will update its pinned timestamp and thus implicit reorder, along with pinned_by)
     room.pin(6, global_admin)
 
-    assert (
-        time.time() - 1
-        < room.pinned_messages[0]['pinned_at']
-        < room.pinned_messages[1]['pinned_at']
-        < time.time() + 1
-    )
+    assert room.pinned_messages[0]['pinned_at'] == from_now.now()
+    assert room.pinned_messages[0]['pinned_at'] < room.pinned_messages[1]['pinned_at']
+    assert room.pinned_messages[1]['pinned_at'] == from_now.now()
     assert old_ts_t == room.pinned_messages[0]['pinned_at']
     rpm = room.pinned_messages.copy()
     for pm in rpm:

tests/util.py

@@ -35,20 +35,27 @@ def __repr__(self):
 
 class from_now:
     @staticmethod
-    def seconds(n):
-        return FuzzyTime(n)
+    def seconds(n, epsilon=None):
+        f = FuzzyTime(n)
+        if epsilon is not None:
+            f.epsilon = epsilon
+        return f
 
     @staticmethod
-    def minutes(n):
-        return from_now.seconds(60) * n
+    def now(epsilon=None):
+        return from_now.seconds(0, epsilon)
 
     @staticmethod
-    def hours(n):
-        return from_now.minutes(60) * n
+    def minutes(n, epsilon=None):
+        return from_now.seconds(60, epsilon) * n
 
     @staticmethod
-    def days(n):
-        return from_now.hours(24) * n
+    def hours(n, epsilon=None):
+        return from_now.minutes(60, epsilon) * n
+
+    @staticmethod
+    def days(n, epsilon=None):
+        return from_now.hours(24, epsilon) * n
 
 
 @contextmanager