补全所有函数参数类型注解 (splitter._read_block + tests helpers)
This commit is contained in:
+1
-1
@@ -18,7 +18,7 @@ import sys
|
||||
from typing import IO, Any
|
||||
|
||||
|
||||
def _read_block(stream, size: int) -> bytes:
|
||||
def _read_block(stream: IO[bytes], size: int) -> bytes:
|
||||
"""循环读取直到攒满 size 字节或 EOF。
|
||||
管道/stdin 不保证一次 read 返回满 size, 必须循环攒块, 否则卷大小不固定。
|
||||
"""
|
||||
|
||||
@@ -46,7 +46,7 @@ class TestRoundTrip(unittest.TestCase):
|
||||
self.eng = CryptoEngine(os.path.join(self._td.name, 'keyring.json'))
|
||||
self.eng.generate_key('k')
|
||||
|
||||
def _roundtrip(self, data):
|
||||
def _roundtrip(self, data: bytes):
|
||||
enc = io.BytesIO()
|
||||
params = self.eng.encrypt_stream(io.BytesIO(data), enc, 'k')
|
||||
enc.seek(0)
|
||||
|
||||
@@ -18,7 +18,7 @@ PROJ = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
PY = os.path.join(PROJ, '.venv/bin/python')
|
||||
|
||||
|
||||
def valid_chunks(n=3, start=1):
|
||||
def valid_chunks(n: int = 3, start: int = 1):
|
||||
return [
|
||||
{"index": i, "size": 100, "sha256": "a" * 64, "filename": f"p{i}.part{i:04d}"}
|
||||
for i in range(start, start + n)
|
||||
|
||||
@@ -15,7 +15,7 @@ PY = os.path.join(PROJ, '.venv/bin/python')
|
||||
SCRIPT = os.path.join(PROJ, 'splitter.py')
|
||||
|
||||
|
||||
def sha256_file(path):
|
||||
def sha256_file(path: str):
|
||||
h = hashlib.sha256()
|
||||
with open(path, 'rb') as f:
|
||||
for chunk in iter(lambda: f.read(1 << 16), b''):
|
||||
@@ -28,7 +28,7 @@ class TestSplitterCli(unittest.TestCase):
|
||||
self._td = tempfile.TemporaryDirectory(prefix='hermes-test-')
|
||||
self.addCleanup(self._td.cleanup)
|
||||
|
||||
def _run(self, args, stdin_data=None):
|
||||
def _run(self, args: list[str], stdin_data: str | None = None):
|
||||
return subprocess.run(
|
||||
[PY, SCRIPT] + args, cwd=PROJ, capture_output=True, text=True,
|
||||
timeout=120, input=stdin_data,
|
||||
|
||||
Reference in New Issue
Block a user