test_rebase.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755
  1. # test_rebase.py -- rebase tests
  2. # Copyright (C) 2025 Dulwich contributors
  3. #
  4. # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
  5. # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
  6. # General Public License as published by the Free Software Foundation; version 2.0
  7. # or (at your option) any later version. You can redistribute it and/or
  8. # modify it under the terms of either of these two licenses.
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. #
  16. # You should have received a copy of the licenses; if not, see
  17. # <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
  18. # and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
  19. # License, Version 2.0.
  20. #
  21. """Tests for dulwich.rebase."""
  22. import os
  23. import tempfile
  24. from dulwich.objects import Blob, Commit, Tree
  25. from dulwich.rebase import (
  26. RebaseConflict,
  27. Rebaser,
  28. RebaseTodo,
  29. RebaseTodoCommand,
  30. RebaseTodoEntry,
  31. process_interactive_rebase,
  32. rebase,
  33. start_interactive,
  34. )
  35. from dulwich.repo import MemoryRepo, Repo
  36. from dulwich.tests.utils import make_commit
  37. from . import TestCase
  38. class RebaserTestCase(TestCase):
  39. """Tests for the Rebaser class."""
  40. def setUp(self):
  41. """Set up test repository."""
  42. super().setUp()
  43. self.repo = MemoryRepo()
  44. def _setup_initial_commit(self):
  45. """Set up initial commit for tests."""
  46. # Create initial commit
  47. blob = Blob.from_string(b"Initial content\n")
  48. self.repo.object_store.add_object(blob)
  49. tree = Tree()
  50. tree.add(b"file.txt", 0o100644, blob.id)
  51. self.repo.object_store.add_object(tree)
  52. self.initial_commit = make_commit(
  53. tree=tree.id,
  54. parents=[],
  55. message=b"Initial commit",
  56. committer=b"Test User <test@example.com>",
  57. author=b"Test User <test@example.com>",
  58. commit_time=1000000,
  59. author_time=1000000,
  60. commit_timezone=0,
  61. author_timezone=0,
  62. )
  63. self.repo.object_store.add_object(self.initial_commit)
  64. # Set up branches
  65. self.repo.refs[b"refs/heads/master"] = self.initial_commit.id
  66. self.repo.refs.set_symbolic_ref(b"HEAD", b"refs/heads/master")
  67. def test_simple_rebase(self):
  68. """Test simple rebase with no conflicts."""
  69. self._setup_initial_commit()
  70. # Create feature branch with one commit
  71. feature_blob = Blob.from_string(b"Feature content\n")
  72. self.repo.object_store.add_object(feature_blob)
  73. feature_tree = Tree()
  74. feature_tree.add(b"feature.txt", 0o100644, feature_blob.id)
  75. feature_tree.add(
  76. b"file.txt", 0o100644, self.repo[self.initial_commit.tree][b"file.txt"][1]
  77. )
  78. self.repo.object_store.add_object(feature_tree)
  79. feature_commit = Commit()
  80. feature_commit.tree = feature_tree.id
  81. feature_commit.parents = [self.initial_commit.id]
  82. feature_commit.message = b"Add feature"
  83. feature_commit.committer = b"Test User <test@example.com>"
  84. feature_commit.author = b"Test User <test@example.com>"
  85. feature_commit.commit_time = 1000100
  86. feature_commit.author_time = 1000100
  87. feature_commit.commit_timezone = 0
  88. feature_commit.author_timezone = 0
  89. self.repo.object_store.add_object(feature_commit)
  90. self.repo.refs[b"refs/heads/feature"] = feature_commit.id
  91. # Create main branch advancement
  92. main_blob = Blob.from_string(b"Main advancement\n")
  93. self.repo.object_store.add_object(main_blob)
  94. main_tree = Tree()
  95. main_tree.add(b"main.txt", 0o100644, main_blob.id)
  96. main_tree.add(
  97. b"file.txt", 0o100644, self.repo[self.initial_commit.tree][b"file.txt"][1]
  98. )
  99. self.repo.object_store.add_object(main_tree)
  100. main_commit = Commit()
  101. main_commit.tree = main_tree.id
  102. main_commit.parents = [self.initial_commit.id]
  103. main_commit.message = b"Main advancement"
  104. main_commit.committer = b"Test User <test@example.com>"
  105. main_commit.author = b"Test User <test@example.com>"
  106. main_commit.commit_time = 1000200
  107. main_commit.author_time = 1000200
  108. main_commit.commit_timezone = 0
  109. main_commit.author_timezone = 0
  110. self.repo.object_store.add_object(main_commit)
  111. self.repo.refs[b"refs/heads/master"] = main_commit.id
  112. # Switch to feature branch
  113. self.repo.refs.set_symbolic_ref(b"HEAD", b"refs/heads/feature")
  114. # Check refs before rebase
  115. main_ref_before = self.repo.refs[b"refs/heads/master"]
  116. feature_ref_before = self.repo.refs[b"refs/heads/feature"]
  117. # Double check that refs are correctly set up
  118. self.assertEqual(main_ref_before, main_commit.id)
  119. self.assertEqual(feature_ref_before, feature_commit.id)
  120. # Perform rebase
  121. rebaser = Rebaser(self.repo)
  122. commits = rebaser.start(b"refs/heads/master", branch=b"refs/heads/feature")
  123. self.assertEqual(len(commits), 1)
  124. self.assertEqual(commits[0].id, feature_commit.id)
  125. # Continue rebase
  126. result = rebaser.continue_()
  127. self.assertIsNone(result) # Rebase complete
  128. # Check that feature branch was updated
  129. new_feature_head = self.repo.refs[b"refs/heads/feature"]
  130. new_commit = self.repo[new_feature_head]
  131. # Should have main commit as parent
  132. self.assertEqual(new_commit.parents, [main_commit.id])
  133. # Should have same tree as original (both files present)
  134. new_tree = self.repo[new_commit.tree]
  135. self.assertIn(b"feature.txt", new_tree)
  136. self.assertIn(b"main.txt", new_tree)
  137. self.assertIn(b"file.txt", new_tree)
  138. def test_rebase_with_conflicts(self):
  139. """Test rebase with merge conflicts."""
  140. self._setup_initial_commit()
  141. # Create feature branch with conflicting change
  142. feature_blob = Blob.from_string(b"Feature change to file\n")
  143. self.repo.object_store.add_object(feature_blob)
  144. feature_tree = Tree()
  145. feature_tree.add(b"file.txt", 0o100644, feature_blob.id)
  146. self.repo.object_store.add_object(feature_tree)
  147. feature_commit = Commit()
  148. feature_commit.tree = feature_tree.id
  149. feature_commit.parents = [self.initial_commit.id]
  150. feature_commit.message = b"Feature change"
  151. feature_commit.committer = b"Test User <test@example.com>"
  152. feature_commit.author = b"Test User <test@example.com>"
  153. feature_commit.commit_time = 1000100
  154. feature_commit.author_time = 1000100
  155. feature_commit.commit_timezone = 0
  156. feature_commit.author_timezone = 0
  157. self.repo.object_store.add_object(feature_commit)
  158. self.repo.refs[b"refs/heads/feature"] = feature_commit.id
  159. # Create main branch with conflicting change
  160. main_blob = Blob.from_string(b"Main change to file\n")
  161. self.repo.object_store.add_object(main_blob)
  162. main_tree = Tree()
  163. main_tree.add(b"file.txt", 0o100644, main_blob.id)
  164. self.repo.object_store.add_object(main_tree)
  165. main_commit = Commit()
  166. main_commit.tree = main_tree.id
  167. main_commit.parents = [self.initial_commit.id]
  168. main_commit.message = b"Main change"
  169. main_commit.committer = b"Test User <test@example.com>"
  170. main_commit.author = b"Test User <test@example.com>"
  171. main_commit.commit_time = 1000200
  172. main_commit.author_time = 1000200
  173. main_commit.commit_timezone = 0
  174. main_commit.author_timezone = 0
  175. self.repo.object_store.add_object(main_commit)
  176. self.repo.refs[b"refs/heads/master"] = main_commit.id
  177. # Attempt rebase - should fail with conflicts
  178. with self.assertRaises(RebaseConflict) as cm:
  179. rebase(self.repo, b"refs/heads/master", branch=b"refs/heads/feature")
  180. self.assertIn(b"file.txt", cm.exception.conflicted_files)
  181. def test_abort_rebase(self):
  182. """Test aborting a rebase."""
  183. self._setup_initial_commit()
  184. # Set up branches similar to simple rebase test
  185. feature_blob = Blob.from_string(b"Feature content\n")
  186. self.repo.object_store.add_object(feature_blob)
  187. feature_tree = Tree()
  188. feature_tree.add(b"feature.txt", 0o100644, feature_blob.id)
  189. feature_tree.add(
  190. b"file.txt", 0o100644, self.repo[self.initial_commit.tree][b"file.txt"][1]
  191. )
  192. self.repo.object_store.add_object(feature_tree)
  193. feature_commit = Commit()
  194. feature_commit.tree = feature_tree.id
  195. feature_commit.parents = [self.initial_commit.id]
  196. feature_commit.message = b"Add feature"
  197. feature_commit.committer = b"Test User <test@example.com>"
  198. feature_commit.author = b"Test User <test@example.com>"
  199. feature_commit.commit_time = 1000100
  200. feature_commit.author_time = 1000100
  201. feature_commit.commit_timezone = 0
  202. feature_commit.author_timezone = 0
  203. self.repo.object_store.add_object(feature_commit)
  204. self.repo.refs[b"refs/heads/feature"] = feature_commit.id
  205. self.repo.refs.set_symbolic_ref(b"HEAD", b"refs/heads/feature")
  206. # Start rebase
  207. rebaser = Rebaser(self.repo)
  208. rebaser.start(b"refs/heads/master")
  209. # Abort rebase
  210. rebaser.abort()
  211. # Check that HEAD is restored
  212. self.assertEqual(self.repo.refs.read_ref(b"HEAD"), b"ref: refs/heads/feature")
  213. self.assertEqual(self.repo.refs[b"refs/heads/feature"], feature_commit.id)
  214. # Check that REBASE_HEAD is cleaned up
  215. self.assertNotIn(b"REBASE_HEAD", self.repo.refs)
  216. def test_rebase_no_commits(self):
  217. """Test rebase when already up to date."""
  218. self._setup_initial_commit()
  219. # Both branches point to same commit
  220. self.repo.refs[b"refs/heads/feature"] = self.initial_commit.id
  221. self.repo.refs.set_symbolic_ref(b"HEAD", b"refs/heads/feature")
  222. # Perform rebase
  223. result = rebase(self.repo, b"refs/heads/master")
  224. # Should return empty list (no new commits)
  225. self.assertEqual(result, [])
  226. def test_rebase_onto(self):
  227. """Test rebase with --onto option."""
  228. self._setup_initial_commit()
  229. # Create a chain of commits: initial -> A -> B -> C
  230. blob_a = Blob.from_string(b"Commit A\n")
  231. self.repo.object_store.add_object(blob_a)
  232. tree_a = Tree()
  233. tree_a.add(b"a.txt", 0o100644, blob_a.id)
  234. tree_a.add(
  235. b"file.txt", 0o100644, self.repo[self.initial_commit.tree][b"file.txt"][1]
  236. )
  237. self.repo.object_store.add_object(tree_a)
  238. commit_a = make_commit(
  239. id=b"a" * 40,
  240. tree=tree_a.id,
  241. parents=[self.initial_commit.id],
  242. message=b"Commit A",
  243. committer=b"Test User <test@example.com>",
  244. author=b"Test User <test@example.com>",
  245. commit_time=1000100,
  246. author_time=1000100,
  247. )
  248. self.repo.object_store.add_object(commit_a)
  249. blob_b = Blob.from_string(b"Commit B\n")
  250. self.repo.object_store.add_object(blob_b)
  251. tree_b = Tree()
  252. tree_b.add(b"b.txt", 0o100644, blob_b.id)
  253. tree_b.add(b"a.txt", 0o100644, blob_a.id)
  254. tree_b.add(
  255. b"file.txt", 0o100644, self.repo[self.initial_commit.tree][b"file.txt"][1]
  256. )
  257. self.repo.object_store.add_object(tree_b)
  258. commit_b = make_commit(
  259. id=b"b" * 40,
  260. tree=tree_b.id,
  261. parents=[commit_a.id],
  262. message=b"Commit B",
  263. committer=b"Test User <test@example.com>",
  264. author=b"Test User <test@example.com>",
  265. commit_time=1000200,
  266. author_time=1000200,
  267. )
  268. self.repo.object_store.add_object(commit_b)
  269. blob_c = Blob.from_string(b"Commit C\n")
  270. self.repo.object_store.add_object(blob_c)
  271. tree_c = Tree()
  272. tree_c.add(b"c.txt", 0o100644, blob_c.id)
  273. tree_c.add(b"b.txt", 0o100644, blob_b.id)
  274. tree_c.add(b"a.txt", 0o100644, blob_a.id)
  275. tree_c.add(
  276. b"file.txt", 0o100644, self.repo[self.initial_commit.tree][b"file.txt"][1]
  277. )
  278. self.repo.object_store.add_object(tree_c)
  279. commit_c = make_commit(
  280. id=b"c" * 40,
  281. tree=tree_c.id,
  282. parents=[commit_b.id],
  283. message=b"Commit C",
  284. committer=b"Test User <test@example.com>",
  285. author=b"Test User <test@example.com>",
  286. commit_time=1000300,
  287. author_time=1000300,
  288. )
  289. self.repo.object_store.add_object(commit_c)
  290. # Create separate branch at commit A
  291. self.repo.refs[b"refs/heads/topic"] = commit_c.id
  292. self.repo.refs[b"refs/heads/newbase"] = commit_a.id
  293. self.repo.refs.set_symbolic_ref(b"HEAD", b"refs/heads/topic")
  294. # Rebase B and C onto initial commit (skipping A)
  295. rebaser = Rebaser(self.repo)
  296. commits = rebaser.start(
  297. upstream=commit_a.id,
  298. onto=self.initial_commit.id,
  299. branch=b"refs/heads/topic",
  300. )
  301. # Should rebase commits B and C
  302. self.assertEqual(len(commits), 2)
  303. self.assertEqual(commits[0].id, commit_b.id)
  304. self.assertEqual(commits[1].id, commit_c.id)
  305. # Continue rebase
  306. result = rebaser.continue_()
  307. self.assertIsNone(result)
  308. # Check result
  309. new_head = self.repo.refs[b"refs/heads/topic"]
  310. new_c = self.repo[new_head]
  311. new_b = self.repo[new_c.parents[0]]
  312. # B should now have initial commit as parent (not A)
  313. self.assertEqual(new_b.parents, [self.initial_commit.id])
  314. # Trees should still have b.txt and c.txt but not a.txt
  315. new_b_tree = self.repo[new_b.tree]
  316. self.assertIn(b"b.txt", new_b_tree)
  317. self.assertNotIn(b"a.txt", new_b_tree)
  318. new_c_tree = self.repo[new_c.tree]
  319. self.assertIn(b"c.txt", new_c_tree)
  320. self.assertIn(b"b.txt", new_c_tree)
  321. self.assertNotIn(b"a.txt", new_c_tree)
  322. class RebasePorcelainTestCase(TestCase):
  323. """Tests for the porcelain rebase function."""
  324. def setUp(self):
  325. """Set up test repository."""
  326. super().setUp()
  327. self.test_dir = tempfile.mkdtemp()
  328. self.repo = Repo.init(self.test_dir)
  329. # Create initial commit
  330. with open(os.path.join(self.test_dir, "README.md"), "wb") as f:
  331. f.write(b"# Test Repository\n")
  332. self.repo.get_worktree().stage(["README.md"])
  333. self.initial_commit = self.repo.get_worktree().commit(
  334. message=b"Initial commit",
  335. committer=b"Test User <test@example.com>",
  336. author=b"Test User <test@example.com>",
  337. )
  338. def tearDown(self):
  339. """Clean up test directory."""
  340. import shutil
  341. shutil.rmtree(self.test_dir)
  342. def test_porcelain_rebase(self):
  343. """Test rebase through porcelain interface."""
  344. from dulwich import porcelain
  345. # Create and checkout feature branch
  346. self.repo.refs[b"refs/heads/feature"] = self.initial_commit
  347. porcelain.checkout(self.repo, "feature")
  348. # Add commit to feature branch
  349. with open(os.path.join(self.test_dir, "feature.txt"), "wb") as f:
  350. f.write(b"Feature file\n")
  351. porcelain.add(self.repo, ["feature.txt"])
  352. porcelain.commit(
  353. self.repo,
  354. message="Add feature",
  355. author="Test User <test@example.com>",
  356. committer="Test User <test@example.com>",
  357. )
  358. # Switch to main and add different commit
  359. porcelain.checkout(self.repo, "master")
  360. with open(os.path.join(self.test_dir, "main.txt"), "wb") as f:
  361. f.write(b"Main file\n")
  362. porcelain.add(self.repo, ["main.txt"])
  363. porcelain.commit(
  364. self.repo,
  365. message="Main update",
  366. author="Test User <test@example.com>",
  367. committer="Test User <test@example.com>",
  368. )
  369. # Switch back to feature and rebase
  370. porcelain.checkout(self.repo, "feature")
  371. # Perform rebase
  372. new_shas = porcelain.rebase(self.repo, "master")
  373. # Should have rebased one commit
  374. self.assertEqual(len(new_shas), 1)
  375. # Check that the rebased commit has the correct parent and tree
  376. feature_head = self.repo.refs[b"refs/heads/feature"]
  377. feature_commit_obj = self.repo[feature_head]
  378. # Should have master as parent
  379. master_head = self.repo.refs[b"refs/heads/master"]
  380. self.assertEqual(feature_commit_obj.parents, [master_head])
  381. # Tree should have both files
  382. tree = self.repo[feature_commit_obj.tree]
  383. self.assertIn(b"feature.txt", tree)
  384. self.assertIn(b"main.txt", tree)
  385. self.assertIn(b"README.md", tree)
  386. class InteractiveRebaseTestCase(TestCase):
  387. """Tests for interactive rebase functionality."""
  388. def setUp(self):
  389. """Set up test repository."""
  390. super().setUp()
  391. self.repo = MemoryRepo()
  392. self._setup_initial_commit()
  393. def _setup_initial_commit(self):
  394. """Set up initial commit for tests."""
  395. # Create initial commit
  396. blob = Blob.from_string(b"Initial content\n")
  397. self.repo.object_store.add_object(blob)
  398. tree = Tree()
  399. tree.add(b"file.txt", 0o100644, blob.id)
  400. self.repo.object_store.add_object(tree)
  401. self.initial_commit = make_commit(
  402. tree=tree.id,
  403. parents=[],
  404. message=b"Initial commit",
  405. committer=b"Test User <test@example.com>",
  406. author=b"Test User <test@example.com>",
  407. commit_time=1000000,
  408. author_time=1000000,
  409. commit_timezone=0,
  410. author_timezone=0,
  411. )
  412. self.repo.object_store.add_object(self.initial_commit)
  413. # Set up branches
  414. self.repo.refs[b"refs/heads/master"] = self.initial_commit.id
  415. self.repo.refs.set_symbolic_ref(b"HEAD", b"refs/heads/master")
  416. def _create_test_commits(self):
  417. """Create a series of test commits for interactive rebase."""
  418. commits = []
  419. parent = self.initial_commit.id
  420. for i in range(3):
  421. blob = Blob.from_string(f"Content {i}\n".encode())
  422. self.repo.object_store.add_object(blob)
  423. tree = Tree()
  424. tree.add(f"file{i}.txt".encode(), 0o100644, blob.id)
  425. self.repo.object_store.add_object(tree)
  426. commit = Commit()
  427. commit.tree = tree.id
  428. commit.parents = [parent]
  429. commit.message = f"Commit {i}".encode()
  430. commit.committer = b"Test User <test@example.com>"
  431. commit.author = b"Test User <test@example.com>"
  432. commit.commit_time = 1000000 + i * 100
  433. commit.author_time = 1000000 + i * 100
  434. commit.commit_timezone = 0
  435. commit.author_timezone = 0
  436. self.repo.object_store.add_object(commit)
  437. commits.append(commit)
  438. parent = commit.id
  439. self.repo.refs[b"refs/heads/feature"] = commits[-1].id
  440. return commits
  441. def test_todo_parsing(self):
  442. """Test parsing of todo file format."""
  443. todo_content = """pick 1234567 First commit
  444. reword 2345678 Second commit
  445. edit 3456789 Third commit
  446. squash 4567890 Fourth commit
  447. fixup 5678901 Fifth commit
  448. drop 6789012 Sixth commit
  449. exec echo "Running test"
  450. break
  451. # This is a comment
  452. """
  453. todo = RebaseTodo.from_string(todo_content)
  454. self.assertEqual(len(todo.entries), 8)
  455. # Check first entry
  456. self.assertEqual(todo.entries[0].command, RebaseTodoCommand.PICK)
  457. self.assertEqual(todo.entries[0].commit_sha, b"1234567")
  458. self.assertEqual(todo.entries[0].short_message, "First commit")
  459. # Check reword
  460. self.assertEqual(todo.entries[1].command, RebaseTodoCommand.REWORD)
  461. # Check exec
  462. self.assertEqual(todo.entries[6].command, RebaseTodoCommand.EXEC)
  463. self.assertEqual(todo.entries[6].arguments, 'echo "Running test"')
  464. # Check break
  465. self.assertEqual(todo.entries[7].command, RebaseTodoCommand.BREAK)
  466. def test_todo_generation(self):
  467. """Test generation of todo list from commits."""
  468. commits = self._create_test_commits()
  469. todo = RebaseTodo.from_commits(commits)
  470. # Should have one pick entry per commit
  471. self.assertEqual(len(todo.entries), 3)
  472. for i, entry in enumerate(todo.entries):
  473. self.assertEqual(entry.command, RebaseTodoCommand.PICK)
  474. # commit_sha stores the full hex SHA as bytes
  475. self.assertEqual(entry.commit_sha, commits[i].id)
  476. self.assertIn(f"Commit {i}", entry.short_message)
  477. def test_todo_serialization(self):
  478. """Test serialization of todo list."""
  479. entries = [
  480. RebaseTodoEntry(
  481. command=RebaseTodoCommand.PICK,
  482. commit_sha=b"1234567890abcdef",
  483. short_message="First commit",
  484. ),
  485. RebaseTodoEntry(
  486. command=RebaseTodoCommand.SQUASH,
  487. commit_sha=b"fedcba0987654321",
  488. short_message="Second commit",
  489. ),
  490. RebaseTodoEntry(command=RebaseTodoCommand.EXEC, arguments="make test"),
  491. ]
  492. todo = RebaseTodo(entries)
  493. content = todo.to_string(include_comments=False)
  494. lines = content.strip().split("\n")
  495. self.assertEqual(len(lines), 3)
  496. self.assertIn("pick 1234567", lines[0])
  497. self.assertIn("squash fedcba0", lines[1])
  498. self.assertIn("exec make test", lines[2])
  499. def test_start_interactive_no_editor(self):
  500. """Test starting interactive rebase without editor."""
  501. self._create_test_commits()
  502. # Start interactive rebase
  503. todo = start_interactive(
  504. self.repo,
  505. b"refs/heads/master",
  506. branch=b"refs/heads/feature",
  507. editor_callback=None,
  508. )
  509. # Should have generated todo list
  510. self.assertEqual(len(todo.entries), 3)
  511. for entry in todo.entries:
  512. self.assertEqual(entry.command, RebaseTodoCommand.PICK)
  513. def test_start_interactive_with_editor(self):
  514. """Test starting interactive rebase with editor callback."""
  515. self._create_test_commits()
  516. def mock_editor(content):
  517. # Simulate user changing pick to squash for second commit
  518. lines = content.decode().splitlines()
  519. new_lines = []
  520. for i, line in enumerate(lines):
  521. if i == 1 and line.startswith("pick"):
  522. new_lines.append(line.replace("pick", "squash"))
  523. else:
  524. new_lines.append(line)
  525. return "\n".join(new_lines).encode()
  526. todo = start_interactive(
  527. self.repo,
  528. b"refs/heads/master",
  529. branch=b"refs/heads/feature",
  530. editor_callback=mock_editor,
  531. )
  532. # Second entry should be squash
  533. self.assertEqual(todo.entries[0].command, RebaseTodoCommand.PICK)
  534. self.assertEqual(todo.entries[1].command, RebaseTodoCommand.SQUASH)
  535. self.assertEqual(todo.entries[2].command, RebaseTodoCommand.PICK)
  536. def test_process_drop_command(self):
  537. """Test processing DROP command in interactive rebase."""
  538. commits = self._create_test_commits()
  539. # Create todo with drop command
  540. entries = [
  541. RebaseTodoEntry(
  542. command=RebaseTodoCommand.PICK,
  543. commit_sha=commits[0].id,
  544. short_message="Commit 0",
  545. ),
  546. RebaseTodoEntry(
  547. command=RebaseTodoCommand.DROP,
  548. commit_sha=commits[1].id,
  549. short_message="Commit 1",
  550. ),
  551. RebaseTodoEntry(
  552. command=RebaseTodoCommand.PICK,
  553. commit_sha=commits[2].id,
  554. short_message="Commit 2",
  555. ),
  556. ]
  557. todo = RebaseTodo(entries)
  558. is_complete, pause_reason = process_interactive_rebase(self.repo, todo)
  559. # Should complete successfully
  560. self.assertTrue(is_complete)
  561. self.assertIsNone(pause_reason)
  562. # Should have only picked 2 commits (dropped one)
  563. # Note: _done list would contain the rebased commits
  564. def test_process_break_command(self):
  565. """Test processing BREAK command in interactive rebase."""
  566. commits = self._create_test_commits()
  567. entries = [
  568. RebaseTodoEntry(
  569. command=RebaseTodoCommand.PICK,
  570. commit_sha=commits[0].id,
  571. short_message="Commit 0",
  572. ),
  573. RebaseTodoEntry(command=RebaseTodoCommand.BREAK),
  574. RebaseTodoEntry(
  575. command=RebaseTodoCommand.PICK,
  576. commit_sha=commits[1].id,
  577. short_message="Commit 1",
  578. ),
  579. ]
  580. todo = RebaseTodo(entries)
  581. is_complete, pause_reason = process_interactive_rebase(self.repo, todo)
  582. # Should pause at break
  583. self.assertFalse(is_complete)
  584. self.assertEqual(pause_reason, "break")
  585. # Todo should be at position after break
  586. self.assertEqual(todo.current_index, 2)
  587. def test_process_edit_command(self):
  588. """Test processing EDIT command in interactive rebase."""
  589. commits = self._create_test_commits()
  590. entries = [
  591. RebaseTodoEntry(
  592. command=RebaseTodoCommand.PICK,
  593. commit_sha=commits[0].id,
  594. short_message="Commit 0",
  595. ),
  596. RebaseTodoEntry(
  597. command=RebaseTodoCommand.EDIT,
  598. commit_sha=commits[1].id,
  599. short_message="Commit 1",
  600. ),
  601. ]
  602. todo = RebaseTodo(entries)
  603. is_complete, pause_reason = process_interactive_rebase(self.repo, todo)
  604. # Should pause for editing
  605. self.assertFalse(is_complete)
  606. self.assertEqual(pause_reason, "edit")
  607. def test_abbreviations(self):
  608. """Test parsing abbreviated commands."""
  609. todo_content = """p 1234567 Pick
  610. r 2345678 Reword
  611. e 3456789 Edit
  612. s 4567890 Squash
  613. f 5678901 Fixup
  614. d 6789012 Drop
  615. x echo test
  616. b
  617. """
  618. todo = RebaseTodo.from_string(todo_content)
  619. self.assertEqual(todo.entries[0].command, RebaseTodoCommand.PICK)
  620. self.assertEqual(todo.entries[1].command, RebaseTodoCommand.REWORD)
  621. self.assertEqual(todo.entries[2].command, RebaseTodoCommand.EDIT)
  622. self.assertEqual(todo.entries[3].command, RebaseTodoCommand.SQUASH)
  623. self.assertEqual(todo.entries[4].command, RebaseTodoCommand.FIXUP)
  624. self.assertEqual(todo.entries[5].command, RebaseTodoCommand.DROP)
  625. self.assertEqual(todo.entries[6].command, RebaseTodoCommand.EXEC)
  626. self.assertEqual(todo.entries[7].command, RebaseTodoCommand.BREAK)