Implement verified household TV updates and release tooling

docker/test_tv_publish.py

@@ -0,0 +1,280 @@
+"""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()