Fix typing

dulwich/attrs.py

@@ -364,6 +364,7 @@ class GitAttributes:
         else:
             # Update the existing pattern in the list
             assert pattern_index >= 0
+            assert attrs_dict is not None
             self._patterns[pattern_index] = (pattern_obj, attrs_dict)
 
         # Update the attribute

dulwich/lfs.py

@@ -25,7 +25,7 @@ import os
 import tempfile
 from collections.abc import Iterable
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, BinaryIO, Optional
+from typing import TYPE_CHECKING, BinaryIO, Optional, Union
 from urllib.error import HTTPError
 from urllib.parse import urljoin
 from urllib.request import Request, urlopen
@@ -122,7 +122,12 @@ class LFSStore:
         path = self._sha_path(sha.hexdigest())
         if not os.path.exists(os.path.dirname(path)):
             os.makedirs(os.path.dirname(path))
-        os.rename(tmppath, path)
+
+        # Handle concurrent writes - if file already exists, just remove temp file
+        if os.path.exists(path):
+            os.remove(tmppath)
+        else:
+            os.rename(tmppath, path)
         return sha.hexdigest()
 
 
@@ -282,7 +287,7 @@ class LFSClient:
     def batch(
         self,
         operation: str,
-        objects: list[dict[str, str | int]],
+        objects: list[dict[str, Union[str, int]]],
         ref: Optional[str] = None,
     ) -> LFSBatchResponse:
         """Perform batch operation to get transfer URLs.
@@ -296,7 +301,7 @@ class LFSClient:
             Batch response from server
         """
         data: dict[
-            str, str | list[str] | list[dict[str, str | int]] | dict[str, str]
+            str, Union[str, list[str], list[dict[str, Union[str, int]]], dict[str, str]]
         ] = {
             "operation": operation,
             "transfers": ["basic"],

dulwich/lfs_server.py

@@ -185,8 +185,10 @@ class LFSRequestHandler(BaseHTTPRequestHandler):
             self.send_error(400, f"OID mismatch: expected {oid}, got {calculated_oid}")
             return
 
-        # Store the object
-        self.server.lfs_store.write_object(chunks)
+        # Check if object already exists
+        if not self._object_exists(oid):
+            # Store the object only if it doesn't exist
+            self.server.lfs_store.write_object(chunks)
 
         self.send_response(200)
         self.end_headers()

dulwich/porcelain.py

@@ -140,7 +140,6 @@ from .objectspec import (
 from .pack import write_pack_from_container, write_pack_index
 from .patch import (
     get_summary,
-    write_blob_diff,
     write_commit_patch,
     write_object_diff,
     write_tree_diff,
@@ -3179,6 +3178,9 @@ def checkout(
             new_branch = new_branch.encode(DEFAULT_ENCODING)
 
         # Parse the target to get the commit
+        assert (
+            original_target is not None
+        )  # Guaranteed by earlier check for normal checkout
         target_commit = parse_commit(r, original_target)
         target_tree_id = target_commit.tree
 

tests/compat/test_lfs.py

@@ -22,7 +22,6 @@
 """Compatibility tests for LFS functionality between dulwich and git-lfs."""
 
 import os
-import shutil
 import subprocess
 import tempfile
 from unittest import skipUnless
@@ -31,7 +30,7 @@ from dulwich import porcelain
 from dulwich.lfs import LFSPointer
 from dulwich.porcelain import lfs_clean, lfs_init, lfs_smudge, lfs_track
 
-from .utils import CompatTestCase, run_git_or_fail
+from .utils import CompatTestCase, rmtree_ro, run_git_or_fail
 
 
 def git_lfs_version():
@@ -63,7 +62,7 @@ class LFSCompatTestCase(CompatTestCase):
     def make_temp_dir(self):
         """Create a temporary directory that will be cleaned up."""
         temp_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, temp_dir)
+        self.addCleanup(rmtree_ro, temp_dir)
         return temp_dir
 
 
@@ -79,18 +78,20 @@ class LFSInitCompatTest(LFSCompatTestCase):
 
         # Verify with git-lfs
         output = run_git_or_fail(["lfs", "env"], cwd=repo_dir)
-        self.assertIn("git config filter.lfs.clean", output)
-        self.assertIn("git config filter.lfs.smudge", output)
+        self.assertIn(b"git config filter.lfs.clean", output)
+        self.assertIn(b"git config filter.lfs.smudge", output)
 
     def test_lfs_init_git(self):
         """Test that git-lfs init is compatible with dulwich."""
         # Initialize with git-lfs
         repo_dir = self.make_temp_dir()
         run_git_or_fail(["init"], cwd=repo_dir)
-        run_git_or_fail(["lfs", "install"], cwd=repo_dir)
+        run_git_or_fail(["lfs", "install", "--local"], cwd=repo_dir)
 
         # Verify with dulwich
-        config = porcelain.get_config_stack(repo_dir)
+        repo = porcelain.open_repo(repo_dir)
+        self.addCleanup(repo.close)
+        config = repo.get_config_stack()
         self.assertEqual(
             config.get(("filter", "lfs"), "clean").decode(), "git-lfs clean -- %f"
         )
@@ -113,14 +114,14 @@ class LFSTrackCompatTest(LFSCompatTestCase):
 
         # Verify with git-lfs
         output = run_git_or_fail(["lfs", "track"], cwd=repo_dir)
-        self.assertIn("*.bin", output)
-        self.assertIn("*.dat", output)
+        self.assertIn(b"*.bin", output)
+        self.assertIn(b"*.dat", output)
 
     def test_track_git(self):
         """Test that git-lfs track is compatible with dulwich."""
         repo_dir = self.make_temp_dir()
         run_git_or_fail(["init"], cwd=repo_dir)
-        run_git_or_fail(["lfs", "install"], cwd=repo_dir)
+        run_git_or_fail(["lfs", "install", "--local"], cwd=repo_dir)
 
         # Track with git-lfs
         run_git_or_fail(["lfs", "track", "*.bin"], cwd=repo_dir)
@@ -156,19 +157,19 @@ class LFSFileOperationsCompatTest(LFSCompatTestCase):
 
         # Verify with git-lfs
         output = run_git_or_fail(["lfs", "ls-files"], cwd=repo_dir)
-        self.assertIn("test.bin", output)
+        self.assertIn(b"test.bin", output)
 
         # Check pointer file in git
         output = run_git_or_fail(["show", "HEAD:test.bin"], cwd=repo_dir)
-        self.assertIn("version https://git-lfs.github.com/spec/v1", output)
-        self.assertIn("oid sha256:", output)
-        self.assertIn("size 1048576", output)
+        self.assertIn(b"version https://git-lfs.github.com/spec/v1", output)
+        self.assertIn(b"oid sha256:", output)
+        self.assertIn(b"size 1048576", output)
 
     def test_add_commit_git(self):
         """Test adding and committing LFS files with git-lfs."""
         repo_dir = self.make_temp_dir()
         run_git_or_fail(["init"], cwd=repo_dir)
-        run_git_or_fail(["lfs", "install"], cwd=repo_dir)
+        run_git_or_fail(["lfs", "install", "--local"], cwd=repo_dir)
         run_git_or_fail(["lfs", "track", "*.bin"], cwd=repo_dir)
         run_git_or_fail(["add", ".gitattributes"], cwd=repo_dir)
         run_git_or_fail(["commit", "-m", "Track .bin files"], cwd=repo_dir)
@@ -185,9 +186,10 @@ class LFSFileOperationsCompatTest(LFSCompatTestCase):
 
         # Verify with dulwich
         repo = porcelain.open_repo(repo_dir)
+        self.addCleanup(repo.close)
         tree = repo[repo.head()].tree
-        entry = repo.object_store[tree][b"test.bin"]
-        blob = repo.object_store[entry.binsha]
+        mode, sha = repo.object_store[tree][b"test.bin"]
+        blob = repo.object_store[sha]
         pointer = LFSPointer.from_bytes(blob.data)
         self.assertEqual(pointer.size, 1048576)
 
@@ -196,7 +198,7 @@ class LFSFileOperationsCompatTest(LFSCompatTestCase):
         # Create repo with git-lfs
         repo_dir = self.make_temp_dir()
         run_git_or_fail(["init"], cwd=repo_dir)
-        run_git_or_fail(["lfs", "install"], cwd=repo_dir)
+        run_git_or_fail(["lfs", "install", "--local"], cwd=repo_dir)
         run_git_or_fail(["lfs", "track", "*.bin"], cwd=repo_dir)
         run_git_or_fail(["add", ".gitattributes"], cwd=repo_dir)
         run_git_or_fail(["commit", "-m", "Track .bin files"], cwd=repo_dir)
@@ -227,10 +229,16 @@ class LFSPointerCompatTest(LFSCompatTestCase):
     def test_pointer_format_dulwich(self):
         """Test that dulwich creates git-lfs compatible pointers."""
         repo_dir = self.make_temp_dir()
+        run_git_or_fail(["init"], cwd=repo_dir)
+        lfs_init(repo_dir)
+
         test_content = b"test content for LFS"
+        test_file = os.path.join(repo_dir, "test.txt")
+        with open(test_file, "wb") as f:
+            f.write(test_content)
 
         # Create pointer with dulwich
-        pointer_data = lfs_clean(test_content)
+        pointer_data = lfs_clean(repo_dir, "test.txt")
 
         # Parse with git-lfs (create a file and check)
         test_file = os.path.join(repo_dir, "test_pointer")
@@ -264,10 +272,17 @@ class LFSFilterCompatTest(LFSCompatTestCase):
 
     def test_clean_filter_compat(self):
         """Test clean filter compatibility between dulwich and git-lfs."""
+        repo_dir = self.make_temp_dir()
+        run_git_or_fail(["init"], cwd=repo_dir)
+        lfs_init(repo_dir)
+
         test_content = b"x" * 1000
+        test_file = os.path.join(repo_dir, "test.txt")
+        with open(test_file, "wb") as f:
+            f.write(test_content)
 
         # Clean with dulwich
-        dulwich_pointer = lfs_clean(test_content)
+        dulwich_pointer = lfs_clean(repo_dir, "test.txt")
 
         # Clean with git-lfs (simulate)
         # Since we can't easily invoke git-lfs clean directly,
@@ -285,7 +300,11 @@ class LFSFilterCompatTest(LFSCompatTestCase):
 
         # Create test content
         test_content = b"test data for smudge filter"
-        pointer_data = lfs_clean(test_content)
+        test_file = os.path.join(repo_dir, "test.txt")
+        with open(test_file, "wb") as f:
+            f.write(test_content)
+
+        pointer_data = lfs_clean(repo_dir, "test.txt")
 
         # Store object in LFS
         lfs_dir = os.path.join(repo_dir, ".git", "lfs")
@@ -318,7 +337,7 @@ class LFSCloneCompatTest(LFSCompatTestCase):
         # Create source repo with LFS
         source_dir = self.make_temp_dir()
         run_git_or_fail(["init"], cwd=source_dir)
-        run_git_or_fail(["lfs", "install"], cwd=source_dir)
+        run_git_or_fail(["lfs", "install", "--local"], cwd=source_dir)
         run_git_or_fail(["lfs", "track", "*.bin"], cwd=source_dir)
         run_git_or_fail(["add", ".gitattributes"], cwd=source_dir)
         run_git_or_fail(["commit", "-m", "Track .bin files"], cwd=source_dir)

tests/test_porcelain_filters.py

@@ -22,13 +22,14 @@
 """Tests for porcelain filter integration."""
 
 import os
-import shutil
 import tempfile
+from io import BytesIO
 
 from dulwich import porcelain
 from dulwich.repo import Repo
 
 from . import TestCase
+from .compat.utils import rmtree_ro
 
 
 class PorcelainFilterTests(TestCase):
@@ -37,8 +38,9 @@ class PorcelainFilterTests(TestCase):
     def setUp(self) -> None:
         super().setUp()
         self.test_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, self.test_dir)
+        self.addCleanup(rmtree_ro, self.test_dir)
         self.repo = Repo.init(self.test_dir)
+        self.addCleanup(self.repo.close)
 
     def test_add_with_autocrlf(self) -> None:
         """Test adding files with autocrlf enabled."""
@@ -132,7 +134,9 @@ class PorcelainFilterTests(TestCase):
             f.write(b"line1\r\nmodified\r\nline3\r\n")
 
         # Get diff - should normalize line endings for comparison
-        diff_output = b"".join(porcelain.diff(self.repo))
+        outstream = BytesIO()
+        porcelain.diff(self.repo, outstream=outstream)
+        diff_output = outstream.getvalue()
         self.assertIn(b"-line2", diff_output)
         self.assertIn(b"+modified", diff_output)
         self.assertIn(b"+line3", diff_output)
@@ -176,8 +180,9 @@ class PorcelainFilterTests(TestCase):
         """Test cloning a repository with filters."""
         # Create a source repository
         source_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, source_dir)
+        self.addCleanup(rmtree_ro, source_dir)
         source_repo = Repo.init(source_dir)
+        self.addCleanup(source_repo.close)
 
         # Add a file with LF endings
         test_file = os.path.join(source_dir, "test.txt")
@@ -189,10 +194,11 @@ class PorcelainFilterTests(TestCase):
 
         # Clone the repository without checkout
         target_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, target_dir)
+        self.addCleanup(rmtree_ro, target_dir)
 
         # Clone without checkout first
         target_repo = porcelain.clone(source_dir, target_dir, checkout=False)
+        self.addCleanup(target_repo.close)
 
         # Configure autocrlf in target repo
         target_config = target_repo.get_config()
@@ -274,8 +280,9 @@ class PorcelainLFSIntegrationTests(TestCase):
     def setUp(self) -> None:
         super().setUp()
         self.test_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, self.test_dir)
+        self.addCleanup(rmtree_ro, self.test_dir)
         self.repo = Repo.init(self.test_dir)
+        self.addCleanup(self.repo.close)
 
         # Set up LFS
         lfs_dir = os.path.join(self.test_dir, ".git", "lfs")
@@ -353,8 +360,9 @@ class FilterEdgeCaseTests(TestCase):
     def setUp(self) -> None:
         super().setUp()
         self.test_dir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, self.test_dir)
+        self.addCleanup(rmtree_ro, self.test_dir)
         self.repo = Repo.init(self.test_dir)
+        self.addCleanup(self.repo.close)
 
     def test_mixed_line_endings(self) -> None:
         """Test handling files with mixed line endings."""