320 lines
11 KiB
Plaintext
320 lines
11 KiB
Plaintext
# main.vp — ViperZipfileProject：zipfile 模块读写测试
|
||
#
|
||
# 覆盖:
|
||
# 1) ZipFile("w") 写入 4 个成员（deflate 文本 / deflate 大文件 / stored 伪随机 / 空文件）
|
||
# 2) ZipFile("r") 读回: 成员数/名称/大小/compress_size 检查 + 逐成员内容比对
|
||
# （CRC32 校验由 zipfile.read 内部完成）
|
||
# 3) 损坏 stored 成员 payload 1 字节 → 再次 read 应返回 None 且错误码 ZIP_ERR_CRC
|
||
import t, c
|
||
from stdint import *
|
||
import stdio
|
||
import stdlib
|
||
import string
|
||
import zipfile
|
||
import w32.win32base as wb
|
||
import w32.win32file as wf
|
||
|
||
|
||
ZIP_PATH: str = "zip_roundtrip.zip"
|
||
BIG_LEN: t.CSizeT = 100000
|
||
RND_LEN: t.CSizeT = 5000
|
||
HELLO_TEXT: str = "Hello, Viper Zipfile!\n"
|
||
|
||
fail_count: t.CInt = 0
|
||
|
||
|
||
def chk(name: str, cond: t.CInt):
|
||
global fail_count
|
||
if cond:
|
||
stdio.printf("[PASS] %s\n", name)
|
||
else:
|
||
fail_count += 1
|
||
stdio.printf("[FAIL] %s\n", name)
|
||
stdio.fflush(None) # 崩溃时也保证已打印行可见（stdout 全缓冲）
|
||
|
||
|
||
def StrLen(s: bytes) -> t.CSizeT:
|
||
n: t.CSizeT = 0
|
||
while s is not None and c.Deref(s) != 0:
|
||
n += 1
|
||
s += 1
|
||
return n
|
||
|
||
|
||
def fill_pattern(buf: bytes, length: t.CSizeT, kind: t.CInt):
|
||
i: t.CSizeT = 0
|
||
while i < length:
|
||
if kind == 1:
|
||
# 'A'..'G' 循环 — 高度可压缩
|
||
buf[i] = 65 + (i % 7)
|
||
else:
|
||
# 伪随机 — 供 ZIP_STORED 成员
|
||
buf[i] = (i * 37 + 11) & 0xFF
|
||
i += 1
|
||
|
||
|
||
def WriteZip() -> t.CInt:
|
||
stdio.printf("\n--- 1) 写 zip（4 个成员） ---\n")
|
||
wf.DeleteFileA(ZIP_PATH)
|
||
w: zipfile.ZipFile | t.CPtr = zipfile.ZipFile(ZIP_PATH, "w", zipfile.ZIP_DEFLATED)
|
||
if w.ok == 0:
|
||
stdio.printf("open(w) failed: %s\n", zipfile.error())
|
||
return 1
|
||
|
||
hello: bytes = stdlib.malloc(64)
|
||
if not hello:
|
||
stdio.printf("malloc hello failed\n")
|
||
w.close()
|
||
return 1
|
||
string.strcpy(hello, HELLO_TEXT)
|
||
hello_len: t.CSizeT = StrLen(hello)
|
||
|
||
big: bytes = stdlib.malloc(BIG_LEN)
|
||
if not big:
|
||
stdio.printf("malloc big failed\n")
|
||
stdlib.free(hello)
|
||
w.close()
|
||
return 1
|
||
fill_pattern(big, BIG_LEN, 1)
|
||
|
||
rnd: bytes = stdlib.malloc(RND_LEN)
|
||
if not rnd:
|
||
stdio.printf("malloc rnd failed\n")
|
||
stdlib.free(hello)
|
||
stdlib.free(big)
|
||
w.close()
|
||
return 1
|
||
fill_pattern(rnd, RND_LEN, 0)
|
||
|
||
rc: t.CInt = w.write("hello.txt", hello, hello_len, zipfile.ZIP_DEFLATED)
|
||
chk("write hello.txt (deflate)", rc == zipfile.ZIP_ERR_OK)
|
||
if rc != zipfile.ZIP_ERR_OK:
|
||
stdio.printf(" hello err: %s\n", zipfile.error())
|
||
rc = w.write("big.bin", big, BIG_LEN, zipfile.ZIP_DEFLATED)
|
||
chk("write big.bin (deflate, 100000B)", rc == zipfile.ZIP_ERR_OK)
|
||
if rc != zipfile.ZIP_ERR_OK:
|
||
stdio.printf(" big err: %s\n", zipfile.error())
|
||
rc = w.write("random.bin", rnd, RND_LEN, zipfile.ZIP_STORED)
|
||
chk("write random.bin (stored)", rc == zipfile.ZIP_ERR_OK)
|
||
if rc != zipfile.ZIP_ERR_OK:
|
||
stdio.printf(" random err: %s\n", zipfile.error())
|
||
rc = w.write("empty.txt", None, 0, zipfile.ZIP_DEFLATED)
|
||
chk("write empty.txt (deflate, 0B)", rc == zipfile.ZIP_ERR_OK)
|
||
if rc != zipfile.ZIP_ERR_OK:
|
||
stdio.printf(" empty err: %s\n", zipfile.error())
|
||
rc = w.close()
|
||
chk("close(w)", rc == zipfile.ZIP_ERR_OK)
|
||
if rc != zipfile.ZIP_ERR_OK:
|
||
stdio.printf(" close err: %s\n", zipfile.error())
|
||
|
||
# close 后用裸文件句柄独立确认落盘大小（不依赖 zipfile 读逻辑）
|
||
ph: wb.HANDLE = wf.CreateFileA(
|
||
ZIP_PATH, wf.GENERIC_READ,
|
||
wf.FILE_SHARE_READ | wf.FILE_SHARE_WRITE | wf.FILE_SHARE_DELETE,
|
||
None, wf.OPEN_EXISTING, wf.FILE_ATTRIBUTE_NORMAL, None)
|
||
if ph == wb.INVALID_HANDLE_VALUE:
|
||
stdio.printf("probe open failed: err=%u\n", wb.GetLastError())
|
||
else:
|
||
pli: wb.LARGE_INTEGER = wb.LARGE_INTEGER()
|
||
if wf.GetFileSizeEx(ph, t.CPtr(c.Addr(pli))) != 0:
|
||
stdio.printf("zip file size on disk = %lld\n", pli.QuadPart)
|
||
wb.CloseHandle(ph)
|
||
|
||
stdlib.free(hello)
|
||
stdlib.free(big)
|
||
stdlib.free(rnd)
|
||
return 0
|
||
|
||
|
||
def CheckReadBack(buf: bytes, expect: bytes, length: t.CSizeT) -> t.CInt:
|
||
# buf: zipfile.read 返回的解压缓冲；expect: 期望内容
|
||
if buf is None:
|
||
return 0
|
||
if length == 0:
|
||
return 1
|
||
return t.CInt(string.memcmp(buf, expect, length) == 0)
|
||
|
||
|
||
def ReadZip() -> t.CInt:
|
||
stdio.printf("\n--- 2) 读回校验 ---\n")
|
||
r: zipfile.ZipFile | t.CPtr = zipfile.ZipFile(ZIP_PATH, "r")
|
||
if r.ok == 0:
|
||
stdio.printf("open(r) failed: %s\n", zipfile.error())
|
||
return 1
|
||
|
||
chk("entry count == 4", r.info_count == 4)
|
||
i: t.CSizeT = 0
|
||
while i < r.info_count:
|
||
stdio.printf(" [%d] %s\n", t.CInt(i), r.namelist_at(i))
|
||
i += 1
|
||
|
||
# ---- hello.txt：deflate 文本 ----
|
||
info: zipfile.ZipInfo | t.CPtr = r.getinfo("hello.txt")
|
||
chk("getinfo(hello.txt)", info is not None)
|
||
hello_expect: bytes = stdlib.malloc(64)
|
||
string.strcpy(hello_expect, HELLO_TEXT)
|
||
exp_len: t.CSizeT = StrLen(hello_expect)
|
||
if info is not None:
|
||
chk("hello file_size == 22", info.file_size == exp_len)
|
||
chk("hello compress_size > 0", info.compress_size > 0)
|
||
olen: t.CSizeT = 0
|
||
data: bytes = r.read("hello.txt", c.Addr(olen))
|
||
chk("read(hello.txt)", data is not None)
|
||
chk("hello content match", CheckReadBack(data, hello_expect, olen) == 1)
|
||
stdlib.free(data)
|
||
stdlib.free(hello_expect)
|
||
|
||
# ---- big.bin：deflate 大文件 ----
|
||
info = r.getinfo("big.bin")
|
||
chk("getinfo(big.bin)", info is not None)
|
||
big_expect: bytes = stdlib.malloc(BIG_LEN)
|
||
fill_pattern(big_expect, BIG_LEN, 1)
|
||
if info is not None:
|
||
chk("big file_size == 100000", info.file_size == BIG_LEN)
|
||
chk("big compress_size < file_size", info.compress_size < info.file_size)
|
||
chk("big crc != 0", info.crc32 != 0)
|
||
olen = 0
|
||
data = r.read("big.bin", c.Addr(olen))
|
||
chk("read(big.bin)", data is not None)
|
||
chk("big content match", data is not None and olen == BIG_LEN
|
||
and CheckReadBack(data, big_expect, BIG_LEN) == 1)
|
||
stdlib.free(data)
|
||
stdlib.free(big_expect)
|
||
|
||
# ---- random.bin：stored ----
|
||
info = r.getinfo("random.bin")
|
||
chk("getinfo(random.bin)", info is not None)
|
||
rnd_expect: bytes = stdlib.malloc(RND_LEN)
|
||
fill_pattern(rnd_expect, RND_LEN, 0)
|
||
if info is not None:
|
||
chk("random file_size == 5000", info.file_size == RND_LEN)
|
||
chk("random stored: compress_size == file_size", info.compress_size == info.file_size)
|
||
olen = 0
|
||
data = r.read("random.bin", c.Addr(olen))
|
||
chk("read(random.bin)", data is not None)
|
||
chk("random content match", data is not None and olen == RND_LEN
|
||
and CheckReadBack(data, rnd_expect, RND_LEN) == 1)
|
||
stdlib.free(data)
|
||
stdlib.free(rnd_expect)
|
||
|
||
# ---- empty.txt：空成员 ----
|
||
info = r.getinfo("empty.txt")
|
||
chk("getinfo(empty.txt)", info is not None)
|
||
olen = 0
|
||
data = r.read("empty.txt", c.Addr(olen))
|
||
chk("read(empty.txt) ok", data is not None)
|
||
chk("empty length == 0", olen == 0)
|
||
stdlib.free(data)
|
||
|
||
# ---- 不存在成员 ----
|
||
olen = 0
|
||
data = r.read("no_such.txt", c.Addr(olen))
|
||
chk("read(missing) returns None", data is None)
|
||
chk("read(missing) code NOT_FOUND", zipfile.error_code() == zipfile.ZIP_ERR_NOT_FOUND)
|
||
|
||
rc: t.CInt = r.close()
|
||
chk("close(r)", rc == zipfile.ZIP_ERR_OK)
|
||
return 0
|
||
|
||
|
||
def CorruptStoredCheck() -> t.CInt:
|
||
stdio.printf("\n--- 3) 损坏 stored 成员后 CRC 校验 ---\n")
|
||
r: zipfile.ZipFile | t.CPtr = zipfile.ZipFile(ZIP_PATH, "r")
|
||
if r.ok == 0:
|
||
stdio.printf("open(r) failed: %s\n", zipfile.error())
|
||
return 1
|
||
info: zipfile.ZipInfo | t.CPtr = r.getinfo("random.bin")
|
||
chk("getinfo(random.bin)", info is not None)
|
||
if info is None or info.compress_size < 2:
|
||
stdio.printf("cannot locate stored member payload\n")
|
||
r.close()
|
||
return 1
|
||
|
||
# 本地头无 extra 时，数据起点 = header_offset + 30 + 名字长
|
||
start: t.CSizeT = t.CSizeT(info.header_offset) + 30 + info.name_len
|
||
flip_pos: t.CSizeT = start + info.compress_size - 1
|
||
|
||
# 方案: 整文件读入内存 → 翻转 stored 数据尾字节 → 整体重写（复用已验证的读写循环）
|
||
hr: wb.HANDLE = wf.CreateFileA(
|
||
ZIP_PATH, wf.GENERIC_READ,
|
||
wf.FILE_SHARE_READ | wf.FILE_SHARE_WRITE | wf.FILE_SHARE_DELETE,
|
||
None, wf.OPEN_EXISTING, wf.FILE_ATTRIBUTE_NORMAL, None)
|
||
chk("open R", hr != wb.INVALID_HANDLE_VALUE)
|
||
if hr != wb.INVALID_HANDLE_VALUE:
|
||
fsz_li: wb.LARGE_INTEGER = wb.LARGE_INTEGER()
|
||
if wf.GetFileSizeEx(hr, t.CPtr(c.Addr(fsz_li))) == 0 or fsz_li.QuadPart <= 0:
|
||
chk("get file size", 0)
|
||
wb.CloseHandle(hr)
|
||
r.close()
|
||
return 1
|
||
fsz: t.CSizeT = t.CSizeT(fsz_li.QuadPart)
|
||
whole: BYTEPTR = BYTEPTR(stdlib.malloc(fsz))
|
||
if not whole:
|
||
wb.CloseHandle(hr)
|
||
r.close()
|
||
return 1
|
||
# 读整个文件
|
||
done: t.CSizeT = 0
|
||
rd_ok: t.CInt = 1
|
||
while done < fsz:
|
||
rdgot: ULONG = 0
|
||
if wf.ReadFile(hr, whole + done, ULONG(fsz - done), c.Addr(rdgot), None) == 0 or rdgot == 0:
|
||
rd_ok = 0
|
||
break
|
||
done += t.CSizeT(rdgot)
|
||
wb.CloseHandle(hr)
|
||
chk("read whole file", rd_ok == 1)
|
||
if rd_ok == 1 and flip_pos < fsz:
|
||
stdio.printf(" flip stored data byte at %lld (size=%lld)\n", flip_pos, fsz_li.QuadPart)
|
||
whole[flip_pos] = (whole[flip_pos] ^ 0xFF) & 0xFF
|
||
# 重写整个文件
|
||
hw: wb.HANDLE = wf.CreateFileA(
|
||
ZIP_PATH, wf.GENERIC_WRITE,
|
||
wf.FILE_SHARE_READ | wf.FILE_SHARE_WRITE | wf.FILE_SHARE_DELETE,
|
||
None, wf.CREATE_ALWAYS, wf.FILE_ATTRIBUTE_NORMAL, None)
|
||
chk("open W rewrite", hw != wb.INVALID_HANDLE_VALUE)
|
||
if hw != wb.INVALID_HANDLE_VALUE:
|
||
wr_ok: t.CInt = 1
|
||
wdone: t.CSizeT = 0
|
||
while wdone < fsz:
|
||
wrgot: ULONG = 0
|
||
if wf.WriteFile(hw, whole + wdone, ULONG(fsz - wdone), c.Addr(wrgot), None) == 0 or wrgot == 0:
|
||
wr_ok = 0
|
||
break
|
||
wdone += t.CSizeT(wrgot)
|
||
wb.CloseHandle(hw)
|
||
chk("rewrite file with flipped byte", wr_ok == 1)
|
||
else:
|
||
chk("flip position in bounds", 0)
|
||
stdlib.free(whole)
|
||
r.close()
|
||
|
||
z2: zipfile.ZipFile | t.CPtr = zipfile.ZipFile(ZIP_PATH, "r")
|
||
chk("reopen after corruption", z2.ok == 1)
|
||
olen: t.CSizeT = 0
|
||
data: bytes = z2.read("random.bin", c.Addr(olen))
|
||
chk("corrupted read returns None", data is None)
|
||
chk("error code == ZIP_ERR_CRC", zipfile.error_code() == zipfile.ZIP_ERR_CRC)
|
||
stdio.printf(" last error: %s\n", zipfile.error())
|
||
if data is not None:
|
||
stdlib.free(data)
|
||
z2.close()
|
||
return 0
|
||
|
||
|
||
@t.CExport
|
||
def main() -> int:
|
||
stdio.printf("ViperZipfileProject: zipfile 模块读写测试\n")
|
||
if WriteZip() != 0:
|
||
return 1
|
||
if ReadZip() != 0:
|
||
return 1
|
||
if CorruptStoredCheck() != 0:
|
||
return 1
|
||
|
||
wf.DeleteFileA(ZIP_PATH)
|
||
stdio.printf("\n失败数: %d\n", fail_count)
|
||
if fail_count == 0:
|
||
stdio.printf("全部通过\n")
|
||
return t.CInt(fail_count != 0)
|