"""Exercise interrupted installs through the actual CLI and relay, plus the shared deadline.""" import json import os from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch import native_dependency_guard as guard import native_pods_retry import test_native_pods PARTIAL = "error: RPC failed; curl 18 transfer closed with outstanding read data remaining\nfatal: early EOF" # Verbatim from ios-zuper-sms Mobile Deploy 34346560202, after the diagnostic's own URL # redaction — the form `retryable` actually sees. TIMEOUT = "\n".join(["CDN: trunk URL couldn't be downloaded: [URL redacted] " "Response: Timeout was reached"] * 30) # Verbatim from printer-ai Mobile Deploy 34374868740, after the diagnostic's own URL # redaction: CocoaPods cloning a dependency when the exit dropped the connection. HANDSHAKE = ("Cloning into '[long value redacted]'...\n" "fatal: unable to access '[URL redacted]': LibreSSL SSL_connect: " "SSL_ERROR_SYSCALL in connection to github.com:443") # The same fault one layer earlier, from 2phone-ios run 34374984763: bundler fetching # specs. It names no curl code and no SSL_ERROR_SYSCALL, only the truncated read. BUNDLER_EOF = ("Retrying fetcher due to error (2/2): Bundler::HTTPError Could not fetch specs " "due to underlying error ") class RetryTests(unittest.TestCase): def setUp(self): self.fixture = test_native_pods.PodsTests() self.fixture.setUp() self.addCleanup(self.fixture.doCleanups) self.fixture.locked() self.root = self.fixture.root def script(self, failures, *, change_lock=False): configuration = json.dumps({"failures": failures, "change_lock": change_lock}) script = self.root / "pod" script.write_text('#!' + sys.executable + '\nimport json,os,sys\nfrom pathlib import Path\n' 'config=json.loads(' + repr(configuration) + ')\n' 'p=Path("attempts.json");rows=json.loads(p.read_text()) if p.exists() else []\n' 'rows.append({"argv":sys.argv[1:]});p.write_text(json.dumps(rows))\n' 'if len(rows)<=len(config["failures"]):\n' ' if config["change_lock"]: Path("Podfile.lock").write_text("changed")\n' ' print(config["failures"][len(rows)-1],file=sys.stderr);raise SystemExit(7)\n' 'p=Path("App.xcworkspace");p.mkdir(exist_ok=True)\n' '(p/"contents.xcworkspacedata").write_text(' '\'\')\n') script.chmod(0o700) self.fixture.env["PATH"] = str(self.root) + os.pathsep + self.fixture.env["PATH"] def test_partial_transfer_recovers_with_same_lock_flags(self): self.script([PARTIAL]) code, result = self.fixture.run_cli() self.assertEqual((code, result["pod_attempts"]), (0, 2)) attempts = json.loads((self.root / "attempts.json").read_text()) self.assertEqual(len(attempts), 2) for attempt in attempts: self.assertEqual(attempt["argv"], ["install", "--deployment", "--no-repo-update"]) self.assertEqual((self.root / "Podfile.lock").read_text(), "fixture-lock") def test_repeat_partial_failure_stops_after_second_attempt(self): self.script([PARTIAL, PARTIAL, PARTIAL]) code, result = self.fixture.run_cli() self.assertEqual((code, result["code"]), (1, "native_pods_install_failed")) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 2) def test_changed_lock_refuses_before_any_retry(self): self.script([PARTIAL], change_lock=True) code, result = self.fixture.run_cli() self.assertEqual((code, result["code"]), (1, "native_bundle_inputs_changed")) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 1) def test_unknown_second_failure_does_not_inherit_first_attempt_classification(self): self.script([PARTIAL, "unclassified failure", PARTIAL]) code, result = self.fixture.run_cli() self.assertEqual(code, 1) self.assertEqual(result["diagnostic"], "unclassified failure") self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 2) def test_terminal_evidence_dominates_partial_transfer(self): for failure in ("certificate verify failed", "authentication failed", "permission denied", "checksum mismatch", "could not find compatible versions", "deployment mode", "The requested URL returned error: 407", "curl: (60) unknown trust failure"): with self.subTest(failure=failure): (self.root / "attempts.json").unlink(missing_ok=True) self.script([PARTIAL + "\n" + failure]) self.assertEqual(self.fixture.run_cli()[0], 1) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 1) def test_timed_out_cdn_fetch_recovers_within_the_same_deadline(self): self.script([TIMEOUT]) code, result = self.fixture.run_cli() self.assertEqual((code, result["pod_attempts"]), (0, 2)) attempts = json.loads((self.root / "attempts.json").read_text()) self.assertEqual(len(attempts), 2) for attempt in attempts: self.assertEqual(attempt["argv"], ["install", "--deployment", "--no-repo-update"]) self.assertEqual((self.root / "Podfile.lock").read_text(), "fixture-lock") def test_repeat_timeout_stops_after_second_attempt(self): self.script([TIMEOUT, TIMEOUT, TIMEOUT]) code, result = self.fixture.run_cli() self.assertEqual((code, result["code"]), (1, "native_pods_install_failed")) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 2) def test_terminal_evidence_dominates_a_timeout(self): for failure in ("certificate verify failed", "The requested URL returned error: 403", "checksum mismatch", "curl: (60) unknown trust failure"): with self.subTest(failure=failure): (self.root / "attempts.json").unlink(missing_ok=True) self.script([TIMEOUT + "\n" + failure]) self.assertEqual(self.fixture.run_cli()[0], 1) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 1) def test_each_transient_class_is_named_and_unrelated_waits_are_not_retried(self): self.assertEqual(native_pods_retry.retryable(TIMEOUT), "transient_timeout") self.assertEqual(native_pods_retry.retryable(PARTIAL), "partial_transfer") self.assertEqual(native_pods_retry.retryable("curl: (28) Operation timed out"), "transient_timeout") for body in ("the timeout was generous", "Response: 404 Not Found", "waiting for the lock", "unknown network error"): with self.subTest(body=body): self.assertEqual(native_pods_retry.retryable(body), "") def test_dropped_handshake_recovers_within_the_same_deadline(self): for failure in (HANDSHAKE, BUNDLER_EOF): with self.subTest(failure=failure.splitlines()[-1][:40]): (self.root / "attempts.json").unlink(missing_ok=True) self.script([failure]) code, result = self.fixture.run_cli() self.assertEqual((code, result["pod_attempts"]), (0, 2)) attempts = json.loads((self.root / "attempts.json").read_text()) self.assertEqual(len(attempts), 2) for attempt in attempts: self.assertEqual(attempt["argv"], ["install", "--deployment", "--no-repo-update"]) self.assertEqual((self.root / "Podfile.lock").read_text(), "fixture-lock") def test_repeat_dropped_handshake_stops_after_second_attempt(self): self.script([HANDSHAKE, HANDSHAKE, HANDSHAKE]) code, result = self.fixture.run_cli() self.assertEqual((code, result["code"]), (1, "native_pods_install_failed")) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 2) def test_a_certificate_verdict_beside_a_dropped_handshake_stays_permanent(self): for failure in ("certificate verify failed", "SSL peer shut down incorrectly", "The requested URL returned error: 403", "curl: (60) unknown trust failure"): with self.subTest(failure=failure): (self.root / "attempts.json").unlink(missing_ok=True) self.script([HANDSHAKE + "\n" + failure]) self.assertEqual(self.fixture.run_cli()[0], 1) self.assertEqual(len(json.loads((self.root / "attempts.json").read_text())), 1) def test_a_dropped_handshake_is_its_own_named_class(self): self.assertEqual(native_pods_retry.retryable(HANDSHAKE), "handshake_reset") self.assertEqual(native_pods_retry.retryable(BUNDLER_EOF), "handshake_reset") for body in ("the peer read the file", "eof while reading the manifest", "connection established", "SSL is configured"): with self.subTest(body=body): self.assertEqual(native_pods_retry.retryable(body), "") def test_one_deadline_and_no_guess_from_generic_eof(self): for body in ("fatal: early EOF", "curl 180 failure", "unknown network error"): self.assertFalse(native_pods_retry.interrupted_transfer(body)) self.assertTrue(native_pods_retry.interrupted_transfer("curl: (18) transfer closed")) def child(_argv, _folder, log, *, timeout): log.write(PARTIAL.encode()) return 7 with tempfile.TemporaryFile() as log, patch.object(native_pods_retry.native_bundle, "command", side_effect=child) as command: with patch.object(native_pods_retry.time, "monotonic", side_effect=[0, 1, 700]): self.assertEqual(native_pods_retry.install(["fixture"], self.root, log, {}), (7, 2)) self.assertEqual([c.kwargs["timeout"] for c in command.call_args_list], [899, 200]) command.reset_mock() with patch.object(native_pods_retry.time, "monotonic", side_effect=[0, 1, 901]): with self.assertRaisesRegex(guard.Refused, "native_pods_operation_interrupted") as caught: native_pods_retry.install(["fixture"], self.root, log, {}) self.assertEqual(command.call_count, 1) self.assertEqual(caught.exception.diagnostic, PARTIAL)