游戏测试机器人搭建 - Player Actor

前言：

第一次写博客，文笔有限，加上半路出身的游测，大佬们别往死里喷，错的地方请指正，在正式介绍之前先简单说下框架：Python 的 pykka(Actor 模型)，测试框架 pytest 和测试报告 allure。粗略框架介绍

阅读前的基础知识

Python 的 socket 编程，多进程，多线程，队列

Actors 做什么

当一个 actor 接收到消息后，它能做如下三件事中的一件：

Actor 的组成

Actor 是由状态（state）、行为（behavior）、邮箱（mailbox）三者组成的。

Actors 一大重要特征在于 actors 之间相互隔离，它们并不互相共享内存。这点区别于上述的对象。也就是说，一个 actor 能维持一个私有的状态，并且这个状态不可能被另一个 actor 所改变。
具体 Actor 介绍可参考：传送门 >> JunChow520 的 Actor 模型介绍

pykka 框架的使用

传送门 >> 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()

环境

正文

机器人架构图

robot

测试机器人要用的 3 个核心 Actor(Player,Send,Recv)

Player Actor

Player 对象初始化

# !/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

封装消息发送，用来给自己发送消息或者其它 Actor 调用

def send_msg(self, msg_type=None, data=None):
    '''
    :param msg_type:消息类型 
    :param data: 数据
    '''
    self.actor_ref.tell({
        'msg': msg_type,
        'data': data
    })

Player Actor 实例化之后第一个执行的地方

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
    })

停止 Player Actor

def on_stop(self):
    self.recv_actor.stop()
    self.send_actor.stop()
    self.socket.close()
    self.socket.shutdown()
    self.stop()

log 收集

# 打印报错消息
@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
或扫描二维码回复机器人脚本即可
微信二维码


↙↙↙阅读原文可查看相关链接，并与作者交流