264 lines
8.3 KiB
Python
264 lines
8.3 KiB
Python
"""The OTA feed: which artifact wins, and what the phone receives."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import zipfile
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.config import env_settings
|
|
from app.services import android_updates
|
|
|
|
|
|
def _configure(monkeypatch, *, token: str = "read-token") -> None:
|
|
monkeypatch.setenv("GITEA_URL", "https://git.example.test")
|
|
monkeypatch.setenv("GITEA_REPO", "timka123/kadence")
|
|
monkeypatch.setenv("GITEA_UPDATE_BRANCH", "master")
|
|
monkeypatch.setenv("GITEA_TOKEN", token)
|
|
env_settings.cache_clear()
|
|
android_updates.reset_cache()
|
|
|
|
|
|
def _artifact(
|
|
artifact_id: int,
|
|
*,
|
|
run_id: int,
|
|
run_number: int,
|
|
name: str = android_updates.METADATA_ARTIFACT,
|
|
branch: str = "master",
|
|
conclusion: str = "success",
|
|
expired: bool = False,
|
|
created_at: str = "2026-10-20T10:00:00Z",
|
|
size: int = 100,
|
|
) -> dict:
|
|
return {
|
|
"id": artifact_id,
|
|
"name": name,
|
|
"created_at": created_at,
|
|
"size_in_bytes": size,
|
|
"expired": expired,
|
|
"workflow_run": {
|
|
"id": run_id,
|
|
"run_number": run_number,
|
|
"head_branch": branch,
|
|
"head_sha": f"sha-{run_number}",
|
|
"conclusion": conclusion,
|
|
},
|
|
}
|
|
|
|
|
|
def _zip(files: dict[str, bytes]) -> bytes:
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w") as archive:
|
|
for name, body in files.items():
|
|
archive.writestr(name, body)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _install_transport(monkeypatch, handler) -> None:
|
|
transport = httpx.MockTransport(handler)
|
|
|
|
def client(_settings, *, timeout: float = 30.0) -> httpx.Client:
|
|
return httpx.Client(
|
|
base_url="https://git.example.test",
|
|
transport=transport,
|
|
timeout=timeout,
|
|
follow_redirects=True,
|
|
)
|
|
|
|
monkeypatch.setattr(android_updates, "_client", client)
|
|
|
|
|
|
def _metadata_body(version_code: int, version_name: str) -> bytes:
|
|
return json.dumps(
|
|
{
|
|
"version_code": version_code,
|
|
"version_name": version_name,
|
|
"commit": f"sha-{version_code}",
|
|
"built_at": "2026-10-20T10:00:00Z",
|
|
"size_bytes": 1234,
|
|
"sha256": "abc",
|
|
"signer_sha256": "def",
|
|
}
|
|
).encode()
|
|
|
|
|
|
def test_newest_successful_master_build_wins(monkeypatch):
|
|
_configure(monkeypatch)
|
|
metadata_artifacts = [
|
|
_artifact(11, run_id=101, run_number=10),
|
|
_artifact(12, run_id=102, run_number=20),
|
|
# A newer tag build must not become the master channel.
|
|
_artifact(13, run_id=103, run_number=30, branch="v0.2.0"),
|
|
# A failed build uploaded an artifact but must never be offered.
|
|
_artifact(14, run_id=104, run_number=40, conclusion="failure"),
|
|
# An expired artifact is not downloadable any more.
|
|
_artifact(15, run_id=105, run_number=50, expired=True),
|
|
]
|
|
apk_artifacts = [
|
|
_artifact(
|
|
artifact_id + 100,
|
|
run_id=run_id,
|
|
run_number=run_number,
|
|
name=android_updates.APK_ARTIFACT,
|
|
)
|
|
for artifact_id, run_id, run_number in (
|
|
(11, 101, 10),
|
|
(12, 102, 20),
|
|
(13, 103, 30),
|
|
(14, 104, 40),
|
|
(15, 105, 50),
|
|
)
|
|
]
|
|
zips = {
|
|
12: _zip({"update.json": _metadata_body(20, "0.1.0-master.20")}),
|
|
112: _zip({"app-release.apk": b"new apk"}),
|
|
}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/actions/artifacts"):
|
|
name = request.url.params.get("name")
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"total_count": 5,
|
|
"artifacts": (
|
|
metadata_artifacts
|
|
if name == android_updates.METADATA_ARTIFACT
|
|
else apk_artifacts
|
|
),
|
|
},
|
|
)
|
|
artifact_id = int(request.url.path.rstrip("/").split("/")[-2])
|
|
return httpx.Response(200, content=zips[artifact_id])
|
|
|
|
_install_transport(monkeypatch, handler)
|
|
|
|
metadata = android_updates.latest_android()
|
|
assert metadata["version_code"] == 20
|
|
assert metadata["version_name"] == "0.1.0-master.20"
|
|
assert metadata["commit"] == "sha-20"
|
|
assert metadata["size_bytes"] == 1234
|
|
assert metadata["signer_sha256"] == "def"
|
|
|
|
path, same = android_updates.android_apk()
|
|
assert path.read_bytes() == b"new apk"
|
|
assert same == metadata
|
|
|
|
|
|
def test_falls_back_to_an_apk_artifact_without_metadata(monkeypatch):
|
|
_configure(monkeypatch)
|
|
apk = _artifact(
|
|
201,
|
|
run_id=301,
|
|
run_number=77,
|
|
name=android_updates.APK_ARTIFACT,
|
|
created_at="2026-10-21T09:30:00Z",
|
|
size=900,
|
|
)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/actions/artifacts"):
|
|
name = request.url.params.get("name")
|
|
return httpx.Response(
|
|
200,
|
|
json={"total_count": 0 if name == android_updates.METADATA_ARTIFACT else 1,
|
|
"artifacts": [] if name == android_updates.METADATA_ARTIFACT else [apk]},
|
|
)
|
|
raise AssertionError("the metadata endpoint should not download anything")
|
|
|
|
_install_transport(monkeypatch, handler)
|
|
|
|
metadata = android_updates.latest_android()
|
|
assert metadata["version_code"] == 77
|
|
assert metadata["version_name"] == "0.1.0-master.77"
|
|
assert metadata["commit"] == "sha-77"
|
|
assert metadata["built_at"] == "2026-10-21T09:30:00Z"
|
|
assert metadata["size_bytes"] == 900
|
|
|
|
|
|
def test_missing_token_is_not_configured(monkeypatch):
|
|
_configure(monkeypatch, token="")
|
|
with pytest.raises(android_updates.UpdateNotConfigured):
|
|
android_updates.latest_android()
|
|
|
|
|
|
def test_gitea_failure_is_reported_as_unavailable(monkeypatch):
|
|
_configure(monkeypatch)
|
|
_install_transport(monkeypatch, lambda _request: httpx.Response(500, text="boom"))
|
|
|
|
with pytest.raises(android_updates.UpdateUnavailable):
|
|
android_updates.latest_android()
|
|
|
|
|
|
def test_the_route_reports_503_when_updates_are_not_configured(client, monkeypatch):
|
|
_configure(monkeypatch, token="")
|
|
|
|
response = client.get("/api/updates/android")
|
|
|
|
assert response.status_code == 503
|
|
assert "GITEA_TOKEN" in response.json()["detail"]
|
|
|
|
|
|
def test_sparse_artifact_run_metadata_is_fetched_from_the_run(monkeypatch):
|
|
_configure(monkeypatch)
|
|
metadata = _artifact(31, run_id=401, run_number=0, branch="", conclusion="")
|
|
metadata["workflow_run"] = {"id": 401, "head_sha": "sha-88"}
|
|
apk = _artifact(
|
|
32,
|
|
run_id=401,
|
|
run_number=0,
|
|
name=android_updates.APK_ARTIFACT,
|
|
branch="",
|
|
conclusion="",
|
|
)
|
|
apk["workflow_run"] = {"id": 401, "head_sha": "sha-88"}
|
|
zips = {
|
|
31: _zip({"update.json": _metadata_body(88, "0.1.0-master.88")}),
|
|
32: _zip({"app-release.apk": b"sparse apk"}),
|
|
}
|
|
run_requests = 0
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal run_requests
|
|
if request.url.path.endswith("/actions/artifacts"):
|
|
name = request.url.params.get("name")
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"total_count": 1,
|
|
"artifacts": [metadata]
|
|
if name == android_updates.METADATA_ARTIFACT
|
|
else [apk],
|
|
},
|
|
)
|
|
if request.url.path.endswith("/actions/runs/401"):
|
|
run_requests += 1
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"id": 401,
|
|
"run_number": 88,
|
|
"head_branch": "master",
|
|
"head_sha": "sha-88",
|
|
"conclusion": "success",
|
|
},
|
|
)
|
|
artifact_id = int(request.url.path.rstrip("/").split("/")[-2])
|
|
return httpx.Response(200, content=zips[artifact_id])
|
|
|
|
_install_transport(monkeypatch, handler)
|
|
|
|
metadata = android_updates.latest_android()
|
|
assert metadata["version_code"] == 88
|
|
assert metadata["version_name"] == "0.1.0-master.88"
|
|
assert metadata["commit"] == "sha-88"
|
|
assert run_requests == 1
|
|
|
|
path, same = android_updates.android_apk()
|
|
assert path.read_bytes() == b"sparse apk"
|
|
assert same == metadata
|