"""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 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.endswith("/latest") and fixture.release.get( "draft", True ): self.respond({}, 404) 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.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(auth for _, auth in self.forge.events)) 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) anonymous = [ event for event, authenticated in self.forge.events if not authenticated ] self.assertEqual(8, len(anonymous)) 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_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()