Skip to content

Commit

Permalink
Supports initializing poco using ip:port, and connecting to android:/…
Browse files Browse the repository at this point in the history
…// by default when the device is not connected.
  • Loading branch information
z committed Nov 22, 2023
1 parent 6cae21c commit 4f572bc
Show file tree
Hide file tree
Showing 6 changed files with 66 additions and 50 deletions.
6 changes: 2 additions & 4 deletions poco/drivers/android/uiautomation.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
import threading
import atexit

from airtest.core.api import connect_device, device as current_device
from airtest.core.android.ime import YosemiteIme
from airtest.core.error import AdbShellError, AirtestError

Expand All @@ -22,6 +21,7 @@
from poco.utils.hrpc.hierarchy import RemotePocoHierarchy
from poco.utils.airtest.input import AirtestInput
from poco.utils import six
from poco.utils.device import default_device
from poco.drivers.android.utils.installation import install, uninstall

__all__ = ['AndroidUiautomationPoco', 'AndroidUiautomationHelper']
Expand Down Expand Up @@ -150,9 +150,7 @@ def __init__(self, device=None, using_proxy=True, force_restart=False, use_airte
if options.get('screenshot_each_action') is False:
self.screenshot_each_action = False

self.device = device or current_device()
if not self.device:
self.device = connect_device("Android:///")
self.device = device or default_device()

self.adb_client = self.device.adb
if using_proxy:
Expand Down
40 changes: 20 additions & 20 deletions poco/drivers/cocosjs/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
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


Expand All @@ -26,25 +27,23 @@


class CocosJsPocoAgent(PocoAgent):
def __init__(self, port, device=None):
self.device = device or current_device()
if not self.device:
self.device = connect_device("Android:///")

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':
# Note: ios is now support for now.
# ip = device.get_ip_address()
# use iproxy first
ip = 'localhost'
local_port, _ = self.device.instruct_helper.setup_proxy(port)
port = local_port
else:
ip = self.device.get_ip_address()
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'
else:
ip = self.device.ip
else:
ip = self.device.get_ip_address()

# transport
self.conn = WebSocketClient('ws://{}:{}'.format(ip, port))
Expand Down Expand Up @@ -90,11 +89,12 @@ def __init__(self, addr=DEFAULT_ADDR, device=None, **options):
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)
agent = CocosJsPocoAgent(port, device, ip=ip)
if 'action_interval' not in options:
options['action_interval'] = 0.5
super(CocosJsPoco, self).__init__(agent, **options)
52 changes: 28 additions & 24 deletions poco/drivers/std/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from poco.utils.simplerpc.rpcclient import RpcClient
from poco.utils.simplerpc.transport.tcp.main import TcpClient
from poco.utils.simplerpc.utils import sync_wrapper
from poco.utils.device import default_device

from airtest.core.api import connect_device, device as current_device
from airtest.core.helper import device_platform
Expand Down Expand Up @@ -70,38 +71,41 @@ class StdPoco(Poco):
device = connect_device('Android:///')
poco = StdPoco(10054, device)
# or use ip:port to initialize poco object
poco = StdPoco(port=10054, ip='xx.xx.xx.xx')
# now you can play with poco
ui = poco('...')
ui.click()
...
"""

def __init__(self, port=DEFAULT_PORT, device=None, use_airtest_input=True, **kwargs):
self.device = device or current_device()
if not self.device:
self.device = connect_device("Android:///")

platform_name = device_platform(self.device)
if platform_name == 'Android':
# always forward for android device to avoid network unreachable
local_port, _ = self.device.adb.setup_forward('tcp:{}'.format(port))
ip = self.device.adb.host or 'localhost'
port = local_port
elif platform_name == 'IOS':
# ip = device.get_ip_address()
# use iproxy first
ip = 'localhost'
port, _ = self.device.instruct_helper.setup_proxy(port)
else:
try:
ip = self.device.get_ip_address()
except AttributeError:
try:
ip = socket.gethostbyname(socket.gethostname())
except socket.gaierror:
# 某些特殊情况下会出现这个error,无法正确获取本机ip地址
def __init__(self, port=DEFAULT_PORT, device=None, use_airtest_input=True, ip=None, **kwargs):
if ip is None or ip == "localhost":
self.device = device or default_device()

platform_name = device_platform(self.device)
if platform_name == 'Android':
# always forward for android device to avoid network unreachable
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'
else:
ip = self.device.ip
else:
try:
ip = self.device.get_ip_address()
except AttributeError:
try:
ip = socket.gethostbyname(socket.gethostname())
except socket.gaierror:
# 某些特殊情况下会出现这个error,无法正确获取本机ip地址
ip = 'localhost'

agent = StdPocoAgent((ip, port), use_airtest_input)
kwargs['reevaluate_volatile_attributes'] = True
Expand Down
2 changes: 1 addition & 1 deletion poco/drivers/ue4/ue4_poco.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,4 +23,4 @@ def __init__(self, addr=DEFAULT_ADDR, ue4_editor=False, connect_default_device=T
if dev is None and connect_default_device and not current_device():
dev = connect_device("Android:///")

super(UE4Poco, self).__init__(addr[1], dev, **options)
super(UE4Poco, self).__init__(addr[1], dev, ip=addr[0], **options)
2 changes: 1 addition & 1 deletion poco/drivers/unity3d/unity3d_poco.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,6 @@ def __init__(self, addr=DEFAULT_ADDR, unity_editor=False, connect_default_device
# can apply auto detection in the future
dev = connect_device("Android:///")

super(UnityPoco, self).__init__(addr[1], dev, **options)
super(UnityPoco, self).__init__(addr[1], dev, ip=addr[0], **options)
# If some devices fail to initialize, the UI tree cannot be obtained
# self.vr = UnityVRSupport(self.agent.rpc)
14 changes: 14 additions & 0 deletions poco/utils/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@
from __future__ import absolute_import

from airtest.core.device import Device
from airtest.core.api import connect_device, device as current_device
from airtest.core.error import NoDeviceError


class VirtualDevice(Device):
Expand All @@ -18,3 +20,15 @@ def get_current_resolution(self):

def get_ip_address(self):
return self.ip


def default_device():
"""
Get default device, if no device connected, connect to first android device.
:return:
"""
try:
return current_device()
except NoDeviceError:
return connect_device('Android:///')

0 comments on commit 4f572bc

Please sign in to comment.