# test_fastexport.py -- Fast export/import functionality # Copyright (C) 2010 Jelmer Vernooij # # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # 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. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # import stat from io import BytesIO from dulwich.object_store import MemoryObjectStore from dulwich.objects import ZERO_SHA, Blob, Commit, Tree from dulwich.repo import MemoryRepo from dulwich.tests.utils import build_commit_graph from . import SkipTest, TestCase class GitFastExporterTests(TestCase): """Tests for the GitFastExporter tests.""" def setUp(self) -> None: super().setUp() self.store = MemoryObjectStore() self.stream = BytesIO() try: from dulwich.fastexport import GitFastExporter except ImportError as exc: raise SkipTest("python-fastimport not available") from exc self.fastexporter = GitFastExporter(self.stream, self.store) def test_emit_blob(self) -> None: b = Blob() b.data = b"fooBAR" self.fastexporter.emit_blob(b) self.assertEqual(b"blob\nmark :1\ndata 6\nfooBAR\n", self.stream.getvalue()) def test_emit_commit(self) -> None: b = Blob() b.data = b"FOO" t = Tree() t.add(b"foo", stat.S_IFREG | 0o644, b.id) c = Commit() c.committer = c.author = b"Jelmer " c.author_time = c.commit_time = 1271345553 c.author_timezone = c.commit_timezone = 0 c.message = b"msg" c.tree = t.id self.store.add_objects([(b, None), (t, None), (c, None)]) self.fastexporter.emit_commit(c, b"refs/heads/master") self.assertEqual( b"""blob mark :1 data 3 FOO commit refs/heads/master mark :2 author Jelmer 1271345553 +0000 committer Jelmer 1271345553 +0000 data 3 msg M 644 :1 foo """, self.stream.getvalue(), ) class GitImportProcessorTests(TestCase): """Tests for the GitImportProcessor tests.""" def setUp(self) -> None: super().setUp() self.repo = MemoryRepo() try: from dulwich.fastexport import GitImportProcessor except ImportError as exc: raise SkipTest("python-fastimport not available") from exc self.processor = GitImportProcessor(self.repo) def test_reset_handler(self) -> None: from fastimport import commands [c1] = build_commit_graph(self.repo.object_store, [[1]]) cmd = commands.ResetCommand(b"refs/heads/foo", c1.id) self.processor.reset_handler(cmd) self.assertEqual(c1.id, self.repo.get_refs()[b"refs/heads/foo"]) self.assertEqual(c1.id, self.processor.last_commit) def test_reset_handler_marker(self) -> None: from fastimport import commands [c1, c2] = build_commit_graph(self.repo.object_store, [[1], [2]]) self.processor.markers[b"10"] = c1.id cmd = commands.ResetCommand(b"refs/heads/foo", b":10") self.processor.reset_handler(cmd) self.assertEqual(c1.id, self.repo.get_refs()[b"refs/heads/foo"]) def test_reset_handler_default(self) -> None: from fastimport import commands [c1, c2] = build_commit_graph(self.repo.object_store, [[1], [2]]) cmd = commands.ResetCommand(b"refs/heads/foo", None) self.processor.reset_handler(cmd) self.assertEqual(ZERO_SHA, self.repo.get_refs()[b"refs/heads/foo"]) def test_commit_handler(self) -> None: from fastimport import commands cmd = commands.CommitCommand( b"refs/heads/foo", b"mrkr", (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), b"FOO", None, [], [], ) self.processor.commit_handler(cmd) commit = self.repo[self.processor.last_commit] self.assertEqual(b"Jelmer ", commit.author) self.assertEqual(b"Jelmer ", commit.committer) self.assertEqual(b"FOO", commit.message) self.assertEqual([], commit.parents) self.assertEqual(432432432.0, commit.commit_time) self.assertEqual(432432432.0, commit.author_time) self.assertEqual(3600, commit.commit_timezone) self.assertEqual(3600, commit.author_timezone) self.assertEqual(commit, self.repo[b"refs/heads/foo"]) def test_commit_handler_markers(self) -> None: from fastimport import commands [c1, c2, c3] = build_commit_graph(self.repo.object_store, [[1], [2], [3]]) self.processor.markers[b"10"] = c1.id self.processor.markers[b"42"] = c2.id self.processor.markers[b"98"] = c3.id cmd = commands.CommitCommand( b"refs/heads/foo", b"mrkr", (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), b"FOO", b":10", [b":42", b":98"], [], ) self.processor.commit_handler(cmd) commit = self.repo[self.processor.last_commit] self.assertEqual(c1.id, commit.parents[0]) self.assertEqual(c2.id, commit.parents[1]) self.assertEqual(c3.id, commit.parents[2]) def test_import_stream(self) -> None: markers = self.processor.import_stream( BytesIO( b"""blob mark :1 data 11 text for a commit refs/heads/master mark :2 committer Joe Foo 1288287382 +0000 data 20 M 100644 :1 a """ ) ) self.assertEqual(2, len(markers)) self.assertIsInstance(self.repo[markers[b"1"]], Blob) self.assertIsInstance(self.repo[markers[b"2"]], Commit) def test_file_add(self) -> None: from fastimport import commands cmd = commands.BlobCommand(b"23", b"data") self.processor.blob_handler(cmd) cmd = commands.CommitCommand( b"refs/heads/foo", b"mrkr", (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), b"FOO", None, [], [commands.FileModifyCommand(b"path", 0o100644, b":23", None)], ) self.processor.commit_handler(cmd) commit = self.repo[self.processor.last_commit] self.assertEqual( [(b"path", 0o100644, b"6320cd248dd8aeaab759d5871f8781b5c0505172")], self.repo[commit.tree].items(), ) def simple_commit(self): from fastimport import commands cmd = commands.BlobCommand(b"23", b"data") self.processor.blob_handler(cmd) cmd = commands.CommitCommand( b"refs/heads/foo", b"mrkr", (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), b"FOO", None, [], [commands.FileModifyCommand(b"path", 0o100644, b":23", None)], ) self.processor.commit_handler(cmd) commit = self.repo[self.processor.last_commit] return commit def make_file_commit(self, file_cmds): """Create a trivial commit with the specified file commands. Args: file_cmds: File commands to run. Returns: The created commit object """ from fastimport import commands cmd = commands.CommitCommand( b"refs/heads/foo", b"mrkr", (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), (b"Jelmer", b"jelmer@samba.org", 432432432.0, 3600), b"FOO", None, [], file_cmds, ) self.processor.commit_handler(cmd) return self.repo[self.processor.last_commit] def test_file_copy(self) -> None: from fastimport import commands self.simple_commit() commit = self.make_file_commit([commands.FileCopyCommand(b"path", b"new_path")]) self.assertEqual( [ ( b"new_path", 0o100644, b"6320cd248dd8aeaab759d5871f8781b5c0505172", ), ( b"path", 0o100644, b"6320cd248dd8aeaab759d5871f8781b5c0505172", ), ], self.repo[commit.tree].items(), ) def test_file_move(self) -> None: from fastimport import commands self.simple_commit() commit = self.make_file_commit( [commands.FileRenameCommand(b"path", b"new_path")] ) self.assertEqual( [ ( b"new_path", 0o100644, b"6320cd248dd8aeaab759d5871f8781b5c0505172", ), ], self.repo[commit.tree].items(), ) def test_file_delete(self) -> None: from fastimport import commands self.simple_commit() commit = self.make_file_commit([commands.FileDeleteCommand(b"path")]) self.assertEqual([], self.repo[commit.tree].items()) def test_file_deleteall(self) -> None: from fastimport import commands self.simple_commit() commit = self.make_file_commit([commands.FileDeleteAllCommand()]) self.assertEqual([], self.repo[commit.tree].items())