feat(points): 实现积分流水列表功能

- 后端新增 GET /api/v1/points/ledger 接口
- 前端新增积分流水列表页面
- 积分中心添加「查看流水」入口
- 重命名 AccountDeleteScreen 为 AccountDataScreen
- 流水列表支持分页加载和空状态展示
This commit is contained in:
ZL-Q
2026-04-28 17:19:08 +08:00
parent a83001de0d
commit 940c67e642
12 changed files with 794 additions and 70 deletions
@@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID, uuid4
@@ -22,12 +23,23 @@ class _FakeAccount:
version: int = 0
@dataclass
class _FakeLedgerRow:
id: UUID
direction: int
amount: int
balance_after: int
change_type: str
created_at: datetime
class _FakePointsRepository:
def __init__(self, *, usage_snapshot: PointsChargeSnapshot | None) -> None:
self.account = _FakeAccount()
self.usage_snapshot = usage_snapshot
self.appended_ledger: list[ApplyPointsChangeCommand] = []
self.appended_audit: list[AppendAuditLedgerCommand] = []
self.ledger_rows: list[_FakeLedgerRow] = []
self.claimed: bool = False
self.claim: RegisterBonusClaims | None = None
@@ -88,6 +100,16 @@ class _FakePointsRepository:
del email_hash
return self.claim
async def list_ledger(
self,
*,
user_id: UUID,
limit: int,
cursor: datetime | None = None,
) -> tuple[list[_FakeLedgerRow], bool]:
del user_id, cursor
return (self.ledger_rows[:limit], len(self.ledger_rows) > limit)
@pytest.mark.asyncio
async def test_consume_successful_run_points_writes_real_usage_to_audit() -> None:
@@ -225,3 +247,68 @@ async def test_grant_register_bonus_if_eligible_restores_balance_snapshot() -> N
assert repository.claimed is False
assert len(repository.appended_ledger) == 0
assert len(repository.appended_audit) == 0
@pytest.mark.asyncio
async def test_get_ledger_list_returns_items_and_next_cursor() -> None:
created_at = datetime(2026, 4, 28, 8, 30, tzinfo=timezone.utc)
repository = _FakePointsRepository(usage_snapshot=None)
repository.ledger_rows = [
_FakeLedgerRow(
id=uuid4(),
direction=1,
amount=60,
balance_after=160,
change_type="purchase",
created_at=created_at,
),
]
service = PointsService(repository=repository) # type: ignore[arg-type]
items, next_cursor, has_more = await service.get_ledger_list(
user_id=uuid4(),
limit=1,
)
assert has_more is False
assert next_cursor is None
assert len(items) == 1
assert items[0].amount == 60
assert items[0].balance_after == 160
assert items[0].change_type == "purchase"
assert items[0].created_at == created_at.isoformat()
@pytest.mark.asyncio
async def test_get_ledger_list_sets_next_cursor_when_more_rows_exist() -> None:
first_created_at = datetime(2026, 4, 28, 8, 30, tzinfo=timezone.utc)
second_created_at = datetime(2026, 4, 27, 8, 30, tzinfo=timezone.utc)
repository = _FakePointsRepository(usage_snapshot=None)
repository.ledger_rows = [
_FakeLedgerRow(
id=uuid4(),
direction=1,
amount=60,
balance_after=160,
change_type="purchase",
created_at=first_created_at,
),
_FakeLedgerRow(
id=uuid4(),
direction=-1,
amount=20,
balance_after=140,
change_type="consume",
created_at=second_created_at,
),
]
service = PointsService(repository=repository) # type: ignore[arg-type]
items, next_cursor, has_more = await service.get_ledger_list(
user_id=uuid4(),
limit=1,
)
assert has_more is True
assert len(items) == 1
assert next_cursor == first_created_at.isoformat()