第一次写博客,文笔有限,加上半路出身的游测,大佬们别往死里喷,错的地方请指正,在正式介绍之前先简单说下框架:Python 的 pykka(Actor 模型),测试框架 pytest 和测试报告 allure。粗略框架介绍
Python 的 socket 编程,多进程,多线程,队列
当一个 actor 接收到消息后,它能做如下三件事中的一件:
Actor 是由状态(state)、行为(behavior)、邮箱(mailbox)三者组成的。
Actors 一大重要特征在于 actors 之间相互隔离,它们并不互相共享内存。这点区别于上述的对象。也就是说,一个 actor 能维持一个私有的状态,并且这个状态不可能被另一个 actor 所改变。
具体 Actor 介绍可参考:传送门 >> JunChow520 的 Actor 模型介绍
传送门 >> pykka 使用说明
引用说明书的例子:
# !/usr/bin/env python3
import pykka
GetMessages = object()
class PlainActor(pykka.ThreadingActor):
def __init__(self):
super().__init__()
self.stored_messages = []
def on_receive(self, message):
if message is GetMessages:
return self.stored_messages
else:
self.stored_messages.append(message)
if __name__ == '__main__':
actor = PlainActor.start()
actor.tell({'no': 'Norway', 'se': 'Sweden'})
actor.tell({'a': 3, 'b': 4, 'c': 5})
print(actor.ask(GetMessages))
actor.stop()
# !/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File : player.py
@Contact : 512759438@qq.com
@Author : Jian
'''
import pykka
import websocket
import traceback as tb
from proto import ProtoHandler
from remote import RemoteHandler
from util import MsgSwitch, RecvActor, SendActor
TEST_CASE_CALL = None
class Player(pykka.ThreadingActor):
def __init__(self, user_name='', server='SERVER_1',sex=1, job=1):
super(Player, self).__init__()
self.host = SERVER_LIST[server]['GAME_HOST']
self.port = SERVER_LIST[server]['GAME_PORT']
self.web_host = "x.x.x.x"
self.recv_actor = None
self.send_actor = None
self.socket = None
self.proto_handler = ProtoHandler(self)
self.remote_handler = RemoteHandler(self)
'''测试用例执行时需要调用player'''
global TEST_CASE_CALL
TEST_CASE_CALL = self
self.player_id = None
self.state_user_id = 0
self.state_user_name = user_name
self.sys_count = 0
def send_msg(self, msg_type=None, data=None):
'''
:param msg_type:消息类型
:param data: 数据
'''
self.actor_ref.tell({
'msg': msg_type,
'data': data
})
def on_start(self):
if self.state_user_name is '':
self.send_msg(MSG_GUEST_LOGIN)
else:
self.send_msg(MSG_LOGIN_INFO)
def on_receive(self, msg):
for case in MsgSwitch(msg):
# 获取用户信息
if case(MSG_LOGIN_INFO):
account_info = Account(self.state_user_name).login_info()
if account_info['code_str'] == 'OK':
user_into = account_info['user']
self.create_player_params = {
'rd3_token': user_into['token'],
'rd3_userId': user_into['userId'],
'server_list_type': 0,
'sid': 1,
'token': user_into['token'],
}
self.create_player_params.update(Account(self.state_user_name).data)
self.create_player_params.pop('password')
self.create_player_params['cmd'] = 'game_login'
self.send_msg(MSG_LOGIN)
else:print(f'获取角色信息ERROR, 原因: {account_info["code_str"]},{account_info["code"]}')
break
# 用户登录
if case(MSG_LOGIN):
self.socket = websocket.create_connection(f'ws://{self.host}:{self.port}/')
self.recv_actor = RecvActor.start(self, self.socket)
self.send_actor = SendActor.start(self, self.socket)
self.send_actor.tell({MSG_PROTO: self.create_player_params})
break
# 用户创角
if case(MSG_CREATE_PLAYER):
create_data = {
'nickname': self.state_user_name,
'rd3_token': self.create_player_params['rd3_token'],
'rd3_userId': self.create_player_params['rd3_userId'],
'sid': self.create_player_params['sid'],
'token': self.create_player_params['token'],
}
self.send_actor.tell({MSG_PROTO: create_data})
break
# 服务端返回协议处理
if case(MSG_PROTO):
method, data = msg['data']
if hasattr(self.proto_handler, method):
getattr(self.proto_handler, method)(data)
else:
print(f"没有为协议: {method} 定义处理方法, 请前往 proto.py 文件中定义!")
break
# 控制台调用命令
if case(MSG_REMOTE_CMD):
method = msg['method']
method = (type(method) is int and "r" + str(method)) or (type(method) is str and method)
if hasattr(self.remote_handler, method):
getattr(self.remote_handler, method)(msg['data'])
else:
print(f"没有为远程命令: {method} 定义处理方法, 请前往 remote.py 文件中定义!")
break
def remote_msg(self, method:str=None, data=None):
'''
调用remote里的方法
:param method: 方法名
:param data: 传入的参数 元组
'''
self.actor_ref.tell({
'msg': MSG_REMOTE_CMD,
'method': method,
'data': data
})
def on_stop(self):
self.recv_actor.stop()
self.send_actor.stop()
self.socket.close()
self.socket.shutdown()
self.stop()
# 打印报错消息
@GetLog(level='fatal')
def on_failure(self, exception_type, exception_value, traceback):
logging.fatal(f'Player: {self.state_user_name} is down.')
logging.fatal(f"ErrorType => {exception_type}")
logging.fatal(f"ErrorValue => {exception_value}")
logging.fatal(f"TraceBack => {tb.print_tb(traceback)}")
self.on_stop()
python 编写游戏测试机器人客户端 (一)
python 编写游戏测试机器人客户端 (二)
python 编写游戏测试机器人客户端 (三)
python 编写游戏测试机器人客户端 (四)
到这里 Player Actor 已经写完了,目前是无法单独运行的,需要结合后面的 Send Actor 和 Recv Actor 才能运行,写的不清晰的欢迎在评论留言
最后的最后,各位的关注、点赞、收藏、碎银子打赏是对我最大的支持,谢谢大家!
需要源码的小伙伴关注微信公众号 ID:gameTesterGz
或扫描二维码回复机器人脚本即可