Dubbo 接口自动化测试框架-526互联

一、核心模块及基础服务类封装

　　1、核心模块介绍

　　2、基础服务类封装设计

　　　　1、创建一个基础服务类（base_service.py）

from dubboclient import DubboClient

class BaseService(object):
    def __init__(self):
        self.dubbo_client = DubboClient("211.103.136.244",6502)

　　　　2、服务对象封装

　　　　　　会员服务类（新建member_service.py）

#导包
import json

from base_service import BaseService

class MemberService(BaseService):
    #重写（扩展式）
    def __init__(self):
        super().__init__()    #继承父类
        self.service_name = "MemberService"    #服务名，多个方法使用
        
    #封装服务类下面的方法
    
    def find_by_telephone(self,tel):
        resp = self.dubbo_client.invoke(self.service_name,"findByTelephone",tel)
        if resp == "null":
            return None
        else:
            #将string类型的数据还原成 字典 或 列表数据
            return json.loads(resp)
        
    def find_member_count_by_months(self,months):
        resp = self.dubbo_client.invoke(self.service_name,"findMemberCountByMonths",months)
        return json.loads(resp)
    
    def add(self,info):               #自定义类型数据
        '''
        info:代表用户传入的测试数据，此时没有class元素
        :return: 
        '''
        #此时自己添加info的 class 类
        info["class"] = "com.itheima.pojo.Member"
        resp = self.dubbo_client.invoke(self.service_name,"add",info)
        if resp == "null":
            return True
        else:
            return False
        
if __name__ == "__main__":
    ms = MemberService()
    print(ms.find_by_telephone())

　　　　　　预约设置服务类（同上）

#导包
import json

from base_service import BaseService

class OrderSettingService(BaseService):
    #重写（扩展式）
    def __init__(self):
        super().__init__()    #继承父类
        self.service_name = "OrderSettingService"    #服务名，多个方法使用

    #封装服务类下面的方法

    def edit_number_by_date(self,date):
        #给date添加class类
        date["class"] = "com.itheima.pojo.OrderSetting"
        resp = self.dubbo_client.invoke(self.service_name,"findByTelephone",date)
        if resp == "Failed":
            return False
        else:
            #将string类型的数据还原成 字典 或 列表数据
            return True

    def get_order_setting_by_month(self,months):
        resp = self.dubbo_client.invoke(self.service_name,"getOrderSettingByMonth",months)
        if resp == "Failed":
            return None
        else:
            return json.loads(resp)

    def add(self,data_list):               #自定义类型数据
      
        resp = self.dubbo_client.invoke(self.service_name,"add",data_list)
        if resp == "Failed":
            return False
        else:
            return True

if __name__ == "__main__":
    ms = OrderSettingService()

websocket runnergo接口dubbo