plezy/scripts/release/test_deploy.py

import json
import os
import plistlib
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))

import deploy
from deploy import (
    DeployError,
    State,
    bump_pubspec_text,
    parse_env_text,
    resolve_phases,
    sanitize_msstore_pricing,
)


class ParseEnvTextTest(unittest.TestCase):
    def test_parses_values_and_skips_comments_and_blanks(self) -> None:
        text = """\
# App Store
DELIVER_USERNAME=user@example.com

export TOKEN=abc123
NOT A VALID LINE
"""
        values = parse_env_text(text)
        self.assertEqual(
            values,
            {"DELIVER_USERNAME": "user@example.com", "TOKEN": "abc123"},
        )

    def test_strips_matching_quotes(self) -> None:
        values = parse_env_text("A=\"quoted value\"\nB='single'\nC=un\"quoted\n")
        self.assertEqual(values["A"], "quoted value")
        self.assertEqual(values["B"], "single")
        self.assertEqual(values["C"], 'un"quoted')

    def test_expands_references_to_earlier_keys(self) -> None:
        # The real .env relies on this: SUPPLY_PACKAGE_NAME=${DELIVER_APP_IDENTIFIER}
        text = "DELIVER_APP_IDENTIFIER=com.edde746.plezy\nSUPPLY_PACKAGE_NAME=${DELIVER_APP_IDENTIFIER}\n"
        values = parse_env_text(text)
        self.assertEqual(values["SUPPLY_PACKAGE_NAME"], "com.edde746.plezy")

    def test_expands_from_base_environment(self) -> None:
        values = parse_env_text("A=${HOME_LIKE}/sub\n", {"HOME_LIKE": "/base"})
        self.assertEqual(values["A"], "/base/sub")

    def test_unknown_reference_expands_empty(self) -> None:
        self.assertEqual(parse_env_text("A=${MISSING}x\n")["A"], "x")

    def test_single_quotes_suppress_expansion(self) -> None:
        values = parse_env_text("A='${NOPE}'\n", {"NOPE": "value"})
        self.assertEqual(values["A"], "${NOPE}")



class LoadEnvFileTest(unittest.TestCase):
    def test_file_values_override_stale_ambient_release_configuration(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / ".env").write_text(
                "DELIVER_APP_IDENTIFIER=com.edde746.plezy\n"
                "SUPPLY_PACKAGE_NAME=${DELIVER_APP_IDENTIFIER}\n"
                "APPLE_TEAM_ID=G88U5B5783\n",
                encoding="utf-8",
            )
            ambient = {
                "DELIVER_APP_IDENTIFIER": "com.example.stale",
                "SUPPLY_PACKAGE_NAME": "com.example.stale",
                "APPLE_TEAM_ID": "STALETEAM1",
            }
            with (
                mock.patch.object(deploy, "ROOT", root),
                mock.patch.dict(os.environ, ambient, clear=True),
            ):
                loaded = deploy.load_env_file()

        self.assertEqual(loaded["DELIVER_APP_IDENTIFIER"], "com.edde746.plezy")
        self.assertEqual(loaded["SUPPLY_PACKAGE_NAME"], "com.edde746.plezy")
        self.assertEqual(loaded["APPLE_TEAM_ID"], "G88U5B5783")


class GoogleRequestTest(unittest.TestCase):
    def test_requests_use_client_retry_policy(self) -> None:
        request = mock.Mock()
        request.execute.return_value = {"ok": True}

        self.assertEqual(deploy._execute_google_request(request), {"ok": True})

        request.execute.assert_called_once_with(num_retries=5)

    def test_resumable_upload_retries_every_chunk(self) -> None:
        status = mock.Mock()
        status.progress.return_value = 0.5
        request = mock.Mock()
        request.next_chunk.side_effect = [(status, None), (None, {"versionCode": 131})]

        response = deploy._execute_resumable_google_upload(request, "play")

        self.assertEqual(response, {"versionCode": 131})
        self.assertEqual(request.next_chunk.call_count, 2)
        request.next_chunk.assert_called_with(num_retries=5)


class AppleUploadTest(unittest.TestCase):
    @staticmethod
    def _context() -> deploy.Context:
        return deploy.Context(
            args=SimpleNamespace(dry_run=False, yes=True),
            env={
                "APPLE_TEAM_ID": "G88U5B5783",
                "ASC_KEY_ID": "KEY1234567",
                "ASC_ISSUER_ID": "issuer",
            },
            state=State(version="2.15.0", build_number=131),
            phases=["ios", "tvos"],
        )

    def test_export_uses_local_xcode_account_and_returns_ipa(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            deploy_dir = Path(tmp) / "deploy"
            archive = Path(tmp) / "Runner.xcarchive"
            archive.mkdir()
            ipa = deploy_dir / "export-tvos/Plezy.ipa"
            with (
                mock.patch.object(deploy, "DEPLOY_DIR", deploy_dir),
                mock.patch.object(deploy, "run") as run_command,
                mock.patch.object(deploy, "_find_single_ipa", return_value=ipa),
            ):
                result = deploy._export_archive(self._context(), archive, "tvos")

            options = plistlib.loads((deploy_dir / "export-options-tvos.plist").read_bytes())

        self.assertEqual(result, ipa)
        self.assertEqual(options["destination"], "export")
        self.assertEqual(options["teamID"], "G88U5B5783")
        command = run_command.call_args.args[0]
        self.assertIn("-allowProvisioningUpdates", command)
        self.assertNotIn("-authenticationKeyPath", command)

    def test_upload_uses_altool_with_api_key(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            ipa = Path(tmp) / "Plezy.ipa"
            ipa.write_bytes(b"ipa")
            with mock.patch.object(deploy, "run") as run_command:
                deploy._upload_ipa(self._context(), ipa, "tvos")

        run_command.assert_called_once_with(
            [
                "xcrun",
                "altool",
                "--upload-app",
                "--type",
                "tvos",
                "-f",
                str(ipa),
                "--apiKey",
                "KEY1234567",
                "--apiIssuer",
                "issuer",
            ]
        )

    def test_ios_resume_skips_an_already_uploaded_build(self) -> None:
        ctx = self._context()
        with (
            mock.patch.object(deploy, "_asc_has_uploaded_build", return_value=True) as has_build,
            mock.patch.object(deploy, "run") as run_command,
        ):
            deploy.phase_ios(ctx)

        has_build.assert_called_once_with(ctx, "IOS")
        run_command.assert_not_called()

    def test_tvos_resume_skips_an_already_uploaded_build(self) -> None:
        ctx = self._context()
        with (
            mock.patch.object(deploy, "_asc_has_uploaded_build", return_value=True) as has_build,
            mock.patch.object(deploy, "run") as run_command,
        ):
            deploy.phase_tvos(ctx)

        has_build.assert_called_once_with(ctx, "TV_OS")
        run_command.assert_not_called()


class BumpPubspecTextTest(unittest.TestCase):
    PUBSPEC = """\
name: plezy
description: A media client.
version: 2.12.1+230
environment:
  sdk: ">=3.12.0 <4.0.0"
dependencies:
  some_package:
    version: 9.9.9+999
"""

    def test_replaces_only_the_top_level_version(self) -> None:
        result = bump_pubspec_text(self.PUBSPEC, "2.13.0", 231)
        self.assertIn("version: 2.13.0+231\n", result)
        self.assertIn("    version: 9.9.9+999\n", result)
        self.assertNotIn("2.12.1+230", result)
        self.assertEqual(result.count("\n"), self.PUBSPEC.count("\n"))

    def test_rejects_pubspec_without_version(self) -> None:
        with self.assertRaises(ValueError):
            bump_pubspec_text("name: plezy\n", "2.13.0", 231)



class ResolvePhasesTest(unittest.TestCase):
    def test_default_selects_all_phases_in_order(self) -> None:
        self.assertEqual(resolve_phases(None, None), deploy.PHASES)

    def test_only_keeps_preflight_and_canonical_order(self) -> None:
        self.assertEqual(
            resolve_phases(["msstore", "play"], None),
            ["preflight", "play", "msstore"],
        )

    def test_skip_removes_phases(self) -> None:
        selected = resolve_phases(None, ["ios", "tvos", "asc"])
        self.assertNotIn("ios", selected)
        self.assertNotIn("tvos", selected)
        self.assertNotIn("asc", selected)
        self.assertIn("play", selected)

    def test_preflight_can_be_skipped_explicitly(self) -> None:
        self.assertEqual(resolve_phases(["release"], ["preflight"]), ["release"])

    def test_unknown_phase_raises(self) -> None:
        with self.assertRaises(DeployError):
            resolve_phases(["appstore"], None)
        with self.assertRaises(DeployError):
            resolve_phases(None, ["nope"])




class BuildFarmTest(unittest.TestCase):
    @staticmethod
    def _context(state: State) -> deploy.Context:
        return deploy.Context(
            args=SimpleNamespace(dry_run=False, yes=True),
            env={},
            state=state,
            phases=["farm_start", "farm_wait"],
        )

    def test_start_passes_release_tag_and_records_only_the_new_run(self) -> None:
        state = State(version="2.15.0", build_number=131)
        state.data["release_sha"] = "abc123"
        responses = [
            SimpleNamespace(returncode=0, stdout=json.dumps([{"databaseId": 10}])),
            SimpleNamespace(returncode=0, stdout=""),
            SimpleNamespace(
                returncode=0,
                stdout=json.dumps(
                    [
                        {
                            "databaseId": 12,
                            "displayTitle": "Build abc123",
                            "headSha": "abc123",
                            "status": "completed",
                            "createdAt": "2026-08-18T00:00:01Z",
                        },
                        {
                            "databaseId": 11,
                            "displayTitle": "Release 2.15.0",
                            "headSha": "abc123",
                            "status": "completed",
                            "createdAt": "2026-08-18T00:00:00Z",
                        },
                        {
                            "databaseId": 10,
                            "headSha": "abc123",
                            "status": "completed",
                            "createdAt": "2026-08-17T00:00:00Z",
                        },
                    ]
                ),
            ),
        ]
        with (
            mock.patch.object(deploy, "run", side_effect=responses) as run_command,
            mock.patch.object(deploy.time, "sleep"),
            mock.patch.object(state, "save"),
        ):
            deploy.phase_farm_start(self._context(state))

        dispatch = run_command.call_args_list[1].args[0]
        self.assertIn("release_tag=2.15.0", dispatch)
        self.assertEqual(state.data["farm_run_id"], 11)

    def test_start_resumes_discovery_without_dispatching_twice(self) -> None:
        state = State(version="2.15.0", build_number=131)
        state.data.update(
            {
                "release_sha": "abc123",
                "farm_dispatch": {
                    "headSha": "abc123",
                    "previousRunIds": [10],
                },
            }
        )
        response = SimpleNamespace(
            returncode=0,
            stdout=json.dumps(
                [
                    {
                        "databaseId": 11,
                        "displayTitle": "Release 2.15.0",
                        "headSha": "abc123",
                        "status": "queued",
                        "createdAt": "2026-08-18T00:00:00Z",
                    }
                ]
            ),
        )
        with (
            mock.patch.object(deploy, "run", return_value=response) as run_command,
            mock.patch.object(deploy.time, "sleep"),
            mock.patch.object(state, "save"),
        ):
            deploy.phase_farm_start(self._context(state))

        run_command.assert_called_once()
        self.assertEqual(run_command.call_args.args[0][:4], ["gh", "run", "list", "--workflow"])
        self.assertEqual(state.data["farm_run_id"], 11)
        self.assertNotIn("farm_dispatch", state.data)

    def test_wait_downloads_only_the_store_artifact(self) -> None:
        state = State(version="2.15.0", build_number=131)
        state.data["farm_run_id"] = 11
        with tempfile.TemporaryDirectory() as tmp:
            artifacts = Path(tmp) / "artifacts"
            responses = [
                SimpleNamespace(
                    returncode=0,
                    stdout=json.dumps({"status": "completed", "conclusion": "success"}),
                ),
                SimpleNamespace(returncode=0, stdout=""),
            ]
            with (
                mock.patch.object(deploy, "ARTIFACTS_DIR", artifacts),
                mock.patch.object(deploy, "run", side_effect=responses) as run_command,
            ):
                deploy.phase_farm_wait(self._context(state))

        download = run_command.call_args_list[1].args[0]
        self.assertEqual(download[:4], ["gh", "run", "download", "11"])
        self.assertIn("windows-msix", download)
        self.assertNotIn("--pattern", download)


class GitHubReleaseTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self._notes_dir = Path(self._tmp.name) / "notes"
        self._notes_dir.mkdir()
        (self._notes_dir / "github.md").write_text("Release notes\n", encoding="utf-8")
        self._notes_patch = mock.patch.object(deploy, "NOTES_DIR", self._notes_dir)
        self._notes_patch.start()

    def tearDown(self) -> None:
        self._notes_patch.stop()
        self._tmp.cleanup()

    @staticmethod
    def _context() -> deploy.Context:
        state = State(version="2.15.0", build_number=131)
        state.data["release_sha"] = "abc123"
        return deploy.Context(
            args=SimpleNamespace(dry_run=False, yes=True),
            env={},
            state=state,
            phases=["release"],
        )

    @staticmethod
    def _release_info(assets: set[str]) -> SimpleNamespace:
        return SimpleNamespace(
            returncode=0,
            stdout=json.dumps(
                {
                    "isDraft": True,
                    "targetCommitish": "abc123",
                    "assets": [{"name": name} for name in sorted(assets)],
                }
            ),
        )

    def test_verifies_workflow_assets_then_attaches_notes(self) -> None:
        responses = [
            self._release_info(set(deploy.RELEASE_ASSET_NAMES)),
            SimpleNamespace(returncode=0, stdout=""),
        ]
        with mock.patch.object(deploy, "run", side_effect=responses) as run_command:
            deploy.phase_release(self._context())

        edit = run_command.call_args_list[1].args[0]
        self.assertEqual(edit[:4], ["gh", "release", "edit", "2.15.0"])
        self.assertIn(str(self._notes_dir / "github.md"), edit)

    def test_rejects_incomplete_draft_before_editing(self) -> None:
        assets = set(deploy.RELEASE_ASSET_NAMES)
        assets.remove("plezy-macos.dmg")
        with mock.patch.object(
            deploy,
            "run",
            return_value=self._release_info(assets),
        ) as run_command:
            with self.assertRaisesRegex(DeployError, "plezy-macos.dmg"):
                deploy.phase_release(self._context())

        run_command.assert_called_once()


class StateTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self._original = deploy.STATE_PATH
        deploy.STATE_PATH = Path(self._tmp.name) / "state.json"

    def tearDown(self) -> None:
        deploy.STATE_PATH = self._original
        self._tmp.cleanup()

    def test_round_trip(self) -> None:
        state = State(version="2.13.0", build_number=231)
        state.data["farm_run_id"] = 42
        state.mark_done("changelog")
        loaded = State.load()
        self.assertIsNotNone(loaded)
        self.assertEqual(loaded.version, "2.13.0")
        self.assertEqual(loaded.build_number, 231)
        self.assertEqual(loaded.done, ["changelog"])
        self.assertEqual(loaded.data, {"farm_run_id": 42})

    def test_mark_done_is_idempotent(self) -> None:
        state = State(version="2.13.0", build_number=231)
        state.mark_done("play")
        state.mark_done("play")
        self.assertEqual(State.load().done, ["play"])

    def test_load_returns_none_without_file(self) -> None:
        self.assertIsNone(State.load())


class InitializeStateTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self._original_root = deploy.ROOT
        self._original_deploy_dir = deploy.DEPLOY_DIR
        self._original_state_path = deploy.STATE_PATH
        deploy.ROOT = Path(self._tmp.name)
        deploy.DEPLOY_DIR = deploy.ROOT / "build" / "deploy"
        deploy.STATE_PATH = deploy.DEPLOY_DIR / "state.json"
        (deploy.ROOT / "pubspec.yaml").write_text("version: 2.14.0+130\n", encoding="utf-8")

    def tearDown(self) -> None:
        deploy.ROOT = self._original_root
        deploy.DEPLOY_DIR = self._original_deploy_dir
        deploy.STATE_PATH = self._original_state_path
        self._tmp.cleanup()

    @staticmethod
    def _args(*, dry_run: bool) -> SimpleNamespace:
        return SimpleNamespace(
            version="2.15.0",
            build_number=None,
            fresh=True,
            dry_run=dry_run,
        )

    def _write_previous_release(self) -> Path:
        State(version="2.14.0", build_number=130, done=["publish"]).save()
        note = deploy.DEPLOY_DIR / "notes" / "play.txt"
        note.parent.mkdir(parents=True)
        note.write_text("old store notes\n", encoding="utf-8")
        return note

    def test_fresh_release_removes_state_and_generated_outputs(self) -> None:
        old_note = self._write_previous_release()

        state = deploy.initialize_state(self._args(dry_run=False))

        self.assertEqual((state.version, state.build_number), ("2.15.0", 131))
        self.assertFalse(old_note.exists())
        self.assertEqual(State.load(), state)

    def test_dry_run_fresh_release_preserves_existing_outputs(self) -> None:
        old_note = self._write_previous_release()

        state = deploy.initialize_state(self._args(dry_run=True))

        self.assertEqual((state.version, state.build_number), ("2.15.0", 131))
        self.assertEqual(old_note.read_text(encoding="utf-8"), "old store notes\n")
        self.assertEqual(State.load().version, "2.14.0")


class DeferredPhaseTest(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self._original_state_path = deploy.STATE_PATH
        deploy.STATE_PATH = Path(self._tmp.name) / "state.json"

    def tearDown(self) -> None:
        deploy.STATE_PATH = self._original_state_path
        self._tmp.cleanup()

    @staticmethod
    def _context(state: State, **args) -> deploy.Context:
        defaults = {
            "dry_run": False,
            "yes": False,
            "amazon_skip_commit": False,
        }
        defaults.update(args)
        return deploy.Context(
            args=SimpleNamespace(**defaults),
            env={
                "AMAZON_APPSTORE_PACKAGE_NAME": "com.example.app",
                "AMAZON_APPSTORE_APP_ID": "amzn1.devportal.mobileapp.fixture",
            },
            state=state,
            phases=["amazon", "publish"],
        )

    def test_deferred_phase_remains_pending_in_release_state(self) -> None:
        state = State(version="2.15.0", build_number=131)
        args = SimpleNamespace(
            only=["publish"],
            skip=None,
            notes="cl.txt",
            dry_run=False,
        )

        def defer(_ctx) -> None:
            raise deploy.PhaseDeferred

        with (
            mock.patch.object(deploy, "load_env_file", return_value={}),
            mock.patch.object(deploy, "initialize_state", return_value=state),
            mock.patch.object(deploy, "resolve_phases", return_value=["publish"]),
            mock.patch.dict(deploy.PHASE_FUNCTIONS, {"publish": defer}, clear=True),
            mock.patch.object(deploy, "print_summary"),
        ):
            self.assertEqual(deploy.cmd_release(args), 0)

        self.assertNotIn("publish", State.load().done)

    def test_publish_decline_defers_phase(self) -> None:
        ctx = self._context(State(version="2.15.0", build_number=131))

        with mock.patch.object(deploy.Context, "confirm", return_value=False):
            with self.assertRaises(deploy.PhaseDeferred):
                deploy.phase_publish(ctx)

    def test_publish_does_not_commit_or_push_package_metadata(self) -> None:
        ctx = self._context(State(version="2.15.0", build_number=131))

        with (
            mock.patch.object(deploy.Context, "confirm", return_value=True),
            mock.patch.object(deploy, "run") as run_command,
        ):
            deploy.phase_publish(ctx)

        run_command.assert_called_once_with(
            ["gh", "release", "edit", "2.15.0", "--draft=false"]
        )

    def test_amazon_resume_commits_existing_edit_without_replacing_apk(self) -> None:
        state = State(version="2.15.0", build_number=131)
        state.data["amazon_pending_edit_id"] = "edit-42"
        ctx = self._context(state)
        client = mock.Mock()
        client.etag.return_value = "etag-42"

        with (
            mock.patch.object(deploy, "_amazon_token", return_value="token"),
            mock.patch.object(deploy, "AmazonClient", return_value=client),
            mock.patch.object(deploy.Context, "confirm", return_value=True),
            mock.patch.object(deploy, "run") as run_command,
        ):
            deploy.phase_amazon(ctx)

        run_command.assert_not_called()
        client.request.assert_called_once_with(
            "POST",
            "/edits/edit-42/commit",
            headers={"If-Match": "etag-42"},
        )
        self.assertNotIn("amazon_pending_edit_id", state.data)

    def test_amazon_resume_decline_keeps_edit_pending(self) -> None:
        state = State(version="2.15.0", build_number=131)
        state.data["amazon_pending_edit_id"] = "edit-42"
        ctx = self._context(state)
        client = mock.Mock()

        with (
            mock.patch.object(deploy, "_amazon_token", return_value="token"),
            mock.patch.object(deploy, "AmazonClient", return_value=client),
            mock.patch.object(deploy.Context, "confirm", return_value=False),
            mock.patch.object(deploy, "run") as run_command,
        ):
            with self.assertRaises(deploy.PhaseDeferred):
                deploy.phase_amazon(ctx)

        run_command.assert_not_called()
        client.request.assert_not_called()
        self.assertEqual(state.data["amazon_pending_edit_id"], "edit-42")


class SplitPhaseListsTest(unittest.TestCase):
    def test_splits_commas_and_repeats(self) -> None:
        self.assertEqual(
            deploy._split_phase_lists(["play,amazon", "ios"]),
            ["play", "amazon", "ios"],
        )

    def test_none_passthrough(self) -> None:
        self.assertIsNone(deploy._split_phase_lists(None))
        self.assertIsNone(deploy._split_phase_lists([]))


class SanitizeMsstorePricingTest(unittest.TestCase):
    def test_drops_price_id_and_advanced_flag_but_keeps_the_rest(self) -> None:
        submission = {
            "pricing": {
                "trialPeriod": "SevenDays",
                "marketSpecificPricings": {"LB": "NotAvailable"},
                "sales": [],
                "priceId": "Base",
                "isAdvancedPricingModel": True,
            }
        }
        sanitize_msstore_pricing(submission)
        self.assertEqual(
            submission["pricing"],
            {
                "trialPeriod": "SevenDays",
                "marketSpecificPricings": {"LB": "NotAvailable"},
                "sales": [],
            },
        )

    def test_tolerates_missing_pricing(self) -> None:
        submission: dict = {}
        sanitize_msstore_pricing(submission)
        self.assertEqual(submission, {})


if __name__ == "__main__":
    unittest.main()