1119 lines
44 KiB
Python
1119 lines
44 KiB
Python
import uuid
|
|
from decimal import Decimal
|
|
|
|
from allauth.account.models import EmailAddress
|
|
from django.conf import settings
|
|
from django.contrib import admin
|
|
from django.contrib.auth import get_user, get_user_model
|
|
from django.contrib.staticfiles import finders
|
|
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.ai.models import AiModel, ModelAlias
|
|
from apps.billing.models import (
|
|
CallRecord,
|
|
ExchangeRate,
|
|
PointsLedger,
|
|
PricingRule,
|
|
RechargeOrder,
|
|
SignupBonusGrant,
|
|
)
|
|
from apps.portal.models import DownloadRelease, ImportTemplate
|
|
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 create_model_alias(
|
|
self,
|
|
*,
|
|
alias: str,
|
|
operation_type: str = ModelAlias.OperationType.TITLE,
|
|
capabilities: list[str] | None = None,
|
|
api_type: str = AiModel.ApiType.CHAT,
|
|
url: str = "https://provider-secret.example/v1/chat/completions",
|
|
model_sku: str = "secret-sku-gpt-5.5",
|
|
) -> ModelAlias:
|
|
ai_model = AiModel.objects.create(
|
|
name=f"{alias}-{uuid.uuid4().hex[:8]}",
|
|
url=url,
|
|
model=model_sku,
|
|
api_type=api_type,
|
|
api_key_encrypted="encrypted-provider-key",
|
|
capabilities=capabilities if capabilities is not None else ["text"],
|
|
extra_body={"internal": "provider-extra-secret"},
|
|
is_active=True,
|
|
)
|
|
return ModelAlias.objects.create(
|
|
alias=alias,
|
|
operation_type=operation_type,
|
|
ai_model=ai_model,
|
|
)
|
|
|
|
def create_download_release(
|
|
self,
|
|
*,
|
|
version: str = "1.0.0",
|
|
platform: str = DownloadRelease.Platform.WINDOWS,
|
|
is_current: bool = True,
|
|
external_url: str = "https://download.example.com/cmhub-desktop.exe",
|
|
sha256: str = "a" * 64,
|
|
release_notes: str = "首版 Windows 客户端",
|
|
) -> DownloadRelease:
|
|
return DownloadRelease.objects.create(
|
|
platform=platform,
|
|
version=version,
|
|
is_current=is_current,
|
|
external_url=external_url,
|
|
sha256=sha256,
|
|
release_notes=release_notes,
|
|
)
|
|
|
|
def create_import_template(
|
|
self,
|
|
*,
|
|
name: str = "导入模板",
|
|
is_current: bool = True,
|
|
external_url: str = "https://download.example.com/import-template.xlsx",
|
|
file_name: str = "",
|
|
sha256: str = "e" * 64,
|
|
notes: str = "商品导入 Excel 模板",
|
|
) -> ImportTemplate:
|
|
return ImportTemplate.objects.create(
|
|
name=name,
|
|
is_current=is_current,
|
|
external_url=external_url,
|
|
file=file_name,
|
|
sha256=sha256,
|
|
notes=notes,
|
|
)
|
|
|
|
def assert_nav_link_active(self, response, *, href: str, label: str):
|
|
html = response.content.decode("utf-8")
|
|
self.assertIn(
|
|
f'<a class="btn btn-sm btn-primary" aria-current="page" href="{href}">{label}</a>',
|
|
html,
|
|
)
|
|
|
|
def assert_nav_link_inactive(self, response, *, href: str, label: str):
|
|
html = response.content.decode("utf-8")
|
|
self.assertIn(
|
|
f'<a class="btn btn-sm btn-outline-secondary" href="{href}">{label}</a>',
|
|
html,
|
|
)
|
|
|
|
def assert_portal_branding(self, response):
|
|
self.assertContains(response, "虾皮圈")
|
|
self.assertNotContains(response, ">cmhub</a>")
|
|
self.assertNotContains(response, " - cmhub</title>")
|
|
self.assertNotContains(response, "cmhub AI 电商生成台")
|
|
self.assertNotContains(response, "cmhub API Key")
|
|
|
|
def test_signup_rate_limit_is_configured(self):
|
|
self.assertEqual(settings.ACCOUNT_RATE_LIMITS["signup"], "20/m/ip")
|
|
|
|
def test_signup_grants_100_points_writes_signup_bonus_ledger_and_can_login(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, 100)
|
|
self.assertEqual(SignupBonusGrant.objects.filter(user=user).count(), 1)
|
|
ledger = PointsLedger.objects.get(
|
|
user=user,
|
|
change_type=PointsLedger.ChangeType.SIGNUP_BONUS,
|
|
)
|
|
self.assertEqual(ledger.points_delta, 100)
|
|
self.assertEqual(ledger.balance_after, 100)
|
|
self.assertEqual(len(mail.outbox), 0)
|
|
self.assertTrue(get_user(self.client).is_authenticated)
|
|
|
|
dashboard_response = self.client.get("/dashboard")
|
|
self.assertEqual(dashboard_response.status_code, 200)
|
|
self.assertEqual(dashboard_response.context["balance"].points_balance, 100)
|
|
self.assertEqual(dashboard_response.context["balance"].ledger_balance, 100)
|
|
self.assertContains(dashboard_response, "100")
|
|
|
|
def test_user_can_login_without_verified_email(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,
|
|
)
|
|
|
|
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)
|
|
|
|
def test_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_homepage_is_public_and_shows_anonymous_onboarding_without_release(self):
|
|
response = self.client.get("/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.resolver_match.url_name, "portal-home")
|
|
self.assertContains(response, "虾皮圈 AI 电商生成台")
|
|
self.assert_portal_branding(response)
|
|
self.assertContains(response, "注册领 100 点")
|
|
self.assertContains(response, "注册送 100 点")
|
|
self.assertContains(response, "登录")
|
|
self.assertContains(response, "客户端暂未发布")
|
|
self.assertContains(response, "暂未发布")
|
|
self.assertNotContains(response, "下载导入模板")
|
|
self.assertNotContains(response, "/login?next=")
|
|
|
|
def test_homepage_shows_current_download_release(self):
|
|
release = self.create_download_release(
|
|
version="1.2.3",
|
|
external_url="https://download.example.com/cmhub-1.2.3.exe",
|
|
sha256="b" * 64,
|
|
release_notes="修复下载入口并补充 SHA256",
|
|
)
|
|
|
|
response = self.client.get("/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context["current_release"], release)
|
|
self.assertContains(response, "Windows 客户端 1.2.3")
|
|
self.assertContains(response, "https://download.example.com/cmhub-1.2.3.exe")
|
|
self.assertContains(response, "b" * 64)
|
|
self.assertContains(response, "修复下载入口并补充 SHA256")
|
|
self.assertContains(response, "未知发布者")
|
|
self.assertContains(response, "填入虾皮圈 API Key 使用")
|
|
self.assertNotContains(response, "填入 cmhub API Key 使用")
|
|
|
|
def test_portal_pages_use_shopee_circle_branding(self):
|
|
user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=10)
|
|
|
|
public_paths = ("/", "/signup", "/login")
|
|
for path in public_paths:
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assert_portal_branding(response)
|
|
|
|
self.client.force_login(user)
|
|
authenticated_paths = (
|
|
"/dashboard",
|
|
"/recharge",
|
|
"/apikeys",
|
|
"/models",
|
|
"/records/recharge",
|
|
"/records/usage",
|
|
"/logout",
|
|
)
|
|
for path in authenticated_paths:
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assert_portal_branding(response)
|
|
|
|
def test_homepage_shows_current_import_template_download_link(self):
|
|
self.create_download_release()
|
|
import_template = self.create_import_template(
|
|
name="蝦皮圈導入模板",
|
|
external_url="https://download.example.com/shopee-import-template.xlsx",
|
|
sha256="f" * 64,
|
|
)
|
|
|
|
response = self.client.get("/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context["current_import_template"], import_template)
|
|
self.assertEqual(
|
|
response.context["current_import_template_download_url"],
|
|
"https://download.example.com/shopee-import-template.xlsx",
|
|
)
|
|
self.assertContains(response, "下载客户端")
|
|
self.assertContains(response, "下载导入模板")
|
|
self.assertContains(
|
|
response,
|
|
"https://download.example.com/shopee-import-template.xlsx",
|
|
)
|
|
|
|
def test_homepage_builds_absolute_import_template_file_url_without_local_path(self):
|
|
self.create_import_template(
|
|
external_url="",
|
|
file_name="import_templates/cmhub-import-template.xlsx",
|
|
)
|
|
|
|
response = self.client.get("/", secure=True)
|
|
|
|
expected_url = "https://testserver/media/import_templates/cmhub-import-template.xlsx"
|
|
response_body = response.content.decode("utf-8")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(
|
|
response.context["current_import_template_download_url"],
|
|
expected_url,
|
|
)
|
|
self.assertContains(response, expected_url)
|
|
self.assertNotIn(str(settings.MEDIA_ROOT), response_body)
|
|
self.assertNotIn("MEDIA_ROOT", response_body)
|
|
|
|
def test_homepage_prefers_import_template_external_url_over_file(self):
|
|
self.create_import_template(
|
|
external_url="https://cdn.example.com/current-template.xlsx",
|
|
file_name="import_templates/local-template.xlsx",
|
|
)
|
|
|
|
response = self.client.get("/", secure=True)
|
|
|
|
self.assertEqual(
|
|
response.context["current_import_template_download_url"],
|
|
"https://cdn.example.com/current-template.xlsx",
|
|
)
|
|
self.assertContains(response, "https://cdn.example.com/current-template.xlsx")
|
|
self.assertNotContains(response, "/media/import_templates/local-template.xlsx")
|
|
|
|
def test_homepage_shows_dashboard_entry_for_authenticated_user(self):
|
|
user = self.create_verified_user()
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.get("/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "进入控制台")
|
|
self.assertNotContains(response, "注册领 100 点")
|
|
|
|
def test_download_release_only_keeps_one_current_per_platform(self):
|
|
old_release = self.create_download_release(version="1.0.0", sha256="c" * 64)
|
|
new_release = self.create_download_release(version="1.1.0", sha256="d" * 64)
|
|
|
|
old_release.refresh_from_db()
|
|
new_release.refresh_from_db()
|
|
|
|
self.assertFalse(old_release.is_current)
|
|
self.assertTrue(new_release.is_current)
|
|
|
|
def test_download_release_admin_requires_sha256_and_size_bytes(self):
|
|
form_class = admin.site._registry[DownloadRelease].form
|
|
base_data = {
|
|
"platform": DownloadRelease.Platform.WINDOWS,
|
|
"version": "1.2.3",
|
|
"external_url": "https://download.example.com/cmhub-1.2.3.zip",
|
|
"release_notes": "发布说明",
|
|
}
|
|
|
|
missing_sha256_form = form_class(data={**base_data, "size_bytes": "123456"})
|
|
missing_size_form = form_class(data={**base_data, "sha256": "a" * 64})
|
|
|
|
self.assertFalse(missing_sha256_form.is_valid())
|
|
self.assertIn("sha256", missing_sha256_form.errors)
|
|
self.assertFalse(missing_size_form.is_valid())
|
|
self.assertIn("size_bytes", missing_size_form.errors)
|
|
|
|
def test_download_release_admin_saves_valid_release_metadata(self):
|
|
form_class = admin.site._registry[DownloadRelease].form
|
|
form = form_class(
|
|
data={
|
|
"platform": DownloadRelease.Platform.WINDOWS,
|
|
"version": "1.2.3",
|
|
"external_url": "https://download.example.com/cmhub-1.2.3.zip",
|
|
"sha256": "b" * 64,
|
|
"size_bytes": "123456",
|
|
"release_notes": "发布说明",
|
|
}
|
|
)
|
|
|
|
self.assertTrue(form.is_valid(), form.errors)
|
|
release = form.save()
|
|
self.assertEqual(release.sha256, "b" * 64)
|
|
self.assertEqual(release.size_bytes, 123456)
|
|
|
|
def test_import_template_only_keeps_one_current(self):
|
|
old_template = self.create_import_template(
|
|
name="旧模板",
|
|
sha256="1" * 64,
|
|
)
|
|
new_template = self.create_import_template(
|
|
name="新模板",
|
|
sha256="2" * 64,
|
|
)
|
|
|
|
old_template.refresh_from_db()
|
|
new_template.refresh_from_db()
|
|
|
|
self.assertFalse(old_template.is_current)
|
|
self.assertTrue(new_template.is_current)
|
|
|
|
def test_import_template_admin_exposes_download_fields(self):
|
|
registered_admin = admin.site._registry[ImportTemplate]
|
|
|
|
self.assertIn("is_current", registered_admin.list_display)
|
|
self.assertIn("download_source", registered_admin.list_display)
|
|
self.assertIn("is_current", registered_admin.list_filter)
|
|
template_fields = registered_admin.fieldsets[0][1]["fields"]
|
|
download_fields = registered_admin.fieldsets[1][1]["fields"]
|
|
self.assertIn("name", template_fields)
|
|
self.assertIn("file", download_fields)
|
|
self.assertIn("external_url", download_fields)
|
|
self.assertIn("sha256", download_fields)
|
|
|
|
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_models_page_requires_session_login(self):
|
|
response = self.client.get("/models")
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertTrue(response["Location"].startswith("/login?next="))
|
|
|
|
def test_models_page_lists_public_aliases_prices_and_unpriced_state(self):
|
|
user = self.create_verified_user()
|
|
title_alias = self.create_model_alias(alias="title-standard", capabilities=["text"])
|
|
image_alias = self.create_model_alias(
|
|
alias="image-edit",
|
|
operation_type=ModelAlias.OperationType.IMAGE,
|
|
capabilities=["image", "vision"],
|
|
api_type=AiModel.ApiType.IMAGES_EDITS,
|
|
url="https://provider-secret.example/v1/images/edits",
|
|
model_sku="secret-sku-image-2",
|
|
)
|
|
vision_alias = self.create_model_alias(
|
|
alias="vision-standard",
|
|
operation_type=ModelAlias.OperationType.VISION,
|
|
capabilities=["text", "vision"],
|
|
model_sku="secret-sku-vision",
|
|
)
|
|
PricingRule.objects.create(
|
|
operation_type=title_alias.operation_type,
|
|
alias=title_alias.alias,
|
|
resolution="",
|
|
points_cost=2,
|
|
)
|
|
PricingRule.objects.create(
|
|
operation_type=image_alias.operation_type,
|
|
alias=image_alias.alias,
|
|
resolution="1k",
|
|
points_cost=12,
|
|
is_active=False,
|
|
)
|
|
PricingRule.objects.create(
|
|
operation_type=vision_alias.operation_type,
|
|
alias=vision_alias.alias,
|
|
resolution="",
|
|
points_cost=3,
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.get("/models")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.context["models"]), 3)
|
|
self.assertContains(response, "可用模型")
|
|
self.assertContains(response, "title-standard")
|
|
self.assertContains(response, "生成标题")
|
|
self.assertContains(response, "默认")
|
|
self.assertContains(response, "2 点")
|
|
self.assertContains(response, "image-edit")
|
|
self.assertContains(response, "生成图片")
|
|
self.assertContains(response, "vision-standard")
|
|
self.assertContains(response, "图片理解")
|
|
self.assertContains(response, "3 点")
|
|
self.assertContains(response, "需要")
|
|
self.assertContains(response, "暂未定价")
|
|
self.assertNotContains(response, "secret-sku")
|
|
self.assertNotContains(response, "provider-secret.example")
|
|
self.assertNotContains(response, "encrypted-provider-key")
|
|
self.assertNotContains(response, "provider-extra-secret")
|
|
self.assertNotContains(response, "api_key")
|
|
self.assertNotContains(response, "api_key_encrypted")
|
|
self.assertNotContains(response, "extra_body")
|
|
|
|
def test_authenticated_nav_highlights_current_page_only(self):
|
|
user = self.create_verified_user()
|
|
self.client.force_login(user)
|
|
cases = (
|
|
("/dashboard", "/dashboard", "控制台"),
|
|
("/recharge", "/recharge", "充值"),
|
|
("/apikeys", "/apikeys", "API Key"),
|
|
("/models", "/models", "可用模型"),
|
|
("/records/recharge", "/records/recharge", "充值记录"),
|
|
("/records/usage", "/records/usage", "点数记录"),
|
|
)
|
|
|
|
for path, href, label in cases:
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assert_nav_link_active(response, href=href, label=label)
|
|
if path != "/recharge":
|
|
self.assert_nav_link_inactive(
|
|
response,
|
|
href="/recharge",
|
|
label="充值",
|
|
)
|
|
|
|
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["credited_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_recharge_records_paginate_more_than_page_size(self):
|
|
user = self.create_verified_user()
|
|
other_user = self.create_verified_user()
|
|
self.create_recharge_order(
|
|
other_user,
|
|
order_no="R-PAGE-OTHER-505",
|
|
amount="99.00",
|
|
points=990,
|
|
)
|
|
for index in range(55):
|
|
self.create_recharge_order(
|
|
user,
|
|
order_no=f"R-PAGE-{index:03d}",
|
|
amount="1.00",
|
|
points=10,
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
first_page = self.client.get("/records/recharge")
|
|
second_page = self.client.get("/records/recharge?page=2")
|
|
|
|
self.assertEqual(first_page.status_code, 200)
|
|
self.assertEqual(first_page.context["records_total_count"], 55)
|
|
self.assertEqual(first_page.context["page_obj"].number, 1)
|
|
self.assertEqual(len(first_page.context["recharge_orders"]), 50)
|
|
self.assertContains(first_page, "共 55 条，每页 50 条")
|
|
self.assertContains(first_page, "R-PAGE-054")
|
|
self.assertNotContains(first_page, "R-PAGE-000")
|
|
self.assertNotContains(first_page, "R-PAGE-OTHER-505")
|
|
|
|
self.assertEqual(second_page.status_code, 200)
|
|
self.assertEqual(second_page.context["page_obj"].number, 2)
|
|
self.assertEqual(len(second_page.context["recharge_orders"]), 5)
|
|
self.assertContains(second_page, "R-PAGE-000")
|
|
self.assertNotContains(second_page, "R-PAGE-054")
|
|
self.assertNotContains(second_page, "R-PAGE-OTHER-505")
|
|
|
|
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.SIGNUP_BONUS,
|
|
points_delta=100,
|
|
balance_after=100,
|
|
reason="new_user_registration",
|
|
)
|
|
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, "注册赠点")
|
|
self.assertContains(response, "100")
|
|
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)
|
|
|
|
def test_usage_records_paginate_more_than_page_size(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")
|
|
other_call = self.create_call_record(
|
|
other_user,
|
|
other_key,
|
|
alias="other-page-alias",
|
|
points_cost=99,
|
|
)
|
|
PointsLedger.objects.create(
|
|
user=other_user,
|
|
change_type=PointsLedger.ChangeType.CONSUME,
|
|
points_delta=-99,
|
|
balance_after=0,
|
|
ref_call=other_call,
|
|
)
|
|
for index in range(55):
|
|
call = self.create_call_record(
|
|
user,
|
|
api_key,
|
|
alias=f"title-page-{index:03d}",
|
|
points_cost=1,
|
|
)
|
|
PointsLedger.objects.create(
|
|
user=user,
|
|
change_type=PointsLedger.ChangeType.CONSUME,
|
|
points_delta=-1,
|
|
balance_after=100 - index,
|
|
ref_call=call,
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
first_page = self.client.get("/records/usage")
|
|
second_page = self.client.get("/records/usage?page=2")
|
|
|
|
self.assertEqual(first_page.status_code, 200)
|
|
self.assertEqual(first_page.context["records_total_count"], 55)
|
|
self.assertEqual(first_page.context["page_obj"].number, 1)
|
|
self.assertEqual(len(first_page.context["usage_entries"]), 50)
|
|
self.assertContains(first_page, "共 55 条，每页 50 条")
|
|
self.assertContains(first_page, "title-page-054")
|
|
self.assertNotContains(first_page, "title-page-000")
|
|
self.assertNotContains(first_page, "other-page-alias")
|
|
|
|
self.assertEqual(second_page.status_code, 200)
|
|
self.assertEqual(second_page.context["page_obj"].number, 2)
|
|
self.assertEqual(len(second_page.context["usage_entries"]), 5)
|
|
self.assertContains(second_page, "title-page-000")
|
|
self.assertNotContains(second_page, "title-page-054")
|
|
self.assertNotContains(second_page, "other-page-alias")
|
|
|
|
def test_recharge_page_requires_login_and_shows_form(self):
|
|
user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=25)
|
|
|
|
anonymous_response = self.client.get("/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("/recharge")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context["balance"].points_balance, 25)
|
|
self.assertContains(response, "充值金额")
|
|
self.assertContains(response, "创建订单")
|
|
self.assertContains(response, "微信")
|
|
self.assertNotContains(response, "支付宝")
|
|
self.assertContains(response, "portal/vendor/bootstrap/bootstrap.min.css")
|
|
self.assertContains(response, "portal/vendor/qrcode/qrcode.js")
|
|
self.assertContains(response, "QRCode.toCanvas")
|
|
self.assertNotContains(response, "cdn.jsdelivr.net")
|
|
self.assertIsNotNone(
|
|
finders.find("portal/vendor/bootstrap/bootstrap.min.css")
|
|
)
|
|
self.assertIsNotNone(finders.find("portal/vendor/qrcode/qrcode.js"))
|
|
|
|
@override_settings(PAYMENT_CALLBACK_MODE="mock")
|
|
def test_recharge_page_post_creates_pending_order_without_crediting_wallet_or_ledger(self):
|
|
user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=5)
|
|
ExchangeRate.objects.create(
|
|
currency="CNY",
|
|
points_per_unit=Decimal("10.0000"),
|
|
effective_from=timezone.now(),
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.post(
|
|
"/recharge",
|
|
{"amount": "20.00", "pay_method": RechargeOrder.PayMethod.WEIXIN},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertTrue(response["Location"].startswith("/recharge?order_no="))
|
|
order = RechargeOrder.objects.get(user=user)
|
|
self.assertEqual(order.amount_money, Decimal("20.00"))
|
|
self.assertEqual(order.exchange_rate, Decimal("10.0000"))
|
|
self.assertEqual(order.points_granted, 200)
|
|
self.assertEqual(order.status, RechargeOrder.Status.PENDING)
|
|
self.assertTrue(order.code_url.startswith("weixin://wxpay/cmhub-mock"))
|
|
self.assertIsNotNone(order.expires_at)
|
|
wallet = UserWallet.objects.get(user=user)
|
|
self.assertEqual(wallet.points_balance, 5)
|
|
self.assertFalse(
|
|
PointsLedger.objects.filter(
|
|
user=user,
|
|
ref_order_id=order.id,
|
|
).exists()
|
|
)
|
|
|
|
follow_response = self.client.get(response["Location"])
|
|
|
|
self.assertEqual(follow_response.status_code, 200)
|
|
self.assertEqual(follow_response.context["current_order"], order)
|
|
self.assertContains(follow_response, order.order_no)
|
|
self.assertContains(follow_response, "weixin://wxpay/cmhub-mock")
|
|
self.assertTrue(follow_response.context["is_mock_payment_mode"])
|
|
self.assertContains(follow_response, "当前为支付测试模式")
|
|
self.assertContains(follow_response, "不能使用微信或支付宝真实付款")
|
|
self.assertContains(follow_response, 'data-status-url="/api/v1/recharge/status"')
|
|
self.assertContains(follow_response, "data-recharge-order")
|
|
|
|
@override_settings(PAYMENT_CALLBACK_MODE="sdk")
|
|
def test_recharge_page_does_not_show_mock_notice_in_sdk_mode(self):
|
|
user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=0)
|
|
order = self.create_recharge_order(
|
|
user,
|
|
order_no="R-SDK-RECHARGE-PAGE-504",
|
|
amount="20.00",
|
|
points=200,
|
|
status=RechargeOrder.Status.PENDING,
|
|
)
|
|
order.code_url = "weixin://wxpay/bizpayurl?pr=real-native-ticket"
|
|
order.save(update_fields=("code_url", "updated_at"))
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.get(f"/recharge?order_no={order.order_no}")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertFalse(response.context["is_mock_payment_mode"])
|
|
self.assertContains(response, "weixin://wxpay/bizpayurl")
|
|
self.assertNotContains(response, "当前为支付测试模式")
|
|
self.assertNotContains(response, "不能使用微信或支付宝真实付款")
|
|
|
|
def test_recharge_page_current_order_only_shows_current_user_order(self):
|
|
user = self.create_verified_user()
|
|
other_user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=0)
|
|
other_order = self.create_recharge_order(
|
|
other_user,
|
|
order_no="R-OTHER-RECHARGE-PAGE-504",
|
|
amount="50.00",
|
|
points=500,
|
|
status=RechargeOrder.Status.PENDING,
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.get(f"/recharge?order_no={other_order.order_no}")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIsNone(response.context["current_order"])
|
|
self.assertNotContains(response, other_order.order_no)
|
|
|
|
@override_settings(RECHARGE_MAX_AMOUNT_CNY="100.00")
|
|
def test_recharge_page_rejects_amount_above_configured_maximum(self):
|
|
user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=0)
|
|
ExchangeRate.objects.create(
|
|
currency="CNY",
|
|
points_per_unit=Decimal("10.0000"),
|
|
effective_from=timezone.now(),
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.post(
|
|
"/recharge",
|
|
{"amount": "100.01", "pay_method": RechargeOrder.PayMethod.WEIXIN},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "单笔充值金额不能超过 100.00 CNY")
|
|
self.assertFalse(RechargeOrder.objects.filter(user=user).exists())
|
|
|
|
def test_recharge_page_rejects_hidden_alipay_submit(self):
|
|
user = self.create_verified_user()
|
|
UserWallet.objects.create(user=user, points_balance=0)
|
|
ExchangeRate.objects.create(
|
|
currency="CNY",
|
|
points_per_unit=Decimal("10.0000"),
|
|
effective_from=timezone.now(),
|
|
)
|
|
self.client.force_login(user)
|
|
|
|
response = self.client.post(
|
|
"/recharge",
|
|
{"amount": "20.00", "pay_method": RechargeOrder.PayMethod.ALIPAY},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "选择一个有效的选项")
|
|
self.assertFalse(RechargeOrder.objects.filter(user=user).exists())
|
|
|
|
def test_recharge_page_post_is_csrf_protected(self):
|
|
user = self.create_verified_user()
|
|
csrf_client = Client(enforce_csrf_checks=True)
|
|
csrf_client.force_login(user)
|
|
|
|
response = csrf_client.post(
|
|
"/recharge",
|
|
{"amount": "20.00", "pay_method": RechargeOrder.PayMethod.WEIXIN},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertFalse(RechargeOrder.objects.filter(user=user).exists())
|