diff --git a/subcmds/wipe.py b/subcmds/wipe.py new file mode 100644 index 000000000..86ad6d3e1 --- /dev/null +++ b/subcmds/wipe.py @@ -0,0 +1,97 @@ +# 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 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): + if not args: + raise UsageError("no projects specified") + + # We need all projects to correctly handle shared object directories. + all_projects = self.GetAllProjects(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) + + objdirs_to_delete = set() + for project in projects_to_wipe: + users = objdir_map.get(project.objdir, {project.name}) + is_shared_with_other_projects = not users.issubset(names_to_wipe) + + if is_shared_with_other_projects and not opt.force: + # Find an example project it's shared with for a helpful error. + other_user = list(users - names_to_wipe)[0] + raise UsageError( + f"project '{project.name}' shares object directory with " + f"'{other_user}' (and possibly others). Use --force to wipe." + ) + + self._WipeProject(project) + + if not is_shared_with_other_projects: + objdirs_to_delete.add(project.objdir) + + for objdir in objdirs_to_delete: + if os.path.exists(objdir): + print(f"Deleting objects directory: {objdir}") + shutil.rmtree(objdir) + + def _WipeProject(self, project): + """Wipes a single project's worktree and git directory.""" + if not project: + return + + # Delete the worktree. + if project.worktree and os.path.exists(project.worktree): + print(f"Deleting worktree: {project.worktree}") + shutil.rmtree(project.worktree) + + # Delete the git directory. + 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..cb2891291 --- /dev/null +++ b/tests/test_subcmds_wipe.py @@ -0,0 +1,180 @@ +# 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): + """Creates a mock project with necessary attributes and directories.""" + proj = MagicMock() + proj.name = name + proj.relpath = name + # 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") + + 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") + p2 = self.create_mock_project("project/two") + + 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) + p2 = self.create_mock_project("project/two", objdir_path=shared_objdir) + + 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) + p2 = self.create_mock_project("project/two", objdir_path=shared_objdir) + + 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) + p2 = self.create_mock_project("project/two", objdir_path=shared_objdir) + + 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))