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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
"""
The browser's ZIP writer, read back by Python's zipfile.
A directory download is assembled in the browser: the node has no idea an
archive is being made, so nothing on this side would notice if the bytes were
malformed. These tests run the real module under Node, hand the output to
zipfile, and check that what comes out is what went in — including the CRCs,
which is the field a streaming writer is most likely to get wrong, since it is
written after the data it describes.
"""
import io
import json
import shutil
import subprocess
import zipfile
from pathlib import Path
import pytest
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
ZIPSTREAM = STATIC / "zipstream.js"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not ZIPSTREAM.exists(),
reason="node or the SPA sources are not available")
def _build(files, force_zip64=False, tmp_path=None):
"""Run zipstream.js over `files` ({name: bytes}) and return the archive."""
# Copied with an .mjs suffix: the browser loads this file with
# <script type="module">, but Node reads a bare .js as CommonJS unless a
# package.json says otherwise, and there is none next to the SPA.
module = tmp_path / "zipstream.mjs"
module.write_text(ZIPSTREAM.read_text())
script = tmp_path / "build.mjs"
out = tmp_path / "out.zip"
script.write_text(f"""
import {{ writeFileSync }} from 'node:fs';
import {{ ZipStream }} from '{module.as_posix()}';
const files = {json.dumps({k: list(v) for k, v in files.items()})};
const parts = [];
const zip = new ZipStream(b => {{ parts.push(Buffer.from(b)); }});
for (const [name, bytes] of Object.entries(files)) {{
const data = Uint8Array.from(bytes);
await zip.begin(name, data.length, new Date('2026-08-15T12:34:56Z'),
{{ forceZip64: {str(force_zip64).lower()} }});
// Written in pieces on purpose: a running CRC that is only correct for a
// single call would pass a friendlier test than this one.
for (let i = 0; i < data.length; i += 7) {{
await zip.write(data.subarray(i, i + 7));
}}
await zip.end();
}}
await zip.finish();
writeFileSync('{out.as_posix()}', Buffer.concat(parts));
""")
proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
assert proc.returncode == 0, proc.stderr
return out.read_bytes()
def test_a_plain_archive_reads_back(tmp_path):
files = {
"readme.txt": b"the quick brown fox\n",
"clip.mp4": bytes(range(256)) * 40,
"notes/deep.txt": b"",
}
data = _build(files, tmp_path=tmp_path)
with zipfile.ZipFile(io.BytesIO(data)) as z:
assert z.testzip() is None, "zipfile found a bad CRC"
assert sorted(z.namelist()) == sorted(files)
for name, content in files.items():
assert z.read(name) == content, f"{name} came back different"
def test_nothing_is_compressed(tmp_path):
"""
Store-only is a decision, not an accident: the payloads are already
compressed, and deflate would burn CPU in the thread doing the decryption.
"""
data = _build({"a.bin": b"x" * 5000}, tmp_path=tmp_path)
with zipfile.ZipFile(io.BytesIO(data)) as z:
info = z.getinfo("a.bin")
assert info.compress_type == zipfile.ZIP_STORED
assert info.compress_size == info.file_size == 5000
def test_sizes_and_crcs_come_after_the_data(tmp_path):
"""
A stream cannot go back and patch a header, so the general purpose bit that
says "look for a data descriptor" has to be set — and the descriptor has to
be there. zipfile reads the central directory, so this checks the flag
explicitly rather than trusting a successful read.
"""
data = _build({"a.bin": b"payload"}, tmp_path=tmp_path)
with zipfile.ZipFile(io.BytesIO(data)) as z:
assert z.getinfo("a.bin").flag_bits & 0x08, "data descriptor bit is not set"
assert z.getinfo("a.bin").flag_bits & 0x800, "names must be marked UTF-8"
assert data.count((0x08074b50).to_bytes(4, "little")) >= 1
def test_a_name_that_is_not_ascii_survives(tmp_path):
files = {"vidéos/été à la mer.txt": "déjà vu\n".encode()}
data = _build(files, tmp_path=tmp_path)
with zipfile.ZipFile(io.BytesIO(data)) as z:
assert z.namelist() == ["vidéos/été à la mer.txt"]
assert z.read("vidéos/été à la mer.txt") == "déjà vu\n".encode()
def test_the_zip64_records_are_right(tmp_path):
"""
Forced rather than fed 4 GiB: the branch is what needs testing, and the
archive it produces has to be readable by something that is not us.
"""
files = {"big.bin": b"a" * 1000, "second.bin": b"b" * 10}
data = _build(files, force_zip64=True, tmp_path=tmp_path)
with zipfile.ZipFile(io.BytesIO(data)) as z:
assert z.testzip() is None
for name, content in files.items():
assert z.read(name) == content
# "version needed to extract" is the field that announces zip64;
# create_version is who wrote it and says nothing about the format.
assert z.getinfo("big.bin").extract_version >= 45, "not marked as zip64"
# And the extra field really is being parsed, not skipped: zipfile takes
# the sizes from it when the 32-bit fields are 0xFFFFFFFF.
assert z.getinfo("big.bin").file_size == 1000
def test_an_empty_archive_is_still_an_archive(tmp_path):
data = _build({}, tmp_path=tmp_path)
with zipfile.ZipFile(io.BytesIO(data)) as z:
assert z.namelist() == []
# ── Which files go in, and under what names ─────────────────────────────────
def _under(entries, dir_, tmp_path):
module = tmp_path / "zipstream.mjs"
module.write_text(ZIPSTREAM.read_text())
script = tmp_path / "under.mjs"
script.write_text(f"""
import {{ entriesUnder }} from '{module.as_posix()}';
const out = entriesUnder({json.dumps(entries)}, {json.dumps(dir_)});
console.log(JSON.stringify(out.map(o => o.name)));
""")
proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def _e(name, path=""):
return {"id": name, "name": name, "path": path, "size": 1}
def test_the_archive_starts_at_the_folder_you_asked_for(tmp_path):
"""
Not at the shared root: an archive of "Holidays/2026" should open as
"2026/…", not as a chain of empty parent folders.
"""
entries = [
_e("beach.jpg", "Holidays/2026"),
_e("hotel.pdf", "Holidays/2026/paperwork"),
_e("old.jpg", "Holidays/2019"),
_e("readme.txt", ""),
]
assert sorted(_under(entries, "Holidays/2026", tmp_path)) == [
"2026/beach.jpg", "2026/paperwork/hotel.pdf",
]
def test_a_sibling_with_a_longer_name_is_not_swept_in(tmp_path):
"""
"Holidays2026" starts with "Holidays" as a string but is a different folder,
which is the mistake a prefix test invites.
"""
entries = [_e("a.txt", "Holidays"), _e("b.txt", "Holidays2026")]
assert _under(entries, "Holidays", tmp_path) == ["Holidays/a.txt"]
def test_the_root_is_wrapped_rather_than_exploded(tmp_path):
"""
The root has no name to give the archive, so entries get a "files/" wrapper.
An archive that unpacks straight into whatever directory it was opened in is
the kind that scatters a hundred files across someone's Downloads folder.
"""
entries = [_e("a.txt", ""), _e("b.txt", "sub")]
assert sorted(_under(entries, "", tmp_path)) == ["files/a.txt", "files/sub/b.txt"]
|