4.29.3

app/tests/test_alias_suffixes.py

@@ -0,0 +1,133 @@
+import re
+
+from app.alias_suffix import get_alias_suffixes
+from app.db import Session
+from app.models import SLDomain, PartnerUser, AliasOptions, CustomDomain
+from app.proton.utils import get_proton_partner
+from init_app import add_sl_domains
+from tests.utils import create_new_user, random_token
+
+
+def setup_module():
+    Session.query(SLDomain).delete()
+    SLDomain.create(
+        domain="hidden", premium_only=False, flush=True, order=5, hidden=True
+    )
+    SLDomain.create(domain="free_non_partner", premium_only=False, flush=True, order=4)
+    SLDomain.create(
+        domain="premium_non_partner", premium_only=True, flush=True, order=3
+    )
+    SLDomain.create(
+        domain="free_partner",
+        premium_only=False,
+        flush=True,
+        partner_id=get_proton_partner().id,
+        order=2,
+    )
+    SLDomain.create(
+        domain="premium_partner",
+        premium_only=True,
+        flush=True,
+        partner_id=get_proton_partner().id,
+        order=1,
+    )
+    Session.commit()
+
+
+def teardown_module():
+    Session.query(SLDomain).delete()
+    add_sl_domains()
+
+
+def test_get_default_domain_even_if_is_not_allowed():
+    user = create_new_user()
+    PartnerUser.create(
+        partner_id=get_proton_partner().id,
+        user_id=user.id,
+        external_user_id=random_token(10),
+        flush=True,
+    )
+    user.trial_end = None
+    default_domain = SLDomain.filter_by(
+        hidden=False, partner_id=None, premium_only=False
+    ).first()
+    user.default_alias_public_domain_id = default_domain.id
+    Session.flush()
+    options = AliasOptions(
+        show_sl_domains=False, show_partner_domains=get_proton_partner()
+    )
+    suffixes = get_alias_suffixes(user, alias_options=options)
+    assert suffixes[0].domain == default_domain.domain
+
+
+def test_get_default_domain_hidden():
+    user = create_new_user()
+    PartnerUser.create(
+        partner_id=get_proton_partner().id,
+        user_id=user.id,
+        external_user_id=random_token(10),
+        flush=True,
+    )
+    user.trial_end = None
+    default_domain = SLDomain.filter_by(
+        hidden=True, partner_id=None, premium_only=False
+    ).first()
+    user.default_alias_public_domain_id = default_domain.id
+    Session.flush()
+    options = AliasOptions(
+        show_sl_domains=False, show_partner_domains=get_proton_partner()
+    )
+    suffixes = get_alias_suffixes(user, alias_options=options)
+    for suffix in suffixes:
+        domain = SLDomain.get_by(domain=suffix.domain)
+        assert not domain.hidden
+    assert suffixes[0].domain != default_domain.domain
+
+
+def test_get_default_domain_is_premium_for_free_user():
+    user = create_new_user()
+    PartnerUser.create(
+        partner_id=get_proton_partner().id,
+        user_id=user.id,
+        external_user_id=random_token(10),
+        flush=True,
+    )
+    user.trial_end = None
+    default_domain = SLDomain.filter_by(partner_id=None, premium_only=True).first()
+    user.default_alias_public_domain_id = default_domain.id
+    Session.flush()
+    options = AliasOptions(
+        show_sl_domains=False, show_partner_domains=get_proton_partner()
+    )
+    suffixes = get_alias_suffixes(user, alias_options=options)
+    for suffix in suffixes:
+        domain = SLDomain.get_by(domain=suffix.domain)
+        assert not domain.premium_only
+    assert suffixes[0].domain != default_domain.domain
+
+
+def test_suffixes_are_valid():
+    user = create_new_user()
+    PartnerUser.create(
+        partner_id=get_proton_partner().id,
+        user_id=user.id,
+        external_user_id=random_token(10),
+        flush=True,
+    )
+    CustomDomain.create(
+        user_id=user.id, domain=f"{random_token(10)}.com", verified=True
+    )
+    user.trial_end = None
+    Session.flush()
+    options = AliasOptions(
+        show_sl_domains=True, show_partner_domains=get_proton_partner()
+    )
+    alias_suffixes = get_alias_suffixes(user, alias_options=options)
+    valid_re = re.compile(r"^(\.[\w_]+)?@[\.\w]+$")
+    has_prefix = 0
+    for suffix in alias_suffixes:
+        match = valid_re.match(suffix.suffix)
+        assert match is not None
+        if len(match.groups()) >= 1:
+            has_prefix += 1
+    assert has_prefix > 0

app/tests/test_subscription_webhook.py

@@ -0,0 +1,113 @@
+import http.server
+import json
+import threading
+
+import arrow
+
+from app import config
+from app.models import (
+    Subscription,
+    AppleSubscription,
+    CoinbaseSubscription,
+    ManualSubscription,
+)
+from tests.utils import create_new_user, random_token
+
+from app.subscription_webhook import execute_subscription_webhook
+
+http_server = None
+last_http_request = None
+
+
+def setup_module():
+    global http_server
+    http_server = http.server.ThreadingHTTPServer(("", 0), HTTPTestServer)
+    print(http_server.server_port)
+    threading.Thread(target=http_server.serve_forever, daemon=True).start()
+    config.SUBSCRIPTION_CHANGE_WEBHOOK = f"http://localhost:{http_server.server_port}"
+
+
+def teardown_module():
+    global http_server
+    config.SUBSCRIPTION_CHANGE_WEBHOOK = None
+    http_server.shutdown()
+
+
+class HTTPTestServer(http.server.BaseHTTPRequestHandler):
+    def do_POST(self):
+        global last_http_request
+        content_len = int(self.headers.get("Content-Length"))
+        body_data = self.rfile.read(content_len)
+        last_http_request = json.loads(body_data)
+        self.send_response(200)
+
+
+def test_webhook_with_trial():
+    user = create_new_user()
+    execute_subscription_webhook(user)
+    assert last_http_request["user_id"] == user.id
+    assert last_http_request["is_premium"]
+    assert last_http_request["active_subscription_end"] is None
+
+
+def test_webhook_with_subscription():
+    user = create_new_user()
+    end_at = arrow.utcnow().shift(days=1).replace(hour=0, minute=0, second=0)
+    Subscription.create(
+        user_id=user.id,
+        cancel_url="",
+        update_url="",
+        subscription_id=random_token(10),
+        event_time=arrow.now(),
+        next_bill_date=end_at.date(),
+        plan="yearly",
+        flush=True,
+    )
+    execute_subscription_webhook(user)
+    assert last_http_request["user_id"] == user.id
+    assert last_http_request["is_premium"]
+    assert last_http_request["active_subscription_end"] == end_at.timestamp
+
+
+def test_webhook_with_apple_subscription():
+    user = create_new_user()
+    end_at = arrow.utcnow().shift(days=2).replace(hour=0, minute=0, second=0)
+    AppleSubscription.create(
+        user_id=user.id,
+        receipt_data=arrow.now().date().strftime("%Y-%m-%d"),
+        expires_date=end_at.date().strftime("%Y-%m-%d"),
+        original_transaction_id=random_token(10),
+        plan="yearly",
+        product_id="",
+        flush=True,
+    )
+    execute_subscription_webhook(user)
+    assert last_http_request["user_id"] == user.id
+    assert last_http_request["is_premium"]
+    assert last_http_request["active_subscription_end"] == end_at.timestamp
+
+
+def test_webhook_with_coinbase_subscription():
+    user = create_new_user()
+    end_at = arrow.utcnow().shift(days=3).replace(hour=0, minute=0, second=0)
+    CoinbaseSubscription.create(
+        user_id=user.id, end_at=end_at.date().strftime("%Y-%m-%d"), flush=True
+    )
+
+    execute_subscription_webhook(user)
+    assert last_http_request["user_id"] == user.id
+    assert last_http_request["is_premium"]
+    assert last_http_request["active_subscription_end"] == end_at.timestamp
+
+
+def test_webhook_with_manual_subscription():
+    user = create_new_user()
+    end_at = arrow.utcnow().shift(days=3).replace(hour=0, minute=0, second=0)
+    ManualSubscription.create(
+        user_id=user.id, end_at=end_at.date().strftime("%Y-%m-%d"), flush=True
+    )
+
+    execute_subscription_webhook(user)
+    assert last_http_request["user_id"] == user.id
+    assert last_http_request["is_premium"]
+    assert last_http_request["active_subscription_end"] == end_at.timestamp