Merge pull request #7303 from ThomasWaldmann/fix-pyi

mypy inspired fixes / updates

src/borg/compress.pyi

@@ -1,4 +1,4 @@
-from typing import Any, Type
+from typing import Any, Type, Dict, Tuple
 
 API_VERSION: str
 
@@ -12,8 +12,8 @@ class CompressionSpec:
 
 class Compressor:
     def __init__(self, name: Any = ..., **kwargs) -> None: ...
-    def compress(self, data: bytes) -> bytes: ...
-    def decompress(self, data: bytes) -> bytes: ...
+    def compress(self, meta: Dict, data: bytes) -> Tuple[Dict, bytes]: ...
+    def decompress(self, meta: Dict, data: bytes) -> Tuple[Dict, bytes]: ...
     @staticmethod
     def detect(data: bytes) -> Any: ...
 

src/borg/crypto/file_integrity.py

@@ -13,6 +13,9 @@ logger = create_logger()
 
 
 class FileLikeWrapper:
+    def __init__(self, fd):
+        self.fd = fd
+
     def __enter__(self):
         self.fd.__enter__()
         return self
@@ -59,7 +62,7 @@ class FileHashingWrapper(FileLikeWrapper):
     FACTORY: Callable = None
 
     def __init__(self, backing_fd, write):
-        self.fd = backing_fd
+        super().__init__(backing_fd)
         self.writing = write
         self.hash = self.FACTORY()
 
@@ -142,7 +145,8 @@ class IntegrityCheckedFile(FileLikeWrapper):
             # TODO: When we're reading but don't have any digests, i.e. no integrity file existed,
             # TODO: then we could just short-circuit.
 
-        self.fd = self.hasher = hash_cls(backing_fd=self.file_fd, write=write)
+        self.hasher = hash_cls(backing_fd=self.file_fd, write=write)
+        super().__init__(self.hasher)
         self.hash_filename(filename)
 
     def load_integrity_data(self, path, integrity_data):

src/borg/crypto/key.py

@@ -3,7 +3,7 @@ import os
 import textwrap
 from binascii import a2b_base64, b2a_base64, hexlify
 from hashlib import sha256, pbkdf2_hmac
-from typing import Literal, Callable
+from typing import Literal, Callable, ClassVar
 
 from ..logger import create_logger
 
@@ -170,7 +170,7 @@ class KeyBase:
     ARG_NAME = "UNDEFINED"
 
     # Storage type (no key blob storage / keyfile / repo)
-    STORAGE = KeyBlobStorage.NO_STORAGE
+    STORAGE: ClassVar[str] = KeyBlobStorage.NO_STORAGE
 
     # Seed for the buzhash chunker (borg.algorithms.chunker.Chunker)
     # type is int
@@ -279,7 +279,6 @@ class PlaintextKey(KeyBase):
     TYPES_ACCEPTABLE = {TYPE}
     NAME = "plaintext"
     ARG_NAME = "none"
-    STORAGE = KeyBlobStorage.NO_STORAGE
 
     chunk_seed = 0
     logically_encrypted = False
@@ -417,6 +416,7 @@ class AESKeyBase(KeyBase):
 
 class FlexiKey:
     FILE_ID = "BORG_KEY"
+    STORAGE: ClassVar[str] = KeyBlobStorage.NO_STORAGE  # override in subclass
 
     @classmethod
     def detect(cls, repository, manifest_data):

src/borg/hashindex.pyi

@@ -28,9 +28,8 @@ class IndexBase:
 class ChunkIndexEntry(NamedTuple):
     refcount: int
     size: int
-    csize: int
 
-CIE = Union[Tuple[int, int, int], Type[ChunkIndexEntry]]
+CIE = Union[Tuple[int, int], Type[ChunkIndexEntry]]
 
 class ChunkKeyIterator:
     def __init__(self, keysize: int) -> None: ...
@@ -38,7 +37,7 @@ class ChunkKeyIterator:
     def __next__(self) -> Tuple[bytes, Type[ChunkIndexEntry]]: ...
 
 class ChunkIndex(IndexBase):
-    def add(self, key: bytes, refs: int, size: int, csize: int) -> None: ...
+    def add(self, key: bytes, refs: int, size: int) -> None: ...
     def decref(self, key: bytes) -> CIE: ...
     def incref(self, key: bytes) -> CIE: ...
     def iteritems(self, marker: bytes = ...) -> Iterator: ...
@@ -65,12 +64,14 @@ class NSIndex(IndexBase):
     def __contains__(self, key: bytes) -> bool: ...
     def __getitem__(self, key: bytes) -> Any: ...
     def __setitem__(self, key: bytes, value: Any) -> None: ...
+    def flags(self, key: bytes, mask: int, value: int = None) -> int: ...
 
 class NSIndex1(IndexBase):  # legacy
     def iteritems(self, *args, **kwargs) -> Iterator: ...
     def __contains__(self, key: bytes) -> bool: ...
     def __getitem__(self, key: bytes) -> Any: ...
     def __setitem__(self, key: bytes, value: Any) -> None: ...
+    def flags(self, key: bytes, mask: int, value: int = None) -> int: ...
 
 class FuseVersionsIndex(IndexBase):
     def __contains__(self, key: bytes) -> bool: ...

src/borg/helpers/fs.py

@@ -230,7 +230,7 @@ class HardLinkManager:
                      chunks / chunks_healthy list
                      hlid
         """
-        assert isinstance(id, self.id_type), f"key is {key!r}, not of type {self.key_type}"
+        assert isinstance(id, self.id_type), f"id is {id!r}, not of type {self.id_type}"
         assert isinstance(info, self.info_type), f"info is {info!r}, not of type {self.info_type}"
         self._map[id] = info
 
@@ -238,7 +238,7 @@ class HardLinkManager:
         """
         retrieve stuff to use it in a (usually contentless) item.
         """
-        assert isinstance(id, self.id_type)
+        assert isinstance(id, self.id_type), f"id is {id!r}, not of type {self.id_type}"
         return self._map.get(id, default)
 
 

src/borg/item.pyi

@@ -4,7 +4,7 @@ from .helpers import StableDict
 
 API_VERSION: str
 
-def want_bytes(v: Any, *, errors: str) -> bytes: ...
+def want_bytes(v: Any, *, errors: str = ...) -> bytes: ...
 def chunks_contents_equal(chunks1: Iterator, chunks2: Iterator) -> bool: ...
 
 class PropDict:
@@ -106,7 +106,6 @@ class ArchiveItem(PropDict):
 class ChunkListEntry(NamedTuple):
     id: bytes
     size: int
-    csize: int
 
 class Item(PropDict):
     @property
@@ -178,6 +177,10 @@ class Item(PropDict):
     @deleted.setter
     def deleted(self, val: bool) -> None: ...
     @property
+    def hlid(self) -> bytes: ...
+    @hlid.setter
+    def hlid(self, val: bytes) -> None: ...
+    @property
     def hardlink_master(self) -> bool: ...
     @hardlink_master.setter
     def hardlink_master(self, val: bool) -> None: ...

src/borg/upgrade.py

@@ -41,6 +41,8 @@ class UpgraderNoOp:
 
 
 class UpgraderFrom12To20:
+    borg1_header_fmt = Struct(">I")
+
     def __init__(self, *, cache):
         self.cache = cache
 
@@ -126,10 +128,9 @@ class UpgraderFrom12To20:
 
         if ctype == ObfuscateSize.ID:
             # in older borg, we used unusual byte order
-            borg1_header_fmt = Struct(">I")
-            hlen = borg1_header_fmt.size
+            hlen = self.borg1_header_fmt.size
             csize_bytes = data[2 : 2 + hlen]
-            csize = borg1_header_fmt.unpack(csize_bytes)
+            csize = self.borg1_header_fmt.unpack(csize_bytes)[0]
             compressed = data[2 + hlen : 2 + hlen + csize]
             meta, compressed = upgrade_zlib_and_level(meta, compressed)
             meta["psize"] = csize