.pouch/skills/tester/tests/test_mobile_bundle.py

#!/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_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_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()