NeteaseUrl/tests/test_billing_scope.py

# -*- coding: utf-8 -*-
"""Billing.scope() 契约：作用域退出自动退还仍 open 的账本行。

设计前提（已由既有实现保证）：refund_open 只退「仍 open」的行——
settle/rename_to 消费过的行天然不在退款范围，因此自动退与成功路径
互不干扰，调用方无需手动取消。测试直接对真实账本（沙箱库）断言。
"""
import asyncio

import pytest

import auth_system as auth
from web.billing import Billing
from web.deps import SiteUser


@pytest.fixture()
def ledger_env(tmp_path, monkeypatch):
    monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path)
    monkeypatch.setattr(auth.db, 'DB_PATH', tmp_path / 'site.db')
    monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups')
    monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
    asyncio.run(auth.init_db())
    row = asyncio.run(auth.create_user('payer', 'Passw0rd12', email='p@example.com', credits=5))
    assert row is not None, '建号应成功'
    yield {'uid': int(row['id'])}
    # 无需清理：tmp_path 由 pytest 回收


async def _row(key: str):
    """取账本行，不存在则测试失败（供「行必须存在」的断言用）。"""
    row = await auth.ledger_get(key)
    assert row is not None, f'账本行 {key} 应存在'
    return row


async def _user_credits() -> int:
    u = await auth.get_user_by_name('payer')
    assert u is not None, '用户 payer 应存在'
    return int(u['duration_seconds'] or 0)


def _make_billing(ledger_env, key: str | None) -> Billing:
    user = SiteUser(id=ledger_env['uid'], username='payer', role='user', credits=5)
    return Billing(user=user, ledger_key=key, need=1, free=False)


async def test_scope_auto_refunds_open_row(ledger_env):
    """正常退出作用域：仍 open 的行被退（rejected），余额恢复。"""
    key = 'req-aaa'
    reserved = await auth.reserve_and_open(ledger_env['uid'], 1, key)
    assert reserved is not None
    assert (await _row(key))['status'] == 'open'

    billing = _make_billing(ledger_env, key)
    async with billing.scope():
        pass  # 业务体正常走完但未结算（等价于路由里的错误早退）

    assert (await _row(key))['status'] == 'rejected'
    assert await _user_credits() == 5


async def test_scope_noop_after_settle(ledger_env):
    """作用域内已结算：退出自动退为 no-op，交付行不受影响。"""
    key = 'req-bbb'
    await auth.reserve_and_open(ledger_env['uid'], 1, key)
    billing = _make_billing(ledger_env, key)

    async with billing.scope():
        await billing.settle(1, 'delivered')

    assert (await _row(key))['status'] == 'delivered'
    assert await _user_credits() == 4  # 恰好扣 1，未被二次退


async def test_scope_noop_after_rename(ledger_env):
    """作用域内账本已改挂 task_id：请求级行不存在，自动退为 no-op；
    任务级行保持 open 交给 worker 结算——与迁移前手工语义逐路径一致。"""
    key = 'req-ccc'
    await auth.reserve_and_open(ledger_env['uid'], 1, key)
    billing = _make_billing(ledger_env, key)

    async with billing.scope():
        ok = await billing.rename_to('task-xyz')
        assert ok is True

    assert await auth.ledger_get(key) is None
    moved = await _row('task-xyz')
    assert moved['status'] == 'open'


async def test_scope_refunds_on_exception_and_reraises(ledger_env):
    """业务体抛异常：先退款，异常原样穿透（路由自行转错误响应）。"""
    key = 'req-ddd'
    await auth.reserve_and_open(ledger_env['uid'], 1, key)
    billing = _make_billing(ledger_env, key)

    with pytest.raises(ValueError, match='boom'):
        async with billing.scope():
            raise ValueError('boom')

    assert (await _row(key))['status'] == 'rejected'


async def test_scope_free_billing_noop(ledger_env):
    """free/unlimited（ledger_key=None）：作用域进出零动作。"""
    billing = _make_billing(ledger_env, None)
    async with billing.scope():
        pass
    # 无账本行可动，不抛异常即为通过
    assert await _open_rows() == []


async def _open_rows():
    rows = await auth._q("SELECT task_id FROM billing_ledger WHERE status='open'")
    return [dict(r) for r in rows]