"""REST IPA upload identity, resume, complete byte coverage and proxy-only transfer.""" import hashlib import io import json import os from pathlib import Path import plistlib import sys import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from unittest import mock import zipfile sys.path.insert(0, str(Path(__file__).resolve().parent)) import upload_build import upload_receipt from upload_receipt import UploadError class BuildUploadTests(unittest.TestCase): def test_ipa_identity_comes_from_the_archived_app(self): with tempfile.TemporaryDirectory() as temporary: ipa = Path(temporary) / "app.ipa" with zipfile.ZipFile(ipa, "w") as archive: archive.writestr("Payload/App.app/Info.plist", plistlib.dumps({ "CFBundleIdentifier": "com.example.app", "CFBundleVersion": "8", "CFBundleShortVersionString": "1.2.0"})) info = upload_build.identity(ipa) self.assertEqual(info["cfBundleVersion"], "8") self.assertEqual(info["cfBundleShortVersionString"], "1.2.0") self.assertEqual(info["sha256"], hashlib.sha256(ipa.read_bytes()).hexdigest()) def test_existing_upload_is_adopted_without_another_create(self): info = {"cfBundleVersion": "8", "cfBundleShortVersionString": "1.2.0"} with mock.patch.object(upload_build, "api", return_value={"data": [{"id": "existing"}]}) as api: self.assertEqual(upload_build.upload_record("app", info)["id"], "existing") self.assertEqual(api.call_count, 1) self.assertEqual(api.call_args.args[0], "GET") def test_existing_different_bytes_are_not_replaced(self): existing = {"data": [{"id": "file", "attributes": { "assetType": "ASSET", "fileName": "other.ipa", "fileSize": 10}}]} with mock.patch.object(upload_build, "api", return_value=existing) as api: with self.assertRaises(UploadError) as caught: upload_build.upload_file({"id": "upload"}, {"fileName": "wanted.ipa", "fileSize": 10}) self.assertEqual(api.call_count, 1) self.assertEqual(caught.exception.category, "upload_identity_conflict") self.assertEqual(caught.exception.details, {"upload_id": "upload", "file_id": "file"}) def test_part_transfer_carries_no_asc_bearer(self): with tempfile.TemporaryDirectory() as temporary: ipa = Path(temporary) / "app.ipa" ipa.write_bytes(b"abcde") operation = {"url": "https://upload.apple.com/part", "method": "PUT", "offset": 1, "length": 3, "requestHeaders": [{"name": "Content-Type", "value": "application/octet-stream"}]} response = mock.Mock(status_code=200) with mock.patch.object(upload_build.requests, "request", return_value=response) as request: upload_build.transfer(ipa, operation) self.assertEqual(request.call_args.kwargs["data"], b"bcd") self.assertNotIn("Authorization", request.call_args.kwargs["headers"]) self.assertFalse(request.call_args.kwargs["allow_redirects"]) # Apple signs the upload URL itself; nothing may resolve a proxy for it. self.assertNotIn("proxies", request.call_args.kwargs) def test_missing_parts_never_mark_a_file_uploaded(self): file = {"id": "file", "attributes": {"uploadOperations": [{"offset": 0, "length": 4}]}} with mock.patch.object(upload_build, "api") as api, mock.patch.object(upload_build, "transfer") as transfer: with self.assertRaises(UploadError): upload_build.transfer_all(Path("unused"), file, {"fileSize": 5, "sha256": "hash"}) api.assert_not_called() transfer.assert_not_called() def test_parts_are_joined_before_committing_the_uploaded_file(self): file = {"id": "file", "attributes": {"uploadOperations": [{"offset": 0, "length": 2}, {"offset": 2, "length": 3}]}} done = [] with mock.patch.object(upload_build, "transfer", side_effect=lambda path, op: done.append(op["offset"])), \ mock.patch.object(upload_build, "api") as api: upload_build.transfer_all(Path("unused"), file, {"fileSize": 5, "sha256": "digest"}) self.assertEqual(sorted(done), [0, 2]) body = api.call_args.args[2]["data"]["attributes"] self.assertEqual(body, {"uploaded": True}) def test_failed_part_never_commits_the_file(self): file = {"id": "file", "attributes": {"uploadOperations": [{"offset": 0, "length": 5}]}} with mock.patch.object(upload_build, "transfer", side_effect=RuntimeError("transport")), \ mock.patch.object(upload_build, "api") as api: with self.assertRaises(RuntimeError): upload_build.transfer_all(Path("unused"), file, {"fileSize": 5, "sha256": "digest"}) api.assert_not_called() class BuildDeliveryTests(unittest.TestCase): @mock.patch.dict(os.environ, {"APP_STORE_APPLE_ID": "app"}) def test_other_app_is_refused_before_creating_uploads(self): with mock.patch.object(upload_build, "identity", return_value={"bundle_id": "com.expected.app"}), \ mock.patch.object(upload_build, "api", return_value={"data": {"attributes": { "bundleId": "com.other.app"}}}) as api, \ mock.patch.object(upload_build, "upload_record") as create, \ tempfile.TemporaryDirectory() as temporary: report = Path(temporary) / "receipt.json" with self.assertRaises(UploadError) as caught: upload_build.deliver(Path("unused"), report, 1) self.assertEqual(api.call_count, 1) create.assert_not_called() self.assertEqual(caught.exception.category, "app_mismatch") self.assertEqual(json.loads(report.read_text())["stage"], "identity") @mock.patch.dict(os.environ, {"APP_STORE_APPLE_ID": "app"}) def test_completed_readback_retains_provider_build_identity(self): reads = [{"data": {"attributes": {"bundleId": "com.expected.app"}}}, {"data": { "attributes": {"state": {"state": "COMPLETE"}}, "relationships": {"build": {"data": {"id": "build-8"}}}}}] with tempfile.TemporaryDirectory() as temporary, \ mock.patch.object(upload_build, "identity", return_value={"bundle_id": "com.expected.app"}), \ mock.patch.object(upload_build, "api", side_effect=reads) as api, \ mock.patch.object(upload_build, "upload_record", return_value={"id": "upload"}), \ mock.patch.object(upload_build, "upload_file", return_value={"id": "file"}), \ mock.patch.object(upload_build, "transfer_all"): report = Path(temporary) / "receipt.json" result = upload_build.deliver(Path("unused"), report, 1) self.assertEqual(result["build_id"], "build-8") self.assertEqual(result["state"], "COMPLETE") self.assertIn('"build-8"', report.read_text()) self.assertEqual(api.call_args.kwargs["params"], {"include": "build"}) IDENTITY = {"bundle_id": "com.expected.app", "cfBundleVersion": "2", "cfBundleShortVersionString": "1.0.0", "sha256": "ab" * 32, "fileName": "ab" * 32 + ".ipa", "fileSize": 10} ENV = {"APP_STORE_APPLE_ID": "app"} class ReceiptStageTests(unittest.TestCase): """The receipt exists from the first local identity on and names every provider id it learns.""" @mock.patch.dict(os.environ, ENV) def test_identity_is_written_before_the_first_provider_request(self): with tempfile.TemporaryDirectory() as temporary, \ mock.patch.object(upload_build, "identity", return_value=dict(IDENTITY)), \ mock.patch.object(upload_build, "api", side_effect=SystemExit("ASC GET failed: 503 body")): report = Path(temporary) / "receipt.json" with self.assertRaises(SystemExit): upload_build.deliver(Path("unused"), report, 1) receipt = json.loads(report.read_text()) self.assertEqual(receipt["stage"], "identity") self.assertEqual(receipt["sha256"], IDENTITY["sha256"]) self.assertEqual(receipt["cfBundleVersion"], "2") self.assertTrue(receipt["proxy_required"]) @mock.patch.dict(os.environ, ENV) def test_upload_id_is_written_before_the_file_is_reserved(self): with tempfile.TemporaryDirectory() as temporary, \ mock.patch.object(upload_build, "identity", return_value=dict(IDENTITY)), \ mock.patch.object(upload_build, "api", return_value={"data": {"attributes": { "bundleId": "com.expected.app"}}}), \ mock.patch.object(upload_build, "upload_record", return_value={"id": "upload-1"}), \ mock.patch.object(upload_build, "upload_file", side_effect=UploadError( "upload_identity_conflict", "different bytes", upload_id="upload-1", file_id="file-old")): report = Path(temporary) / "receipt.json" with self.assertRaises(UploadError): upload_build.deliver(Path("unused"), report, 1) receipt = json.loads(report.read_text()) self.assertEqual(receipt["stage"], "upload_reserved") self.assertEqual(receipt["upload_id"], "upload-1") self.assertEqual(receipt["app_id"], "app") class FailureAnnotationTests(unittest.TestCase): """The typed annotation carries the category and identifiers, never exception text.""" def run_main(self, exc: BaseException, report: Path) -> tuple[dict, str, str]: out, err = io.StringIO(), io.StringIO() argv = ["upload_build.py", "app.ipa", "--report", str(report)] with mock.patch.object(upload_build, "deliver", side_effect=exc), \ mock.patch.object(sys, "argv", argv), redirect_stdout(out), redirect_stderr(err): with self.assertRaises(SystemExit) as caught: upload_build.main() self.assertIn("retain its receipt and IPA", str(caught.exception.code)) prefix = "::error title=apple_build_upload_failed::" line = next(row for row in out.getvalue().splitlines() if row.startswith(prefix)) return json.loads(line.split("::", 2)[2]), out.getvalue(), str(caught.exception.code) def test_provider_exit_text_stays_out_of_the_annotation_and_receipt(self): with tempfile.TemporaryDirectory() as temporary: report = Path(temporary) / "receipt.json" report.write_text(json.dumps({"stage": "file_reserved", "upload_id": "upload-1", "file_id": "file-1", "cfBundleVersion": "2"})) secret = "Bearer eyJ-secret-token-value" failure, out, err = self.run_main(SystemExit(f"ASC PATCH /x failed: 409 {secret}"), report) receipt = json.loads(report.read_text()) self.assertEqual(failure["schema"], "gowalk-cicd/apple-build-upload-failed.v2") self.assertEqual((failure["category"], failure["stage"]), ("provider_request_failed", "file_reserved")) self.assertEqual((failure["upload_id"], failure["file_id"], failure["cfBundleVersion"]), ("upload-1", "file-1", "2")) self.assertTrue(failure["receipt_available"]) self.assertEqual(receipt["failure"], {"category": "provider_request_failed", "exception": "SystemExit"}) for text in (out, err, json.dumps(receipt)): self.assertNotIn("secret-token-value", text) self.assertNotIn("ASC PATCH", text) def test_classified_details_are_published(self): with tempfile.TemporaryDirectory() as temporary: report = Path(temporary) / "receipt.json" exc = UploadError("provider_request_failed", "refused", provider_status=409, provider_code="ENTITY_ERROR.ATTRIBUTE.INVALID", provider_pointer="/data/attributes/sourceFileChecksums") failure, _, err = self.run_main(exc, report) self.assertEqual(failure["provider_status"], 409) self.assertEqual(failure["provider_code"], "ENTITY_ERROR.ATTRIBUTE.INVALID") self.assertEqual(failure["stage"], "start") self.assertTrue(failure["receipt_available"]) self.assertIn("provider_request_failed at stage start", err) self.assertNotIn("refused", err) def test_unknown_failures_fall_back_to_a_safe_category(self): self.assertEqual(upload_receipt.classify(KeyError("data")), "provider_response_shape") self.assertEqual(upload_receipt.classify(OSError("disk")), "local_io") self.assertEqual(upload_receipt.classify(ValueError("x")), "invalid_state") self.assertEqual(upload_receipt.classify(RuntimeError("x")), "upload_failed") self.assertEqual(upload_receipt.classify(ZeroDivisionError()), "unexpected") class ProviderRefusalTests(unittest.TestCase): @mock.patch.dict(os.environ, {"ASC_KEY_ID": "k", "ASC_ISSUER_ID": "i", "ASC_KEY_PATH": "p"}) def test_provider_refusal_becomes_a_classified_error_with_the_apple_code(self): response = mock.Mock(status_code=409) response.json.return_value = {"errors": [{"code": "ENTITY_ERROR.ATTRIBUTE.INVALID", "detail": "body text that must not leak", "source": {"pointer": "/data/attributes/sourceFileChecksums"}}]} with mock.patch.object(upload_build.asc_common, "make_jwt", return_value="jwt"), \ mock.patch.object(upload_build.asc_common, "request", return_value=response) as request: with self.assertRaises(UploadError) as caught: upload_build.api("PATCH", "/buildUploadFiles/f", {"data": {}}) self.assertEqual(caught.exception.category, "provider_request_failed") self.assertEqual(caught.exception.details, { "provider_status": 409, "provider_code": "ENTITY_ERROR.ATTRIBUTE.INVALID", "provider_pointer": "/data/attributes/sourceFileChecksums"}) self.assertNotIn("body text", json.dumps(caught.exception.details)) self.assertEqual(request.call_args.kwargs["allow_status"], upload_build.CLASSIFIED_STATUSES) self.assertEqual(request.call_args.kwargs["max_attempts"], 1) def test_unreadable_ipa_is_classified(self): with tempfile.TemporaryDirectory() as temporary: ipa = Path(temporary) / "app.ipa" ipa.write_bytes(b"not a zip") with self.assertRaises(UploadError) as caught: upload_build.identity(ipa) self.assertEqual(caught.exception.category, "ipa_invalid") class ActionWiringTests(unittest.TestCase): def test_failed_upload_retains_the_exported_ipa_next_to_its_receipt(self): text = (Path(__file__).resolve().parents[1] / "action.yml").read_text() upload = text.index("- name: Upload to TestFlight") self.assertIn("id: upload", text[upload:upload + 200]) retain = text.index("- name: Retain the exported IPA for upload recovery") self.assertLess(text.index("- name: Retain Apple upload receipt"), retain) block = text[retain:retain + 700] self.assertIn("steps.upload.outcome != 'success'", block) self.assertIn("name: apple-ipa-${{ github.run_id }}-${{ github.run_attempt }}", block) self.assertIn("path: ${{ runner.temp }}/export/*.ipa", block) self.assertIn("if-no-files-found: ignore", block) if __name__ == "__main__": unittest.main()