accounting/test/test_rf.py

from appPublic.registerfunction import RegisterFunction

rf = RegisterFunction()

def get_module_database(module_name):
	return 'sage'

rf.register('get_module_database', get_module_database)

async def get_providers_by_orgid(sor, accounting_orgid):
	sql = """select * from organization a, orgtypes b
where a.parentid = ${accounting_orgid}$ and
	b.orgtypeid = 'provider' and
	a.id = b.orgid """
	recs = await sor.sqlExe(sql, {'accounting_orgid':accounting_orgid})
	return recs

rf.register('get_providers_by_orgid', get_providers_by_orgid)

async def get_resellers_by_orgid(sor, accounting_orgid):
	sql = """select * from organization a, orgtypes b
where a.parentid=${accounting_orgid}$ and
	a.id = b.orgid and
	b.orgtypeid = 'reseller'"""
	recs = await sor.sqlExe(sql, {'accounting_orgid':accounting_orgid})
	return recs

rf.register('get_resellers_by_orgid', get_resellers_by_orgid)

async def get_customers_by_orgid(sor, accounting_orgid):
	sql = """select * from organization a,
	(select orgid, count(*) cnt
		from orgtypes
		where orgtypeid in ('customer', 'personalcustomer', 'agencycustomer')
		group by orgid
	) b
where a.parentid=${accounting_orgid}$ and
	a.id = b.orgid and
	b.cnt > 0
	"""
	recs = await sor.sqlExe(sql, {'accounting_orgid':accounting_orgid})
	return recs

rf.register('get_customers_by_orgid', get_customers_by_orgid)