From 681d705da06d8497af743939cae49d0bdab9d3ae Mon Sep 17 00:00:00 2001 From: Hans-Christoph Steiner Date: Tue, 5 Mar 2024 11:47:58 +0100 Subject: [PATCH] install: reliable algorithm for picking devices from adb output Versions of this algorithm are used elsewhere: * https://github.com/openatx/adbutils/blob/master/adbutils/_adb.py --- fdroidserver/install.py | 91 ++++++++++++++----------- tests/install.TestCase | 144 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 193 insertions(+), 42 deletions(-) diff --git a/fdroidserver/install.py b/fdroidserver/install.py index 3c9316e9..4f65e050 100644 --- a/fdroidserver/install.py +++ b/fdroidserver/install.py @@ -28,7 +28,6 @@ from urllib.parse import urlencode, urlparse, urlunparse from . import _ from . import common, index, net -from .common import SdkToolsPopen from .exception import FDroidException config = None @@ -85,14 +84,57 @@ def download_fdroid_apk(): def devices(): - p = SdkToolsPopen(['adb', "devices"]) + """Get the list of device serials for use with adb commands.""" + p = common.SdkToolsPopen(['adb', "devices"]) if p.returncode != 0: raise FDroidException("An error occured when finding devices: %s" % p.output) - lines = [line for line in p.output.splitlines() if not line.startswith('* ')] - if len(lines) < 3: - return [] - lines = lines[1:-1] - return [line.split()[0] for line in lines] + serials = list() + for line in p.output.splitlines(): + columns = line.strip().split("\t", maxsplit=1) + if len(columns) == 2: + serial, status = columns + if status == 'device': + serials.append(serial) + else: + d = {'serial': serial, 'status': status} + logging.warning(_('adb reports {serial} is "{status}"!'.format(**d))) + return serials + + +def install_apks_to_devices(apks): + """Install the list of APKs to all Android devices reported by `adb devices`.""" + for apk in apks: + # Get device list each time to avoid device not found errors + devs = devices() + if not devs: + raise FDroidException(_("No attached devices found")) + logging.info(_("Installing %s...") % apk) + for dev in devs: + logging.info( + _("Installing '{apkfilename}' on {dev}...").format( + apkfilename=apk, dev=dev + ) + ) + p = common.SdkToolsPopen(['adb', "-s", dev, "install", apk]) + fail = "" + for line in p.output.splitlines(): + if line.startswith("Failure"): + fail = line[9:-1] + if not fail: + continue + + if fail == "INSTALL_FAILED_ALREADY_EXISTS": + logging.warning( + _('"{apkfilename}" is already installed on {dev}.').format( + apkfilename=apk, dev=dev + ) + ) + else: + raise FDroidException( + _("Failed to install '{apkfilename}' on {dev}: {error}").format( + apkfilename=apk, dev=dev, error=fail + ) + ) def main(): @@ -152,45 +194,14 @@ def main(): for appid, apk in apks.items(): if not apk: raise FDroidException(_("No signed APK available for %s") % appid) + install_apks_to_devices(apks.values()) else: apks = { common.publishednameinfo(apkfile)[0]: apkfile for apkfile in sorted(glob.glob(os.path.join(output_dir, '*.apk'))) } - - for appid, apk in apks.items(): - # Get device list each time to avoid device not found errors - devs = devices() - if not devs: - raise FDroidException(_("No attached devices found")) - logging.info(_("Installing %s...") % apk) - for dev in devs: - logging.info( - _("Installing '{apkfilename}' on {dev}...").format( - apkfilename=apk, dev=dev - ) - ) - p = SdkToolsPopen(['adb', "-s", dev, "install", apk]) - fail = "" - for line in p.output.splitlines(): - if line.startswith("Failure"): - fail = line[9:-1] - if not fail: - continue - - if fail == "INSTALL_FAILED_ALREADY_EXISTS": - logging.warning( - _('"{apkfilename}" is already installed on {dev}.').format( - apkfilename=apk, dev=dev - ) - ) - else: - raise FDroidException( - _("Failed to install '{apkfilename}' on {dev}: {error}").format( - apkfilename=apk, dev=dev, error=fail - ) - ) + install_apks_to_devices(apks.values()) logging.info('\n' + _('Finished')) diff --git a/tests/install.TestCase b/tests/install.TestCase index 540a2bf0..6e701469 100755 --- a/tests/install.TestCase +++ b/tests/install.TestCase @@ -5,9 +5,11 @@ import inspect import os import sys +import textwrap import unittest from pathlib import Path +from unittest.mock import Mock, patch localmodule = os.path.realpath( os.path.join(os.path.dirname(inspect.getfile(inspect.currentframe())), '..') @@ -16,13 +18,17 @@ print('localmodule: ' + localmodule) if localmodule not in sys.path: sys.path.insert(0, localmodule) -import fdroidserver.common -import fdroidserver.install +import fdroidserver +from fdroidserver import common, install +from fdroidserver.exception import BuildException, FDroidException class InstallTest(unittest.TestCase): '''fdroidserver/install.py''' + def tearDown(self): + common.config = None + def test_devices(self): config = dict() fdroidserver.common.fill_config_defaults(config) @@ -35,6 +41,140 @@ class InstallTest(unittest.TestCase): for device in devices: self.assertIsInstance(device, str) + def test_devices_fail(self): + common.config = dict() + common.fill_config_defaults(common.config) + common.config['adb'] = '/bin/false' + with self.assertRaises(FDroidException): + fdroidserver.install.devices() + + def test_devices_fail_nonexistent(self): + """This is mostly just to document this strange difference in behavior""" + common.config = dict() + common.fill_config_defaults(common.config) + common.config['adb'] = '/nonexistent' + with self.assertRaises(BuildException): + fdroidserver.install.devices() + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_none(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = 'List of devices attached\n\n' + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual([], fdroidserver.install.devices()) + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_one(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = 'List of devices attached\n05995813\tdevice\n\n' + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual(['05995813'], fdroidserver.install.devices()) + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_many(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = textwrap.dedent( + """* daemon not running; starting now at tcp:5037 + * daemon started successfully + List of devices attached + RZCT809FTQM device + 05995813 device + emulator-5556 device + emulator-5554 unauthorized + 0a388e93 no permissions (missing udev rules? user is in the plugdev group); see [http://developer.android.com/tools/device.html] + 986AY133QL device + 09301JEC215064 device + 015d165c3010200e device + 4DCESKVGUC85VOTO device + + """ + ) + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual( + [ + 'RZCT809FTQM', + '05995813', + 'emulator-5556', + '986AY133QL', + '09301JEC215064', + '015d165c3010200e', + '4DCESKVGUC85VOTO', + ], + fdroidserver.install.devices(), + ) + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_error(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = textwrap.dedent( + """* daemon not running. starting it now on port 5037 * + * daemon started successfully * + ** daemon still not running + error: cannot connect to daemon + """ + ) + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual([], fdroidserver.install.devices()) + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_no_permissions(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = textwrap.dedent( + """List of devices attached + ???????????????? no permissions + """ + ) + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual([], fdroidserver.install.devices()) + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_unauthorized(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = textwrap.dedent( + """List of devices attached + aeef5e4e unauthorized + """ + ) + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual([], fdroidserver.install.devices()) + + @patch('fdroidserver.common.SdkToolsPopen') + def test_devices_with_mock_no_permissions_with_serial(self, mock_SdkToolsPopen): + p = Mock() + mock_SdkToolsPopen.return_value = p + p.output = textwrap.dedent( + """List of devices attached + 4DCESKVGUC85VOTO no permissions (missing udev rules? user is in the plugdev group); see [http://developer.android.com/tools/device.html] + + """ + ) + p.returncode = 0 + common.config = dict() + common.fill_config_defaults(common.config) + self.assertEqual([], fdroidserver.install.devices()) + + @patch('fdroidserver.net.download_using_mirrors', lambda m: 'testvalue') + def test_download_fdroid_apk_smokecheck(self): + self.assertEqual('testvalue', install.download_fdroid_apk()) + @unittest.skipUnless(os.getenv('test_download_fdroid_apk'), 'requires net access') def test_download_fdroid_apk(self): f = fdroidserver.install.download_fdroid_apk()