From 31f83de4164a4b491b959af1e2c1fff96735cddd Mon Sep 17 00:00:00 2001 From: Gavin Mak Date: Tue, 15 Jul 2025 19:06:08 +0000 Subject: [PATCH] wipe: Add new `repo wipe` subcommand This new command allows users to delete projects from the worktree and from the `.repo` directory. It is a destructive operation. It handles shared projects by refusing to wipe them unless the `--force` flag is used. It also checks for uncommitted changes before wiping. Bug: 393383056 Change-Id: Ia30d8ffdc781a3f179af56310ce31c9dae331bbe --- subcmds/wipe.py | 121 ++++++++++++++++++++ tests/test_subcmds_wipe.py | 229 +++++++++++++++++++++++++++++++++++++ 2 files changed, 350 insertions(+) create mode 100644 subcmds/wipe.py create mode 100644 tests/test_subcmds_wipe.py diff --git a/subcmds/wipe.py b/subcmds/wipe.py new file mode 100644 index 000000000..546f5efae --- /dev/null +++ b/subcmds/wipe.py @@ -0,0 +1,121 @@ +# Copyright (C) 2025 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import os +import shutil +from typing import List + +from command import Command +from command import UsageError +from git_command import GitCommand +from project import Project + + +class Wipe(Command, GitCommand): + """Delete projects from the worktree and .repo""" + + COMMON = True + helpSummary = "Wipe projects from the worktree" + helpUsage = """ +%prog ... +""" + helpDescription = """ +The `repo wipe` command removes the specified projects from the worktree, +and deletes the project's git data from `.repo`. + +This is a destructive operation and cannot be undone. +""" + + def _Options(self, p): + p.add_option( + "-f", + "--force", + action="store_true", + help="force wipe in the case of shared projects", + ) + + def Execute(self, opt, args: List[str]): + if not args: + raise UsageError("no projects specified") + + # We need all projects to correctly handle shared object directories. + all_projects = self.GetProjects(None, all_manifests=True) + projects_to_wipe = self.GetProjects(args, all_manifests=True) + names_to_wipe = {p.name for p in projects_to_wipe} + + # Build a map from objdir to the names of projects that use it. + objdir_map = {} + for p in all_projects: + objdir_map.setdefault(p.objdir, set()).add(p.name) + + uncommitted_projects = [] + shared_objdirs = {} + objdirs_to_delete = set() + + for project in projects_to_wipe: + if project.HasChanges(): + uncommitted_projects.append(project.name) + + users = objdir_map.get(project.objdir, {project.name}) + is_shared = not users.issubset(names_to_wipe) + if is_shared: + shared_objdirs.setdefault(project.objdir, set()).update(users) + else: + objdirs_to_delete.add(project.objdir) + + if (uncommitted_projects or shared_objdirs) and not opt.force: + error_messages = [] + if uncommitted_projects: + error_messages.append( + "The following projects have uncommitted changes:\n" + + "\n - ".join(sorted(uncommitted_projects)) + ) + if shared_objdirs: + shared_dir_messages = [] + for objdir, users in sorted(shared_objdirs.items()): + other_users = users - names_to_wipe + projects_to_wipe_in_dir = users & names_to_wipe + message = f"""Object directory {objdir} is shared by: + Projects to be wiped: {', '.join(sorted(list(projects_to_wipe_in_dir)))} + Projects not to be wiped: {', '.join(sorted(list(other_users)))}""" + shared_dir_messages.append(message) + error_messages.append( + "The following projects have shared object directories:\n" + + "\n\n".join(sorted(shared_dir_messages)) + ) + error_messages.append("\nUse --force to wipe anyway.") + raise UsageError("\n\n".join(error_messages)) + + # If we are here, either there were no issues, or --force was used. + # Proceed with wiping. + for project in projects_to_wipe: + self._WipeProject(project) + + for objdir in objdirs_to_delete: + if os.path.exists(objdir): + print(f"Deleting objects directory: {objdir}") + shutil.rmtree(objdir) + + def _WipeProject(self, project: Project): + """Wipes a single project's worktree and git directory.""" + if not project: + return + + if project.worktree and os.path.exists(project.worktree): + print(f"Deleting worktree: {project.worktree}") + shutil.rmtree(project.worktree) + + if os.path.exists(project.gitdir): + print(f"Deleting git directory: {project.gitdir}") + shutil.rmtree(project.gitdir) diff --git a/tests/test_subcmds_wipe.py b/tests/test_subcmds_wipe.py new file mode 100644 index 000000000..35ba14d85 --- /dev/null +++ b/tests/test_subcmds_wipe.py @@ -0,0 +1,229 @@ +# Copyright (C) 2025 The Android Open Source Project +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import os +import shutil +import tempfile +import unittest +from unittest.mock import MagicMock + +from command import UsageError +from subcmds.wipe import Wipe + + +class WipeUnitTest(unittest.TestCase): + """Test the wipe subcommand.""" + + def setUp(self): + # Create a temporary directory for the entire test run + self.tempdir = tempfile.mkdtemp(prefix="repo_wipe_test_") + self.addCleanup(shutil.rmtree, self.tempdir) + + # Helper to create mock project objects + def create_mock_project(name, objdir_path=None, has_changes=False): + """Creates a mock project with necessary attributes and directories.""" + proj = MagicMock() + proj.name = name + proj.relpath = name + proj.HasChanges = MagicMock(return_value=has_changes) + # Simulate paths based on the temp directory + proj.worktree = os.path.join(self.tempdir, name) + proj.gitdir = os.path.join( + self.tempdir, ".repo/projects", name + ".git" + ) + # Allow sharing objdir by specifying a path + if objdir_path: + proj.objdir = objdir_path + else: + proj.objdir = os.path.join( + self.tempdir, ".repo/project-objects", name + ".git" + ) + + os.makedirs(proj.worktree, exist_ok=True) + os.makedirs(proj.gitdir, exist_ok=True) + os.makedirs(proj.objdir, exist_ok=True) + return proj + + self.create_mock_project = create_mock_project + + def test_wipe_single_unshared_project(self): + """Test wiping a single project that is not shared.""" + p1 = self.create_mock_project("project/one", has_changes=False) + + cmd = Wipe() + # Simulate repo's project discovery + cmd.GetAllProjects = MagicMock(return_value=[p1]) + cmd.GetProjects = MagicMock(return_value=[p1]) + + cmd.Execute( + cmd.OptionParser.parse_args(["project/one"])[0], ["project/one"] + ) + + # All directories should be gone + self.assertFalse(os.path.exists(p1.worktree)) + self.assertFalse(os.path.exists(p1.gitdir)) + self.assertFalse(os.path.exists(p1.objdir)) + + def test_wipe_multiple_unshared_projects(self): + """Test wiping multiple projects that are not shared.""" + p1 = self.create_mock_project("project/one", has_changes=False) + p2 = self.create_mock_project("project/two", has_changes=False) + + cmd = Wipe() + cmd.GetAllProjects = MagicMock(return_value=[p1, p2]) + cmd.GetProjects = MagicMock(return_value=[p1, p2]) + + cmd.Execute( + cmd.OptionParser.parse_args(["project/one", "project/two"])[0], + ["project/one", "project/two"], + ) + + # All directories for both projects should be gone + self.assertFalse(os.path.exists(p1.worktree)) + self.assertFalse(os.path.exists(p1.gitdir)) + self.assertFalse(os.path.exists(p1.objdir)) + self.assertFalse(os.path.exists(p2.worktree)) + self.assertFalse(os.path.exists(p2.gitdir)) + self.assertFalse(os.path.exists(p2.objdir)) + + def test_wipe_shared_project_no_force_raises_error(self): + """Test that wiping a shared project without --force raises an error.""" + # p1 and p2 share an object directory + shared_objdir = os.path.join( + self.tempdir, ".repo/project-objects", "shared.git" + ) + p1 = self.create_mock_project( + "project/one", objdir_path=shared_objdir, has_changes=False + ) + p2 = self.create_mock_project( + "project/two", objdir_path=shared_objdir, has_changes=False + ) + + cmd = Wipe() + cmd.GetAllProjects = MagicMock(return_value=[p1, p2]) + # We are only trying to wipe p1 + cmd.GetProjects = MagicMock(return_value=[p1]) + + with self.assertRaises(UsageError): + cmd.Execute( + cmd.OptionParser.parse_args(["project/one"])[0], ["project/one"] + ) + + # Nothing should have been deleted + self.assertTrue(os.path.exists(p1.worktree)) + self.assertTrue(os.path.exists(p1.gitdir)) + self.assertTrue(os.path.exists(p2.worktree)) + self.assertTrue(os.path.exists(p2.gitdir)) + self.assertTrue(os.path.exists(shared_objdir)) + + def test_wipe_shared_project_with_force(self): + """ + Test wiping a shared project with --force. + It should remove the project but leave the shared object directory. + """ + shared_objdir = os.path.join( + self.tempdir, ".repo/project-objects", "shared.git" + ) + p1 = self.create_mock_project( + "project/one", objdir_path=shared_objdir, has_changes=False + ) + p2 = self.create_mock_project( + "project/two", objdir_path=shared_objdir, has_changes=False + ) + + cmd = Wipe() + cmd.GetAllProjects = MagicMock(return_value=[p1, p2]) + cmd.GetProjects = MagicMock(return_value=[p1]) + + cmd.Execute( + cmd.OptionParser.parse_args(["--force", "project/one"])[0], + ["project/one"], + ) + + # p1's specific dirs are gone + self.assertFalse(os.path.exists(p1.worktree)) + self.assertFalse(os.path.exists(p1.gitdir)) + + # The shared object directory and p2's dirs must remain + self.assertTrue(os.path.exists(shared_objdir)) + self.assertTrue(os.path.exists(p2.worktree)) + self.assertTrue(os.path.exists(p2.gitdir)) + + def test_wipe_all_sharing_projects(self): + """ + Test wiping all projects that share an object directory. + This should remove the shared object directory as well. + """ + shared_objdir = os.path.join( + self.tempdir, ".repo/project-objects", "shared.git" + ) + p1 = self.create_mock_project( + "project/one", objdir_path=shared_objdir, has_changes=False + ) + p2 = self.create_mock_project( + "project/two", objdir_path=shared_objdir, has_changes=False + ) + + cmd = Wipe() + cmd.GetAllProjects = MagicMock(return_value=[p1, p2]) + # Wiping both projects + cmd.GetProjects = MagicMock(return_value=[p1, p2]) + + cmd.Execute( + cmd.OptionParser.parse_args(["project/one", "project/two"])[0], + ["project/one", "project/two"], + ) + + # Everything should be gone, including the shared directory + self.assertFalse(os.path.exists(p1.worktree)) + self.assertFalse(os.path.exists(p1.gitdir)) + self.assertFalse(os.path.exists(p2.worktree)) + self.assertFalse(os.path.exists(p2.gitdir)) + self.assertFalse(os.path.exists(shared_objdir)) + + def test_wipe_with_uncommitted_changes_raises_error(self): + """Test that wiping a project with uncommitted changes raises an error.""" + p1 = self.create_mock_project("project/one", has_changes=True) + + cmd = Wipe() + cmd.GetAllProjects = MagicMock(return_value=[p1]) + cmd.GetProjects = MagicMock(return_value=[p1]) + + with self.assertRaises(UsageError): + cmd.Execute( + cmd.OptionParser.parse_args(["project/one"])[0], ["project/one"] + ) + + # Nothing should have been deleted + self.assertTrue(os.path.exists(p1.worktree)) + self.assertTrue(os.path.exists(p1.gitdir)) + self.assertTrue(os.path.exists(p1.objdir)) + + def test_wipe_with_uncommitted_changes_with_force(self): + """Test wiping a project with uncommitted changes with --force.""" + p1 = self.create_mock_project("project/one", has_changes=True) + + cmd = Wipe() + cmd.GetAllProjects = MagicMock(return_value=[p1]) + cmd.GetProjects = MagicMock(return_value=[p1]) + + cmd.Execute( + cmd.OptionParser.parse_args(["--force", "project/one"])[0], + ["project/one"], + ) + + # All directories should be gone + self.assertFalse(os.path.exists(p1.worktree)) + self.assertFalse(os.path.exists(p1.gitdir)) + self.assertFalse(os.path.exists(p1.objdir))