92 lines
3.7 KiB
Python
92 lines
3.7 KiB
Python
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.services.finance_service import FinanceService
|
|
|
|
|
|
def _db():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
db = Session(engine)
|
|
db.execute(text("""
|
|
CREATE TABLE users (id TEXT PRIMARY KEY, nickname TEXT, phone TEXT)
|
|
"""))
|
|
db.execute(text("""
|
|
CREATE TABLE token_payment_orders (
|
|
id TEXT, order_no TEXT PRIMARY KEY, user_id TEXT, plan_id TEXT,
|
|
payment_method TEXT, pay_type TEXT, pay_way TEXT, points_amount INTEGER,
|
|
price_cents INTEGER, status TEXT, provider TEXT, provider_order_id TEXT,
|
|
provider_order_no TEXT, provider_status TEXT, pay_message TEXT,
|
|
failure_reason TEXT, refund_status TEXT, created_at TEXT, updated_at TEXT,
|
|
paid_at TEXT, refunded_at TEXT
|
|
)
|
|
"""))
|
|
db.execute(text("""
|
|
CREATE TABLE payment_refunds (
|
|
id TEXT, refund_no TEXT, order_no TEXT, amount_cents INTEGER,
|
|
points_amount INTEGER, reason TEXT, status TEXT, provider_refund_no TEXT,
|
|
requested_by TEXT, failure_reason TEXT, created_at TEXT, updated_at TEXT,
|
|
completed_at TEXT
|
|
)
|
|
"""))
|
|
db.execute(text("""
|
|
CREATE TABLE invoice_applications (
|
|
id TEXT, order_no TEXT, user_id TEXT, amount_cents INTEGER, title TEXT,
|
|
invoice_type TEXT, tax_number TEXT, email TEXT, status TEXT,
|
|
invoice_no TEXT, invoice_url TEXT, remark TEXT, created_at TEXT,
|
|
updated_at TEXT, issued_at TEXT
|
|
)
|
|
"""))
|
|
db.execute(text("INSERT INTO users VALUES ('u1','测试用户','13800000000')"))
|
|
db.execute(text("""
|
|
INSERT INTO token_payment_orders VALUES (
|
|
'o1','AV1','u1','1','wechat','WECHAT','APP',2000000,1000,'paid','huihui',
|
|
'','','SUCCESS','secret-payment-message','','none','2026-09-08 12:00:00',
|
|
'2026-09-08 12:01:00','2026-09-08 12:01:00',NULL
|
|
)
|
|
"""))
|
|
db.execute(text("""
|
|
INSERT INTO invoice_applications VALUES (
|
|
'i1','AV1','u1',1000,'测试用户','personal','','u@example.com','pending',
|
|
'','','','2026-09-08 12:02:00','2026-09-08 12:02:00',NULL
|
|
)
|
|
"""))
|
|
db.commit()
|
|
return db
|
|
|
|
|
|
def test_finance_summary_and_orders_hide_provider_payment_payload():
|
|
db = _db()
|
|
try:
|
|
summary = FinanceService.summary(db)
|
|
assert summary == {
|
|
"paid_revenue": 10.0,
|
|
"paid_orders": 1,
|
|
"pending_orders": 0,
|
|
"processing_refunds": 0,
|
|
"pending_invoices": 1,
|
|
}
|
|
total, orders = FinanceService.list_orders(db, keyword="测试用户")
|
|
assert total == 1
|
|
assert orders[0]["price"] == 10.0
|
|
assert orders[0]["invoice_status"] == "pending"
|
|
assert "pay_message" not in orders[0]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_invoice_can_be_issued_and_pending_order_can_be_closed():
|
|
db = _db()
|
|
try:
|
|
FinanceService.update_invoice(db, "i1", status="issued", invoice_no="FP-001", invoice_url="", remark="")
|
|
assert db.execute(text("SELECT status, invoice_no FROM invoice_applications WHERE id='i1'" )).fetchone() == ("issued", "FP-001")
|
|
db.execute(text("""
|
|
INSERT INTO token_payment_orders
|
|
(id,order_no,user_id,plan_id,payment_method,pay_type,pay_way,points_amount,price_cents,status,provider,refund_status)
|
|
VALUES ('o2','AV2','u1','1','alipay','ALIPAY','H5',1,100,'pending','huihui','none')
|
|
"""))
|
|
db.commit()
|
|
FinanceService.close_order(db, "AV2", status="closed", reason="超时")
|
|
assert db.execute(text("SELECT status FROM token_payment_orders WHERE order_no='AV2'" )).scalar() == "closed"
|
|
finally:
|
|
db.close()
|