84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""core_ro.sum_trades_on_date 单测（A2 · sqlite 内存库验证 SQL 逻辑，MySQL 对照留灌库后）。"""
|
|
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from app.repository.core_ro import CoreReadOnlyRepository
|
|
|
|
|
|
@pytest.fixture()
|
|
def repo():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
CREATE TABLE core_trade (
|
|
trade_id VARCHAR(64) PRIMARY KEY,
|
|
customer_id VARCHAR(64),
|
|
product_id VARCHAR(64),
|
|
trade_type VARCHAR(16),
|
|
amount NUMERIC(18,2),
|
|
trade_status VARCHAR(16) DEFAULT 'confirmed',
|
|
traded_at TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
|
|
def insert(trade_id, amount, traded_at, trade_type="subscribe", status="confirmed", cid="C1"):
|
|
conn = engine.connect()
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type,"
|
|
" amount, trade_status, traded_at)"
|
|
" VALUES (:tid, :cid, 'P1', :tt, :amt, :st, :at)"
|
|
),
|
|
{"tid": trade_id, "cid": cid, "tt": trade_type, "amt": amount, "st": status, "at": traded_at},
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
yield CoreReadOnlyRepository(engine=engine), insert
|
|
engine.dispose()
|
|
|
|
|
|
DAY = date(2026, 9, 6)
|
|
|
|
|
|
def _at(hour, minute=0):
|
|
return datetime(2026, 9, 6, hour, minute)
|
|
|
|
|
|
def test_sum_only_confirmed_subscribe_redeem(repo):
|
|
r, insert = repo
|
|
insert("T1", "600000", _at(9)) # subscribe 计入
|
|
insert("T2", "400000", _at(10), "redeem") # redeem 计入
|
|
insert("T3", "999999", _at(11), status="pending") # 未确认不计
|
|
insert("T4", "999999", _at(12), "convert") # convert 不计
|
|
assert r.sum_trades_on_date("C1", DAY) == Decimal("1000000")
|
|
|
|
|
|
def test_sum_excludes_other_days(repo):
|
|
r, insert = repo
|
|
insert("T1", "500000", _at(9))
|
|
insert("T2", "500000", datetime(2026, 9, 5, 23)) # 前一日不计
|
|
insert("T3", "500000", datetime(2026, 9, 7, 0)) # 次日不计
|
|
assert r.sum_trades_on_date("C1", DAY) == Decimal("500000")
|
|
|
|
|
|
def test_sum_empty_is_zero(repo):
|
|
r, _ = repo
|
|
assert r.sum_trades_on_date("C1", DAY) == Decimal(0)
|
|
|
|
|
|
def test_sum_per_customer(repo):
|
|
r, insert = repo
|
|
insert("T1", "500000", _at(9), cid="C1")
|
|
insert("T2", "500000", _at(9), cid="C2")
|
|
assert r.sum_trades_on_date("C1", DAY) == Decimal("500000")
|
|
assert r.sum_trades_on_date("C2", DAY) == Decimal("500000")
|