poco.drivers.cocosjs 源代码

# -*- coding: utf-8 -*-
# @Author: gzliuxin
# @Email:  gzliuxin@corp.netease.com
# @Date:   2017-07-14 19:47:51

from poco.pocofw import Poco
from poco.agent import PocoAgent
from poco.freezeui.hierarchy import FrozenUIHierarchy, FrozenUIDumper
from poco.utils.simplerpc.utils import sync_wrapper
from poco.utils.airtest import AirtestInput, AirtestScreen
from poco.utils.simplerpc.rpcclient import RpcClient
from poco.utils.simplerpc.transport.ws import WebSocketClient
from poco.utils import six
if six.PY3:
    from urllib.parse import urlparse
    from urlparse import urlparse

from airtest.core.api import connect_device, device as current_device
from poco.utils.device import default_device
from airtest.core.helper import device_platform

__all__ = ['CocosJsPoco']

class CocosJsPocoAgent(PocoAgent):
    def __init__(self, port, device=None, ip=None):
        if ip is None or ip == "localhost":
            self.device = device or default_device()

            platform_name = device_platform(self.device)
            if platform_name == 'Android':
                local_port, _ = self.device.adb.setup_forward('tcp:{}'.format(port))
                ip = self.device.adb.host or 'localhost'
                port = local_port
            elif platform_name == 'IOS':
                port, _ = self.device.setup_forward(port)
                if self.device.is_local_device:
                    ip = 'localhost'
                    ip = self.device.ip
                ip = self.device.get_ip_address()

        # transport
        self.conn = WebSocketClient('ws://{}:{}'.format(ip, port))
        self.c = RpcClient(self.conn)

        hierarchy = FrozenUIHierarchy(Dumper(self.c))
        screen = AirtestScreen()
        inputs = AirtestInput()
        super(CocosJsPocoAgent, self).__init__(hierarchy, inputs, screen, None)

    def rpc(self):
        return self.c

    def get_sdk_version(self):
        return self.rpc.call("getSDKVersion")

class Dumper(FrozenUIDumper):
    def __init__(self, rpcclient):
        super(Dumper, self).__init__()
        self.rpcclient = rpcclient

    def dumpHierarchy(self, onlyVisibleNode=True):
        # NOTE: cocosjs 的driver里,这个rpc方法名首字母是小写,特别注意!
        return self.rpcclient.call("dump", onlyVisibleNode)

[文档]class CocosJsPoco(Poco): """docstring for CocosJsPoco""" def __init__(self, addr=DEFAULT_ADDR, device=None, **options): if not isinstance(addr, (tuple, list, six.string_types)): raise TypeError('Argument "addr" should be `tuple[2]`, `list[2]` or `string` only. Got {}' .format(type(addr))) try: if isinstance(addr, (list, tuple)): ip, port = addr else: port = urlparse(addr).port if not port: raise ValueError ip = urlparse(addr).hostname except ValueError: raise ValueError('Argument "addr" should be a tuple[2] or string format. e.g. ' '["localhost", 5003] or "ws://localhost:5003". Got {}'.format(repr(addr))) agent = CocosJsPocoAgent(port, device, ip=ip) if 'action_interval' not in options: options['action_interval'] = 0.5 super(CocosJsPoco, self).__init__(agent, **options)