def printSystemNotifications(): import HTMLParser notifications = call( 'SystemNotification.getNotifications', last_notifications=100, only_unread=True, ) notifications.sort( key = lambda row: -row['notification_id'] ) for row in notifications: print 'ID: %s'%row['notification_id'] print 'Date: %s'%row['date'] print 'Type: %s'%row['type'] link = row['links'] if link: if not '://' in link: if not link.startswith('/'): link = '/' + link link = 'http://127.0.0.1' + link print 'Link: %s'%link print 'Message:\n%s'%HTMLParser.HTMLParser().unescape(row['message']) print '_______________________________________________________\n'
1 Comment
Nima Tehrani
#!/usr/bin/env python2.7
import requests
import json
url = 'http://127.0.0.1:1237'
auth_type, auth_name, auth_pass = 'ADMIN', 'test_admin', 'test_password'
#auth_type, auth_name, auth_pass = 'NORMAL_USER', 'test_user', 'test_password'
auth_remoteaddr = '127.0.0.1'
#################################################
def baseCall(method, **params):
response = requests.post(
url,
data=json.dumps({
'method': method,
'params': params,
#'jsonrpc': '2.0',
#'id': 0,
}),
headers={
'content-type': 'application/json'
},
).json()
error = response.get('error')
if error:
raise Exception(error)
return response['result']
def call(method, **params):
params.update(dict(
auth_type = auth_type,
auth_name = auth_name,
auth_pass = auth_pass,
auth_remoteaddr = auth_remoteaddr,
date_type = 'gregorian',
))
return baseCall(method, **params)
def callSaveJson(fpath, method, **params):
data = call(method, **params)
text = json.dumps(
data,
sort_keys=True,
indent=2,
ensure_ascii=False,
).encode('utf-8')
open(fpath, 'w').write(text)
return data
def testUssdPayment():
call(
'online_payment.ussdPaymentVerifiedForUser',
user_id = 107455,
amount = 1234,
comment = 'saeed'
)
#################################################
def getOnlineUsers():
internet_onlines, voip_onlines = callSaveJson(
'/tmp/onlines.json',
'report.getOnlineUsers',
normal_sort_by='user_id', normal_desc=False, voip_sort_by='user_id', voip_desc=False, conds={},
)
print '%4d Internet Onlines'%len(internet_onlines)
print '%4d VoIP Onlines'%len(voip_onlines)
def updateUsernamePassword(user_id, username, password):
call(
'user.updateUserAttrs',
user_id=str(user_id),
attrs={
'normal_user_spec': {
'normal_username': username,
'normal_password': password,
}
},
to_del_attrs=[],
)
def getLockedUsers():
params = {
'conds': {
'lock': '',
},
'from': 0,
'to': 10,
'order_by': 'user_id',
'desc': False,
}
result = call(
'user.searchUser',
**params
)
total_count = result[0]
total_credit = result[1]
user_ids = result[2]
print total_count, len(user_ids)
print user_ids
def getUsernameByFailedUserIP(ip):
return call(
'util.getUsernameByFailedUserIP',
ip=ip,
)
def test_getInOutUsages():
params = {
'conds': {
'logout_time_from': '2',
'logout_time_from_unit': 'minutes',
},
'from': 0,
'to': 5,
'order_by': 'sum',
}
pprint(call(
'report.getInOutUsages',
**params
))
def test_prefix_usage():
params = {
'conds': {
'logout_time_from': '1',
'logout_time_from_unit': 'minutes',
'group_by': 'prefix_name',
},
'from': 0,
'to': 5,
'order_by': 'count',
'desc': True,
}
pprint(call(
'report.getPrefixNameUsage',
**params
))
def test_credit_changes():
params = {
'conds': {
"user_view": "",
#'action': [
# 'ADD_USER',
# 'CHANGE_CREDIT',
# 'RENEW',
#],
#'change_time_from': '5',
#'change_time_from_unit': 'hours',
},
'from': 0,
'to': 10,
'sort_by': 'change_time',
'desc': True,
}
pprint(call(
'report.getCreditChanges',
**params
))
def test_save_credit_changes():
params = {
'conds': {
'action': [
'ADD_USER',
'CHANGE_CREDIT',
'RENEW',
],
'change_time_from': '5',
'change_time_from_unit': 'hours',
},
'cols': [
'change_time_formatted',
'admin_name',
'change_time',
'credit_change_id',
'action',
'action_text',
'admin_id',
'comment',
'per_user_credit',
'remote_addr',
'isp_credit',
'isp_id',
'user_ids',
'isp_name',
'name',
'serial',
'user_id',
'username',
'before_credit',
'group',
'group_id',
],
'sort_by': 'change_time',
'desc': True,
'output_type': 'csv',
}
pprint(call(
'report.saveCreditChanges',
**params
))
def test_user_deposit_changes():
params = {
'conds': {
#'deposit': 1000,
#'deposit_op': '<',
#'action': [
# 'RENEW',
#],
'change_time_from': '5',
'change_time_from_unit': 'hours',
'isp_names': [
'Dejkhah',
'Hemat',
'KH',
],
},
'from': 0,
'to': 10,
'sort_by': 'change_time',
'desc': True,
}
pprint(call(
'report.getUserDepositChanges',
**params
))
def test_user_audit_logs():
params = {
'conds': {
},
'from': 0,
'to': 10,
'sort_by': 'change_time',
'desc': True,
}
pprint(call(
'report.getUserAuditLogs',
**params
))
def test_isp_deposit_changes():
params = {
'conds': {
'change_time_from': '5',
'change_time_from_unit': 'hours',
'isp_names': [
'Dejkhah',
'Hemat',
'KH',
],
},
'from': 0,
'to': 10,
'sort_by': 'isp_deposit_change_time',
'desc': True,
}
pprint(call(
'report.getISPDepositChangeLogs',
**params
))
def test_temp_extend_users():
params = {
'conds': {
#'temporary_extend_date_from': '10',
#'temporary_extend_date_from_unit': 'days',
},
'from': 0,
'to': 10,
'sort_by': 'temporary_extend_date',
'desc': True,
}
pprint(call(
'report.getTemporaryExtendLogs',
**params
))
def test_management_summary():
params = {
'conds': {
'view_period': 'daily',
'included_objects': [
'isp',
'isp_mapped_user',
'group',
'ras',
'ras_group',
'ras_isp',
'service',
],
'report_targets': [
'duration',
'credit',
'in_bytes',
'out_bytes',
],
'login_time_from': '5',
'login_time_from_unit': 'minutes',
},
'from': 0,
'to': 10,
'sort_by': '',
'desc': True,
}
pprint(call(
'report.getManagementSummaryReport',
**params
))
def test_system_audit_logs():
params = {
'conds': {
'change_time_from': '5',
'change_time_from_unit': 'hours',
},
'from': 0,
'to': 10,
'sort_by': 'change_time',
'desc': True,
}
pprint(call(
'report.getSystemAuditLogs',
**params
))
def test_online_payment_report():
params = {
'conds': {
#'payment_date_from': '5',
#'payment_date_from_unit': 'hours',
},
'from': 0,
'to': 10,
'sort_by': 'payment_date',
'desc': True,
}
pprint(call(
'report.getOnlinePaymentReport',
**params
))
def test_save_connection_usage():
params = {
'report_type': 'inout_usages',
'conds': {
'login_time_from': '5',
'login_time_from_unit': 'minutes',
},
'cols': [
'creation_date',
'user_id',
'name',
'user_name',
'isp_name',
'group_name',
'serial',
],
'sort_by': 'sum',
'output_type': 'csv',
}
pprint(call(
'report.saveConnectionUsages',
**params
))
def getNotifications():
params = {
}
pprint(call(
'SystemNotification.getNotifications',
**params
))
def changeNotificationStatus():
pprint(call(
'SystemNotification.changeNotificationStatus',
notification_id=4,
read=True,
))
def removeNotifications():
pprint(call(
'SystemNotification.removeNotifications',
notification_ids=[
9,
],
))
def getRemainingByteDuration():
pprint(call(
'telephony_support.getRemainingByteDuration',
user_id=786,
))
def test_adminSearchSentMessage():
params = {
'conds': {
'msg_types': [
'SMS',
],
},
'from': 0,
'to': 10,
'order_by': 'message_id',
'desc': True,
}
pprint(call(
'mc.adminSearchSentMessage',
**params
))
def test_updateUserAttrs(user_id):
from time import strptime, mktime
## real_first_login: no change, no delete
## delete assign_dns_1, deletes both
## update assign_dns_1, requires assign_dns_2
call(
'user.updateUserAttrs',
user_id=str(user_id),
attrs={
'limit_usage_days': 5,
#'save_bw_usage': '',# true_if_exists
#'allow_user_disconnect': '',# true_if_exists
#'deny_change_password': '',# true_if_exists
#'auto_renew': '',# true_if_exists
#'reset_negative_credit_on_renew': '',# true_if_exists
##'renew_do_not_reset_credit': '',# true_if_exists
#'renew_do_not_reset_credit_on_state_package': '',# true_if_exists
#'renew_do_not_reset_credit_on_state_recharged': '',# true_if_exists
#'auto_renew_on_credit_finish': '',# true_if_exists
#'auto_renew_do_not_reset_credit': '',# true_if_exists
#'auto_recharge': '',# true_if_exists
#'allow_recharge_by_voucher': '',# true_if_exists
#'prevent_web_login': '',# true_if_exists
#'enable_failed_login': '',# true_if_exists
#'ignore_idle_threshold_static_ip': '',# true_if_exists
#'ignore_idle_threshold': '',# true_if_exists
#'caller_id_bind_on_login': '',# true_if_exists
#'limit_mac_bind_on_login': '',# true_if_exists
#'limit_port_bind_on_login': '',# true_if_exists
#'limit_ras_bind_on_login': '',# true_if_exists
#'limit_station_ip_bind_on_login': '',# true_if_exists
##'is_deposit_transferee': '',# true_if_exists
##'is_deposit_transferer': '',# true_if_exists
#'normal_user_spec': {
# 'normal_username': 'pptest',
# 'normal_password': 'pptest',
#}
#'voip_user_spec': {
# 'voip_username': 'pptest',
# 'voip_password': 'pptest',
#},
#'first_login': int(mktime(strptime('2012-01-01 13:23', '%Y-%m-%d %H:%M'))),
#'charge': 'charge2',
#'abs_exp_date': '2017-01-01 13:23',
#'abs_exp_date_unit': 'gregorian',
#'abs_exp_date': '2',
#'abs_exp_date_unit': 'years',
#'assign_dns_1': '1.1.1.1',
#'assign_dns_2': '1.1.1.2',
#'assign_ip': '1.2.3.4',
#'assign_netmask': '255.255.255.1',
#'assign_route_ip': '1.2.3.4',
#'caller_id': '12345',
#'fast_dial': ['1' for i in range(20)],
#'forward_number': '12345',
#'asterisk_extension': 'test5',
#'asterisk_ras_ip': '1.2.3.4',
#'asterisk_sip_details': {
# 'allow': 'none',
# 'canreinvite': 'no',
#},
#'asterisk_iax_details': {},
#'negative_credit_start': '',
},
to_del_attrs=[
## true_if_exists
#'auto_renew',
#'auto_recharge',
#'auto_renew_on_credit_finish',
#'auto_renew_do_not_reset_credit',
#'reset_negative_credit_on_renew',
#'allow_recharge_by_voucher',
#'allow_user_disconnect',
#'caller_id_bind_on_login',
#'deny_change_password',
#'enable_failed_login',
#'is_deposit_transferee',
#'is_deposit_transferer',
#'limit_mac_bind_on_login',
#'limit_port_bind_on_login',
#'limit_ras_bind_on_login',
#'limit_station_ip_bind_on_login',
#'prevent_web_login',
#'negative_credit_start',
#'save_bw_usage',
#'ignore_idle_threshold',
#'forward_number',
#'fast_dial',
#'assign_route_ip',
#'assign_netmask',
#'assign_ip',
#'assign_dns_1',
#'abs_exp_date_unit',
#'abs_exp_date',
#'charge',
#'first_login',
],
)
def test_saveSearchExpiredUsers():
from time import sleep
params = {
'output_type': 'csv',
'conds': {
'group_name': 'pptest',
},
'order_by': 'user_id',
'desc': False,
'basic': 'user_id',
'attrs': ','.join([
'save_bw_usage',# true_if_exists
'normal_username',
'allow_user_disconnect',# true_if_exists
'charge',
'deny_change_password',# true_if_exists
'auto_renew',# true_if_exists
'reset_negative_credit_on_renew',# true_if_exists
##'renew_do_not_reset_credit',# true_if_exists
'renew_do_not_reset_credit_on_state_package',# true_if_exists
'renew_do_not_reset_credit_on_state_recharged',# true_if_exists
'auto_renew_on_credit_finish',# true_if_exists
'auto_renew_do_not_reset_credit',# true_if_exists
'auto_recharge',# true_if_exists
'allow_recharge_by_voucher',# true_if_exists
'prevent_web_login',# true_if_exists
'enable_failed_login',# true_if_exists
'ignore_idle_threshold_static_ip',# true_if_exists
'ignore_idle_threshold',# true_if_exists
'caller_id_bind_on_login',# true_if_exists
'limit_mac_bind_on_login',# true_if_exists
'limit_port_bind_on_login',# true_if_exists
'limit_ras_bind_on_login',# true_if_exists
'limit_station_ip_bind_on_login',# true_if_exists
##'is_deposit_transferee',# true_if_exists
##'is_deposit_transferer',# true_if_exists
])
}
call(
'user.saveSearchExpiredUsers',
**params
)
sleep(2)
printSystemNotifications()
def printSystemNotifications():
import HTMLParser
notifications = call(
'SystemNotification.getNotifications',
last_notifications=100,
only_unread=True,
)
notifications.sort(
key = lambda row: -row['notification_id']
)
for row in notifications:
print 'ID: %s'%row['notification_id']
print 'Date: %s'%row['date']
print 'Type: %s'%row['type']
link = row['links']
if link:
if not '://' in link:
if not link.startswith('/'):
link = '/' + link
link = 'http://127.0.0.1' + link
print 'Link: %s'%link
print 'Message:\n%s'%HTMLParser.HTMLParser().unescape(row['message'])
print '_______________________________________________________\n'
def test_searchUser():
params = {
'conds': {
#'charge': 'charge2',
#'group_name': [
# 'pptest',
# '30-days-from-renew',
#],
#'ippool': ['test'],
#'isp_names': 'test-isp',
#'isp_id': '3',
#'limit_usage_days': [5, 6, 7],
#'mikrotik_address_list': ['Mik', 'aaaaaaaaa'],
#'mikrotik_address_list_op': 'like',
#'notification_profile': ['test'],
'user_status': [1],
},
'from': 0,
'to': 10,
'order_by': 'user_id',
'desc': False,
}
count, total_credit, user_ids = call(
'user.searchUser',
**params
)
print 'Found %s users'%count
print 'User IDs: %s'%user_ids
def test_searchAdminLoginHistory():
return call(
'login.searchAdminLoginHistory',
conds={
'admin': 'parspooyesh',
},
_from=0,
to=2,
sort_by='login_time',
desc=True,
)
def test_webLogin():
from pprint import pprint
pprint(baseCall(
'login.webLogin',
auth_type="ANONYMOUS",
auth_name="",
auth_pass="",
login_auth_type=sys.argv[1].upper(),
login_auth_name=sys.argv[2],
login_auth_pass=sys.argv[3],
auth_remoteaddr='127.0.0.1',
), width=10)
def test_login():
from pprint import pprint
pprint(baseCall(
'login.login',
auth_type="ANONYMOUS",
auth_name="",
auth_pass="",
login_auth_type=sys.argv[1].upper(),
login_auth_name=sys.argv[2],
login_auth_pass=sys.argv[3],
auth_remoteaddr='127.0.0.1',
create_session=True,# default is False
))
def test_setFeshfesheParams():
user_id = 1
charge_rule_id = 1
'''
call(
'user.setOneChargeRuleUsage',
user_id=user_id,
charge_rule_id=charge_rule_id,
credit_usage=1234,
time_usage=3600,
traffic_usage=2345,
)'''
call(
'user.setFeshfesheParams',
user_id=user_id,
traffic=1200,
date_from='2016-01-02',
date_to='2017-01-02',
)
def test_updateUserAttrs_service_price():
from pprint import pprint
call(
'user.updateUserAttrs',
user_id='1',
attrs={
'extra_traffic_options': '1=1234',
},
to_del_attrs=[],
)
pprint(call(
'user.getUserInfo',
user_id='1',
))
if __name__=='__main__':
import sys
from pprint import pprint
#pprint(call('user.getUserInfo')) # auth_type == 'NORMAL_USER'
#pprint(call('user.getUserInfo', user_id = sys.argv[1]))
pprint(call('user.getUserInfo', normal_username = sys.argv[1]))