371 lines
15 KiB
Python
371 lines
15 KiB
Python
"""Exercise the release CLI against a local Forgejo-shaped HTTP server."""
|
|
|
|
import email.parser
|
|
import email.policy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import urllib.parse
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from tv_publish import REQUIRED, Forge
|
|
from tv_release import PACKAGE, SIGNER
|
|
|
|
|
|
class FakeForge:
|
|
"""Draft assets require authentication; anonymous reads work only after promotion."""
|
|
|
|
def __init__(self) -> None:
|
|
self.release: dict[str, Any] = {}
|
|
self.assets: dict[str, bytes] = {}
|
|
self.events: list[tuple[str, bool]] = []
|
|
self.corrupt_draft = False
|
|
self.corrupt_public = False
|
|
self.redirect_url: str | None = None
|
|
self.published_code: int | None = None
|
|
self.channel_manifest: dict[str, Any] = {}
|
|
self.latest_status = 200
|
|
self.release_list: list[dict[str, Any]] = []
|
|
self.latest_reads = 0
|
|
self.advance_before_promote: int | None = None
|
|
fixture = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, format: str, *args: Any) -> None:
|
|
pass
|
|
|
|
def respond(self, value: Any, code: int = 200) -> None:
|
|
data = value if isinstance(value, bytes) else json.dumps(value).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_POST(self) -> None:
|
|
authenticated = bool(self.headers.get("Authorization"))
|
|
fixture.events.append(("POST " + self.path, authenticated))
|
|
if not authenticated:
|
|
self.respond({}, 401)
|
|
return
|
|
data = self.rfile.read(int(self.headers["Content-Length"]))
|
|
if self.path.endswith("/releases"):
|
|
fixture.release = json.loads(data) | {"id": 17, "assets": []}
|
|
self.respond(fixture.release)
|
|
else:
|
|
name = urllib.parse.parse_qs(
|
|
urllib.parse.urlsplit(self.path).query
|
|
)["name"][0]
|
|
message = email.parser.BytesParser(
|
|
policy=email.policy.default
|
|
).parsebytes(
|
|
b"Content-Type: "
|
|
+ self.headers["Content-Type"].encode()
|
|
+ b"\r\n\r\n"
|
|
+ data
|
|
)
|
|
fixture.assets[name] = next(message.iter_parts()).get_payload(
|
|
decode=True
|
|
)
|
|
asset = {
|
|
"name": name,
|
|
"browser_download_url": fixture.base + "/assets/" + name,
|
|
"size": len(fixture.assets[name]),
|
|
}
|
|
fixture.release["assets"].append(asset)
|
|
self.respond(asset)
|
|
|
|
def do_PATCH(self) -> None:
|
|
fixture.events.append(
|
|
("PATCH", bool(self.headers.get("Authorization")))
|
|
)
|
|
fixture.release.update(
|
|
json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
)
|
|
self.respond(fixture.release)
|
|
|
|
def do_GET(self) -> None:
|
|
authenticated = bool(self.headers.get("Authorization"))
|
|
fixture.events.append(("GET " + self.path, authenticated))
|
|
if fixture.redirect_url:
|
|
self.send_response(302)
|
|
self.send_header("Location", fixture.redirect_url)
|
|
self.end_headers()
|
|
return
|
|
if self.path.startswith("/assets/"):
|
|
if fixture.release["draft"] and not authenticated:
|
|
self.respond({}, 404)
|
|
return
|
|
data = fixture.assets[self.path.removeprefix("/assets/")]
|
|
corrupt = (
|
|
fixture.corrupt_draft
|
|
if fixture.release["draft"]
|
|
else fixture.corrupt_public
|
|
)
|
|
self.respond(data + b"corrupt" if corrupt else data)
|
|
elif self.path == "/channel/wolfskin-update.json":
|
|
self.respond(
|
|
fixture.channel_manifest
|
|
| {"versionCode": fixture.published_code}
|
|
)
|
|
elif self.path.endswith("/latest"):
|
|
fixture.latest_reads += 1
|
|
if (
|
|
fixture.latest_reads == 2
|
|
and fixture.advance_before_promote is not None
|
|
):
|
|
fixture.published_code = fixture.advance_before_promote
|
|
if fixture.latest_status != 200:
|
|
self.respond({}, fixture.latest_status)
|
|
elif fixture.release and not fixture.release["draft"]:
|
|
self.respond(fixture.release)
|
|
elif fixture.published_code is not None:
|
|
self.respond(
|
|
{
|
|
"id": 9,
|
|
"assets": [
|
|
{
|
|
"name": "wolfskin-update.json",
|
|
"browser_download_url": fixture.base
|
|
+ "/channel/wolfskin-update.json",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
else:
|
|
self.respond({}, 404)
|
|
elif urllib.parse.urlsplit(self.path).path.endswith("/releases"):
|
|
self.respond(fixture.release_list)
|
|
else:
|
|
self.respond(fixture.release)
|
|
|
|
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
self.base = f"http://127.0.0.1:{self.server.server_port}"
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def close(self) -> None:
|
|
"""Release fixture resources after each independent publication scenario."""
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join()
|
|
|
|
|
|
class PublishTest(unittest.TestCase):
|
|
"""Permanent tests assert safety transitions and real uploaded bytes, not script wording."""
|
|
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.directory = Path(self.temporary.name)
|
|
self.forge = FakeForge()
|
|
self.addCleanup(self.forge.close)
|
|
self.addCleanup(self.temporary.cleanup)
|
|
assets = []
|
|
for abi in ("armeabi-v7a", "arm64-v8a", "x86_64", "universal"):
|
|
name = "Wolfskin.apk" if abi == "universal" else f"Wolfskin-{abi}.apk"
|
|
data = f"fixture APK bytes for {abi}".encode()
|
|
(self.directory / name).write_bytes(data)
|
|
assets.append(
|
|
{
|
|
"name": name,
|
|
"abi": abi,
|
|
"bytes": len(data),
|
|
"sha256": hashlib.sha256(data).hexdigest(),
|
|
"minSdk": 23,
|
|
}
|
|
)
|
|
manifest = {
|
|
"schemaVersion": 1,
|
|
"packageId": PACKAGE,
|
|
"versionName": "1.1.1",
|
|
"versionCode": 100001,
|
|
"publishedAt": "2026-11-07T12:00:00Z",
|
|
"sourceCommit": "a" * 40,
|
|
"sourceTag": "pwolfskin-opt-v1.1.1",
|
|
"signerSha256": SIGNER,
|
|
"assets": assets,
|
|
}
|
|
self.forge.channel_manifest = manifest
|
|
(self.directory / "wolfskin-update.json").write_text(json.dumps(manifest))
|
|
(self.directory / "RELEASE-NOTES.md").write_text("Fixture release notes")
|
|
self.checksums()
|
|
|
|
def checksums(self) -> None:
|
|
"""Hash fixture files after deliberately constructing or mutating the bundle."""
|
|
(self.directory / "SHA256SUMS").write_text(
|
|
"".join(
|
|
f"{hashlib.sha256((self.directory / name).read_bytes()).hexdigest()} {name}\n"
|
|
for name in sorted(REQUIRED - {"SHA256SUMS"})
|
|
)
|
|
)
|
|
|
|
def cli(self, *flags: str) -> subprocess.CompletedProcess[str]:
|
|
"""Drive the real CLI and real git credential-helper protocol with a fake credential."""
|
|
environment = os.environ | {
|
|
"GIT_CONFIG_COUNT": "2",
|
|
"GIT_CONFIG_KEY_0": "credential.helper",
|
|
"GIT_CONFIG_VALUE_0": "",
|
|
"GIT_CONFIG_KEY_1": "credential.helper",
|
|
"GIT_CONFIG_VALUE_1": "!printf 'username=fixture\npassword=fixture-secret\n'",
|
|
}
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(Path(__file__).with_name("tv_publish.py")),
|
|
"--directory",
|
|
str(self.directory),
|
|
"--test-server",
|
|
self.forge.base,
|
|
*flags,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=environment,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
self.assertNotIn("fixture-secret", result.stdout + result.stderr)
|
|
return result
|
|
|
|
def test_dry_run_never_requests_or_fetches_credentials(self) -> None:
|
|
result = self.cli("--dry-run", "--promote")
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual([], self.forge.events)
|
|
|
|
def test_draft_then_explicit_promotion_reverifies_all_bytes(self) -> None:
|
|
result = self.cli()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertTrue(self.forge.release["draft"])
|
|
self.assertEqual(REQUIRED, set(self.forge.assets))
|
|
self.assertTrue(
|
|
all(
|
|
not auth
|
|
for path, auth in self.forge.events
|
|
if "/latest" in path or "?limit=" in path
|
|
)
|
|
)
|
|
result = self.cli("--draft-id", "17", "--promote")
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertFalse(self.forge.release["draft"])
|
|
for name, data in self.forge.assets.items():
|
|
self.assertEqual((self.directory / name).read_bytes(), data)
|
|
|
|
def test_corrupt_uploaded_bytes_remain_draft(self) -> None:
|
|
self.forge.corrupt_draft = True
|
|
result = self.cli("--promote")
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertTrue(self.forge.release["draft"])
|
|
self.assertFalse(any(event == "PATCH" for event, _ in self.forge.events))
|
|
|
|
def test_failed_anonymous_verification_withdraws_promoted_release(self) -> None:
|
|
self.forge.corrupt_public = True
|
|
result = self.cli("--promote")
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertTrue(self.forge.release["draft"])
|
|
self.assertEqual(2, sum(event == "PATCH" for event, _ in self.forge.events))
|
|
|
|
def test_invalid_local_checksum_never_creates_draft(self) -> None:
|
|
(self.directory / "Wolfskin.apk").write_bytes(b"tampered")
|
|
result = self.cli()
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertEqual([], self.forge.events)
|
|
|
|
def test_manifest_size_mismatch_never_creates_draft(self) -> None:
|
|
path = self.directory / "wolfskin-update.json"
|
|
manifest = json.loads(path.read_text())
|
|
manifest["assets"][0]["bytes"] += 1
|
|
path.write_text(json.dumps(manifest))
|
|
self.checksums()
|
|
result = self.cli()
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertEqual([], self.forge.events)
|
|
|
|
def test_empty_manifest_asset_set_never_creates_draft(self) -> None:
|
|
path = self.directory / "wolfskin-update.json"
|
|
manifest = json.loads(path.read_text())
|
|
manifest["assets"] = []
|
|
path.write_text(json.dumps(manifest))
|
|
self.checksums()
|
|
result = self.cli()
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertEqual([], self.forge.events)
|
|
|
|
def test_credential_helper_tracing_is_disabled(self) -> None:
|
|
trace = self.directory / "credential.trace"
|
|
with patch.dict(os.environ, {"GIT_TRACE": str(trace)}):
|
|
result = self.cli()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertFalse(trace.exists())
|
|
|
|
def test_equal_and_lower_bundle_codes_never_create_a_draft(self) -> None:
|
|
for published_code in (100001, 100002):
|
|
with self.subTest(published_code=published_code):
|
|
self.forge.published_code = published_code
|
|
result = self.cli("--promote")
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertEqual({}, self.forge.release)
|
|
self.assertTrue(all(not auth for _, auth in self.forge.events))
|
|
|
|
def test_newer_bundle_reads_live_manifest_anonymously(self) -> None:
|
|
self.forge.published_code = 100000
|
|
result = self.cli()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertTrue(self.forge.release["draft"])
|
|
live_reads = [
|
|
auth
|
|
for path, auth in self.forge.events
|
|
if path.endswith("/latest") or path.startswith("GET /channel/")
|
|
]
|
|
self.assertTrue(live_reads)
|
|
self.assertFalse(any(live_reads))
|
|
|
|
def test_channel_advance_during_upload_prevents_promotion(self) -> None:
|
|
self.forge.advance_before_promote = 100001
|
|
result = self.cli("--promote")
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertTrue(self.forge.release["draft"])
|
|
self.assertFalse(any(path == "PATCH" for path, _ in self.forge.events))
|
|
|
|
def test_empty_channel_404_permits_first_draft(self) -> None:
|
|
result = self.cli()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertTrue(self.forge.release["draft"])
|
|
self.assertIn(
|
|
(
|
|
"GET /api/v1/repos/justinzeus/wolfskin-tv-updates/releases?limit=1",
|
|
False,
|
|
),
|
|
self.forge.events,
|
|
)
|
|
|
|
def test_missing_latest_with_nonempty_channel_is_not_first_release(self) -> None:
|
|
self.forge.release_list = [{"id": 9}]
|
|
result = self.cli()
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertEqual({}, self.forge.release)
|
|
|
|
def test_refused_live_channel_never_creates_draft(self) -> None:
|
|
self.forge.latest_status = 403
|
|
result = self.cli()
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertEqual({}, self.forge.release)
|
|
|
|
def test_authenticated_redirect_never_reaches_another_origin(self) -> None:
|
|
other = FakeForge()
|
|
self.addCleanup(other.close)
|
|
self.forge.redirect_url = other.base + "/asset"
|
|
forge = Forge(self.forge.base, "Basic fixture-authorization", test_server=True)
|
|
with self.assertRaises(ValueError):
|
|
forge.verify(self.forge.base + "/redirect", "0" * 64, authenticated=True)
|
|
self.assertEqual([], other.events)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|