Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 19 additions & 7 deletions scripts/sync_gitee_ota_release.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,21 +70,27 @@ def with_retry(description: str, action):
raise AssertionError("unreachable")


def request_json(request: urllib.request.Request, timeout: float = 60) -> dict[str, object]:
def read_json(request: urllib.request.Request, timeout: float = 60) -> object:
# get_method(), not .method: Request only sets .method when one is passed.
method = request.get_method()
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
payload = response.read()
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
raise RequestFailed(
f"{request.method} {request.full_url} failed ({error.code}): {body[:500]}", error.code
f"{method} {request.full_url} failed ({error.code}): {body[:500]}", error.code
) from error
except (OSError, urllib.error.URLError) as error:
# A dropped connection has no status; it is the transient case retry exists for.
raise RequestFailed(f"{request.method} {request.full_url} failed: {error}", None) from error
result = json.loads(payload) if payload else {}
raise RequestFailed(f"{method} {request.full_url} failed: {error}", None) from error
return json.loads(payload) if payload else {}


def request_json(request: urllib.request.Request, timeout: float = 60) -> dict[str, object]:
result = read_json(request, timeout)
if not isinstance(result, dict):
raise RuntimeError(f"{request.method} {request.full_url} returned a non-object response")
raise RuntimeError(f"{request.get_method()} {request.full_url} returned a non-object response")
return result


Expand All @@ -102,12 +108,18 @@ def gitee_release(owner: str, repo: str, tag: str, token: str) -> dict[str, obje
url = f"{GITEE_API}/repos/{owner}/{repo}/releases/tags/{urllib.parse.quote(tag)}"
request = urllib.request.Request(url, headers={"Authorization": f"token {token}", "Accept": "application/json"})
try:
return with_retry("read Gitee release", lambda: request_json(request))
result = with_retry("read Gitee release", lambda: read_json(request))
except RequestFailed as error:
# No release for this tag yet, which is the normal first-sync case.
if error.status == 404:
return None
raise
# Gitee answers a tag with no release as HTTP 200 and a literal `null`, not a
# 404: that is the normal first-sync case for a new version.
if result is None:
return None
if not isinstance(result, dict):
raise RuntimeError(f"GET {url} returned a non-object response")
return result


def create_gitee_release(
Expand Down
36 changes: 36 additions & 0 deletions scripts/test_sync_gitee_ota_release.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,27 @@
import io
import unittest
import urllib.error
import urllib.request
from unittest import mock

from scripts.sync_gitee_ota_release import (
ATTEMPTS,
RequestFailed,
gitee_release,
request_json,
selected_assets,
wanted_asset,
with_retry,
)


def respond(payload: bytes):
"""Stand in for urlopen, answering every request with HTTP 200 and payload."""
response = mock.MagicMock()
response.__enter__.return_value.read.return_value = payload
return mock.patch("scripts.sync_gitee_ota_release.urllib.request.urlopen", return_value=response)


class SyncGiteeOTAReleaseTests(unittest.TestCase):
def test_only_updater_assets_are_selected(self) -> None:
release = {
Expand Down Expand Up @@ -87,5 +99,29 @@ def action() -> str:
self.assertEqual(len(attempts), 2)


class GiteeReleaseTests(unittest.TestCase):
def test_a_tag_without_a_release_reads_as_missing(self) -> None:
"""Gitee answers HTTP 200 with `null` here, observed for v0.8.6 before its first sync."""
with respond(b"null"):
self.assertIsNone(gitee_release("maimory", "BootAgent", "v0.8.6", "token"))

def test_a_404_reads_as_missing(self) -> None:
error = urllib.error.HTTPError("https://gitee.com", 404, "Not Found", {}, io.BytesIO(b"{}"))
with mock.patch("scripts.sync_gitee_ota_release.urllib.request.urlopen", side_effect=error):
self.assertIsNone(gitee_release("maimory", "BootAgent", "v0.8.6", "token"))

def test_an_existing_release_is_returned(self) -> None:
with respond(b'{"id": 1132764, "tag_name": "v0.8.5", "assets": []}'):
release = gitee_release("maimory", "BootAgent", "v0.8.5", "token")
self.assertEqual(release["id"], 1132764)

def test_a_non_object_response_is_reported_not_crashed_on(self) -> None:
"""The message must name the request; Request has no .method unless one was passed."""
request = urllib.request.Request("https://gitee.com/api/v5/x")
with respond(b"[]"):
with self.assertRaisesRegex(RuntimeError, r"^GET https://gitee\.com/api/v5/x returned a non-object"):
request_json(request)


if __name__ == "__main__":
unittest.main()
Loading