260 lines
8.4 KiB
Plaintext
260 lines
8.4 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.fileio as fio
|
||
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)
|
||
|
||
|
||
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)
|
||
rc = w.write("big.bin", big, BIG_LEN, zipfile.ZIP_DEFLATED)
|
||
chk("write big.bin (deflate, 100000B)", rc == zipfile.ZIP_ERR_OK)
|
||
rc = w.write("random.bin", rnd, RND_LEN, zipfile.ZIP_STORED)
|
||
chk("write random.bin (stored)", rc == zipfile.ZIP_ERR_OK)
|
||
rc = w.write("empty.txt", None, 0, zipfile.ZIP_DEFLATED)
|
||
chk("write empty.txt (deflate, 0B)", rc == zipfile.ZIP_ERR_OK)
|
||
rc = w.close()
|
||
chk("close(w)", rc == zipfile.ZIP_ERR_OK)
|
||
|
||
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
|
||
stdio.printf(" flip byte at offset %lld (raw)\n", flip_pos)
|
||
|
||
f: fio.File | t.CPtr = fio.File(ZIP_PATH, fio.MODE.RP)
|
||
chk("open RP", f.closed == 0)
|
||
if f.closed == 0:
|
||
byte_buf: bytes = stdlib.malloc(1)
|
||
if not byte_buf:
|
||
r.close()
|
||
return 1
|
||
f.seek(LONG(flip_pos), fio.SEEK_SET)
|
||
rd: LONG = f.read(byte_buf, 1)
|
||
chk("read back byte", rd == 1)
|
||
f.seek(LONG(flip_pos), fio.SEEK_SET)
|
||
byte_buf[0] = (byte_buf[0] ^ 0xFF) & 0xFF
|
||
wr: LONG = f.write(byte_buf, 1)
|
||
chk("flip byte written", wr == 1)
|
||
f.close()
|
||
stdlib.free(byte_buf)
|
||
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)
|