#!/usr/bin/env python3 import json import plistlib import subprocess import tempfile import unittest import zipfile from pathlib import Path from unittest import mock import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import mobile_bundle as bundle def init_repo(root: Path) -> str: run = lambda args: subprocess.run(args, check=True, stdout=subprocess.DEVNULL) run(["git", "init", "-q", str(root)]) run(["git", "-C", str(root), "config", "user.email", "tester@example.invalid"]) run(["git", "-C", str(root), "config", "user.name", "Tester"]) (root / "marker").write_text("x") run(["git", "-C", str(root), "add", "marker"]) run(["git", "-C", str(root), "commit", "-qm", "bundle"]) return subprocess.check_output(["git", "-C", str(root), "rev-parse", "HEAD"], text=True).strip() class MobileBundleTests(unittest.TestCase): def test_artifact_locator_is_required(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "repo" root.mkdir() init_repo(root) args = bundle.parser().parse_args(["run", "--platform", "android,ios", "--root", str(root)]) self.assertEqual(bundle.execute(args), 1) runs = list((root / ".pouch" / "tester" / "runs").glob("*.yaml")) self.assertEqual(len(runs), 1) record = json.loads(runs[0].read_text(encoding="utf-8")) self.assertIn("缺少调用方产物位置", record["error"]) def test_parse_artifacts_accepts_file_and_url(self): parsed = bundle.parse_artifacts([ "android=/tmp/app.apk", "ios=https://ci.example/app.ipa", ]) self.assertEqual(parsed["android"], "/tmp/app.apk") self.assertEqual(parsed["ios"], "https://ci.example/app.ipa") with self.assertRaises(bundle.ArtifactFailure): bundle.parse_artifacts(["android"]) with self.assertRaises(bundle.ArtifactFailure): bundle.parse_artifacts(["android=/tmp/a.apk", "android=/tmp/b.apk"]) def test_parse_bindings_keeps_platforms_separate(self): self.assertEqual( bundle.parse_bindings(["android=tester.android", "ios=tester.ios"], ["android", "ios"]), {"android": "tester.android", "ios": "tester.ios"}, ) self.assertEqual( bundle.parse_bindings([], ["android"]), {"android": "tester.android"}, ) with self.assertRaises(bundle.ArtifactFailure): bundle.parse_bindings(["android=tester.test", "ios=tester.test"], ["android", "ios"]) def test_preflight_matrix_rejects_remote_platform_using_local_runner(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) agent_dir = root / ".pouch" agent_dir.mkdir() (agent_dir / "agent.yaml").write_text( """ schemaVersion: 1 locations: local-app: environment: local host: local project: test path: /tmp/test remote-ios: environment: macmini host: runtime:12345678-1234-1234-1234-123456789012 project: test path: /tmp/test-ios profiles: test-agent: role: test cli: codex tier: standard model: gpt-5 reasoningEffort: medium permissionMode: workspace-write bindings: tester.android: {adapter: tester, location: local-app, profile: test-agent} tester.ios: {adapter: tester, location: remote-ios, profile: test-agent} """.strip() + "\n", encoding="utf-8", ) config = root / "matrix.yaml" config.write_text( """ matrix: - platform: android tester: {binding: tester.android, node: local, runner: {kind: local}} - platform: ios tester: {binding: tester.ios, node: local, runner: {kind: local}} """.strip() + "\n", encoding="utf-8", ) with self.assertRaisesRegex(bundle.EnvironmentFailure, "ios.*contract runner"): bundle.validate_preflight_matrix( root, config, {"android": "tester.android", "ios": "tester.ios"}, ["android", "ios"], ) def test_new_worktree_each_run(self): with tempfile.TemporaryDirectory() as temp: repo, worktrees = Path(temp) / "repo", Path(temp) / "worktrees" repo.mkdir() commit = init_repo(repo) first, first_commit = bundle.ensure_worktree(commit, bundle.RuntimeContext(repo), worktrees) second, second_commit = bundle.ensure_worktree(commit, bundle.RuntimeContext(repo), worktrees) self.assertEqual(first_commit, commit) self.assertEqual(second_commit, commit) self.assertNotEqual(first, second) self.assertTrue(first.is_dir()) self.assertTrue(second.is_dir()) def test_ipa_identity_and_install(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) ipa = root / "app.ipa" with zipfile.ZipFile(ipa, "w") as archive: archive.writestr("Payload/Writes.app/Info.plist", plistlib.dumps({"CFBundleIdentifier": "page.writes"})) self.assertEqual(bundle.ios_identity(ipa), "page.writes") context = bundle.RuntimeContext(root) calls = [] def fake(args, **kwargs): calls.append(args) return bundle.CommandResult(0, "/container" if "get_app_container" in args else "") with mock.patch.object(bundle, "run_command", side_effect=fake): result = bundle.install_ios({"xcrun": "xcrun", "udid": "UDID", "name": "Test"}, ipa, "page.writes", context) self.assertTrue(result["verified"]) self.assertTrue(any("install" in call for call in calls)) def test_android_identity_accepts_bare_apkanalyzer_output(self): with tempfile.TemporaryDirectory() as temp: apk = Path(temp) / "app.apk" apk.write_bytes(b"apk") context = bundle.RuntimeContext(Path(temp)) with mock.patch.object(bundle.shutil, "which", side_effect=lambda name: "apkanalyzer" if name == "apkanalyzer" else None), mock.patch.object( bundle, "run_command", return_value=bundle.CommandResult(0, "page.writes\n") ): self.assertEqual(bundle.android_identity(apk, context), "page.writes") def test_android_missing_device_is_environment_failure(self): context = bundle.RuntimeContext(Path(".")) with mock.patch.object(bundle.shutil, "which", return_value="/sdk/adb"), mock.patch.object( bundle, "run_command", side_effect=[bundle.CommandResult(0, ""), bundle.CommandResult(0, "List of devices attached")] ): with self.assertRaises(bundle.EnvironmentFailure): bundle.android_preflight(context) def test_network_failure_is_actionable(self): context = bundle.RuntimeContext(Path(".")) with mock.patch.object(bundle.urllib.request, "urlopen", side_effect=OSError("offline")), mock.patch.object(bundle.time, "sleep"): with mock.patch.dict(bundle.os.environ, {"HOST": "test.invalid"}, clear=False): with self.assertRaisesRegex(bundle.EnvironmentFailure, "VPN"): bundle.check_network(context) def test_retry_recovers(self): context = bundle.RuntimeContext(Path(".")) responses = [bundle.CommandResult(1, "busy"), bundle.CommandResult(0, "ok")] with mock.patch.object(bundle, "run_command", side_effect=responses), mock.patch.object(bundle.time, "sleep"): self.assertEqual(bundle.retry_command(["fake"], context=context, action="install").code, 0) def test_full_gate_blocks_install_when_ios_fails(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "repo" root.mkdir() commit = init_repo(root) apk, ipa = Path(temp) / "a.apk", Path(temp) / "a.ipa" apk.write_bytes(b"apk") ipa.write_bytes(b"ipa") args = bundle.parser().parse_args([ "run", "--artifact", f"android={apk}", "--artifact", f"ios={ipa}", "--root", str(root), "--worktree-root", str(Path(temp) / "wt"), ]) with mock.patch.object(bundle, "android_identity", return_value="page.writes"), mock.patch.object(bundle, "ios_identity", return_value="page.writes"), mock.patch.object(bundle, "android_preflight", return_value={"adb": "adb", "serial": "device"}), mock.patch.object(bundle, "ios_preflight", side_effect=bundle.EnvironmentFailure("Mac unavailable")), mock.patch.object(bundle, "check_network"), mock.patch.object(bundle, "install_android") as install: self.assertEqual(bundle.execute(args), 1) install.assert_not_called() self.assertEqual(args.commit, "") self.assertTrue(COMMIT_HEAD(root, commit)) def COMMIT_HEAD(root: Path, commit: str) -> bool: actual = subprocess.check_output(["git", "-C", str(root), "rev-parse", "HEAD"], text=True).strip() return actual == commit if __name__ == "__main__": unittest.main()