import uuid from decimal import Decimal 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 django.utils import timezone from rest_framework.test import APIClient from apps.billing.models import CallRecord, PointsLedger, RechargeOrder 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 create_recharge_order( self, user, *, order_no: str, amount: str, points: int, status: str = RechargeOrder.Status.PAID, ): return RechargeOrder.objects.create( user=user, order_no=order_no, amount_money=Decimal(amount), currency="CNY", pay_method=RechargeOrder.PayMethod.WEIXIN, exchange_rate=Decimal("10.0000"), points_granted=points, status=status, payment_txn_no=f"txn-{order_no}" if status == RechargeOrder.Status.PAID else "", paid_at=timezone.now() if status == RechargeOrder.Status.PAID else None, ) def create_call_record( self, user, api_key, *, alias: str, points_cost: int, status: str = CallRecord.Status.SUCCESS, ): return CallRecord.objects.create( user=user, api_key=api_key, operation_type=CallRecord.OperationType.TITLE, alias=alias, model_used="gpt-5.5", points_cost=points_cost, status=status, ) 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) def test_dashboard_shows_account_summary_recent_records_for_current_user_only(self): user = self.create_verified_user() other_user = self.create_verified_user() UserWallet.objects.create(user=user, points_balance=160) UserWallet.objects.create(user=other_user, points_balance=999) api_key, _raw_key = ApiKey.create_for_user(user, name="desktop") other_key, _other_raw_key = ApiKey.create_for_user(other_user, name="other") order = self.create_recharge_order( user, order_no="R-OWN-PAID-503", amount="20.00", points=200, ) self.create_recharge_order( other_user, order_no="R-OTHER-PAID-503", amount="99.00", points=990, ) PointsLedger.objects.create( user=user, change_type=PointsLedger.ChangeType.RECHARGE, points_delta=200, balance_after=200, ref_order_id=order.id, ) call = self.create_call_record( user, api_key, alias="title-standard", points_cost=50, status=CallRecord.Status.FAILED, ) other_call = self.create_call_record( other_user, other_key, alias="other-alias", points_cost=70, ) PointsLedger.objects.create( user=user, change_type=PointsLedger.ChangeType.CONSUME, points_delta=-50, balance_after=150, ref_call=call, ) PointsLedger.objects.create( user=user, change_type=PointsLedger.ChangeType.REFUND, points_delta=10, balance_after=160, ref_call=call, ) PointsLedger.objects.create( user=other_user, change_type=PointsLedger.ChangeType.CONSUME, points_delta=-70, balance_after=929, ref_call=other_call, ) self.client.force_login(user) response = self.client.get("/dashboard") self.assertEqual(response.status_code, 200) self.assertEqual(response.context["balance"].points_balance, 160) self.assertEqual(response.context["balance"].ledger_balance, 160) self.assertEqual(response.context["recharge_total_amount"], Decimal("20.00")) self.assertEqual(response.context["recharge_points_total"], 200) self.assertEqual(response.context["consumed_points_total"], 50) self.assertEqual(response.context["refunded_points_total"], 10) self.assertEqual(response.context["net_used_points"], 40) self.assertContains(response, "R-OWN-PAID-503") self.assertContains(response, "title-standard") self.assertNotContains(response, "R-OTHER-PAID-503") self.assertNotContains(response, "other-alias") def test_recharge_records_require_login_and_only_show_current_user_orders(self): user = self.create_verified_user() other_user = self.create_verified_user() paid_order = self.create_recharge_order( user, order_no="R-OWN-RECHARGE-503", amount="30.00", points=300, ) self.create_recharge_order( user, order_no="R-OWN-PENDING-503", amount="88.00", points=880, status=RechargeOrder.Status.PENDING, ) self.create_recharge_order( other_user, order_no="R-OTHER-RECHARGE-503", amount="99.00", points=990, ) PointsLedger.objects.create( user=user, change_type=PointsLedger.ChangeType.RECHARGE, points_delta=300, balance_after=300, ref_order_id=paid_order.id, ) anonymous_response = self.client.get("/records/recharge") self.assertEqual(anonymous_response.status_code, 302) self.assertTrue(anonymous_response["Location"].startswith("/login?next=")) self.client.force_login(user) response = self.client.get("/records/recharge") self.assertEqual(response.status_code, 200) self.assertEqual(response.context["recharge_total_amount"], Decimal("30.00")) self.assertEqual(response.context["recharge_points_total"], 300) self.assertEqual(response.context["paid_recharge_count"], 1) self.assertContains(response, "R-OWN-RECHARGE-503") self.assertContains(response, "R-OWN-PENDING-503") self.assertNotContains(response, "R-OTHER-RECHARGE-503") def test_usage_records_require_login_and_only_show_current_user_usage(self): user = self.create_verified_user() other_user = self.create_verified_user() api_key, _raw_key = ApiKey.create_for_user(user, name="desktop") other_key, _other_raw_key = ApiKey.create_for_user(other_user, name="other") call = self.create_call_record( user, api_key, alias="title-standard", points_cost=40, status=CallRecord.Status.FAILED, ) other_call = self.create_call_record( other_user, other_key, alias="other-alias", points_cost=90, ) PointsLedger.objects.create( user=user, change_type=PointsLedger.ChangeType.CONSUME, points_delta=-40, balance_after=60, ref_call=call, ) PointsLedger.objects.create( user=user, change_type=PointsLedger.ChangeType.REFUND, points_delta=15, balance_after=75, ref_call=call, ) PointsLedger.objects.create( user=other_user, change_type=PointsLedger.ChangeType.CONSUME, points_delta=-90, balance_after=10, ref_call=other_call, ) anonymous_response = self.client.get("/records/usage") self.assertEqual(anonymous_response.status_code, 302) self.assertTrue(anonymous_response["Location"].startswith("/login?next=")) self.client.force_login(user) response = self.client.get("/records/usage") self.assertEqual(response.status_code, 200) self.assertEqual(response.context["consumed_points_total"], 40) self.assertEqual(response.context["refunded_points_total"], 15) self.assertEqual(response.context["net_used_points"], 25) self.assertContains(response, "title-standard") self.assertContains(response, api_key.key_prefix) self.assertContains(response, "-40") self.assertContains(response, "15") self.assertNotContains(response, "other-alias") self.assertNotContains(response, other_key.key_prefix)