Issue: Device Becomes Unavailable After First Test Execution in Appium Device Farm

[raghuns2311]

Issue: Device Becomes Unavailable After First Test Execution in Appium Device Farm

• Is there a way to fix this?
• Is there a way we can block the device from detecting from appium server and appium device farm?
• The reason I'm triggering the appium server here is I have my both AI and Appium automation in same code base

Description

When using Appium Device Farm with a physical Android device, the device is initially detected and can be used for test execution. However, after the first test run, the device disappears from the device farm UI and becomes unavailable for subsequent test runs, even though the device is still physically connected and visible via ADB.

Expected Behavior

• Device should remain available in the device farm after test execution
• Should be able to run multiple test sessions on the same device
• Device status should properly reset between test runs

Actual Behavior

1. Device is initially detected and shown as available in device farm UI
2. First test execution works correctly and the device is in blocked state
3. After test completion, device disappears from device farm UI
4. Subsequent test executions fail with "device not found" errors
5. Only restarting the Appium server makes the device available again

Environment

• Appium: 2.17.1
• Appium Device Farm Plugin: 9.8.4
• Platform: Linux
• Device: Samsung Tab Android 14/15

Steps to Reproduce

1. Start Appium with device farm:

   appium --use-plugins=device-farm,images -ka 800 --relaxed-security --base-path /wd/hub

2. Run a test that uses the device
3. After test completion, check device farm UI
4. Try to run another test - device is no longer available

Code to connect and initialize driver with running server

class DeviceFarmDriver:
    def __init__(self, device_udid: str, device_tag: str = None, appium_server: str = APPIUM_SERVER):
        self.device_udid = device_udid
        self.device_tag = device_tag or device_udid  # Use UDID as tag if none provided
        self.appium_server = appium_server
        self.driver = None
        self._device_farm_url = "http://10.40.144.31:4723/device-farm/api"
        self._device_blocked = False

    def _call_device_farm_api(self, endpoint: str, data: dict) -> dict:
        """Helper method to call device farm API endpoints."""
        url = f"{self._device_farm_url}/{endpoint}"
        headers = {'Content-Type': 'application/json'}
        
        try:
            response = requests.post(url, json=data, headers=headers, timeout=10)
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            logger.error(f"Device farm API call to {endpoint} failed: {e}")
            if hasattr(e, 'response') and e.response is not None:
                try:
                    error_response = e.response.json()
                    logger.error(f"Error response: {error_response}")
                except:
                    logger.error(f"Response: {e.response.text}")
            return {"status": "error", "message": str(e)}

    def _block_device(self) -> bool:
        """Block the device in the device farm."""
        try:
            logger.info(f"Blocking device {self.device_tag} in device farm")
            
            # Get the host IP from the appium server URL
            host = self.appium_server.split('//')[-1].split('/')[0]
            if ':' in host:
                host = host.split(':')[0]
            host_url = f"http://{host}:4723"
            
            result = self._call_device_farm_api(
                "block",
                {"udid": self.device_udid, "host": host_url}
            )
            
            if result and result.get('status') == 'success':
                self._device_blocked = True
                logger.info(f"Successfully blocked device {self.device_tag}")
                return True
            else:
                error_msg = result.get('message', 'Unknown error') if isinstance(result, dict) else 'Invalid response format'
                logger.error(f"Failed to block device: {error_msg}")
                return False
                
        except Exception as e:
            logger.error(f"Exception while blocking device {self.device_tag}: {e}")
            return False

    def _unblock_device(self) -> bool:
        """Unblock the device in the device farm."""
        if not self._device_blocked:
            return True
            
        try:
            logger.info(f"Unblocking device {self.device_tag} in device farm")
            
            # Get the host IP from the appium server URL
            host = self.appium_server.split('//')[-1].split('/')[0]
            if ':' in host:
                host = host.split(':')[0]
            host_url = f"http://{host}:4723"
            
            result = self._call_device_farm_api(
                "unblock",
                {"udid": self.device_udid, "host": host_url}
            )
            
            if result and result.get('status') == 'success':
                self._device_blocked = False
                logger.info(f"Successfully unblocked device {self.device_tag}")
                return True
            else:
                error_msg = result.get('message', 'Unknown error') if isinstance(result, dict) else 'Invalid response format'
                logger.error(f"Failed to unblock device: {error_msg}")
                return False
                
        except Exception as e:
            logger.error(f"Exception while unblocking device {self.device_tag}: {e}")
            return False
        finally:
            self._device_blocked = False

    def initialize_driver(self) -> bool:
        """Initialize Appium driver with the given device UDID."""
        try:
            options = UiAutomator2Options()

            # Required capabilities
            options.platform_name = 'Android'
            options.automation_name = 'UiAutomator2'
            options.device_name = self.device_tag  # For display
            options.udid = self.device_udid       # Actual device UDID

            # Session settings
            options.new_command_timeout = 300      # 5 minutes
            options.auto_grant_permissions = True  # Auto-accept permissions

            # Reset behavior
            options.no_reset = True               # Don't reset app state
            options.full_reset = False             # Don't uninstall app

            # Performance optimizations
            options.disable_id_locator_autocompletion = True
            options.set_capability('newCommandTimeout', 300)
            options.set_capability('ignoreHiddenApiPolicyError', True)

            logger.info(f"Initializing Appium driver for device: {self.device_tag}")
            logger.info(f"Connecting to Appium server at: {self.appium_server}")
            logger.info(f"Device UDID: {self.device_udid}")

            try:
                import socket
                socket.setdefaulttimeout(30)
                logger.info("Creating WebDriver instance")
                # Try to initialize the driver with a connection timeout
                self.driver = webdriver.Remote(
                    command_executor=self.appium_server,
                    options=options,
                )
                logger.info("WebDriver created, setting implicit wait...")
                self.driver.implicitly_wait(30)  # 30 seconds

                # Test basic device communication
                logger.info("Testing device communication...")
                current_activity = self.driver.current_activity
                logger.info(f"Current activity: {current_activity}")

                # Wait for device to be ready with a timeout
                logger.info("Waiting for device to be ready...")
                wait = WebDriverWait(self.driver, 30)
                wait.until(EC.presence_of_element_located((
                    AppiumBy.ANDROID_UIAUTOMATOR,
                    'new UiSelector().className("android.widget.FrameLayout")'
                )))

                if not self._block_device():
                    logger.warning(f"Failed to block device {self.device_tag} in device farm")
                # Continue even if blocking fails, as the driver is already initialized

                logger.info("Appium driver initialized successfully")
                return True

            except Exception as e:
                logger.error(f"Failed to initialize WebDriver: {str(e)}")
                logger.error("Possible issues:")
                logger.error("1. Appium server is not running or not accessible")
                logger.error("2. Device is not connected or not authorized")
                logger.error(f"3. Device with UDID {self.device_tag} not found")
                logger.error("4. Appium server version mismatch with client")
                return False

        except Exception as e:
            logger.error(f"Error in driver initialization: {str(e)}")
            self.quit_driver()
            return False

    def quit_driver(self) -> None:
        """Safely quit the driver if it exists and unblock the device."""
        driver_quit_success = False
        
        try:
            # First try to unblock the device
            try:
                self._unblock_device()
            except Exception as e:
                logger.error(f"Error while unblocking device: {e}")
            
            # Then quit the driver if it exists
            if self.driver:
                try:
                    # Try to close the app first
                    try:
                        self.driver.close_app()
                    except:
                        pass  # Ignore errors in close_app
                        
                    # Then quit the driver
                    self.driver.quit()
                    driver_quit_success = True
                    logger.info("Appium driver stopped")

                except Exception as e:
                    logger.error(f"Error while quitting driver: {e}")
                finally:
                    self.driver = None
            
        except Exception as e:
            logger.error(f"Unexpected error during driver cleanup: {e}")
        finally:
            # Final attempt to unblock the device if not already done
            if not driver_quit_success:
                try:
                    self._unblock_device()
                except Exception as e:
                    logger.error(f"Final attempt to unblock device failed: {e}")

    def __enter__(self):
        """Context manager entry point."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit - ensure driver is properly closed."""
        self.quit_driver()

[raghuns2311]

@saikrishna321 @sudharsan-selvaraj

[saikrishna321]

@raghuns2311 Please provide logs as gist to under further what is going on.

[raghuns2311]

Sure

[sudharsan-selvaraj]

@raghuns2311 Since you're currently using an older version of Device Farm (9.8.4), I recommend updating to the latest version 10.0.5 to check if the issue persists.