accounting/test/recharge.py
2025-01-08 16:50:37 +08:00

48 lines
1.1 KiB
Python

import asyncio
from sqlor.dbpools import DBPools
from appPublic.jsonConfig import getConfig
from appPublic.dictObject import DictObject
from accounting.recharge import recharge_accounting
async def test1():
db = DBPools()
dbname = 'sage'
async with db.sqlorContext(dbname) as sor:
await recharge_accounting(sor, '4twr0MGHMZ4aFHYzQ9Lxh', 'RECHARGE', 'oEdlkg4SfX6JH5xQnCHW', '2025-01-10', 100)
async def test2():
rl = DictObject()
rl.customerid = '4zXVMkBCEaTmR0xwneUBX'
rl.recharge_date = '2024-09-21'
rl.recharge_amt = 100
rl.action = 'RECHARGE_REVERSE'
rl.orderid = '1'
rl.recharge_channel = 'alipay'
ra = RechargeAccounting(rl)
db = DBPools()
async with db.sqlorContext('sage') as sor:
await ra.accounting(sor)
async def test():
await test1()
await test2()
if __name__ == '__main__':
DBPools({
"sage":{
"driver":"aiomysql",
"async_mode":True,
"coding":"utf8",
"maxconn":100,
"dbname":"sage",
"kwargs":{
"user":"test",
"db":"sage",
"password":"QUZVcXg5V1p1STMybG5Ia6mX9D0v7+g=",
"host":"localhost"
}
}
})
asyncio.get_event_loop().run_until_complete(test1())