Enforce monotonic household releases and staged APK integrity

docker/test_tv_publish.py

@@ -30,6 +30,12 @@ class FakeForge:
         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):
@@ -104,10 +110,39 @@ class FakeForge:
                         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)
+                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)
 
@@ -157,6 +192,7 @@ class PublishTest(unittest.TestCase):
             "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()
@@ -208,16 +244,18 @@ class PublishTest(unittest.TestCase):
         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))
+        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)
-        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
@@ -266,6 +304,59 @@ class PublishTest(unittest.TestCase):
         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)