4.42.0

app/tests/cron/test_get_alias_for_hibp.py

@@ -28,7 +28,7 @@ def test_get_alias_for_free_user_has_no_alias():
     assert len(aliases) == 0
 
 
-def test_get_alias_for_lifetime():
+def test_get_alias_for_lifetime_with_null_hibp_date():
     user = create_new_user()
     user.lifetime = True
     alias_id = Alias.create_new_random(user).id
@@ -39,6 +39,19 @@ def test_get_alias_for_lifetime():
     assert alias_id == aliases[0].id
 
 
+def test_get_alias_for_lifetime_with_old_hibp_date():
+    user = create_new_user()
+    user.lifetime = True
+    alias = Alias.create_new_random(user)
+    alias.hibp_last_check = arrow.now().shift(days=-1)
+    alias_id = alias.id
+    Session.commit()
+    aliases = list(
+        cron.get_alias_to_check_hibp(arrow.now(), [], alias_id, alias_id + 1)
+    )
+    assert alias_id == aliases[0].id
+
+
 def create_partner_sub(user: User):
     pu = PartnerUser.create(
         partner_id=get_proton_partner().id,
@@ -114,3 +127,16 @@ def test_skipped_user_is_not_checked():
         cron.get_alias_to_check_hibp(arrow.now(), [user.id], alias_id, alias_id + 1)
     )
     assert len(aliases) == 0
+
+
+def test_already_checked_is_not_checked():
+    user = create_new_user()
+    user.lifetime = True
+    alias = Alias.create_new_random(user)
+    alias.hibp_last_check = arrow.now().shift(days=1)
+    alias_id = alias.id
+    Session.commit()
+    aliases = list(
+        cron.get_alias_to_check_hibp(arrow.now(), [user.id], alias_id, alias_id + 1)
+    )
+    assert len(aliases) == 0

app/tests/tasks/test_cleanup_old_imports.py

@@ -0,0 +1,35 @@
+import tempfile
+from io import BytesIO
+
+import arrow
+
+from app import s3, config
+from app.models import File, BatchImport
+from tasks.cleanup_old_imports import cleanup_old_imports
+from tests.utils import random_token, create_new_user
+
+
+def test_cleanup_old_imports():
+    BatchImport.filter().delete()
+    with tempfile.TemporaryDirectory() as tmpdir:
+        config.UPLOAD_DIR = tmpdir
+        user = create_new_user()
+        path = random_token()
+        s3.upload_from_bytesio(path, BytesIO("data".encode("utf-8")))
+        file = File.create(path=path, commit=True)  # noqa: F821
+        now = arrow.now()
+        delete_batch_import_id = BatchImport.create(
+            user_id=user.id,
+            file_id=file.id,
+            created_at=now.shift(minutes=-1),
+            flush=True,
+        ).id
+        keep_batch_import_id = BatchImport.create(
+            user_id=user.id,
+            file_id=file.id,
+            created_at=now.shift(minutes=+1),
+            commit=True,
+        ).id
+        cleanup_old_imports(now)
+        assert BatchImport.get(id=delete_batch_import_id) is None
+        assert BatchImport.get(id=keep_batch_import_id) is not None

app/tests/tasks/test_cleanup_old_jobs.py

@@ -0,0 +1,72 @@
+import arrow
+
+from app import config
+from app.models import Job, JobState
+from tasks.cleanup_old_jobs import cleanup_old_jobs
+
+
+def test_cleanup_old_jobs():
+    Job.filter().delete()
+    now = arrow.now()
+    delete_ids = [
+        Job.create(
+            updated_at=now.shift(minutes=-1),
+            state=JobState.done.value,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+        Job.create(
+            updated_at=now.shift(minutes=-1),
+            state=JobState.error.value,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+        Job.create(
+            updated_at=now.shift(minutes=-1),
+            state=JobState.taken.value,
+            attempts=config.JOB_MAX_ATTEMPTS,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+    ]
+
+    keep_ids = [
+        Job.create(
+            updated_at=now.shift(minutes=+1),
+            state=JobState.done.value,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+        Job.create(
+            updated_at=now.shift(minutes=+1),
+            state=JobState.error.value,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+        Job.create(
+            updated_at=now.shift(minutes=+1),
+            state=JobState.taken.value,
+            attempts=config.JOB_MAX_ATTEMPTS,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+        Job.create(
+            updated_at=now.shift(minutes=-1),
+            state=JobState.taken.value,
+            attempts=config.JOB_MAX_ATTEMPTS - 1,
+            name="",
+            payload="",
+            flush=True,
+        ).id,
+    ]
+    cleanup_old_jobs(now)
+    for delete_id in delete_ids:
+        assert Job.get(id=delete_id) is None
+    for keep_id in keep_ids:
+        assert Job.get(id=keep_id) is not None

app/tests/tasks/test_cleanup_old_notifications.py

@@ -0,0 +1,26 @@
+import arrow
+
+from app.models import Notification
+from tasks.cleanup_old_notifications import cleanup_old_notifications
+from tests.utils import create_new_user
+
+
+def test_cleanup_old_notifications():
+    Notification.filter().delete()
+    user = create_new_user()
+    now = arrow.now()
+    delete_id = Notification.create(
+        user_id=user.id,
+        created_at=now.shift(minutes=-1),
+        message="",
+        flush=True,
+    ).id
+    keep_id = Notification.create(
+        user_id=user.id,
+        created_at=now.shift(minutes=+1),
+        message="",
+        flush=True,
+    ).id
+    cleanup_old_notifications(now)
+    assert Notification.get(id=delete_id) is None
+    assert Notification.get(id=keep_id) is not None