• Appium 点击操作梳理

            Appium框架的客户端脚本中执行UI操作的原理是：脚本中需要执行UI操作时，会发送一条http请求（请求数据中包括了控件相关信息）到Appium的服务器，服务器再把接收到的数据转义一下，然后转发给安装在手机端的插桩程序。这时候插桩程序调用android sdk提供的uiautomator相关ui操作库来执行真正的UI操作。然后再把结果沿路一直返回到脚本中形成闭环。

            下面注意分析下脚本中如何给Appium服务器发送http请求。

            1、启动Appim服务

            首先需要启动Appium服务，让Appium服务监听端口4723，这样脚本就可以往这个端口发送http请求了；

            2、脚本中执行用例前需要创建webDriver对象

            这个对象可以理解为appium提供给脚本中执行UI操作的封装函数库。

            webDriver类的构造方法中会根据”desired capabilities“信息向appium服务器发起了一次请求，服务器拿到”desired capabilities“后会根据这些信息创建一个SessionId并返回给用例脚本。

            并且Appium拿到”desired capabilities“后就能知道要和哪个连接PC的手机设备进行连接了。Appium服务在这里会做很多事情来确保和手机端的插桩服务程序连接成功。具体做了哪些事情可以参考我的另外一篇文章”appium 从启动到测试再到结束流程梳理“。

            也许你会问什么要执行用例前要获取session id呢？

            因为执行测试时，脚本用例本质上是给服务端发送http请求，但是http请求是无状态的，服务器收到的每条http请求都被认为和之前的请求没有任何关系。这会导致每一条http请求都必须带上”desired capabilities“信息，这样服务器才知道要和哪个手机设备通信，并且”desired capabilities“还有很多其他信息，才能确保appium服务按照”desired capabilities“指定的参数运行。即需要保证每条http请求的运行环境是一致的。

            每一条http请求都带上”desired capabilities“信息这显然是不可取的。所以appium采取的是session机制。appium服务第一次拿到”desired capabilities“后会存在本地，并返回一个session id给脚本用例，这样后续的用例再发送http请求到appium服务，则appium服务根据session id就能知道对应哪个”desired capabilities“，然后就能知道运行环境是怎样的了（运行环境指的是和哪台设手机设备通信，用例执行延时或重试这些在”desired capabilities“中指定的信息）。

    一、获取webdriver对象，并得到session id

    1. self.caps = {}
    2. self.caps["platformName"] = "Android"
    3. self.caps["platformVersion"] = devices.dev[Constant.phone]["platformVersion"]
    4. self.caps["deviceName"] = devices.dev[Constant.phone]["phone"]
    5. self.caps["appPackage"] = Constant.appPackage
    6. self.caps["appActivity"] = Constant.appActivity
    7. self.caps['app'] = Constant.app
    8. self.caps["unicodeKeyboard"] = True
    9. self.caps["autoAcceptAlerts"] = True  # 对权限弹窗进行授权
    10. self.caps["resetKeyboard"] = True
    11. self.caps["noReset"] = True
    12. self.caps["newCommandTimeout"] = 6000
    13. self.driver = webdriver.Remote('http://127.0.0.1:4723/wd/hub', self.caps)  # localhost

    上面是在appium的脚本中用一个字典来存储”desired capabilities“，然后创建了一个webDriver对象。

            'http://127.0.0.1:4723/wd/hub'代表的是向本机的4723端口发送请求，appium服务运行时监听的端口就是4723，url中的路径部分/wd/hub，其中wd是webdriver的缩写，hub表示中心节点。这些在appium服务的node.js源码中能找到对应路径。

    1. class WebDriver(webdriver.Remote):
    2.     def __init__(self, command_executor='http://127.0.0.1:4444/wd/hub',
    3.                  desired_capabilities=None, browser_profile=None, proxy=None, keep_alive=False):
    4.  
    5.         super(WebDriver, self).__init__(command_executor, desired_capabilities, browser_profile, proxy, keep_alive)
    6.  
    7.         if self.command_executor is not None:
    8.             self._addCommands()
    1. class WebDriver(object):
    2.     def __init__(self, command_executor='http://127.0.0.1:4444/wd/hub',
    3.                  desired_capabilities=None, browser_profile=None, proxy=None,
    4.                  keep_alive=False, file_detector=None):
    5.         ...
    6.         if type(self.command_executor) is bytes or isinstance(self.command_executor, str):
    7.             self.command_executor = RemoteConnection(command_executor, keep_alive=keep_alive)
    8.         ...
    9.         self.start_session(desired_capabilities, browser_profile)
    10.         ...

            先看简单的吧，当self.command_executor不为None时，调用self._addCommands()往self.command_executor._commands命令映射列表中新增一些命令。因为appium是基于selenium二次开发的，self.command_executor._commands是selenium框架中原有的命令字映射表，appium在这基础上新增了一些。

            然后构造器中主要就是创建RemoteConnection对象和start_session()。

    1、创建RemoteConnection对象并赋值给command_executor属性

            来看看RemoteConnection的构造器，发现其内部主要是检查我们最开始传入的url的参数格式是否正确。并把url解析出来后赋值给自己的属性保存。并使用_commands字典来保存http请求中请求方法和url路径（这个应该是为后面做准备吧）。形式为：命令描述字符串-->(http请求方法，http请求路径)。 这样只需要只要命令描述字符串就知道了命令请求方法和路径。

    1. self._commands = {
    2. 	Command.STATUS: ('GET', '/status'),
    3. 	Command.NEW_SESSION: ('POST', '/session'),
    4. 	Command.GET_ALL_SESSIONS: ('GET', '/sessions'),
    5. 	Command.QUIT: ('DELETE', '/session/$sessionId'),
    6. 	Command.GET_CURRENT_WINDOW_HANDLE:
    7. 	   ('GET', '/session/$sessionId/window_handle'),
    8. 	Command.GET_WINDOW_HANDLES:
    9. 	   ('GET', '/session/$sessionId/window_handles'),
    10. 	Command.GET: ('POST', '/session/$sessionId/url'),
    11. 	Command.GO_FORWARD: ('POST', '/session/$sessionId/forward'),
    12. 	Command.GO_BACK: ('POST', '/session/$sessionId/back'),
    13. 	Command.REFRESH: ('POST', '/session/$sessionId/refresh'),
    14. 	Command.EXECUTE_SCRIPT: ('POST', '/session/$sessionId/execute'),
    15. 	Command.GET_CURRENT_URL: ('GET', '/session/$sessionId/url'),
    16. 	Command.GET_TITLE: ('GET', '/session/$sessionId/title'),
    17. 	Command.GET_PAGE_SOURCE: ('GET', '/session/$sessionId/source'),
    18. 	Command.SCREENSHOT: ('GET', '/session/$sessionId/screenshot'),
    19. 	Command.ELEMENT_SCREENSHOT: ('GET', '/session/$sessionId/element/$id/screenshot'),
    20. 	Command.FIND_ELEMENT: ('POST', '/session/$sessionId/element'),
    21. 	Command.FIND_ELEMENTS: ('POST', '/session/$sessionId/elements'),
    22.     ...

           Command类保存了用例中所有的命令

    1. class Command(object):
    2.     """
    3.     Defines constants for the standard WebDriver commands.
    5.     While these constants have no meaning in and of themselves, they are
    6.     used to marshal commands through a service that implements WebDriver's
    7.     remote wire protocol:
    9.         https://github.com/SeleniumHQ/selenium/wiki/JsonWireProtocol
    11.     """
    12.  
    13.     # Keep in sync with org.openqa.selenium.remote.DriverCommand
    14.  
    15.     STATUS = "status"
    16.     NEW_SESSION = "newSession"
    17.     GET_ALL_SESSIONS = "getAllSessions"
    18.     DELETE_SESSION = "deleteSession"
    19.     CLOSE = "close"
    20.     QUIT = "quit"
    21.     GET = "get"
    22.     GO_BACK = "goBack"
    23.     GO_FORWARD = "goForward"
    24.     REFRESH = "refresh"
    25.     ADD_COOKIE = "addCookie"
    26.     GET_COOKIE = "getCookie"
    27.     GET_ALL_COOKIES = "getCookies"
    28.     DELETE_COOKIE = "deleteCookie"
    29.     DELETE_ALL_COOKIES = "deleteAllCookies"
    30.     FIND_ELEMENT = "findElement"
    31.     FIND_ELEMENTS = "findElements"
    32.     FIND_CHILD_ELEMENT = "findChildElement"
    33.     FIND_CHILD_ELEMENTS = "findChildElements"
    34.     CLEAR_ELEMENT = "clearElement"
    35.     CLICK_ELEMENT = "clickElement"
    36.     SEND_KEYS_TO_ELEMENT = "sendKeysToElement"
    37.     SEND_KEYS_TO_ACTIVE_ELEMENT = "sendKeysToActiveElement"
    38.     SUBMIT_ELEMENT = "submitElement"
    39.     UPLOAD_FILE = "uploadFile"
    40.     GET_CURRENT_WINDOW_HANDLE = "getCurrentWindowHandle"
    41.     GET_WINDOW_HANDLES = "getWindowHandles"
    42.     ...

    这样就把命令关键字和命令关联起来起来了。

    2、执行start_session(desired_capabilities, browser_profile)生成session id

    1.  def start_session(self, desired_capabilities, browser_profile=None):
    2. 	"""
    3. 	Creates a new session with the desired capabilities.
    5. 	:Args:
    6. 	 - browser_name - The name of the browser to request.
    7. 	 - version - Which browser version to request.
    8. 	 - platform - Which platform to request the browser on.
    9. 	 - javascript_enabled - Whether the new session should support JavaScript.
    10. 	 - browser_profile - A selenium.webdriver.firefox.firefox_profile.FirefoxProfile object. Only used if Firefox is requested.
    11. 	"""
    12. 	capabilities = {'desiredCapabilities': {}, 'requiredCapabilities': {}}
    13. 	for k, v in desired_capabilities.items():
    14. 	    if k not in ('desiredCapabilities', 'requiredCapabilities'):
    15. 	        capabilities['desiredCapabilities'][k] = v
    16. 	    else:
    17. 	        capabilities[k].update(v)
    18. 	if browser_profile:
    19. 	    capabilities['desiredCapabilities']['firefox_profile'] = browser_profile.encoded
    20. 	response = self.execute(Command.NEW_SESSION, capabilities)
    21. 	if 'sessionId' not in response:
    22. 	    response = response['value']
    23. 	self.session_id = response['sessionId']
    24. 	self.capabilities = response['value']
    25.  
    26. 	# Quick check to see if we have a W3C Compliant browser
    27. 	self.w3c = response.get('status') is None

    其中response = self.execute(Command.NEW_SESSION, capabilities)执行。

    appium中发送http请求都是在excute()方法中执行。excute()中又执行了command_executor.execute(), 最终在这个execute()中调用request发送http请求。

    1. def execute(self, driver_command, params=None):
    2. 	"""
    3. 	Sends a command to be executed by a command.CommandExecutor.
    4. 	
    5. 	:Args:
    6. 	 - driver_command: The name of the command to execute as a string.
    7. 	 - params: A dictionary of named parameters to send with the command.
    8. 	
    9. 	:Returns:
    10. 	  The command's JSON response loaded into a dictionary object.
    11. 	"""
    12. 	if self.session_id is not None:
    13. 	    if not params:
    14. 	        params = {'sessionId': self.session_id}
    15. 	    elif 'sessionId' not in params:
    16. 	        params['sessionId'] = self.session_id
    17. 	
    18. 	params = self._wrap_value(params)
    19. 	response = self.command_executor.execute(driver_command, params)
    20. 	if response:
    21. 	    self.error_handler.check_response(response)
    22. 	    response['value'] = self._unwrap_value(
    23. 	        response.get('value', None))
    24. 	    return response
    25. 	# If the server doesn't send a response, assume the command was
    26. 	# a success
    27. 	return {'success': 0, 'value': None, 'sessionId': self.session_id}

            可以看到，函数内部首先会检查session_id是不是为none。在获取sessioid的时候，这个session_id是none，直接走下面的逻辑。 获取完session之后的所有请求，sessioid不为null，则会检查参数params加上sessionid参数。所以服务器就知道了请求来自哪个客户端。

            原来客户端的session id是在这里获取，并每次请求时在这里加上session id的呀！

    调用流程有点复杂，来个流程图吧。

     

    二、执行查找控件和执行UI操作

            用一个基本的点击操作来梳理这个过程。

    self.driver.find_element_by_id("xxx").click()

    1、driver.find_element_by_id()获取控件       

    1. def find_element_by_id(self, id_):
    2.     """Finds an element by id.
    4.     :Args:
    5.      - id\_ - The id of the element to be found.
    7.     :Usage:
    8.         driver.find_element_by_id('foo')
    9.     """
    10.     return self.find_element(by=By.ID, value=id_)

    其内部调用的是自身的find_element()方法

    1. def find_element(self, by=By.ID, value=None):
    2.     """
    3.     'Private' method used by the find_element_by_* methods.
    5.     :Usage:
    6.         Use the corresponding find_element_by_* instead of this.
    8.     :rtype: WebElement
    9.     """
    10.     if self.w3c:
    11.         if by == By.ID:
    12.             by = By.CSS_SELECTOR
    13.             value = '[id="%s"]' % value
    14.         elif by == By.TAG_NAME:
    15.             by = By.CSS_SELECTOR
    16.         elif by == By.CLASS_NAME:
    17.             by = By.CSS_SELECTOR
    18.             value = ".%s" % value
    19.         elif by == By.NAME:
    20.             by = By.CSS_SELECTOR
    21.             value = '[name="%s"]' % value
    22.     return self.execute(Command.FIND_ELEMENT, {
    23.         'using': by,
    24.         'value': value})['value']

    这里根据不同查找方式，对by和value参数进行了处理，然后再调用自身的excute()方法。注意看注释里的rtype: WebElement，说明find_element()返回的是一个WebElement对象。

    self.execute(Command.FIND_ELEMENT, {'using': by,'value': value})['value']

    就需要去excute()中看看是如何返回一个WebElement对象了。

    1. def execute(self, driver_command, params=None):
    2.     """
    3.     Sends a command to be executed by a command.CommandExecutor.
    5.     :Args:
    6.      - driver_command: The name of the command to execute as a string.
    7.      - params: A dictionary of named parameters to send with the command.
    9.     :Returns:
    10.       The command's JSON response loaded into a dictionary object.
    11.     """
    12.     if self.session_id is not None:
    13.         if not params:
    14.             params = {'sessionId': self.session_id}
    15.         elif 'sessionId' not in params:
    16.             params['sessionId'] = self.session_id
    17.  
    18.     params = self._wrap_value(params)
    19.     response = self.command_executor.execute(driver_command, params)
    20.     if response:
    21.         self.error_handler.check_response(response)
    22.         response['value'] = self._unwrap_value(
    23.             response.get('value', None))
    24.         return response
    25.     # If the server doesn't send a response, assume the command was
    26.     # a success
    27.     return {'success': 0, 'value': None, 'sessionId': self.session_id}

            通过注释可以看到，excute()发送了一条需要被执行的命令到command.CommandExecutor，然后得到返回结果response。并且response是json格式的字典类型。

            excute()方法前面是给params参数加上session id，这个在前面已经分析过了。

            然后是包装params参数，并执行command_executor.execute(driver_command, params)得到返回response。然后再检查response格式这些是否正常，如果response中不包含错误，则对response中的value进行解包装。就是在这个地方生成WebElement对象的。

    1. def _unwrap_value(self, value):
    2.     if isinstance(value, dict) and ('ELEMENT' in value or 'element-6066-11e4-a52e-4f735466cecf' in value):
    3.         wrapped_id = value.get('ELEMENT', None)
    4.         if wrapped_id:
    5.             return self.create_web_element(value['ELEMENT'])
    6.         else:
    7.             return self.create_web_element(value['element-6066-11e4-a52e-4f735466cecf'])
    8.  
    9.     elif isinstance(value, list):
    10.         return list(self._unwrap_value(item) for item in value)
    11.     else:
    12.         return value

            其中create_web_element()内部实现为

    1. def create_web_element(self, element_id):
    2.     """Creates a web element with the specified `element_id`."""
    3.     return self._web_element_cls(self, element_id, w3c=self.w3c)

            而_web_element_cls为

    _web_element_cls = WebElement

          所以，终于明白了是如何调用find_element_by_id()一步步如何最终获取到WebElement了。

    2、WebElement对象上执行click()点击

    1. def click(self):
    2.     """Clicks the element."""
    3.     self._execute(Command.CLICK_ELEMENT)

    进入到_excute(）

    1. # Private Methods
    2. def _execute(self, command, params=None):
    3.     """Executes a command against the underlying HTML element.
    5.     Args:
    6.       command: The name of the command to _execute as a string.
    7.       params: A dictionary of named parameters to send with the command.
    9.     Returns:
    10.       The command's JSON response loaded into a dictionary object.
    11.     """
    12.     if not params:
    13.         params = {}
    14.     params['id'] = self._id
    15.     return self._parent.execute(command, params)

    self._parent是什么对象呢？在WebElement的构造器中，self._parent是构造器传入的第一个参数

    1. class WebElement(object):
    2.  
    3.     def __init__(self, parent, id_, w3c=False):
    4.         self._parent = parent
    5.         self._id = id_
    6.         self._w3c = w3c

    那么再回到上面创建WebElement的地方，发现传入的是WebDriver对象

    1. def create_web_element(self, element_id):
    2.     """Creates a web element with the specified `element_id`."""
    3.     return self._web_element_cls(self, element_id, w3c=self.w3c)
    4.  
    5. def _unwrap_value(self, value):
    6.     if isinstance(value, dict) and ('ELEMENT' in value or 'element-6066-11e4-a52e-4f735466cecf' in value):
    7.         wrapped_id = value.get('ELEMENT', None)
    8.         if wrapped_id:
    9.             return self.create_web_element(value['ELEMENT'])
    10.         else:
    11.             return self.create_web_element(value['element-6066-11e4-a52e-4f735466cecf'])
    12.  
    13.     elif isinstance(value, list):
    14.         return list(self._unwrap_value(item) for item in value)
    15.     else:
    16.         return value

    所以self._parent.execute(command, params)调用的还是webdriver对象的excute()方法。

    到这里，就理清了控件是如何点击的，其本质也是向appium服务发送一个http请求。

  • 相关阅读:
    C语言char与short取反以及符号判断问题
    OpenCV中的边缘检测技术及实现
    Kafka 之 KRaft —— 配置、存储工具、部署注意事项、缺失的特性
    传智杯第二届javaB组例题
    Python的基础语法(八)（持续更新）
    LayUI之CRUD
    软件测试面试题之自动化测试题大合集（上）
    DAST 黑盒漏洞扫描器 第五篇：漏洞扫描引擎与服务能力
    EasyNTS上云网关断电重启后设备离线是什么原因？
    熊市下PLATO如何通过Elephant Swap，获得溢价收益？
  • 原文地址：https://blog.csdn.net/liuqinhou/article/details/126009276