4.41.2

app/tests/cron/test_cron.py

@@ -0,0 +1,55 @@
+import arrow
+
+import cron
+from app.db import Session
+from app.models import CoinbaseSubscription, ApiToCookieToken, ApiKey, User
+from tests.utils import create_new_user
+
+
+def test_notify_manual_sub_end(flask_client):
+    user = create_new_user()
+    CoinbaseSubscription.create(
+        user_id=user.id, end_at=arrow.now().shift(days=13, hours=2), commit=True
+    )
+    cron.notify_manual_sub_end()
+
+
+def test_cleanup_tokens(flask_client):
+    user = create_new_user()
+    api_key = ApiKey.create(
+        user_id=user.id,
+        commit=True,
+    )
+    id_to_clean = ApiToCookieToken.create(
+        user_id=user.id,
+        api_key_id=api_key.id,
+        commit=True,
+        created_at=arrow.now().shift(days=-1),
+    ).id
+
+    id_to_keep = ApiToCookieToken.create(
+        user_id=user.id,
+        api_key_id=api_key.id,
+        commit=True,
+    ).id
+    cron.delete_expired_tokens()
+    assert ApiToCookieToken.get(id_to_clean) is None
+    assert ApiToCookieToken.get(id_to_keep) is not None
+
+
+def test_cleanup_users():
+    u_delete_none_id = create_new_user().id
+    u_delete_grace_has_expired = create_new_user()
+    u_delete_grace_has_expired_id = u_delete_grace_has_expired.id
+    u_delete_grace_has_not_expired = create_new_user()
+    u_delete_grace_has_not_expired_id = u_delete_grace_has_not_expired.id
+    now = arrow.now()
+    u_delete_grace_has_expired.delete_on = now.shift(days=-(cron.DELETE_GRACE_DAYS + 1))
+    u_delete_grace_has_not_expired.delete_on = now.shift(
+        days=-(cron.DELETE_GRACE_DAYS - 1)
+    )
+    Session.flush()
+    cron.clear_users_scheduled_to_be_deleted()
+    assert User.get(u_delete_none_id) is not None
+    assert User.get(u_delete_grace_has_not_expired_id) is not None
+    assert User.get(u_delete_grace_has_expired_id) is None

app/tests/cron/test_get_alias_for_hibp.py

@@ -0,0 +1,116 @@
+import arrow
+import pytest
+
+import cron
+from app.db import Session
+from app.models import (
+    Alias,
+    AppleSubscription,
+    PlanEnum,
+    CoinbaseSubscription,
+    ManualSubscription,
+    Subscription,
+    PartnerUser,
+    PartnerSubscription,
+    User,
+)
+from app.proton.utils import get_proton_partner
+from tests.utils import create_new_user, random_token
+
+
+def test_get_alias_for_free_user_has_no_alias():
+    user = create_new_user()
+    alias_id = Alias.create_new_random(user).id
+    Session.commit()
+    aliases = list(
+        cron.get_alias_to_check_hibp(arrow.now(), [], alias_id, alias_id + 1)
+    )
+    assert len(aliases) == 0
+
+
+def test_get_alias_for_lifetime():
+    user = create_new_user()
+    user.lifetime = True
+    alias_id = Alias.create_new_random(user).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,
+        partner_email=user.email,
+        external_user_id=random_token(10),
+        user_id=user.id,
+        flush=True,
+    )
+    PartnerSubscription.create(
+        partner_user_id=pu.id, end_at=arrow.utcnow().shift(days=15)
+    )
+
+
+sub_generator_list = [
+    lambda u: AppleSubscription.create(
+        user_id=u.id,
+        expires_date=arrow.now().shift(days=15),
+        original_transaction_id=random_token(10),
+        receipt_data=random_token(10),
+        plan=PlanEnum.monthly,
+    ),
+    lambda u: CoinbaseSubscription.create(
+        user_id=u.id,
+        end_at=arrow.now().shift(days=15),
+    ),
+    lambda u: ManualSubscription.create(
+        user_id=u.id,
+        end_at=arrow.now().shift(days=15),
+    ),
+    lambda u: Subscription.create(
+        user_id=u.id,
+        cancel_url="",
+        update_url="",
+        subscription_id=random_token(10),
+        event_time=arrow.now(),
+        next_bill_date=arrow.now().shift(days=15).date(),
+        plan=PlanEnum.monthly,
+    ),
+    create_partner_sub,
+]
+
+
+@pytest.mark.parametrize("sub_generator", sub_generator_list)
+def test_get_alias_for_sub(sub_generator):
+    user = create_new_user()
+    sub_generator(user)
+    alias_id = Alias.create_new_random(user).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 test_disabled_user_is_not_checked():
+    user = create_new_user()
+    user.lifetime = True
+    user.disabled = True
+    alias_id = Alias.create_new_random(user).id
+    Session.commit()
+    aliases = list(
+        cron.get_alias_to_check_hibp(arrow.now(), [], alias_id, alias_id + 1)
+    )
+    assert len(aliases) == 0
+
+
+def test_skipped_user_is_not_checked():
+    user = create_new_user()
+    user.lifetime = True
+    alias_id = Alias.create_new_random(user).id
+    Session.commit()
+    aliases = list(
+        cron.get_alias_to_check_hibp(arrow.now(), [user.id], alias_id, alias_id + 1)
+    )
+    assert len(aliases) == 0