Impr upgrade

src/ffx/cli.py

@@ -112,6 +112,24 @@ def getBundleRepoPath():
     return getRepoRootPath()
 
 
+def getTrackedGitChanges(repoPath):
+    completed = subprocess.run(
+        ['git', 'status', '--porcelain', '--untracked-files=no'],
+        cwd=repoPath,
+        capture_output=True,
+        text=True,
+    )
+
+    if completed.returncode != 0:
+        commandLabel = 'git status --porcelain --untracked-files=no'
+        errorOutput = completed.stderr.strip() or completed.stdout.strip()
+        raise click.ClickException(
+            f"Unable to inspect bundle repository state using '{commandLabel}': {errorOutput}"
+        )
+
+    return [line for line in completed.stdout.splitlines() if line.strip()]
+
+
 @ffx.command(name='configure_workstation')
 @click.pass_context
 @click.option('--check', is_flag=True, default=False, help='Only verify workstation-configuration readiness')
@@ -152,6 +170,24 @@ def upgrade(ctx, branch):
         raise click.ClickException(f"Bundle pip not found at {bundlePipPath}")
 
     commandSequences = []
+    trackedChanges = getTrackedGitChanges(bundleRepoPath)
+
+    if trackedChanges:
+        click.echo("Tracked local changes detected in the bundle repository:")
+        for trackedChange in trackedChanges:
+            click.echo(f"  {trackedChange}")
+
+        shouldReset = click.confirm(
+            "Discard these tracked changes with 'git reset --hard HEAD' before upgrade?",
+            default=False,
+        )
+
+        if not shouldReset:
+            raise click.ClickException(
+                "Upgrade aborted because tracked local changes are present."
+            )
+
+        commandSequences.append(['git', 'reset', '--hard', 'HEAD'])
 
     if branch:
         commandSequences.append(['git', 'checkout', branch])

tests/unit/test_cli_upgrade.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from pathlib import Path
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+
+from click.testing import CliRunner
+
+
+SRC_ROOT = Path(__file__).resolve().parents[2] / "src"
+
+if str(SRC_ROOT) not in sys.path:
+    sys.path.insert(0, str(SRC_ROOT))
+
+
+from ffx import cli  # noqa: E402
+
+
+class UpgradeCommandTests(unittest.TestCase):
+    def make_completed(self, args, *, stdout: str = "", stderr: str = "", returncode: int = 0):
+        return subprocess.CompletedProcess(args=args, returncode=returncode, stdout=stdout, stderr=stderr)
+
+    def test_upgrade_aborts_when_tracked_changes_are_present_and_reset_is_declined(self):
+        runner = CliRunner()
+        repo_path = "/tmp/ffx-repo"
+        pip_path = "/tmp/ffx-venv/bin/pip"
+
+        subprocess_calls = []
+
+        def fake_run(args, **kwargs):
+            subprocess_calls.append((args, kwargs))
+            if args == ['git', 'status', '--porcelain', '--untracked-files=no']:
+                return self.make_completed(args, stdout=" M src/ffx/cli.py\n")
+            raise AssertionError(f"Unexpected subprocess invocation: args={args} kwargs={kwargs}")
+
+        with (
+            patch.object(cli, "getBundleRepoPath", return_value=repo_path),
+            patch.object(cli, "getBundlePipPath", return_value=pip_path),
+            patch.object(cli.os.path, "isdir", return_value=True),
+            patch.object(cli.os.path, "isfile", return_value=True),
+            patch.object(cli.subprocess, "run", side_effect=fake_run),
+        ):
+            result = runner.invoke(cli.ffx, ["upgrade"], input="n\n")
+
+        self.assertNotEqual(0, result.exit_code)
+        self.assertIn("Tracked local changes detected in the bundle repository:", result.output)
+        self.assertIn("Discard these tracked changes with 'git reset --hard HEAD' before upgrade?", result.output)
+        self.assertIn("Upgrade aborted because tracked local changes are present.", result.output)
+        self.assertEqual(1, len(subprocess_calls))
+        self.assertEqual(
+            ['git', 'status', '--porcelain', '--untracked-files=no'],
+            subprocess_calls[0][0],
+        )
+        self.assertEqual(repo_path, subprocess_calls[0][1]["cwd"])
+        self.assertTrue(subprocess_calls[0][1]["capture_output"])
+        self.assertTrue(subprocess_calls[0][1]["text"])
+
+    def test_upgrade_resets_before_checkout_and_pull_when_user_confirms(self):
+        runner = CliRunner()
+        repo_path = "/tmp/ffx-repo"
+        pip_path = "/tmp/ffx-venv/bin/pip"
+
+        subprocess_calls = []
+
+        def fake_run(args, **kwargs):
+            subprocess_calls.append((args, kwargs))
+            if args == ['git', 'status', '--porcelain', '--untracked-files=no']:
+                return self.make_completed(args, stdout="M  src/ffx/constants.py\n")
+            return self.make_completed(args)
+
+        with (
+            patch.object(cli, "getBundleRepoPath", return_value=repo_path),
+            patch.object(cli, "getBundlePipPath", return_value=pip_path),
+            patch.object(cli.os.path, "isdir", return_value=True),
+            patch.object(cli.os.path, "isfile", return_value=True),
+            patch.object(cli.subprocess, "run", side_effect=fake_run),
+        ):
+            result = runner.invoke(cli.ffx, ["upgrade", "--branch", "main"], input="y\n")
+
+        self.assertEqual(0, result.exit_code, result.output)
+        self.assertIn("Tracked local changes detected in the bundle repository:", result.output)
+        self.assertEqual(
+            [
+                ['git', 'status', '--porcelain', '--untracked-files=no'],
+                ['git', 'reset', '--hard', 'HEAD'],
+                ['git', 'checkout', 'main'],
+                ['git', 'pull'],
+                [pip_path, 'install', '--editable', '.'],
+            ],
+            [call[0] for call in subprocess_calls],
+        )
+        for args, kwargs in subprocess_calls[1:]:
+            self.assertEqual(repo_path, kwargs["cwd"], args)
+
+
+if __name__ == "__main__":
+    unittest.main()