cmhub/apps/portal/tests.py

import uuid

from allauth.account.models import EmailAddress
from django.contrib.auth import get_user, get_user_model
from django.core import mail
from django.core.cache import cache
from django.test import Client, TestCase, override_settings
from rest_framework.test import APIClient

from apps.billing.models import PointsLedger
from apps.users.models import ApiKey, UserWallet


@override_settings(EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend")
class PortalAccountFlowTests(TestCase):
    password = "Strong-pass-123"

    def setUp(self):
        cache.clear()

    def create_verified_user(self):
        suffix = uuid.uuid4().hex[:8]
        user = get_user_model().objects.create_user(
            username=f"portal-user-{suffix}",
            email=f"portal-user-{suffix}@example.com",
            password=self.password,
        )
        EmailAddress.objects.create(
            user=user,
            email=user.email,
            verified=True,
            primary=True,
        )
        return user

    def test_signup_creates_unverified_user_wallet_with_zero_points_and_no_ledger(self):
        suffix = uuid.uuid4().hex[:8]
        email = f"signup-{suffix}@example.com"

        response = self.client.post(
            "/signup",
            {
                "username": f"signup-{suffix}",
                "email": email,
                "password1": self.password,
                "password2": self.password,
            },
        )

        self.assertEqual(response.status_code, 302)
        user = get_user_model().objects.get(email=email)
        email_address = EmailAddress.objects.get(user=user, email=email)
        wallet = UserWallet.objects.get(user=user)
        self.assertFalse(email_address.verified)
        self.assertTrue(email_address.primary)
        self.assertEqual(wallet.points_balance, 0)
        self.assertFalse(PointsLedger.objects.filter(user=user).exists())
        self.assertEqual(len(mail.outbox), 1)

    def test_unverified_email_cannot_establish_login_session(self):
        suffix = uuid.uuid4().hex[:8]
        user = get_user_model().objects.create_user(
            username=f"unverified-{suffix}",
            email=f"unverified-{suffix}@example.com",
            password=self.password,
        )
        EmailAddress.objects.create(
            user=user,
            email=user.email,
            verified=False,
            primary=True,
        )

        self.client.post("/login", {"login": user.email, "password": self.password})

        self.assertFalse(get_user(self.client).is_authenticated)

    def test_verified_user_can_login_view_dashboard_and_logout(self):
        user = self.create_verified_user()
        UserWallet.objects.create(user=user, points_balance=42)

        login_response = self.client.post(
            "/login",
            {"login": user.email, "password": self.password},
        )

        self.assertEqual(login_response.status_code, 302)
        self.assertEqual(login_response["Location"], "/dashboard")
        self.assertTrue(get_user(self.client).is_authenticated)

        dashboard_response = self.client.get("/dashboard")
        self.assertEqual(dashboard_response.status_code, 200)
        self.assertContains(dashboard_response, user.email)
        self.assertContains(dashboard_response, "42")

        logout_response = self.client.post("/logout")
        self.assertEqual(logout_response.status_code, 302)
        self.assertEqual(logout_response["Location"], "/login")
        self.assertFalse(get_user(self.client).is_authenticated)

    def test_dashboard_requires_session_login(self):
        response = self.client.get("/dashboard")

        self.assertEqual(response.status_code, 302)
        self.assertTrue(response["Location"].startswith("/login?next="))

    def test_login_post_is_csrf_protected(self):
        csrf_client = Client(enforce_csrf_checks=True)

        response = csrf_client.post(
            "/login",
            {"login": "nobody@example.com", "password": "password"},
        )

        self.assertEqual(response.status_code, 403)

    def test_apikeys_requires_session_login(self):
        response = self.client.get("/apikeys")

        self.assertEqual(response.status_code, 302)
        self.assertTrue(response["Location"].startswith("/login?next="))

    def test_create_api_key_shows_plaintext_once_and_stores_only_hash(self):
        user = self.create_verified_user()
        self.client.force_login(user)

        response = self.client.post("/apikeys", {"name": "desktop"}, follow=True)

        self.assertEqual(response.status_code, 200)
        api_key = ApiKey.objects.get(user=user)
        new_api_key = response.context["new_api_key"]
        raw_key = new_api_key["raw_key"]
        self.assertTrue(raw_key.startswith("sk_cmhub_"))
        self.assertEqual(api_key.name, "desktop")
        self.assertEqual(api_key.key_prefix, raw_key[: ApiKey.KEY_PREFIX_LENGTH])
        self.assertNotEqual(api_key.key_hash, raw_key)
        self.assertNotIn(raw_key, str(api_key.__dict__))
        self.assertTrue(api_key.matches_key(raw_key))
        self.assertContains(response, raw_key)
        self.assertContains(response, api_key.key_prefix)
        self.assertNotContains(response, api_key.key_hash)

        second_response = self.client.get("/apikeys")

        self.assertEqual(second_response.status_code, 200)
        self.assertNotContains(second_response, raw_key)
        self.assertContains(second_response, api_key.key_prefix)
        self.assertNotContains(second_response, api_key.key_hash)

    def test_apikey_list_only_shows_current_user_prefix(self):
        user = self.create_verified_user()
        other_user = self.create_verified_user()
        own_key, own_raw_key = ApiKey.create_for_user(user, name="desktop")
        other_key, other_raw_key = ApiKey.create_for_user(other_user, name="other")
        self.client.force_login(user)

        response = self.client.get("/apikeys")

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, own_key.key_prefix)
        self.assertContains(response, "desktop")
        self.assertNotContains(response, own_raw_key)
        self.assertNotContains(response, own_key.key_hash)
        self.assertNotContains(response, other_key.key_prefix)
        self.assertNotContains(response, other_raw_key)
        self.assertNotContains(response, other_key.key_hash)

    def test_delete_api_key_revokes_it_and_external_api_returns_403(self):
        user = self.create_verified_user()
        api_key, raw_key = ApiKey.create_for_user(user, name="desktop")
        self.client.force_login(user)

        response = self.client.post(f"/apikeys/{api_key.pk}/delete")

        self.assertEqual(response.status_code, 302)
        self.assertEqual(response["Location"], "/apikeys")
        api_key.refresh_from_db()
        self.assertEqual(api_key.status, ApiKey.Status.REVOKED)

        api_client = APIClient()
        api_response = api_client.get(
            "/api/v1/balance",
            HTTP_AUTHORIZATION=f"Bearer {raw_key}",
        )

        self.assertEqual(api_response.status_code, 403)
        self.assertEqual(api_response.data["error"]["code"], "account_disabled")

    def test_delete_api_key_does_not_allow_other_users_key(self):
        user = self.create_verified_user()
        other_user = self.create_verified_user()
        other_key, _raw_key = ApiKey.create_for_user(other_user, name="other")
        self.client.force_login(user)

        response = self.client.post(f"/apikeys/{other_key.pk}/delete")

        self.assertEqual(response.status_code, 404)
        other_key.refresh_from_db()
        self.assertEqual(other_key.status, ApiKey.Status.ACTIVE)

    def test_apikey_create_and_delete_posts_are_csrf_protected(self):
        user = self.create_verified_user()
        api_key, _raw_key = ApiKey.create_for_user(user, name="desktop")
        csrf_client = Client(enforce_csrf_checks=True)
        csrf_client.force_login(user)

        create_response = csrf_client.post("/apikeys", {"name": "new"})
        delete_response = csrf_client.post(f"/apikeys/{api_key.pk}/delete")

        self.assertEqual(create_response.status_code, 403)
        self.assertEqual(delete_response.status_code, 403)
        api_key.refresh_from_db()
        self.assertEqual(api_key.status, ApiKey.Status.ACTIVE)