#!/usr/bin/env python3 """ Tests for the POST /appStoreVersions 409-classifier path. The single-source-of-truth fix means CI must NEVER bump MARKETING_VERSION on its own; every 409 from POST that is not an actual race-recovered REUSE-by-fetch must surface the underlying ASC error and exit non-zero. This file pins: - 409 RELATIONSHIP.INVALID / unknown-code / version-collision under the no-bump default -> SystemExit(4) without bump. - The pre-flight request body is logged before POST so failures point at the offending payload immediately. - Failure response bodies log untruncated (full ASC errors). - 409 race-recovery via fetch_versions reconcile (REUSE-by-fetch). """ from __future__ import annotations import io import json import sys import unittest from contextlib import redirect_stderr from pathlib import Path from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parent)) import manage_marketing_version as mmv # noqa: E402 import asc_build_history # noqa: E402 import asc_version_create # noqa: E402 from tests_common import _v # noqa: E402 def _patch_empty_fetches(): """Common patch set for create_or_reuse 409-classifier tests: empty fetch_versions + empty preReleaseVersions/builds so the floor never blocks before the 409 path runs.""" return ( mock.patch.object(mmv, "fetch_versions", return_value=[]), mock.patch.object( asc_build_history, "fetch_prerelease_versions", return_value=[], ), mock.patch.object( asc_build_history, "fetch_builds_prerelease_versions", return_value=[], ), ) class Status409ClassifierTests(unittest.TestCase): """Guards: 409 RELATIONSHIP.INVALID/etc. must fail fast (NOT bump).""" def _conflict(self, code: str, pointer: str | None = None, detail: str = "conflict") -> mock.Mock: err: dict = {"status": "409", "code": code, "detail": detail} if pointer: err["source"] = {"pointer": pointer} resp = mock.Mock() resp.status_code = 409 resp.headers = {"X-ASC-Request-Id": "abc-123"} resp.text = json.dumps({"errors": [err]}) return resp def test_create_version_payload_shape(self): captured: dict = {} def fake_request(method, path, token, **kwargs): captured["method"] = method captured["path"] = path captured["json_body"] = kwargs.get("json_body") out = mock.Mock() out.status_code = 201 out.json.return_value = {"data": {"id": "ok"}} return out with mock.patch.object(asc_version_create, "request", side_effect=fake_request): asc_version_create.create_version("6755308530", "1.0.6", "tok") self.assertEqual(captured["method"], "POST") self.assertEqual(captured["path"], "/appStoreVersions") body = captured["json_body"] data = body["data"] self.assertEqual(data["type"], "appStoreVersions") self.assertEqual(data["attributes"]["platform"], "IOS") self.assertEqual(data["attributes"]["versionString"], "1.0.6") app_rel = data["relationships"]["app"]["data"] self.assertEqual(app_rel["type"], "apps") self.assertEqual(app_rel["id"], "6755308530") self.assertIsInstance(app_rel["id"], str) def test_request_body_logged_before_post(self): buf = io.StringIO() logged_before_post: list[bool] = [] def fake_request(method, path, token, **kwargs): logged_before_post.append( "[create-version] request body" in buf.getvalue() ) out = mock.Mock() out.status_code = 201 out.json.return_value = {"data": {"id": "ok"}} return out with mock.patch.object(asc_version_create, "request", side_effect=fake_request), \ redirect_stderr(buf): asc_version_create.create_version("111", "1.0.6", "tok") self.assertTrue(logged_before_post and logged_before_post[0]) stderr = buf.getvalue() self.assertIn("[create-version] request body", stderr) self.assertIn("1.0.6", stderr) self.assertIn("appStoreVersions", stderr) def test_response_body_logged_on_failure_untruncated(self): big_detail = "x" * 2500 resp = self._conflict( "ENTITY_ERROR.RELATIONSHIP.INVALID", pointer="/data/relationships/app", detail=big_detail, ) buf = io.StringIO() fv, fpr, fbpr = _patch_empty_fetches() with mock.patch.object(mmv, "create_version", return_value=resp), \ fv, fpr, fbpr, redirect_stderr(buf): with self.assertRaises(SystemExit): mmv.create_or_reuse("111", "1.0.6", "tok") stderr = buf.getvalue() self.assertIn(big_detail, stderr) self.assertIn("/data/relationships/app", stderr) def test_409_relationship_invalid_fails_fast(self): """409 RELATIONSHIP.INVALID -> SystemExit(4); never reaches bump_fn.""" resp = self._conflict( "ENTITY_ERROR.RELATIONSHIP.INVALID", pointer="/data/relationships/app", ) attempts: list[str] = [] def fake_create(app_id, version, token): attempts.append(version) return resp buf = io.StringIO() fv, fpr, fbpr = _patch_empty_fetches() with mock.patch.object(mmv, "create_version", side_effect=fake_create), \ fv, fpr, fbpr, redirect_stderr(buf): with self.assertRaises(SystemExit) as ctx: mmv.create_or_reuse("111", "1.0.6", "tok") self.assertEqual(attempts, ["1.0.6"]) self.assertEqual(ctx.exception.code, 4) stderr = buf.getvalue() self.assertIn("RELATIONSHIP.INVALID", stderr) self.assertIn("/data/relationships/app", stderr) def test_409_unknown_code_fails_fast_with_diagnostic(self): resp = self._conflict( "ENTITY_ERROR.SOMETHING_NEW", pointer="/data/attributes/platform", detail="unexpected thing", ) attempts: list[str] = [] def fake_create(app_id, version, token): attempts.append(version) return resp buf = io.StringIO() fv, fpr, fbpr = _patch_empty_fetches() with mock.patch.object(mmv, "create_version", side_effect=fake_create), \ fv, fpr, fbpr, redirect_stderr(buf): with self.assertRaises(SystemExit): mmv.create_or_reuse("111", "1.0.6", "tok") self.assertEqual(attempts, ["1.0.6"]) stderr = buf.getvalue() self.assertIn("ENTITY_ERROR.SOMETHING_NEW", stderr) def test_409_version_collision_no_bump_exits_4(self): """Even a real version-collision 409 must exit 4 now.""" collision_resp = self._conflict( "CONFLICT.VERSION_EXISTS", detail="A version with this versionString already exists.", ) attempts: list[str] = [] def fake_create(app_id, version, token): attempts.append(version) return collision_resp buf = io.StringIO() fv, fpr, fbpr = _patch_empty_fetches() with mock.patch.object(mmv, "create_version", side_effect=fake_create), \ fv, fpr, fbpr, redirect_stderr(buf): with self.assertRaises(SystemExit) as ctx: mmv.create_or_reuse("111", "1.0.6", "tok") self.assertEqual(attempts, ["1.0.6"]) self.assertEqual(ctx.exception.code, 4) stderr = buf.getvalue() self.assertIn("hidden", stderr.lower()) def test_409_reuses_existing_version_by_versionString(self): """If POST returns 409 and fetch_versions surfaces a matching id, create_or_reuse returns that id (REUSE-by-fetch path).""" fake_resp = mock.Mock() fake_resp.status_code = 409 fake_resp.headers = {} fake_resp.text = "" with mock.patch.object(mmv, "create_version", return_value=fake_resp), \ mock.patch.object( mmv, "fetch_versions", return_value=[_v("1.2.3", "PREPARE_FOR_SUBMISSION", vid="race-id")], ), \ mock.patch.object( asc_build_history, "fetch_prerelease_versions", return_value=[]), \ mock.patch.object( asc_build_history, "fetch_builds_prerelease_versions", return_value=[]): got = mmv.create_or_reuse("111", "1.2.3", "tok") self.assertEqual(got, "race-id") if __name__ == "__main__": unittest.main()