diff --git a/README.md b/README.md index 7151cec..ad92bc7 100644 --- a/README.md +++ b/README.md @@ -20,13 +20,15 @@ python -m sideload ## Usage ``` -sideload [APP] [--list] [--device SERIAL] [--dry-run] [--config PATH] +sideload [APP] [--list] [--update-all] [--device SERIAL] [--dry-run] [--config PATH] ``` - `APP` — app name from config (omit for interactive selection) - `--list` — list configured apps - `--device`, `-s` — ADB device serial - `--dry-run` — resolve, download, validate without installing +- `--update-all` — inspect every configured app on one selected device and update only + installed apps whose remote APK has a greater `versionCode` (incompatible with `APP` and `--list`) - `--config`, `-c` — path to config.toml (default: ./config.toml) ## Configuration diff --git a/src/sideload.py b/src/sideload.py index d28d25b..26806f4 100644 --- a/src/sideload.py +++ b/src/sideload.py @@ -118,6 +118,10 @@ class AdbDevice: transport: str # "usb" or "wifi" +class InstalledPackageError(RuntimeError): + """The installed package state could not be read safely.""" + + # ── utilities ───────────────────────────────────────────────────────────────── @@ -210,7 +214,8 @@ def _run( *args: str, check: bool = True, timeout: int = 30, **kwargs ) -> subprocess.CompletedProcess: return subprocess.run( - list(args), check=check, timeout=timeout, capture_output=True, text=True, **kwargs + list(args), check=check, timeout=timeout, capture_output=True, text=True, + shell=False, **kwargs ) @@ -907,6 +912,24 @@ class ApkValidator: ) return None + def version_code(self, apk_path: Path) -> int: + """Read the APK versionCode, using aapt2 then the legacy aapt tool.""" + errors: list[str] = [] + for binary in (self.aapt2_bin, self.aapt_bin): + try: + result = _run(binary, "dump", "badging", str(apk_path), check=False, timeout=15) + except FileNotFoundError: + errors.append(f"{binary} was not found") + continue + if result.returncode != 0: + errors.append(result.stderr.strip() or f"{binary} failed") + continue + match = re.search(r"versionCode\s*=\s*['\"](\d+)['\"]", result.stdout) + if match: + return int(match.group(1)) + errors.append(f"{binary} output has no parseable versionCode") + raise ValueError("unable to read APK versionCode: " + "; ".join(errors)) + # ── ADB management ──────────────────────────────────────────────────────────── @@ -951,6 +974,23 @@ class AdbManager: return f"install failed: {err}" return None + def installed_version_code(self, package: str, serial: str | None = None) -> int | None: + """Return versionCode, or None when dumpsys confirms the app is absent.""" + cmd = [self.binary] + if serial: + cmd += ["-s", serial] + cmd += ["shell", "dumpsys", "package", package] + result = _run(*cmd, check=False, timeout=30) + output = (result.stdout or "") + "\n" + (result.stderr or "") + if result.returncode != 0: + raise InstalledPackageError(output.strip() or "dumpsys failed") + if re.search(r"unable to find package|package .* not found", output, re.IGNORECASE): + return None + match = re.search(r"\bversionCode=(\d+)\b", result.stdout or "") + if not match: + raise InstalledPackageError("installed package versionCode is missing or invalid") + return int(match.group(1)) + # ── interactive helpers ─────────────────────────────────────────────────────── @@ -1035,6 +1075,115 @@ def _download_progress(downloaded: int, total: int | None) -> None: sys.stderr.flush() +def _prepare_device( + config: Config, adb: AdbManager, device_serial: str | None = None +) -> AdbDevice: + """Connect (at most once) and select the single device used by an operation.""" + if config.adb_mode == "wifi": + if not config.adb_address or not _validate_adb_address(config.adb_address): + raise ValueError("Invalid adb.address; expected HOST:PORT") + error = adb.connect_wifi(config.adb_address) + if error: + raise RuntimeError(error) + devices = adb.list_devices() + if not devices: + raise RuntimeError("No devices found") + if device_serial: + for device in devices: + if device.serial == device_serial: + return device + serials = ", ".join(d.serial for d in devices) + raise RuntimeError(f"Device '{device_serial}' not found. Available: {serials}") + return _choose_device(adb, devices) + + +def _download_and_validate( + app: AppEntry, + config: Config, + adb_validator: ApkValidator, + *, + progress: Callable[[int, int | None], None] | None = None, +) -> tuple[ReleaseInfo, Path]: + source = _make_source(app) + release = source.get_latest_release() + apk_path = source.download_apk(release, DownloadCache(config.cache_dir), progress=progress) + validation_error = adb_validator.validate(apk_path, app.app_id) + if validation_error: + raise ValueError(validation_error) + return release, apk_path + + +def update_all( + config: Config, + *, + device_serial: str | None = None, + dry_run: bool = False, + adb_manager: AdbManager | None = None, + validator: ApkValidator | None = None, +) -> int: + """Update installed configured apps, continuing after per-app failures.""" + adb = adb_manager or AdbManager(config.adb_binary) + apk_validator = validator or ApkValidator() + try: + device = _prepare_device(config, adb, device_serial) + except Exception as exc: + print(f"Error detecting devices: {exc}", file=sys.stderr) + return 1 + print(f"Updating {len(config.apps)} apps on {device.model} ({device.serial})…") + failures = 0 + counts: dict[str, int] = {key: 0 for key in ("updated", "current", "absent", "failed")} + + for app in config.apps: + print(f"\n{app.name} ({app.app_id})") + try: + installed = adb.installed_version_code(app.app_id, device.serial) + if installed is None: + counts["absent"] += 1 + print(" not installed (skipped; no download)") + continue + print(f" installed versionCode: {installed}") + progress_seen = False + + def report_progress(downloaded: int, total: int | None) -> None: + nonlocal progress_seen + progress_seen = True + _download_progress(downloaded, total) + + release, apk_path = _download_and_validate( + app, config, apk_validator, progress=report_progress + ) + if progress_seen: + sys.stderr.write("\n") + sys.stderr.flush() + remote_code = apk_validator.version_code(apk_path) + print(f" remote versionCode: {remote_code}") + if remote_code <= installed: + counts["current"] += 1 + state = "already up to date" if remote_code == installed else "local version is newer" + print(f" {state}; skipped") + continue + if dry_run: + counts["updated"] += 1 + print(" [dry-run] would install") + continue + error = adb.install(apk_path, device.serial) + if error: + raise RuntimeError(error) + counts["updated"] += 1 + print(f" updated to {remote_code}") + except Exception as exc: + counts["failed"] += 1 + failures += 1 + print(f" FAILED: {exc}", file=sys.stderr) + + print( + "\nSummary: " + f"{counts['updated']} updated, {counts['current']} unchanged, " + f"{counts['absent']} not installed, {counts['failed']} failed" + ) + return 1 if failures else 0 + + def sideload( app: AppEntry, config: Config, @@ -1168,6 +1317,11 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="resolve/download/validate without installing", ) + p.add_argument( + "--update-all", + action="store_true", + help="update every configured app already installed on one ADB device", + ) p.add_argument( "--config", "-c", @@ -1185,6 +1339,11 @@ def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) + if args.update_all and args.app: + parser.error("--update-all cannot be combined with the APP argument") + if args.update_all and args.list: + parser.error("--update-all cannot be combined with --list") + if not args.config.is_file(): print(f"Config not found: {args.config}", file=sys.stderr) print( @@ -1204,6 +1363,9 @@ def main(argv: list[str] | None = None) -> int: print(f" {a.name} ({a.source}) -> {a.app_id}") return 0 + if args.update_all: + return update_all(config, device_serial=args.device, dry_run=args.dry_run) + if args.app: matches = [a for a in config.apps if a.name.lower() == args.app.lower()] if not matches: diff --git a/tests/test_sideload.py b/tests/test_sideload.py index 9c7a856..a9c4c8b 100644 --- a/tests/test_sideload.py +++ b/tests/test_sideload.py @@ -844,6 +844,33 @@ class TestApkValidator: assert err is not None assert "failed" in err + def test_version_code_aapt2(self, apk_file: Path): + validator = sideload.ApkValidator(aapt2_bin="aapt2", aapt_bin="aapt") + with mock.patch( + "sideload._run", + return_value=subprocess.CompletedProcess( + args=["aapt2"], returncode=0, + stdout="package: name='org.example.test' versionCode='42' versionName='x'\n", + stderr="", + ), + ): + assert validator.version_code(apk_file) == 42 + + def test_version_code_falls_back_to_aapt(self, apk_file: Path): + validator = sideload.ApkValidator(aapt2_bin="aapt2", aapt_bin="aapt") + with mock.patch( + "sideload._run", + side_effect=[ + subprocess.CompletedProcess(args=["aapt2"], returncode=1, stdout="", stderr="bad"), + subprocess.CompletedProcess( + args=["aapt"], returncode=0, + stdout="package: name='org.example.test' versionCode='7' versionName='x'\n", + stderr="", + ), + ], + ): + assert validator.version_code(apk_file) == 7 + # ── AdbManager tests ────────────────────────────────────────────────────────── @@ -944,6 +971,32 @@ class TestAdbManager: assert err is not None assert "INSTALL_FAILED" in err + @pytest.mark.parametrize( + ("output", "expected"), + [ + ("Packages:\n versionCode=123 minSdk=26\n", 123), + ("Unable to find package org.missing\n", None), + ], + ) + def test_installed_version_code(self, output: str, expected: int | None): + adb = sideload.AdbManager(binary="fake-adb") + with mock.patch( + "sideload._run", + return_value=subprocess.CompletedProcess( + args=[], returncode=0, stdout=output, stderr="" + ), + ): + assert adb.installed_version_code("org.example.test", "SERIAL") == expected + + def test_installed_version_code_failure(self): + adb = sideload.AdbManager(binary="fake-adb") + with mock.patch( + "sideload._run", + return_value=subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="offline"), + ): + with pytest.raises(sideload.InstalledPackageError): + adb.installed_version_code("org.example.test", "SERIAL") + # ── _make_source tests ──────────────────────────────────────────────────────── @@ -982,6 +1035,58 @@ class TestMakeSource: class TestSideload: + def test_update_all_mixed_batch_continues(self, tmp_path: Path, apk_file: Path, capsys): + apps = [ + sideload.AppEntry(f"App{i}", "fdroid", f"org.example.{i}", {}) + for i in range(4) + ] + config = sideload.Config(apps, tmp_path / "cache", "fake-adb") + device = sideload.AdbDevice("SERIAL", "Pixel", "usb") + + class FakeAdb: + def __init__(self): + self.installed = iter([10, 5, None]) + self.installs: list[str] = [] + + def list_devices(self): + return [device] + + def installed_version_code(self, package, serial): + if package.endswith("3"): + raise RuntimeError("dumpsys unavailable") + return next(self.installed) + + def install(self, path, serial): + self.installs.append(serial) + return None + + class FakeValidator: + def validate(self, path, package): + return None + + def version_code(self, path): + return 11 if path.name == "remote-1.apk" else 5 + + adb = FakeAdb() + release = sideload.ReleaseInfo(apps[0], "x", "https://example.test/x.apk", None) + downloaded: list[str] = [] + + def fake_download(app, cfg, validator, *, progress=None): + downloaded.append(app.name) + path = tmp_path / ("remote-1.apk" if app.name == "App0" else "remote-2.apk") + path.write_bytes(b"apk") + return release, path + + with mock.patch("sideload._download_and_validate", side_effect=fake_download): + result = sideload.update_all(config, adb_manager=adb, validator=FakeValidator()) + assert result == 1 + assert downloaded == ["App0", "App1"] + assert adb.installs == ["SERIAL"] + captured = capsys.readouterr() + assert "1 updated" in captured.out + # The complete four-counter summary is printed on a single line. + assert "Summary: 1 updated, 1 unchanged, 1 not installed, 1 failed" in captured.out + def test_dry_run_success( self, app_entry: sideload.AppEntry, config: sideload.Config, apk_file: Path ): @@ -1199,6 +1304,169 @@ class TestSideload: assert result == 0 +class TestUpdateAll: + """Focused coverage for update_all decision paths and failure handling.""" + + def _device(self) -> sideload.AdbDevice: + return sideload.AdbDevice("SERIAL", "Pixel", "usb") + + def _run_updates( + self, + tmp_path: Path, + capsys, + *, + installed: int | None = 10, + remote_code: int = 11, + install_error: str | None = None, + version_code_error: Exception | None = None, + dry_run: bool = False, + ) -> tuple[int, list[tuple[Path, str]], pytest.CaptureFixture]: + apps = [sideload.AppEntry("App0", "fdroid", "org.example.0", {})] + config = sideload.Config(apps, tmp_path / "cache", "fake-adb") + device = self._device() + + class FakeAdb: + def __init__(self): + self.install_calls: list[tuple[Path, str]] = [] + + def list_devices(self): + return [device] + + def installed_version_code(self, package, serial): + return installed + + def install(self, path, serial): + self.install_calls.append((path, serial)) + return install_error + + class FakeValidator: + def validate(self, path, package): + return None + + def version_code(self, path): + if version_code_error is not None: + raise version_code_error + return remote_code + + adb = FakeAdb() + apk_path = tmp_path / "remote.apk" + apk_path.write_bytes(b"apk") + release = sideload.ReleaseInfo(apps[0], "x", "https://example.test/x.apk", None) + with mock.patch("sideload._download_and_validate", return_value=(release, apk_path)): + result = sideload.update_all( + config, + adb_manager=adb, + validator=FakeValidator(), + dry_run=dry_run, + ) + return result, adb.install_calls, capsys.readouterr() + + def test_dry_run_no_install_but_update_status(self, tmp_path, capsys): + result, install_calls, captured = self._run_updates( + tmp_path, capsys, installed=10, remote_code=11, dry_run=True + ) + assert result == 0 + assert install_calls == [] + assert " remote versionCode: 11" in captured.out + assert "[dry-run] would install" in captured.out + assert "Summary: 1 updated, 0 unchanged, 0 not installed, 0 failed" in captured.out + + def test_version_code_read_error_counts_failed(self, tmp_path, capsys): + error = ValueError("unable to read APK versionCode: no tools available") + result, install_calls, captured = self._run_updates( + tmp_path, capsys, installed=10, remote_code=11, version_code_error=error + ) + assert result == 1 + assert install_calls == [] + assert "unable to read APK versionCode" in captured.err + assert "Summary: 0 updated, 0 unchanged, 0 not installed, 1 failed" in captured.out + + def test_remote_version_equal_is_current(self, tmp_path, capsys): + result, install_calls, captured = self._run_updates( + tmp_path, capsys, installed=10, remote_code=10 + ) + assert result == 0 + assert install_calls == [] + assert " already up to date; skipped" in captured.out + assert "Summary: 0 updated, 1 unchanged, 0 not installed, 0 failed" in captured.out + + def test_remote_version_lower_is_current(self, tmp_path, capsys): + result, install_calls, captured = self._run_updates( + tmp_path, capsys, installed=10, remote_code=5 + ) + assert result == 0 + assert install_calls == [] + assert " local version is newer; skipped" in captured.out + assert "Summary: 0 updated, 1 unchanged, 0 not installed, 0 failed" in captured.out + + def test_install_error_returned_counts_failed(self, tmp_path, capsys): + result, install_calls, captured = self._run_updates( + tmp_path, + capsys, + installed=10, + remote_code=11, + install_error="install failed: INSTALL_FAILED_TEST", + ) + assert result == 1 + assert len(install_calls) == 1 + assert "FAILED: install failed: INSTALL_FAILED_TEST" in captured.err + assert "Summary: 0 updated, 0 unchanged, 0 not installed, 1 failed" in captured.out + + def test_absent_app_skips_download(self, tmp_path, capsys): + result, install_calls, captured = self._run_updates(tmp_path, capsys, installed=None) + assert result == 0 + assert install_calls == [] + assert " not installed (skipped; no download)" in captured.out + assert "Summary: 0 updated, 0 unchanged, 1 not installed, 0 failed" in captured.out + + def test_device_prep_failure_no_devices(self, tmp_path, capsys): + apps = [sideload.AppEntry("App0", "fdroid", "org.example.0", {})] + config = sideload.Config(apps, tmp_path / "cache", "fake-adb") + + class FakeAdb: + def list_devices(self): + return [] + + result = sideload.update_all(config, adb_manager=FakeAdb()) + assert result == 1 + captured = capsys.readouterr() + assert "Error detecting devices: No devices found" in captured.err + assert "Summary:" not in captured.out + + def test_device_prep_failure_serial_not_found(self, tmp_path, capsys): + apps = [sideload.AppEntry("App0", "fdroid", "org.example.0", {})] + config = sideload.Config(apps, tmp_path / "cache", "fake-adb") + device = self._device() + + class FakeAdb: + def list_devices(self): + return [device] + + result = sideload.update_all(config, adb_manager=FakeAdb(), device_serial="GONE") + assert result == 1 + captured = capsys.readouterr() + assert "Error detecting devices: Device 'GONE' not found. Available: SERIAL" in captured.err + + def test_device_prep_failure_wifi_connect(self, tmp_path, capsys): + apps = [sideload.AppEntry("App0", "fdroid", "org.example.0", {})] + config = sideload.Config( + apps, + tmp_path / "cache", + "fake-adb", + adb_mode="wifi", + adb_address="192.168.1.10:5555", + ) + + class FakeAdb: + def connect_wifi(self, address): + return "failed to connect: timeout" + + result = sideload.update_all(config, adb_manager=FakeAdb()) + assert result == 1 + captured = capsys.readouterr() + assert "Error detecting devices: failed to connect: timeout" in captured.err + + # ── CLI tests ───────────────────────────────────────────────────────────────── @@ -1254,6 +1522,54 @@ class TestCLI: assert args.app == "MyApp" assert args.dry_run is True + def test_update_all_parser_and_conflicts(self): + parser = sideload.build_parser() + assert parser.parse_args(["--update-all", "--dry-run"]).update_all is True + # The positional argument is optional, so these semantic conflicts are + # rejected by main() after argparse has parsed the options. + with pytest.raises(SystemExit): + sideload.main(["--update-all", "MyApp"]) + with pytest.raises(SystemExit): + sideload.main(["--update-all", "--list"]) + + def test_main_update_all_dry_run_skips_install(self, tmp_config: Path, tmp_path: Path, capsys): + device = sideload.AdbDevice("SERIAL", "Pixel", "usb") + apk_path = tmp_path / "remote.apk" + apk_path.write_bytes(b"apk") + config = sideload.Config.from_file(tmp_config) + app = config.apps[0] + release = sideload.ReleaseInfo(app, "x", "https://example.test/x.apk", None) + + class FakeAdb: + def __init__(self, binary): + self.binary = binary + + def list_devices(self): + return [device] + + def installed_version_code(self, package, serial): + return 10 + + def install(self, path, serial): + raise AssertionError("install must not run in dry-run") + + class FakeValidator: + def validate(self, path, package): + return None + + def version_code(self, path): + return 11 + + with mock.patch("sideload.AdbManager", return_value=FakeAdb("adb")), \ + mock.patch("sideload.ApkValidator", return_value=FakeValidator()), \ + mock.patch("sideload._download_and_validate", return_value=(release, apk_path)): + result = sideload.main(["--config", str(tmp_config), "--update-all", "--dry-run"]) + + assert result == 0 + captured = capsys.readouterr() + assert "[dry-run] would install" in captured.out + assert "Summary: 1 updated, 0 unchanged, 0 not installed, 0 failed" in captured.out + # ── utility tests ─────────────────────────────────────────────────────────────