test.py

import requests
from time import time
import json
import random
host = 'http://192.168.56.101:8080'
default_account = {
    'account_id': 'account',
    'password': 'password',
    'udid': 'udid',
    'idfa': 'idfa',
    'serial': 'serial',
    'imei': 'imei',
    'ecid': 'ecid'
}
default_task = {
    't_appid': 't_appid',
    't_appkey': 't_appkey',
    't_devicetype': 'iPhone',
    't_thread': 10,
    't_data_count': 20,
    't_pc_num': 20,
    'unknown': 'unknown',
}
def add_account(acc):
    acc_ = default_account.copy()
    acc_.update(acc)
    rsp = requests.post(host + '/account', data=acc_)
    print(rsp.json())
    return rsp.json()
def delete_account(acc_id):
    rsp = requests.delete(host + '/account/%s'%acc_id)
    print(rsp.json())
def get_account(acc):
    rsp = requests.get(host + '/account/%s'%acc)
    print(rsp.json())
def get_avalible_account_num(app_id, app_key):
    rsp = requests.get(host + '/account_avaliable?app_id=%s&app_key=%s'%(app_id, app_key))
    print(rsp.json())
def get_account_list(page=1, rows = 20):
    rsp = requests.get(host + '/accounts?page=%s&rows=%s'%(page, rows))
    ret = rsp.json()
    print('Account Total: %s'%ret['total'])
    print('Account Got: %s'%len(ret['rows']))
def get_computers():
    rsp = requests.get(host + '/computers')
    ret = rsp.json()
    print('Computer Online: %s'%ret['onlinenum'])
    print('Computer Got: %s'%len(ret['rows']))
def delete_computer(c_id):
    rsp = requests.delete(host + '/computer/%s'%c_id)
    print(rsp.json())
def get_relys():
    rsp = requests.get(host + '/relys')
    ret = rsp.json()
    print('Computer Rely Got: %s'%len(ret['rows']))
def get_rely_hists(page=1, rows=20):
    rsp = requests.get(host + '/rely_hists?page=%s&rows=%s'%(page, rows))
    ret = rsp.json()
    print('Computer Rely History Total: %s'%ret['total'])
    print('Computer Rely History Got: %s'%len(ret['rows']))
def delete_rely_hist(r_id):
    rsp = requests.delete(host + '/rely_hist/%s'%r_id)
    print(rsp.json())
def clear_rely_hist(app_id):
    rsp = requests.post(host + '/rely_hists/clear', data = { 'cx_appid': app_id })
    print(rsp.json())
def get_settings():
    rsp = requests.get(host + '/settings')
    print(rsp.json())
def save_settings():
    keys = ['accountlimittype', 'programversion', 'accountlimitdays', 'offlinetimeout']
    setting = {
        'accountlimittype': '2',
        'accountlimitdays': '35',
        'programversion': 'A',
        'offlinetimeout': 1
    }
    rsp = requests.post(host + '/settings', data = setting)
    print(rsp.json())
def save_task(task):
    data = default_task.copy()
    data.update(task)
    rsp = requests.post(host + '/task', data = data)
    return rsp.json()
def change_task_state(t_id, state):
    rsp = requests.post(host + '/task/%s'%t_id, data = { 'state': state})
    print(rsp.json())
def delete_task(t_id):
    rsp = requests.delete(host + '/task/%s'%t_id)
    print(rsp.json())
def get_task_list():
    rsp = requests.get(host + '/tasks')
    ret = rsp.json()
    print('Task Got: %s'%len(ret['rows']))
    print('Task unfinished Got: %s'%ret['unfinishtasknum'])
def get_task(key, name):
    rsp = requests.get(host + '/request_task?task_key=%s&task_name=%s'%(key, name))
    return rsp.json()
def rely(data):
    rsp = requests.post(host + '/rely_data', data=json.dumps(data))
    return rsp.json()
def test_account():
    ret = add_account({ 'account_id': 'first_account_%s'%time() })
    a_id = ret['a_id']
    get_account(a_id)
    delete_account(a_id)
    get_avalible_account_num(105, 'task1')
    get_account_list()
def test_computer():
    get_computers()
    delete_computer(1)
    get_relys()
    get_rely_hists()
    delete_rely_hist(1)
    clear_rely_hist('106')
def test_settings():
    get_settings()
    save_settings()
def test_task():
    ret = save_task({ 't_appid': 't_appid_%s'%time(), 't_appkey': 't_appkey_%s'%time() })
    t_id = ret['t_id']
    change_task_state(t_id, 1)
    get_task_list()
    delete_task(t_id)
def test_get_task():
    task_key = 'A'
    task_name = 'macpc_%s'%random.randint(0, 200)
    ret = get_task(task_key, task_name)
    if ret['status'] != 100:
        ret = save_task({ 't_appid': 't_appid_%s'%time(), 't_appkey': 't_appkey_%s'%time(), 't_pc_num': 100 })
        t_id = ret['t_id']
        change_task_state(t_id, 1)
        ret = get_task(task_key, task_name)
    def _rely_acc(acc):
        return { 'account': acc['account_id'] }
    data = {
        'task_key': task_key,
        'task_name': task_name,
        'status': 200,
        'data_count': ret['data_count'] - 10,
        'data': list(map(_rely_acc, ret['data']))
    }
    ret = rely(data)
    ret = rely(data)
    # delete_task(t_id)
if __name__ == '__main__':
    test_account()
    test_computer()
    test_settings()
    test_task()
    for _ in range(10000):
        test_get_task()