accounting/test/recharge.py

48 lines
1.1 KiB
Python
Raw Normal View History

2025-01-08 16:50:37 +08:00
import asyncio
2024-09-21 14:07:27 +08:00
from sqlor.dbpools import DBPools
from appPublic.jsonConfig import getConfig
from appPublic.dictObject import DictObject
2025-01-08 16:50:37 +08:00
from accounting.recharge import recharge_accounting
2024-09-21 14:07:27 +08:00
2024-09-22 12:05:37 +08:00
async def test1():
2024-09-21 14:07:27 +08:00
db = DBPools()
2025-01-08 16:50:37 +08:00
dbname = 'sage'
async with db.sqlorContext(dbname) as sor:
await recharge_accounting(sor, '4twr0MGHMZ4aFHYzQ9Lxh', 'RECHARGE', 'oEdlkg4SfX6JH5xQnCHW', '2025-01-10', 100)
2024-09-21 14:07:27 +08:00
2024-09-22 12:05:37 +08:00
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()
2024-09-21 14:07:27 +08:00
if __name__ == '__main__':
2025-01-08 16:50:37 +08:00
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())