test: Update Android tests to the canonical pytest format (#1180)

Author: mykola-mokhnach

.github/workflows/functional-test.yml

@@ -134,6 +134,11 @@ jobs:
         distribution: 'temurin'
         java-version: '17'
 
+    - name: Install ffmpeg
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y ffmpeg
+
     - name: Install Node.js
       uses: actions/setup-node@v4
       with:

test/apps/ApiDemos-debug.apk.zip

@@ -22,6 +22,7 @@
 
 @pytest.fixture
 def appium_service() -> Generator[AppiumService, None, None]:
+    """Create and configure Appium service for testing."""
     service = AppiumService()
     service.start(
         args=[
@@ -33,12 +34,13 @@ def appium_service() -> Generator[AppiumService, None, None]:
             '/wd/hub',
         ]
     )
-    try:
-        yield service
-    finally:
-        service.stop()
+
+    yield service
+
+    service.stop()
 
 
 def test_appium_service(appium_service: AppiumService) -> None:
+    """Test that Appium service is running and listening."""
     assert appium_service.is_running
     assert appium_service.is_listening

test/functional/android/appium_service_tests.py

@@ -12,19 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import TYPE_CHECKING, Generator
+
 import pytest
 from selenium.webdriver.common.bidi.common import command_builder
 
 from appium import webdriver
-from appium.options.common import AppiumOptions
 from appium.webdriver.client_config import AppiumClientConfig
 from test.functional.test_helper import is_ci
 from test.helpers.constants import SERVER_URL_BASE
 
-from .helper.desired_capabilities import get_desired_capabilities
+from .options import make_options
+
+if TYPE_CHECKING:
+    from appium.webdriver.webdriver import WebDriver
 
 
 class AppiumLogEntry:
+    """Represents a log entry from Appium BiDi."""
+
     event_class = 'log.entryAdded'
 
     def __init__(self, level, text, timestamp, source, type):
@@ -49,36 +55,36 @@ def from_json(cls, json: dict):
         )
 
 
-class TestChromeWithBiDi:
-    """This test requires selenium python client which supports 'command_builder'"""
+@pytest.fixture
+def driver() -> Generator['WebDriver', None, None]:
+    """Create and configure Chrome driver with BiDi support for testing."""
+    client_config = AppiumClientConfig(remote_server_addr=SERVER_URL_BASE)
+    client_config.timeout = 600
+    options = make_options()
+    options.web_socket_url = True
+    driver = webdriver.Remote(SERVER_URL_BASE, options=options, client_config=client_config)
 
-    def setup_method(self) -> None:
-        client_config = AppiumClientConfig(remote_server_addr=SERVER_URL_BASE)
-        client_config.timeout = 600
-        caps = get_desired_capabilities()
-        caps['webSocketUrl'] = True
-        self.driver = webdriver.Remote(
-            SERVER_URL_BASE, options=AppiumOptions().load_capabilities(caps), client_config=client_config
-        )
+    yield driver
+
+    driver.quit()
 
-    def teardown_method(self) -> None:
-        self.driver.quit()
 
-    @pytest.mark.skipif(is_ci(), reason='Flaky on CI')
-    def test_bidi_log(self) -> None:
-        log_entries = []
-        bidi_log_param = {'events': ['log.entryAdded'], 'contexts': ['NATIVE_APP']}
+@pytest.mark.skipif(is_ci(), reason='Flaky on CI')
+def test_bidi_log(driver: 'WebDriver') -> None:
+    """Test BiDi logging functionality with Chrome driver."""
+    log_entries = []
+    bidi_log_param = {'events': ['log.entryAdded'], 'contexts': ['NATIVE_APP']}
 
-        self.driver.script.conn.execute(command_builder('session.subscribe', bidi_log_param))
+    driver.script.conn.execute(command_builder('session.subscribe', bidi_log_param))
 
-        def _log(entry: AppiumLogEntry):
-            # e.g. {'type': 'syslog', 'level': 'info', 'source': {'realm': ''}, 'text': '08-05 13:30:32.617 29677 29709 I appium  : channel read: GET /session/d7c38859-8930-4eb0-960a-8f917c9e6a38/source', 'timestamp': 1754368241565}
-            log_entries.append(entry.json)
+    def _log(entry: AppiumLogEntry):
+        # e.g. {'type': 'syslog', 'level': 'info', 'source': {'realm': ''}, 'text': '08-05 13:30:32.617 29677 29709 I appium  : channel read: GET /session/d7c38859-8930-4eb0-960a-8f917c9e6a38/source', 'timestamp': 1754368241565}
+        log_entries.append(entry.json)
 
-        try:
-            callback_id = self.driver.script.conn.add_callback(AppiumLogEntry, _log)
-            self.driver.page_source
-            assert len(log_entries) != 0
-            self.driver.script.conn.remove_callback(AppiumLogEntry, callback_id)
-        finally:
-            self.driver.script.conn.execute(command_builder('session.unsubscribe', bidi_log_param))
+    try:
+        callback_id = driver.script.conn.add_callback(AppiumLogEntry, _log)
+        driver.page_source
+        assert len(log_entries) != 0
+        driver.script.conn.remove_callback(AppiumLogEntry, callback_id)
+    finally:
+        driver.script.conn.execute(command_builder('session.unsubscribe', bidi_log_param))

test/functional/android/bidi_tests.py

@@ -12,31 +12,39 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import TYPE_CHECKING, Generator
+
+import pytest
+
 from appium import webdriver
-from appium.options.common import AppiumOptions
 from appium.webdriver.client_config import AppiumClientConfig
 from appium.webdriver.common.appiumby import AppiumBy
 from test.helpers.constants import SERVER_URL_BASE
 
-from .helper.desired_capabilities import get_desired_capabilities
+from .options import make_options
+
+if TYPE_CHECKING:
+    from appium.webdriver.webdriver import WebDriver
+
+
+@pytest.fixture
+def driver() -> Generator['WebDriver', None, None]:
+    """Create and configure Chrome driver for testing."""
+    client_config = AppiumClientConfig(remote_server_addr=SERVER_URL_BASE)
+    client_config.timeout = 600
+    options = make_options()
+    options.browser_name = 'Chrome'
+    driver = webdriver.Remote(SERVER_URL_BASE, options=options, client_config=client_config)
 
+    yield driver
 
-class TestChrome(object):
-    def setup_method(self) -> None:
-        client_config = AppiumClientConfig(remote_server_addr=SERVER_URL_BASE)
-        client_config.timeout = 600
-        caps = get_desired_capabilities()
-        caps['browserName'] = 'Chrome'
-        self.driver = webdriver.Remote(
-            SERVER_URL_BASE, options=AppiumOptions().load_capabilities(caps), client_config=client_config
-        )
+    driver.quit()
 
-    def teardown_method(self) -> None:
-        self.driver.quit()
 
-    def test_find_single_element(self) -> None:
-        e = self.driver.find_element(by=AppiumBy.XPATH, value='//body')
-        assert e.text == ''
+def test_find_single_element(driver: 'WebDriver') -> None:
+    """Test finding a single element in Chrome browser."""
+    e = driver.find_element(by=AppiumBy.XPATH, value='//body')
+    assert e.text == ''
 
-        # Chrome browser's default page
-        assert '<html><head></head><body></body></html>' in self.driver.page_source
+    # Chrome browser's default page
+    assert '<html><head></head><body></body></html>' in driver.page_source

test/functional/android/chrome_tests.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from typing import Optional
+
+from appium.options.android import UiAutomator2Options
+from test.functional.test_helper import get_worker_info
+
+
+def make_options(app: Optional[str] = None) -> UiAutomator2Options:
+    """Get UiAutomator2 options configured for Android testing with parallel execution support."""
+    options = UiAutomator2Options()
+
+    # Set basic Android capabilities
+    options.device_name = android_device_name()
+    options.platform_name = 'Android'
+    options.automation_name = 'UIAutomator2'
+    options.new_command_timeout = 240
+    options.uiautomator2_server_install_timeout = 120000
+    options.adb_exec_timeout = 120000
+
+    if app is not None:
+        options.app = app
+
+    return options
+
+
+def android_device_name() -> str:
+    """
+    Get a unique device name for the current worker.
+    Uses the base device name and appends the port number for uniqueness.
+    """
+    prefix = os.getenv('ANDROID_MODEL') or 'Android Emulator'
+    worker_info = get_worker_info()
+
+    if worker_info.is_parallel:
+        # For parallel execution, we can use different device names or ports
+        # This is a simplified approach - in practice you might want to use different emulators
+        return f'{prefix} - Worker {worker_info.worker_id}'
+
+    return prefix

test/functional/android/helper/__init__.py

@@ -19,11 +19,6 @@
 from test.functional.test_helper import get_wda_port, get_worker_info
 
 
-def PATH(p: str) -> str:
-    """Get the absolute path of a file relative to the folder where this file is located."""
-    return os.path.abspath(os.path.join(os.path.dirname(__file__), p))
-
-
 def make_options(app: Optional[str] = None) -> XCUITestOptions:
     """Get XCUITest options configured for iOS testing with parallel execution support."""
     options = XCUITestOptions()
@@ -36,7 +31,7 @@ def make_options(app: Optional[str] = None) -> XCUITestOptions:
     options.simple_is_visible_check = True
 
     if app is not None:
-        options.app = PATH(os.path.join('..', '..', '..', 'apps', app))
+        options.app = app
 
     if local_prebuilt_wda := os.getenv('LOCAL_PREBUILT_WDA'):
         options.use_preinstalled_wda = True