aqua/zlib_test.lua at master · semyon422/aqua · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
local zlib = require("zlib")

local test = {}

---@param t testing.T
function test.zlib(t)
	local test_string = ("test"):rep(100)

	-- Basic compress/uncompress
	t:eq(zlib.uncompress(zlib.compress(test_string), #test_string), test_string)

	-- Uncompress without size (uses inflate)
	t:eq(zlib.uncompress(zlib.compress(test_string)), test_string)

	-- Deflate/Inflate with chunks
	t:eq(zlib.inflate(zlib.deflate(test_string, nil, nil, 10), nil, 10), test_string)
	t:eq(zlib.inflate(zlib.deflate(test_string, 0, nil, 10), nil, 10), test_string)
	t:eq(zlib.inflate(zlib.deflate(test_string, 9, nil, 10), nil, 10), test_string)
	t:eq(zlib.inflate(zlib.deflate(test_string)), test_string)

	-- Raw deflate
	local raw = zlib.deflate_raw(test_string)
	t:eq(zlib.inflate_raw(raw), test_string)

	-- Gzip
	local gz = zlib.gzip(test_string)
	t:eq(zlib.gunzip(gz), test_string)

	-- Auto detection (zlib/gzip)
	t:eq(zlib.inflate(zlib.compress(test_string), zlib.AUTO_WBITS), test_string)
	t:eq(zlib.inflate(gz, zlib.AUTO_WBITS), test_string)
end

---@param t testing.T
function test.streaming_edge_case(t)
	-- This test uses 1-byte chunks to force frequent yielding.
	-- If there is a bug in apply_filter (missing the last chunk), this will fail.
	local test_string = ("abcdefghijklmnopqrstuvwxyz"):rep(10)
	local compressed = zlib.deflate(test_string, nil, nil, 1)
	local decompressed = zlib.inflate(compressed, nil, 1)
	t:eq(#decompressed, #test_string, "Decompressed length mismatch")
	t:eq(decompressed, test_string, "Decompression content mismatch")
end

---@param t testing.T
function test.data_loss_proof(t)
	-- Use random data to ensure no compression patterns hide the bug.
	-- If the last chunk is lost, the CRC will definitely mismatch.
	math.randomseed(42)

	---@type string[]
	local data = {}
	for i = 1, 100 do data[i] = string.char(math.random(0, 255)) end
	local test_string = table.concat(data)

	local compressed = zlib.deflate(test_string, nil, nil, 5)
	local decompressed = zlib.inflate(compressed, nil, 5)

	t:eq(#decompressed, #test_string, "Length mismatch - potential trailing data loss")
	t:eq(zlib.crc32(nil, decompressed), zlib.crc32(nil, test_string), "CRC mismatch - data was corrupted or truncated")
end

---@param t testing.T
function test.gzip_uncompress(t)
	local test_string = "hello gzip world"
	local gz = zlib.deflate(test_string, nil, nil, zlib.GZIP_WBITS)

	-- This should ideally work by auto-detecting gzip vs zlib
	t:eq(zlib.uncompress(gz), test_string, "uncompress should auto-detect gzip format")
end

---@param t testing.T
function test.stream_interface(t)
	local test_string = "streaming interface test"

	-- Deflate stream
	local ds = zlib.deflate_stream()
	local compressed = ds:write("streaming ")
	compressed = compressed .. ds:write("interface ")
	compressed = compressed .. ds:write("test")
	compressed = compressed .. ds:finish()

	-- Inflate stream
	local is = zlib.inflate_stream()
	local decompressed = is:write(compressed:sub(1, 10))
	decompressed = decompressed .. is:write(compressed:sub(11))
	decompressed = decompressed .. is:finish()

	t:eq(decompressed, test_string, "Stream interface failed")
end

---@param t testing.T
function test.checksums(t)
	local test_string = "hello world"

	-- Adler32
	t:eq(zlib.adler32(nil, test_string), 436929629)
	t:eq(zlib.adler32(zlib.adler32(nil, "hello "), "world"), 436929629)

	-- CRC32
	t:eq(zlib.crc32(nil, test_string), 222957957)
	t:eq(zlib.crc32(zlib.crc32(nil, "hello "), "world"), 222957957)
end

---@param t testing.T
function test.multiple_finish(t)
	local ds = zlib.deflate_stream()
	ds:write("test")
	local c = ds:finish()
	t:assert(#c > 0, "Finish should return data")
	t:eq(ds:finish(), "", "Subsequent finishes should return empty string")
	t:eq(ds:write("more"), "", "Writing to closed stream should return empty string")
end

---@param t testing.T
function test.error_reporting(t)
	local is = zlib.inflate_stream()
	t:has_error(function()
		is:write("invalid data")
		is:finish()
	end)
end

return test