众所周知,Dubbo是Java分布式服务的一个非常重要的框架,在实际生产中有着广泛的应用。作为软件测试攻城狮,如果有在使用Java提供后台服务的大型项目中效力过,那么可能已经接触过Dubbo。
Dubbo框架自带了telnet的测试方案。那我们是否可以利用这个原理来实现对Dubbo服务接口的简单调用,从而实现使用python来测试dubbo接口呢?答案是肯定的,话不多说,直接上代码(放马)
import json
import telnetlib
class Dubbo(telnetlib.Telnet):
prompt = 'dubbo>'
coding = 'utf8'
def __init__(self, host=None, port=0):
super().__init__(host, port)
self.write(b'\n')
def command(self, flag, str_=""):
data = self.read_until(flag.encode())
self.write(str_.encode(self.coding) + b"\n")
return data
def invoke(self, service_name, method_name, parms, parmstype='String'):
items = ""
lengthparmss = len(parms)
i = 1
for item in parms:
if i < lengthparmss:
if isinstance(item, str):
items += '\'{0}\','.format(item)
else:
items += "{0},".format(item)
else:
if isinstance(item, str):
items += '\'{0}\''.format(item)
else:
items += "{0}".format(item)
i += 1
command_str = "invoke %s.%s(" % (service_name, method_name) + items + ")"
try:
self.command(Dubbo.prompt, command_str)
data = self.command(Dubbo.prompt, "")
data = json.loads(data.decode(Dubbo.coding, errors='ignore').split('\n')[0].strip())
return data
except Exception as e:
print(e)
return "Failed!"
if __name__ == '__main__':
# 服务ip地址和端口
conn = Dubbo('*.*.*.*', ***)
Dubbo.coding = 'gbk' # 在这个可以给出具体的参数和形式
result = conn.invoke(
# "连到dubbo ls下面的某个服务"
"cn.touna.fund.api.PartnerRepayService",
# "服务下的具体方法"
"sendSmsCode",
# 方法入参
[{'applyid': 99200669, 'phone': "13534187661"}],
)
print(result)
上面的代码可以实现对Dubbo接口简单的调用,只需要知道服务,方法,参数就可以了。
是不是很简单,如果有需要可以参考一下,当然你还可以扩展,通过对不同dubbo参数类型和格式的扩展和封装,你也许可以将代码写的更加完善和健壮。
网友评论