#!/usr/bin/env python3 """ Tests for next_build_number.py. All ASC HTTP is stubbed via unittest.mock; no network calls. Focus: * ``/builds`` alone returns the correct max + 1. * ``/preReleaseVersions`` surfaces fresh uploads missing from ``/builds``. * Union takes the max across both sources. * Non-integer / malformed version strings are ignored, not fatal. * ASC failure falls back to the local pbxproj CURRENT_PROJECT_VERSION. * Both sources empty + no pbxproj -> exit non-zero with a clear error. """ from __future__ import annotations import io import sys import unittest from contextlib import redirect_stdout from pathlib import Path from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parent)) import next_build_number as nb # noqa: E402 def _build(version: str, state: str = "VALID", uploaded: str = "") -> dict: return { "type": "builds", "attributes": { "version": version, "processingState": state, "uploadedDate": uploaded, }, } class BuildVersionIntTests(unittest.TestCase): def test_parses_integer_version_string(self): self.assertEqual(nb._build_version_int(_build("12")), 12) def test_returns_none_for_non_integer(self): self.assertIsNone(nb._build_version_int(_build("1.0.3"))) def test_returns_none_for_missing_attributes(self): self.assertIsNone(nb._build_version_int({})) class FetchBuildsTests(unittest.TestCase): def test_returns_versions_from_builds_endpoint(self): payload = {"data": [_build("3"), _build("7"), _build("1")]} with mock.patch.object(nb, "get_json", return_value=payload) as mget: got = nb.fetch_build_versions("111", "tok") self.assertEqual(got, {1, 3, 7}) # Confirm the call used filter[app] and no state filter. _args, kwargs = mget.call_args self.assertEqual(kwargs["params"]["filter[app]"], "111") self.assertNotIn("filter[processingState]", kwargs["params"]) def test_empty_data_returns_empty_set(self): with mock.patch.object(nb, "get_json", return_value={"data": []}): self.assertEqual(nb.fetch_build_versions("111", "tok"), set()) class FetchPreReleaseVersionsTests(unittest.TestCase): def test_harvests_included_builds(self): payload = { "data": [{"type": "preReleaseVersions", "id": "P1"}], "included": [ _build("5", "PROCESSING", "2026-04-18T10:00:00Z"), {"type": "apps", "id": "X"}, # non-build resource ignored _build("9"), ], } with mock.patch.object(nb, "get_json", return_value=payload): got = nb.fetch_prerelease_versions("111", "tok") self.assertEqual(got, {5, 9}) def test_missing_included_is_handled(self): with mock.patch.object(nb, "get_json", return_value={"data": []}): got = nb.fetch_prerelease_versions("111", "tok") self.assertEqual(got, set()) class FetchBuildUploadVersionsTests(unittest.TestCase): def _upload(self, version: str, state: str) -> dict: return {"type": "buildUploads", "attributes": {"cfBundleVersion": version, "platform": "IOS", "state": {"state": state}}} def test_every_reservation_counts_whatever_its_state(self): payload = {"data": [self._upload("2", "AWAITING_UPLOAD"), self._upload("3", "FAILED"), self._upload("4", "COMPLETE"), self._upload("x", "PROCESSING")]} with mock.patch.object(nb, "get_json", return_value=payload) as mget: got = nb.fetch_build_upload_versions("111", "tok") self.assertEqual(got, {2, 3, 4}) args, kwargs = mget.call_args self.assertEqual(args[0], "/apps/111/buildUploads") self.assertEqual(kwargs["params"]["filter[platform]"], "IOS") def test_no_reservations_is_empty(self): with mock.patch.object(nb, "get_json", return_value={"data": []}): self.assertEqual(nb.fetch_build_upload_versions("111", "tok"), set()) class ResolveNextBuildNumberTests(unittest.TestCase): def setUp(self): patcher = mock.patch.object(nb, "fetch_build_upload_versions", return_value=set()) self.uploads = patcher.start() self.addCleanup(patcher.stop) def test_max_across_both_sources_plus_one(self): with mock.patch.object(nb, "fetch_build_versions", return_value={1, 2}), \ mock.patch.object(nb, "fetch_prerelease_versions", return_value={5}): self.assertEqual(nb.resolve_next_build_number("111", "tok"), 6) def test_unfinished_upload_reservation_is_never_reused(self): # The exact production collision: build 1 processed, build 2 reserved by a REST # upload whose commit was refused, so /builds and /preReleaseVersions know only 1. self.uploads.return_value = {2} with mock.patch.object(nb, "fetch_build_versions", return_value={1}), \ mock.patch.object(nb, "fetch_prerelease_versions", return_value={1}): self.assertEqual(nb.resolve_next_build_number("111", "tok"), 3) def test_prerelease_fills_gap_when_builds_empty(self): # Simulates the exact bug: /builds misses the fresh upload but # /preReleaseVersions still sees it. with mock.patch.object(nb, "fetch_build_versions", return_value=set()), \ mock.patch.object(nb, "fetch_prerelease_versions", return_value={1}): self.assertEqual(nb.resolve_next_build_number("111", "tok"), 2) def test_both_empty_returns_one(self): with mock.patch.object(nb, "fetch_build_versions", return_value=set()), \ mock.patch.object(nb, "fetch_prerelease_versions", return_value=set()): self.assertEqual(nb.resolve_next_build_number("111", "tok"), 1) def test_ignores_non_integer_versions(self): # fetch_* functions already filter; resolve just unions and maxes. with mock.patch.object(nb, "fetch_build_versions", return_value={3}), \ mock.patch.object(nb, "fetch_prerelease_versions", return_value={3}): self.assertEqual(nb.resolve_next_build_number("111", "tok"), 4) class MainFallbackTests(unittest.TestCase): def _run_main(self) -> str: buf = io.StringIO() with redirect_stdout(buf): nb.main() return buf.getvalue().strip() def test_asc_success_prints_next(self): env = { "APP_STORE_APPLE_ID": "111", "ASC_KEY_ID": "K", "ASC_ISSUER_ID": "I", "ASC_KEY_PATH": "/tmp/x", } with mock.patch.dict(nb.os.environ, env, clear=False), \ mock.patch.object(nb, "make_jwt", return_value="tok"), \ mock.patch.object(nb, "resolve_next_build_number", return_value=42): self.assertEqual(self._run_main(), "42") def test_asc_failure_falls_back_to_pbxproj(self): env = { "APP_STORE_APPLE_ID": "111", "ASC_KEY_ID": "K", "ASC_ISSUER_ID": "I", "ASC_KEY_PATH": "/tmp/x", } with mock.patch.dict(nb.os.environ, env, clear=False), \ mock.patch.object(nb, "make_jwt", return_value="tok"), \ mock.patch.object( nb, "resolve_next_build_number", side_effect=RuntimeError("ASC 503"), ), \ mock.patch.object(nb, "_read_pbxproj_build_number", return_value=7): self.assertEqual(self._run_main(), "8") def test_asc_failure_and_no_pbxproj_exits_nonzero(self): env = { "APP_STORE_APPLE_ID": "111", "ASC_KEY_ID": "K", "ASC_ISSUER_ID": "I", "ASC_KEY_PATH": "/tmp/x", } with mock.patch.dict(nb.os.environ, env, clear=False), \ mock.patch.object(nb, "make_jwt", return_value="tok"), \ mock.patch.object( nb, "resolve_next_build_number", side_effect=RuntimeError("ASC 503"), ), \ mock.patch.object(nb, "_read_pbxproj_build_number", return_value=None): with self.assertRaises(SystemExit): nb.main() def test_missing_apple_id_raises_actionable_error(self): env_missing = {k: "" for k in [ "APP_STORE_APPLE_ID", "ASC_KEY_ID", "ASC_ISSUER_ID", "ASC_KEY_PATH", ]} with mock.patch.dict(nb.os.environ, env_missing, clear=False): with self.assertRaises(SystemExit) as ctx: nb.main() self.assertIn("APP_STORE_APPLE_ID", str(ctx.exception)) if __name__ == "__main__": unittest.main()