48 lines
1.1 KiB
Python
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())
|
|
|