lou
116 lines
4.4 KiB
Python
116 lines
4.4 KiB
Python
"""metadata 模块行为测试 (组装/校验/集成管线, 非 mock)
|
|
|
|
运行: cd /home/lou/文档/7z-encrypt && .venv/bin/python -m unittest discover -s tests -v
|
|
"""
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from metadata import MetadataError, build_init_json, sha256_of
|
|
|
|
PROJ = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
PY = os.path.join(PROJ, '.venv/bin/python')
|
|
|
|
|
|
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)
|
|
]
|
|
|
|
|
|
class TestBuildInitJson(unittest.TestCase):
|
|
def setUp(self):
|
|
self.enc = {"alg": "cobblestone-aes256gcm", "key_id": "k", "context": "7z-encrypt:v1"}
|
|
|
|
def test_valid(self):
|
|
out = build_init_json("a.mp4", 300, "b" * 64, self.enc, valid_chunks())
|
|
self.assertEqual(out["file_name"], "a.mp4")
|
|
self.assertEqual(out["file_size"], 300)
|
|
self.assertEqual(out["chunk_count"], 3)
|
|
self.assertEqual(out["total_sha256"], "b" * 64)
|
|
self.assertEqual(out["enc"], self.enc)
|
|
# chunks 剥离 filename
|
|
self.assertEqual(out["chunks"], [
|
|
{"index": 1, "size": 100, "sha256": "a" * 64},
|
|
{"index": 2, "size": 100, "sha256": "a" * 64},
|
|
{"index": 3, "size": 100, "sha256": "a" * 64},
|
|
])
|
|
|
|
def test_enc_missing(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 300, "b" * 64, {}, valid_chunks())
|
|
|
|
def test_enc_key_missing(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 300, "b" * 64, {"alg": "x"}, valid_chunks())
|
|
|
|
def test_chunks_empty(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 300, "b" * 64, self.enc, [])
|
|
|
|
def test_chunks_index_gap(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 300, "b" * 64, self.enc, valid_chunks(start=2))
|
|
|
|
def test_chunks_bad_sha256_len(self):
|
|
ch = valid_chunks()
|
|
ch[0]["sha256"] = "abc"
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 300, "b" * 64, self.enc, ch)
|
|
|
|
def test_bad_total_sha256(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 300, "short", self.enc, valid_chunks())
|
|
|
|
def test_file_size_zero(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("a.mp4", 0, "b" * 64, self.enc, valid_chunks())
|
|
|
|
def test_empty_file_name(self):
|
|
with self.assertRaises(MetadataError):
|
|
build_init_json("", 300, "b" * 64, self.enc, valid_chunks())
|
|
|
|
|
|
class TestIntegration(unittest.TestCase):
|
|
"""集成: 小文件 加密 -> 分卷 -> init json, 全部真实跑"""
|
|
|
|
def test_pipeline(self):
|
|
with tempfile.TemporaryDirectory(prefix='hermes-test-') as td:
|
|
data = os.urandom(3 << 20) # 3MB
|
|
src = os.path.join(td, 'sample.bin')
|
|
with open(src, 'wb') as f:
|
|
f.write(data)
|
|
r = subprocess.run(
|
|
[PY, os.path.join(PROJ, 'metadata.py'), src, '--chunk-size', '1'],
|
|
cwd=PROJ, capture_output=True, text=True, timeout=180,
|
|
)
|
|
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
|
|
init_path = src + '.init.json'
|
|
with open(init_path, encoding='utf-8') as f:
|
|
init = json.load(f)
|
|
self.assertEqual(init['file_name'], 'sample.bin')
|
|
self.assertEqual(init['file_size'], len(data))
|
|
enc_size = os.path.getsize(src + '.enc')
|
|
# 密文含 header+tag 开销, 卷数按密文实际大小算
|
|
import math
|
|
self.assertEqual(init['chunk_count'], math.ceil(enc_size / (1 << 20)))
|
|
self.assertEqual(len(init['chunks']), init['chunk_count'])
|
|
self.assertEqual(len(init['total_sha256']), 64)
|
|
# 密文总哈希 = init 里的 total_sha256
|
|
self.assertEqual(init['total_sha256'], sha256_of(src + '.enc'))
|
|
# 卷清单哈希与真实卷文件一致
|
|
for ch in init['chunks']:
|
|
part = os.path.join(td, os.path.basename(src) + '.enc.part%04d' % ch['index'])
|
|
self.assertEqual(ch['sha256'], sha256_of(part))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|