"""Unit tests for sideload module — no network, ADB, or Android SDK required."""
from __future__ import annotations
import hashlib
from http.client import HTTPMessage
import io
import json
import subprocess
import sys
import urllib.error
import urllib.request
from contextlib import contextmanager
from pathlib import Path
from unittest import mock
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
import sideload # noqa: E402
# ── helpers ───────────────────────────────────────────────────────────────────
def _make_apk_bytes(content: bytes = b"fake-apk-content") -> bytes:
return content
def _sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _write_config(path: Path, content: str) -> Path:
path.write_text(content)
return path
def _mock_urlopen_response(data: bytes):
"""Return a MagicMock that acts as a urlopen context manager returning *data*."""
m = mock.MagicMock()
m.__enter__ = mock.MagicMock(return_value=m)
m.__exit__ = mock.MagicMock(return_value=False)
m.read.side_effect = [data, b""]
return m
def _generic_response(data: str | bytes, url: str):
response = _mock_urlopen_response(data.encode() if isinstance(data, str) else data)
response.geturl.return_value = url
return response
@contextmanager
def _mock_download_opener(response=None, *, error=None):
with mock.patch("sideload._generic_opener") as opener_factory:
if error is not None:
opener_factory.return_value.open.side_effect = error
else:
opener_factory.return_value.open.return_value = response
yield
# ── fixtures ──────────────────────────────────────────────────────────────────
@pytest.fixture
def tmp_config(tmp_path: Path) -> Path:
return _write_config(
tmp_path / "config.toml",
"""\
[cache]
directory = "/tmp/test-cache"
[[apps]]
name = "TestApp"
source = "fdroid"
app_id = "org.example.test"
repo_url = "https://f-droid.org/repo"
""",
)
@pytest.fixture
def tmp_config_multi(tmp_path: Path) -> Path:
return _write_config(
tmp_path / "config.toml",
"""\
[[apps]]
name = "Fennec"
source = "fdroid"
app_id = "org.mozilla.fennec_fdroid"
repo_url = "https://f-droid.org/repo"
[[apps]]
name = "NewPipe"
source = "github"
app_id = "org.schabi.newpipe"
owner = "TeamNewPipe"
repo = "NewPipe"
[[apps]]
name = "Aurora"
source = "gitlab"
app_id = "com.aurora.store"
project_url = "https://gitlab.com/AuroraOSS/AuroraStore"
""",
)
@pytest.fixture
def app_entry() -> sideload.AppEntry:
return sideload.AppEntry(
name="TestApp",
source="fdroid",
app_id="org.example.test",
config={"repo_url": "https://f-droid.org/repo"},
)
@pytest.fixture
def github_entry() -> sideload.AppEntry:
return sideload.AppEntry(
name="NewPipe",
source="github",
app_id="org.schabi.newpipe",
config={"owner": "TeamNewPipe", "repo": "NewPipe"},
)
@pytest.fixture
def gitlab_entry() -> sideload.AppEntry:
return sideload.AppEntry(
name="Aurora",
source="gitlab",
app_id="com.aurora.store",
config={"project_url": "https://gitlab.com/AuroraOSS/AuroraStore"},
)
@pytest.fixture
def codeberg_entry() -> sideload.AppEntry:
return sideload.AppEntry(
name="CoMaps",
source="codeberg",
app_id="app.comaps",
config={
"url": "https://codeberg.org/comaps/comaps",
"asset_pattern": r"^CoMaps-.*-main-release\.apk$",
},
)
@pytest.fixture
def config(tmp_config: Path) -> sideload.Config:
return sideload.Config.from_file(tmp_config)
@pytest.fixture
def cache(tmp_path: Path) -> sideload.DownloadCache:
return sideload.DownloadCache(tmp_path / "cache")
@pytest.fixture
def apk_bytes() -> bytes:
return _make_apk_bytes()
@pytest.fixture
def apk_file(tmp_path: Path, apk_bytes: bytes) -> Path:
p = tmp_path / "test.apk"
p.write_bytes(apk_bytes)
return p
# ── Config tests ──────────────────────────────────────────────────────────────
class TestConfig:
def test_from_file_basic(self, tmp_config: Path):
cfg = sideload.Config.from_file(tmp_config)
assert len(cfg.apps) == 1
assert cfg.apps[0].name == "TestApp"
assert cfg.apps[0].source == "fdroid"
assert cfg.apps[0].app_id == "org.example.test"
assert cfg.apps[0].config["repo_url"] == "https://f-droid.org/repo"
assert cfg.cache_dir == Path("/tmp/test-cache")
assert cfg.adb_binary == "adb"
def test_from_file_multi(self, tmp_config_multi: Path):
cfg = sideload.Config.from_file(tmp_config_multi)
assert len(cfg.apps) == 3
sources = {a.name: a.source for a in cfg.apps}
assert sources == {"Fennec": "fdroid", "NewPipe": "github", "Aurora": "gitlab"}
def test_from_file_default_cache(self, tmp_path: Path):
p = _write_config(
tmp_path / "cfg.toml",
'[[apps]]\nname="X"\nsource="fdroid"\napp_id="x"\nrepo_url="https://r"',
)
cfg = sideload.Config.from_file(p)
assert cfg.cache_dir == Path("~/.cache/adb-sideload").expanduser().resolve()
def test_from_file_no_apps_raises(self, tmp_path: Path):
p = _write_config(tmp_path / "cfg.toml", "")
with pytest.raises(ValueError, match="no apps configured"):
sideload.Config.from_file(p)
def test_from_file_missing_required_field(self, tmp_path: Path):
p = _write_config(
tmp_path / "cfg.toml",
'[[apps]]\nname="X"\nsource="fdroid"\napp_id="x"',
)
cfg = sideload.Config.from_file(p)
assert cfg.apps[0].config.get("repo_url") is None
def test_from_file_package_and_url_format(self, tmp_path: Path):
p = _write_config(
tmp_path / "cfg.toml",
"""\
[[apps]]
name = "NewPipe"
source = "github"
package = "org.schabi.newpipe"
url = "https://github.com/TeamNewPipe/NewPipe"
asset_pattern = ".*\\\\.apk$"
""",
)
cfg = sideload.Config.from_file(p)
assert cfg.apps[0].app_id == "org.schabi.newpipe"
assert cfg.apps[0].config["url"].startswith("https://github.com/")
def test_generic_requires_https_url_and_asset_pattern(self, tmp_path: Path):
p = _write_config(
tmp_path / "cfg.toml",
'[[apps]]\nname="X"\nsource="generic"\npackage="x"\nurl="http://example.test/"\nasset_pattern=".*\\\\.apk"',
)
with pytest.raises(ValueError, match="absolute HTTPS"):
sideload.Config.from_file(p)
p.write_text(
'[[apps]]\nname="X"\nsource="generic"\npackage="x"\n'
'url="https://example.test/"\nasset_pattern=".*\\\\.apk"\nsha256="not-a-sha"'
)
with pytest.raises(ValueError, match="64 hexadecimal"):
sideload.Config.from_file(p)
# ── DownloadCache tests ───────────────────────────────────────────────────────
class TestDownloadCache:
def test_cached_path(self, cache: sideload.DownloadCache):
p = cache.cached_path("org.example", "abc123")
assert p == cache.root / "org.example" / "abc123.apk"
def test_find_cached_missing(self, cache: sideload.DownloadCache):
assert cache.find_cached("org.example", "abc123") is None
def test_find_cached_present(self, cache: sideload.DownloadCache, apk_file: Path):
sha = sideload._sha256_file(apk_file)
cache.store("org.example", apk_file, sha)
found = cache.find_cached("org.example", sha)
assert found is not None
assert found.is_file()
def test_find_cached_wrong_sha(self, cache: sideload.DownloadCache, apk_file: Path):
cache.store("org.example", apk_file, "abc123")
assert cache.find_cached("org.example", "abc123") is None
def test_store_computes_sha(self, cache: sideload.DownloadCache, apk_file: Path):
dest = cache.store("org.example", apk_file)
assert dest.is_file()
assert dest.read_bytes() == apk_file.read_bytes()
def test_store_idempotent(self, cache: sideload.DownloadCache, apk_file: Path):
sha = sideload._sha256_file(apk_file)
d1 = cache.store("org.example", apk_file, sha)
d2 = cache.store("org.example", apk_file, sha)
assert d1 == d2
# ── FdroidSource tests ────────────────────────────────────────────────────────
class TestFdroidSource:
FDROID_INDEX = {
"packages": {
"org.example.test": {
"12345": {
"file": {
"name": "/org.example.test_12345.apk",
"sha256": "abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890",
"size": 1234567,
}
},
"12344": {
"file": {
"name": "/org.example.test_12344.apk",
"sha256": "0000000000000000000000000000000000000000000000000000000000000000",
}
},
}
}
}
def test_get_latest_release(self, app_entry: sideload.AppEntry):
source = sideload.FdroidSource(app_entry)
with mock.patch("sideload._http_get", return_value=json.dumps(self.FDROID_INDEX).encode()):
release = source.get_latest_release()
assert release.version == "12345"
assert release.download_url == "https://f-droid.org/repo/org.example.test_12345.apk"
assert release.expected_sha256 == self.FDROID_INDEX["packages"]["org.example.test"]["12345"]["file"]["sha256"]
assert release.file_size == 1234567
def test_get_latest_release_app_not_found(self, app_entry: sideload.AppEntry):
source = sideload.FdroidSource(app_entry)
index = {"packages": {}}
with mock.patch("sideload._http_get", return_value=json.dumps(index).encode()):
with pytest.raises(ValueError, match="not found in F-Droid index"):
source.get_latest_release()
def test_get_latest_release_no_versions(self, app_entry: sideload.AppEntry):
source = sideload.FdroidSource(app_entry)
index = {"packages": {"org.example.test": {}}}
with mock.patch("sideload._http_get", return_value=json.dumps(index).encode()):
with pytest.raises(ValueError, match="no versions"):
source.get_latest_release()
def test_get_latest_release_no_file_name(self, app_entry: sideload.AppEntry):
source = sideload.FdroidSource(app_entry)
index = {"packages": {"org.example.test": {"1": {"file": {}}}}}
with mock.patch("sideload._http_get", return_value=json.dumps(index).encode()):
with pytest.raises(ValueError, match="no file name"):
source.get_latest_release()
def test_get_latest_release_index_v2_versions_key(self, app_entry: sideload.AppEntry):
index = {
"packages": {
"org.example.test": {
"metadata": {"name": "Test"},
"versions": {
"12345": {
"file": {
"name": "org.example.test_12345.apk",
"sha256": "a" * 64,
}
}
},
}
}
}
source = sideload.FdroidSource(app_entry)
with mock.patch("sideload._http_get", return_value=json.dumps(index).encode()):
release = source.get_latest_release()
assert release.version == "12345"
assert release.download_url.endswith("/repo/org.example.test_12345.apk")
def test_get_latest_release_uses_manifest_version_code(self, app_entry: sideload.AppEntry):
index = {
"packages": {
"org.example.test": {
"versions": {
"sha256-of-version": {
"file": {"name": "/org.example.test_36.apk", "sha256": "a" * 64},
"manifest": {"versionCode": 36, "versionName": "5.1.1"},
},
"older-version": {
"file": {"name": "/org.example.test_34.apk", "sha256": "b" * 64},
"manifest": {"versionCode": 34, "versionName": "5.0.0"},
},
}
}
}
}
source = sideload.FdroidSource(app_entry)
with mock.patch("sideload._http_get", return_value=json.dumps(index).encode()):
release = source.get_latest_release()
assert release.version == "5.1.1"
assert release.download_url.endswith("/org.example.test_36.apk")
def test_missing_repo_url_raises(self):
entry = sideload.AppEntry(name="X", source="fdroid", app_id="x", config={})
with pytest.raises(ValueError, match="repo_url is required"):
sideload.FdroidSource(entry)
def test_download_apk_cached(self, app_entry: sideload.AppEntry, cache: sideload.DownloadCache, apk_file: Path):
sha = sideload._sha256_file(apk_file)
cache.store(app_entry.app_id, apk_file, sha)
source = sideload.FdroidSource(app_entry)
release = sideload.ReleaseInfo(
app_entry=app_entry,
version="1",
download_url="https://example.com/x.apk",
expected_sha256=sha,
)
progress = mock.Mock()
result = source.download_apk(release, cache, progress=progress)
assert result == cache.cached_path(app_entry.app_id, sha)
progress.assert_not_called()
def test_download_apk_sha256_mismatch(self, app_entry: sideload.AppEntry, cache: sideload.DownloadCache):
apk_data = b"real-content"
wrong_sha = "0" * 64
source = sideload.FdroidSource(app_entry)
release = sideload.ReleaseInfo(
app_entry=app_entry,
version="1",
download_url="https://example.com/x.apk",
expected_sha256=wrong_sha,
)
mock_resp = _mock_urlopen_response(apk_data)
with _mock_download_opener(mock_resp):
with pytest.raises(ValueError, match="SHA-256 mismatch"):
source.download_apk(release, cache)
# ── GitHubSource tests ────────────────────────────────────────────────────────
class TestGitHubSource:
GITHUB_RELEASE = {
"tag_name": "v0.27.0",
"assets": [
{
"name": "NewPipe_v0.27.0.apk",
"browser_download_url": "https://github.com/TeamNewPipe/NewPipe/releases/download/v0.27.0/NewPipe_v0.27.0.apk",
"size": 12345678,
},
{
"name": "Source code (zip)",
"browser_download_url": "https://github.com/TeamNewPipe/NewPipe/archive/v0.27.0.zip",
},
],
}
def test_get_latest_release(self, github_entry: sideload.AppEntry):
source = sideload.GitHubSource(github_entry)
with mock.patch("sideload._http_get", return_value=json.dumps(self.GITHUB_RELEASE).encode()):
release = source.get_latest_release()
assert release.version == "0.27.0"
assert "NewPipe_v0.27.0.apk" in release.download_url
assert release.expected_sha256 is None
assert release.file_size == 12345678
def test_get_latest_release_no_apk(self, github_entry: sideload.AppEntry):
source = sideload.GitHubSource(github_entry)
data = {"tag_name": "v1.0", "assets": [{"name": "source.zip"}]}
with mock.patch("sideload._http_get", return_value=json.dumps(data).encode()):
with pytest.raises(ValueError, match="no APK asset"):
source.get_latest_release()
def test_missing_owner_raises(self):
entry = sideload.AppEntry(name="X", source="github", app_id="x", config={"repo": "r"})
with pytest.raises(ValueError, match="owner and repo are required"):
sideload.GitHubSource(entry)
def test_missing_repo_raises(self):
entry = sideload.AppEntry(name="X", source="github", app_id="x", config={"owner": "o"})
with pytest.raises(ValueError, match="owner and repo are required"):
sideload.GitHubSource(entry)
def test_download_apk(self, github_entry: sideload.AppEntry, cache: sideload.DownloadCache):
source = sideload.GitHubSource(github_entry)
release = sideload.ReleaseInfo(
app_entry=github_entry,
version="0.27.0",
download_url="https://example.com/x.apk",
expected_sha256=None,
)
apk_data = b"github-apk-content"
mock_resp = _mock_urlopen_response(apk_data)
with _mock_download_opener(mock_resp):
result = source.download_apk(release, cache)
assert result.is_file()
assert result.read_bytes() == apk_data
# ── CodebergSource tests ──────────────────────────────────────────────────────
class TestCodebergSource:
CODEBERG_RELEASE = {
"tag_name": "v2026.08.0",
"assets": [
{
"name": "CoMaps-2026.08.0-main-release.apk",
"browser_download_url": "https://codeberg.org/comaps/comaps/releases/download/v2026.08.0/CoMaps-2026.08.0-main-release.apk",
"size": 12345678,
},
{
"name": "CoMaps-2026.08.0-debug.apk",
"browser_download_url": "https://codeberg.org/comaps/comaps/releases/download/v2026.08.0/CoMaps-2026.08.0-debug.apk",
},
{
"name": "CoMaps-2026.08.0-test.apk",
"browser_download_url": "https://codeberg.org/comaps/comaps/releases/download/v2026.08.0/CoMaps-2026.08.0-test.apk",
},
{
"name": "CoMaps-2026.08.0-unsigned.apk",
"browser_download_url": "https://codeberg.org/comaps/comaps/releases/download/v2026.08.0/CoMaps-2026.08.0-unsigned.apk",
},
{
"name": "Source code (zip)",
"browser_download_url": "https://codeberg.org/comaps/comaps/archive/v2026.08.0.zip",
},
],
}
def test_get_latest_release(self, codeberg_entry: sideload.AppEntry):
source = sideload.CodebergSource(codeberg_entry)
with mock.patch("sideload._http_get", return_value=json.dumps(self.CODEBERG_RELEASE).encode()) as http_get:
release = source.get_latest_release()
http_get.assert_called_once_with(
"https://codeberg.org/api/v1/repos/comaps/comaps/releases/latest"
)
assert release.version == "2026.08.0"
assert release.download_url.endswith("CoMaps-2026.08.0-main-release.apk")
assert release.file_size == 12345678
def test_get_latest_release_ambiguous_assets(self, codeberg_entry: sideload.AppEntry):
source = sideload.CodebergSource(codeberg_entry)
release = {
"tag_name": "v1.0",
"assets": [
{"name": "CoMaps-one-main-release.apk", "browser_download_url": "https://example.com/one.apk"},
{"name": "CoMaps-two-main-release.apk", "browser_download_url": "https://example.com/two.apk"},
],
}
with mock.patch("sideload._http_get", return_value=json.dumps(release).encode()):
with pytest.raises(ValueError, match="multiple APK assets match"):
source.get_latest_release()
def test_get_latest_release_no_apk(self, codeberg_entry: sideload.AppEntry):
source = sideload.CodebergSource(codeberg_entry)
release = {"tag_name": "v1.0", "assets": [{"name": "Source code (tar.gz)"}]}
with mock.patch("sideload._http_get", return_value=json.dumps(release).encode()):
with pytest.raises(ValueError, match="no APK asset"):
source.get_latest_release()
@pytest.mark.parametrize(
"url",
[
"http://codeberg.org/comaps/comaps",
"https://example.com/comaps/comaps",
"https://codeberg.org/comaps",
"https://codeberg.org/comaps/comaps/releases",
],
)
def test_invalid_url_raises(self, url: str):
entry = sideload.AppEntry(
name="CoMaps", source="codeberg", app_id="app.comaps", config={"url": url}
)
with pytest.raises(ValueError, match="HTTPS Codeberg repository URL"):
sideload.CodebergSource(entry)
def test_get_latest_release_network_error(self, codeberg_entry: sideload.AppEntry):
source = sideload.CodebergSource(codeberg_entry)
with mock.patch("sideload._http_get", side_effect=RuntimeError("network down")):
with pytest.raises(RuntimeError, match="network down"):
source.get_latest_release()
# ── GitLabSource tests ────────────────────────────────────────────────────────
class TestGitLabSource:
GITLAB_RELEASE = [
{
"tag_name": "v4.4.1",
"name": "Aurora Store 4.4.1",
"assets": {
"links": [
{
"name": "AuroraStore-4.4.1.apk",
"url": "https://gitlab.com/AuroraOSS/AuroraStore/-/releases/v4.4.1/downloads/AuroraStore-4.4.1.apk",
}
]
},
}
]
def test_get_latest_release(self, gitlab_entry: sideload.AppEntry):
source = sideload.GitLabSource(gitlab_entry)
with mock.patch("sideload._http_get", return_value=json.dumps(self.GITLAB_RELEASE).encode()):
release = source.get_latest_release()
assert release.version == "4.4.1"
assert "AuroraStore-4.4.1.apk" in release.download_url
assert release.expected_sha256 is None
def test_get_latest_release_no_releases(self, gitlab_entry: sideload.AppEntry):
source = sideload.GitLabSource(gitlab_entry)
with mock.patch("sideload._http_get", return_value=b"[]"):
with pytest.raises(ValueError, match="no releases found"):
source.get_latest_release()
def test_get_latest_release_no_apk(self, gitlab_entry: sideload.AppEntry):
source = sideload.GitLabSource(gitlab_entry)
data = [{"tag_name": "v1.0", "assets": {"links": [{"name": "source.zip"}]}}]
with mock.patch("sideload._http_get", return_value=json.dumps(data).encode()):
with pytest.raises(ValueError, match="no APK link"):
source.get_latest_release()
def test_missing_project_url_raises(self):
entry = sideload.AppEntry(name="X", source="gitlab", app_id="x", config={})
with pytest.raises(ValueError, match="project_url is required"):
sideload.GitLabSource(entry)
def test_download_apk(self, gitlab_entry: sideload.AppEntry, cache: sideload.DownloadCache):
source = sideload.GitLabSource(gitlab_entry)
release = sideload.ReleaseInfo(
app_entry=gitlab_entry,
version="4.4.1",
download_url="https://example.com/x.apk",
expected_sha256=None,
)
apk_data = b"gitlab-apk-content"
mock_resp = _mock_urlopen_response(apk_data)
with _mock_download_opener(mock_resp):
result = source.download_apk(release, cache)
assert result.is_file()
assert result.read_bytes() == apk_data
# ── ApkValidator tests ────────────────────────────────────────────────────────
class TestGenericSource:
def _entry(self, **config):
return sideload.AppEntry(
name="Firefox", source="generic", app_id="org.mozilla.firefox",
config={
"url": "https://downloads.example/releases/",
"intermediate_patterns": [r"(\d+)\/$", "android/$"],
"asset_pattern": r"^fenix-.*\.apk$",
**config,
},
)
def test_directory_navigation_and_numeric_version_selection(self):
entry = self._entry()
pages = [
_generic_response('110bad', entry.config["url"]),
_generic_response('android', "https://downloads.example/releases/10/"),
_generic_response('apk', "https://downloads.example/releases/10/android/"),
]
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.side_effect = pages
release = sideload.GenericSource(entry).get_latest_release()
assert release.download_url.endswith(".apk?download=1")
assert release.version == "fenix-1.multi.android-arm64-v8a.apk"
def test_ambiguous_intermediate_links(self):
entry = self._entry(intermediate_patterns=[r"release/$"])
page = _generic_response('ab', entry.config["url"])
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.return_value = page
with pytest.raises(ValueError, match="multiple intermediate"):
sideload.GenericSource(entry).get_latest_release()
def test_excludes_debug_and_rejects_http_assets(self):
entry = self._entry(intermediate_patterns=[], asset_pattern=r"^good\.apk$")
page = _generic_response(
'http'
'debuggood',
entry.config["url"],
)
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.return_value = page
release = sideload.GenericSource(entry).get_latest_release()
assert release.download_url.endswith("good.apk")
def test_initial_http_and_network_error(self):
bad = self._entry(url="http://downloads.example/releases/")
with pytest.raises(ValueError, match="absolute HTTPS"):
sideload.GenericSource(bad)
entry = self._entry()
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.side_effect = urllib.error.URLError("offline")
with pytest.raises(RuntimeError, match="network error"):
sideload.GenericSource(entry).get_latest_release()
def test_headers_are_sent_but_not_exposed(self, capsys):
entry = self._entry(intermediate_patterns=[], asset_pattern=r"^good\.apk$", headers={"Authorization": "secret-token"})
response = _generic_response('good', entry.config["url"])
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.return_value = response
sideload.GenericSource(entry).get_latest_release()
request = opener_factory.return_value.open.call_args.args[0]
assert request.get_header("Authorization") == "secret-token"
captured = capsys.readouterr()
assert "secret-token" not in captured.out + captured.err
def test_https_redirect_is_blocked_before_following(self):
request = urllib.request.Request(
"https://downloads.example/start", headers={"Authorization": "secret-token"}
)
with pytest.raises(ValueError, match="non-HTTPS"):
sideload._HttpsOnlyRedirectHandler().redirect_request(
request, mock.Mock(), 302, "Found", {}, "http://evil.example/"
)
def test_cross_host_redirect_is_blocked_and_same_origin_keeps_headers(self):
request = urllib.request.Request(
"https://downloads.example/start", headers={"Authorization": "secret-token"}
)
handler = sideload._HttpsOnlyRedirectHandler(
"https://downloads.example/start", same_origin=True
)
with pytest.raises(ValueError, match="changed origin"):
handler.redirect_request(
request, mock.Mock(), 302, "Found", {}, "https://cdn.example/file"
)
redirected = handler.redirect_request(
request, mock.Mock(), 302, "Found", {}, "https://downloads.example/file"
)
assert redirected.get_header("Authorization") == "secret-token"
def test_github_cdn_redirect_drops_unredirected_host_and_keeps_authorization(self):
request = urllib.request.Request(
"https://github.com/example/app/releases/download/v1/app.apk",
headers={"Authorization": "secret-token"},
)
request.unredirected_hdrs["Host"] = "github.com"
handler = sideload._HttpsOnlyRedirectHandler(same_origin=False)
redirected = handler.redirect_request(
request,
mock.Mock(),
302,
"Found",
{},
"https://objects.githubusercontent.com/app.apk",
)
assert redirected.get_header("Host") is None
assert redirected.get_header("Authorization") == "secret-token"
def test_https_redirect_accepts_http_message_headers(self):
request = urllib.request.Request(
"https://downloads.example/start", headers={"Authorization": "secret-token"}
)
headers = HTTPMessage()
headers.add_header("Location", "https://downloads.example/file.apk")
handler = sideload._HttpsOnlyRedirectHandler(
"https://downloads.example/start", same_origin=True
)
redirected = handler.redirect_request(
request,
mock.Mock(),
302,
"Found",
headers,
"https://downloads.example/file.apk",
)
assert redirected.full_url == "https://downloads.example/file.apk"
assert redirected.get_header("Authorization") == "secret-token"
def test_atomic_download_rejects_downgrade_before_response_body(self):
request = urllib.request.Request("https://downloads.example/file.apk")
handler = sideload._HttpsOnlyRedirectHandler(
"https://downloads.example/file.apk", same_origin=False
)
with pytest.raises(ValueError, match="non-HTTPS"):
handler.redirect_request(
request, mock.Mock(read=mock.Mock(side_effect=AssertionError)),
302, "Found", {}, "http://downloads.example/file.apk"
)
def test_redirected_page_url_is_used_for_relative_links(self):
entry = self._entry(intermediate_patterns=[] , asset_pattern=r"^good\.apk$")
response = _generic_response('good', "https://cdn.example/final/")
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.return_value = response
release = sideload.GenericSource(entry).get_latest_release()
assert release.download_url == "https://cdn.example/final/good.apk"
def test_generic_cache_uses_conditional_request_without_sha(self, tmp_path: Path):
entry = self._entry(intermediate_patterns=[] , asset_pattern=r"^good\.apk$")
source = sideload.GenericSource(entry)
release = sideload.ReleaseInfo(entry, "good.apk", "https://downloads.example/good.apk", None)
first = _generic_response(b"apk-content", release.download_url)
first.headers = {"ETag": "v1"}
first.getcode.return_value = 200
second = _generic_response(b"", release.download_url)
second.headers = {}
second.getcode.return_value = 304
with mock.patch("sideload._generic_opener") as opener_factory:
opener_factory.return_value.open.side_effect = [first, second]
cache = sideload.DownloadCache(tmp_path / "cache")
cached_first = source.download_apk(release, cache)
cached_second = source.download_apk(release, cache)
assert cached_second == cached_first
assert cached_second.read_bytes() == b"apk-content"
second_request = opener_factory.return_value.open.call_args_list[1].args[0]
assert second_request.get_header("If-none-match") == "v1"
def test_generic_cache_rejects_304_without_valid_cache(self, tmp_path: Path):
entry = self._entry(intermediate_patterns=[], asset_pattern=r"^good\.apk$")
release = sideload.ReleaseInfo(entry, "good.apk", "https://downloads.example/good.apk", None)
response = _generic_response(b"", release.download_url)
response.getcode.return_value = 304
with _mock_download_opener(response):
with pytest.raises(ValueError, match="304"):
sideload.GenericSource(entry).download_apk(
release, sideload.DownloadCache(tmp_path / "cache")
)
class TestApkValidator:
def test_validate_success(self, apk_file: Path):
validator = sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout="org.example.test\n", stderr=""
),
):
err = validator.validate(apk_file, "org.example.test")
assert err is None
def test_validate_mismatch(self, apk_file: Path):
validator = sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout="com.other.app\n", stderr=""
),
):
err = validator.validate(apk_file, "org.example.test")
assert err is not None
assert "mismatch" in err
def test_validate_aapt_fails(self, apk_file: Path):
validator = sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr="invalid APK"
),
):
err = validator.validate(apk_file, "org.example.test")
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 ──────────────────────────────────────────────────────────
class TestAdbManager:
ADB_DEVICES_OUTPUT = (
"List of devices attached\n"
"R5CT1234ABCD device usb:1-1 product:blueline model:Pixel_3 device:blueline transport_id:1\n"
"192.168.1.100:5555 device product:redfin model:Pixel_5 device:redfin transport_id:2\n"
)
def test_list_devices(self):
adb = sideload.AdbManager(binary="fake-adb")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout=self.ADB_DEVICES_OUTPUT, stderr=""
),
):
devices = adb.list_devices()
assert len(devices) == 2
assert devices[0].serial == "R5CT1234ABCD"
assert devices[0].model == "Pixel_3"
assert devices[0].transport == "usb"
assert devices[1].serial == "192.168.1.100:5555"
assert devices[1].model == "Pixel_5"
assert devices[1].transport == "wifi"
def test_list_devices_empty(self):
adb = sideload.AdbManager(binary="fake-adb")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout="List of devices attached\n", stderr=""
),
):
devices = adb.list_devices()
assert devices == []
def test_list_devices_skips_offline(self):
adb = sideload.AdbManager(binary="fake-adb")
output = "List of devices attached\nABCD1234 offline\n"
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout=output, stderr=""
),
):
devices = adb.list_devices()
assert devices == []
def test_connect_wifi_success(self):
adb = sideload.AdbManager(binary="fake-adb")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout="connected to 192.168.1.100:5555\n", stderr=""
),
):
err = adb.connect_wifi("192.168.1.100:5555")
assert err is None
def test_connect_wifi_failure(self):
adb = sideload.AdbManager(binary="fake-adb")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=1, stdout="cannot connect\n", stderr=""
),
):
err = adb.connect_wifi("192.168.1.100:5555")
assert err is not None
assert "failed to connect" in err
def test_install_success(self, apk_file: Path):
adb = sideload.AdbManager(binary="fake-adb")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout="Performing Streamed Install\nSuccess\n", stderr=""
),
):
err = adb.install(apk_file, serial="R5CT1234ABCD")
assert err is None
def test_install_failure(self, apk_file: Path):
adb = sideload.AdbManager(binary="fake-adb")
with mock.patch(
"sideload._run",
return_value=subprocess.CompletedProcess(
args=[],
returncode=1,
stdout="",
stderr="INSTALL_FAILED_INSUFFICIENT_STORAGE",
),
):
err = adb.install(apk_file)
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 ────────────────────────────────────────────────────────
class TestMakeSource:
def test_fdroid(self, app_entry: sideload.AppEntry):
s = sideload._make_source(app_entry)
assert isinstance(s, sideload.FdroidSource)
def test_github(self, github_entry: sideload.AppEntry):
s = sideload._make_source(github_entry)
assert isinstance(s, sideload.GitHubSource)
def test_gitlab(self, gitlab_entry: sideload.AppEntry):
s = sideload._make_source(gitlab_entry)
assert isinstance(s, sideload.GitLabSource)
def test_codeberg(self, codeberg_entry: sideload.AppEntry):
s = sideload._make_source(codeberg_entry)
assert isinstance(s, sideload.CodebergSource)
def test_generic(self):
entry = sideload.AppEntry(
name="X", source="generic", app_id="x",
config={"url": "https://example.test/", "asset_pattern": r"x\.apk"},
)
assert isinstance(sideload._make_source(entry), sideload.GenericSource)
def test_unknown_source_raises(self):
entry = sideload.AppEntry(name="X", source="unknown", app_id="x", config={})
with pytest.raises(ValueError, match="unknown source type"):
sideload._make_source(entry)
# ── sideload orchestration tests ──────────────────────────────────────────────
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
):
"""Full dry-run flow: resolve, download, validate, skip install."""
fdroid_index = {
"packages": {
"org.example.test": {
"1": {
"file": {
"name": "/repo/org.example.test_1.apk",
"sha256": _sha256(apk_file.read_bytes()),
}
}
}
}
}
mock_resp = _mock_urlopen_response(apk_file.read_bytes())
def mock_run(*args, **kwargs):
return subprocess.CompletedProcess(
args=list(args), returncode=0, stdout="org.example.test\n", stderr=""
)
with mock.patch(
"sideload._http_get", return_value=json.dumps(fdroid_index).encode()
):
with _mock_download_opener(mock_resp):
with mock.patch("sideload._run", side_effect=mock_run):
result = sideload.sideload(
app_entry, config, dry_run=True,
validator=sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt"),
)
assert result == 0
def test_resolve_error(self, app_entry: sideload.AppEntry, config: sideload.Config):
with mock.patch("sideload._http_get", side_effect=RuntimeError("network down")):
result = sideload.sideload(app_entry, config, dry_run=True)
assert result == 1
def test_validation_fails(
self, app_entry: sideload.AppEntry, config: sideload.Config, apk_file: Path
):
fdroid_index = {
"packages": {
"org.example.test": {
"1": {
"file": {
"name": "/repo/org.example.test_1.apk",
"sha256": _sha256(apk_file.read_bytes()),
}
}
}
}
}
mock_resp = _mock_urlopen_response(apk_file.read_bytes())
def mock_run(*args, **kwargs):
return subprocess.CompletedProcess(
args=list(args), returncode=0, stdout="com.wrong.id\n", stderr=""
)
with mock.patch(
"sideload._http_get", return_value=json.dumps(fdroid_index).encode()
):
with _mock_download_opener(mock_resp):
with mock.patch("sideload._run", side_effect=mock_run):
result = sideload.sideload(
app_entry, config, dry_run=True,
validator=sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt"),
)
assert result == 1
def test_no_devices(
self, app_entry: sideload.AppEntry, config: sideload.Config, apk_file: Path
):
fdroid_index = {
"packages": {
"org.example.test": {
"1": {
"file": {
"name": "/repo/org.example.test_1.apk",
"sha256": _sha256(apk_file.read_bytes()),
}
}
}
}
}
mock_resp = _mock_urlopen_response(apk_file.read_bytes())
call_count = 0
def mock_run(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
return subprocess.CompletedProcess(
args=list(args), returncode=0, stdout="org.example.test\n", stderr=""
)
return subprocess.CompletedProcess(
args=list(args),
returncode=0,
stdout="List of devices attached\n",
stderr="",
)
with mock.patch(
"sideload._http_get", return_value=json.dumps(fdroid_index).encode()
):
with _mock_download_opener(mock_resp):
with mock.patch("sideload._run", side_effect=mock_run):
result = sideload.sideload(
app_entry, config,
validator=sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt"),
adb_manager=sideload.AdbManager(binary="fake-adb"),
)
assert result == 1
def test_device_not_found_by_serial(
self, app_entry: sideload.AppEntry, config: sideload.Config, apk_file: Path
):
fdroid_index = {
"packages": {
"org.example.test": {
"1": {
"file": {
"name": "/repo/org.example.test_1.apk",
"sha256": _sha256(apk_file.read_bytes()),
}
}
}
}
}
mock_resp = _mock_urlopen_response(apk_file.read_bytes())
call_count = 0
def mock_run(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
return subprocess.CompletedProcess(
args=list(args), returncode=0, stdout="org.example.test\n", stderr=""
)
return subprocess.CompletedProcess(
args=list(args),
returncode=0,
stdout="List of devices attached\nABCD1234 device model:X\n",
stderr="",
)
with mock.patch(
"sideload._http_get", return_value=json.dumps(fdroid_index).encode()
):
with _mock_download_opener(mock_resp):
with mock.patch("sideload._run", side_effect=mock_run):
result = sideload.sideload(
app_entry, config,
device_serial="NONEXISTENT",
validator=sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt"),
adb_manager=sideload.AdbManager(binary="fake-adb"),
)
assert result == 1
def test_full_install_success(
self, app_entry: sideload.AppEntry, config: sideload.Config, apk_file: Path
):
fdroid_index = {
"packages": {
"org.example.test": {
"1": {
"file": {
"name": "/repo/org.example.test_1.apk",
"sha256": _sha256(apk_file.read_bytes()),
}
}
}
}
}
mock_resp = _mock_urlopen_response(apk_file.read_bytes())
call_count = 0
def mock_run(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
return subprocess.CompletedProcess(
args=list(args), returncode=0, stdout="org.example.test\n", stderr=""
)
if call_count == 2:
return subprocess.CompletedProcess(
args=list(args),
returncode=0,
stdout="List of devices attached\nABCD1234 device model:Pixel\n",
stderr="",
)
return subprocess.CompletedProcess(
args=list(args),
returncode=0,
stdout="Performing Streamed Install\nSuccess\n",
stderr="",
)
with mock.patch(
"sideload._http_get", return_value=json.dumps(fdroid_index).encode()
):
with _mock_download_opener(mock_resp):
with mock.patch("sideload._run", side_effect=mock_run):
result = sideload.sideload(
app_entry, config,
device_serial="ABCD1234",
validator=sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt"),
adb_manager=sideload.AdbManager(binary="fake-adb"),
)
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 ─────────────────────────────────────────────────────────────────
class TestCLI:
def test_list(self, tmp_config: Path, capsys):
result = sideload.main(["--config", str(tmp_config), "--list"])
assert result == 0
captured = capsys.readouterr()
assert "TestApp" in captured.out
def test_config_not_found(self, capsys):
result = sideload.main(["--config", "/nonexistent/config.toml"])
assert result == 1
captured = capsys.readouterr()
assert "Config not found" in captured.err
def test_app_not_found(self, tmp_config: Path, capsys):
result = sideload.main(["--config", str(tmp_config), "NonExistent"])
assert result == 1
captured = capsys.readouterr()
assert "not found" in captured.err
def test_ambiguous_name(self, tmp_config_multi: Path, capsys):
apk_data = b"fake"
real_sha = _sha256(apk_data)
fdroid_index = {
"packages": {
"org.mozilla.fennec_fdroid": {
"1": {"file": {"name": "/repo/fennec.apk", "sha256": real_sha}}
}
}
}
mock_resp = _mock_urlopen_response(apk_data)
with mock.patch("sideload._http_get", return_value=json.dumps(fdroid_index).encode()):
with _mock_download_opener(mock_resp):
with mock.patch("sideload._run", return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout="org.mozilla.fennec_fdroid\n", stderr=""
)):
result = sideload.main(
["--config", str(tmp_config_multi), "fennec", "--dry-run"]
)
assert result == 0
def test_version(self, capsys):
with pytest.raises(SystemExit):
sideload.main(["--version"])
def test_build_parser(self):
parser = sideload.build_parser()
args = parser.parse_args(["--list"])
assert args.list is True
args = parser.parse_args(["MyApp", "--dry-run"])
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 ─────────────────────────────────────────────────────────────
class TestUtilities:
def test_sha256_file(self, tmp_path: Path):
p = tmp_path / "test.bin"
p.write_bytes(b"hello world")
expected = hashlib.sha256(b"hello world").hexdigest()
assert sideload._sha256_file(p) == expected
def test_make_apk_stem(self, app_entry: sideload.AppEntry):
release = sideload.ReleaseInfo(
app_entry=app_entry,
version="1.2.3",
download_url="https://example.com/x.apk",
expected_sha256=None,
)
stem = sideload._make_apk_stem(release)
assert stem == "org.example.test_1.2.3"
def test_make_apk_stem_sanitizes_slashes(self, app_entry: sideload.AppEntry):
release = sideload.ReleaseInfo(
app_entry=app_entry,
version="feature/branch",
download_url="https://example.com/x.apk",
expected_sha256=None,
)
stem = sideload._make_apk_stem(release)
assert "/" not in stem
def test_atomic_download(self, tmp_path: Path):
dest = tmp_path / "sub" / "file.apk"
data = b"downloaded-content"
mock_resp = _mock_urlopen_response(data)
with _mock_download_opener(mock_resp):
result = sideload._atomic_download("https://example.com/x.apk", dest)
assert result == dest
assert dest.read_bytes() == data
assert not dest.with_suffix(".apk.tmp").exists()
def test_atomic_download_reports_content_length_progress(self, tmp_path: Path):
data = b"downloaded-content"
response = _mock_urlopen_response(data)
response.headers = {"Content-Length": str(len(data))}
updates = []
with _mock_download_opener(response):
result = sideload._atomic_download(
"https://example.com/x.apk",
tmp_path / "test.apk",
progress=lambda downloaded, total: updates.append((downloaded, total)),
)
assert result.read_bytes() == data
assert updates[0] == (0, len(data))
assert updates[-1] == (len(data), len(data))
def test_atomic_download_reports_indeterminate_progress(self, tmp_path: Path):
data = b"downloaded-content"
response = _mock_urlopen_response(data)
response.headers = {}
updates = []
with _mock_download_opener(response):
result = sideload._atomic_download(
"https://example.com/x.apk",
tmp_path / "test.apk",
expected_size=None,
progress=lambda downloaded, total: updates.append((downloaded, total)),
)
assert result.read_bytes() == data
assert updates[0] == (0, None)
assert updates[-1] == (len(data), None)
def test_atomic_download_uses_expected_size_without_content_length(self, tmp_path: Path):
data = b"downloaded-content"
response = _mock_urlopen_response(data)
response.headers = {}
updates = []
with _mock_download_opener(response):
sideload._atomic_download(
"https://example.com/x.apk",
tmp_path / "test.apk",
expected_size=len(data),
progress=lambda downloaded, total: updates.append((downloaded, total)),
)
assert updates[0] == (0, len(data))
assert updates[-1] == (len(data), len(data))
def test_atomic_download_cleans_up_on_error(self, tmp_path: Path):
dest = tmp_path / "file.apk"
with _mock_download_opener(error=RuntimeError("fail")):
with pytest.raises(RuntimeError):
sideload._atomic_download("https://example.com/x.apk", dest)
assert not dest.exists()
assert not dest.with_suffix(".apk.tmp").exists()
def test_http_get_error(self):
with mock.patch(
"urllib.request.urlopen",
side_effect=urllib.error.HTTPError(
"https://example.com", 404, "Not Found", {}, None
),
):
with pytest.raises(RuntimeError, match="HTTP 404"):
sideload._http_get("https://example.com")
def test_http_get_network_error(self):
with mock.patch(
"urllib.request.urlopen",
side_effect=urllib.error.URLError("connection refused"),
):
with pytest.raises(RuntimeError, match="network error"):
sideload._http_get("https://example.com")
def test_run_passes_args_correctly(self):
with mock.patch("subprocess.run") as mock_run:
mock_run.return_value = subprocess.CompletedProcess(
args=["echo", "hello"], returncode=0, stdout="hello\n", stderr=""
)
sideload._run("echo", "hello")
mock_run.assert_called_once()
call_args = mock_run.call_args[0][0]
assert call_args == ["echo", "hello"]
# ── AdbDevice dataclass ───────────────────────────────────────────────────────
class TestAdbDevice:
def test_creation(self):
d = sideload.AdbDevice(serial="ABC123", model="Pixel", transport="usb")
assert d.serial == "ABC123"
assert d.model == "Pixel"
assert d.transport == "usb"
# ── ReleaseInfo dataclass ─────────────────────────────────────────────────────
class TestReleaseInfo:
def test_creation(self, app_entry: sideload.AppEntry):
r = sideload.ReleaseInfo(
app_entry=app_entry,
version="1.0",
download_url="https://example.com/x.apk",
expected_sha256="abc",
file_size=1000,
)
assert r.version == "1.0"
assert r.file_size == 1000