Single-file, stdlib-only CLI that writes the CSV format WinDirStat's CsvLoader accepts (parent-before-child rows, subtree totals, localized headers for all 25 UI languages). Includes install.sh, pyproject entry point, unit tests and a generator for the language table. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
148 lines
6.1 KiB
Python
148 lines
6.1 KiB
Python
import contextlib
|
|
import csv
|
|
import io
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
|
import wdsexport # noqa: E402
|
|
|
|
SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "wdsexport.py")
|
|
|
|
|
|
def read_rows(path_or_text):
|
|
text = path_or_text if "\n" in path_or_text else open(path_or_text, encoding="utf-8").read()
|
|
assert text.startswith("\ufeff"), "missing UTF-8 BOM"
|
|
rows = list(csv.reader(io.StringIO(text[1:])))
|
|
return rows[0], rows[1:]
|
|
|
|
|
|
def check_loader_invariants(testcase, rows):
|
|
"""Mirror the constraints of WinDirStat's CsvLoader: one root container first,
|
|
every other row's parent (prefix before last backslash) already seen as a container."""
|
|
containers = set()
|
|
for i, r in enumerate(rows):
|
|
name, itype = r[0], int(r[7], 16)
|
|
if i == 0:
|
|
testcase.assertTrue(itype & wdsexport.ITF_ROOTITEM)
|
|
else:
|
|
testcase.assertFalse(itype & wdsexport.ITF_ROOTITEM)
|
|
testcase.assertIn(name.rsplit("\\", 1)[0], containers, f"orphan row {name}")
|
|
if itype & wdsexport.IT_DIRECTORY:
|
|
containers.add(name)
|
|
|
|
|
|
class ExportTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = os.path.join(self.tmp.name, "tree")
|
|
os.makedirs(os.path.join(self.root, "a", "b"))
|
|
os.makedirs(os.path.join(self.root, "skipme"))
|
|
os.makedirs(os.path.join(self.root, ".hidden"))
|
|
self.write("a/one.bin", 1000)
|
|
self.write("a/b/two.bin", 5000)
|
|
self.write('a/b/we,ird "na\\me".txt', 10)
|
|
self.write("skipme/big.bin", 100000)
|
|
self.write(".hidden/x", 7)
|
|
os.link(os.path.join(self.root, "a/one.bin"), os.path.join(self.root, "a/b/hardlink.bin"))
|
|
os.symlink("a/one.bin", os.path.join(self.root, "link"))
|
|
self.locked = os.path.join(self.root, "locked")
|
|
os.makedirs(self.locked)
|
|
self.write("locked/secret", 3)
|
|
os.chmod(self.locked, 0)
|
|
self.out = os.path.join(self.tmp.name, "out.csv")
|
|
|
|
def tearDown(self):
|
|
os.chmod(self.locked, 0o755)
|
|
self.tmp.cleanup()
|
|
|
|
def write(self, rel, size):
|
|
with open(os.path.join(self.root, rel), "wb") as f:
|
|
f.write(b"x" * size)
|
|
|
|
def run_main(self, *args):
|
|
err = io.StringIO()
|
|
with contextlib.redirect_stderr(err):
|
|
rc = wdsexport.main(list(args))
|
|
self.assertEqual(rc, 0, err.getvalue())
|
|
return err.getvalue()
|
|
|
|
def test_structure_and_totals(self):
|
|
self.run_main(self.root, "-o", self.out, "--name", "test", "--exclude",
|
|
os.path.join(self.root, "skipme"), "-q")
|
|
header, rows = read_rows(self.out)
|
|
self.assertEqual(header, wdsexport.header("en"))
|
|
check_loader_invariants(self, rows)
|
|
by_name = {r[0]: r for r in rows}
|
|
|
|
root = rows[0]
|
|
self.assertEqual(root[0], "test")
|
|
symlink_size = os.lstat(os.path.join(self.root, "link")).st_size
|
|
# one.bin counted once despite the hard link; symlink counted as itself
|
|
self.assertEqual(int(root[3]), 1000 + 5000 + 10 + 7 + symlink_size)
|
|
# files: one.bin, two.bin, weird, x, link (hardlink and locked contents not counted)
|
|
self.assertEqual(int(root[1]), 5)
|
|
# folders: a, a/b, .hidden, locked
|
|
self.assertEqual(int(root[2]), 4)
|
|
|
|
self.assertNotIn("test\\skipme", by_name)
|
|
# which hard-link path gets the bytes depends on scandir order; both live under a/
|
|
self.assertEqual(by_name["test\\a"][3], "6010")
|
|
self.assertIn("test\\a\\b\\we,ird 'na_me'.txt", by_name)
|
|
self.assertEqual(by_name["test\\.hidden"][5], "H")
|
|
self.assertEqual(by_name["test\\locked"][9], "partial")
|
|
self.assertTrue(int(by_name["test\\locked"][7], 16) & wdsexport.ITF_SCANERROR)
|
|
self.assertRegex(root[6], r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$")
|
|
|
|
# each CSV line must have exactly 8 double quotes (loader has no escaping)
|
|
with open(self.out, encoding="utf-8") as f:
|
|
next(f)
|
|
for line in f:
|
|
self.assertEqual(line.count('"'), 8, line)
|
|
|
|
def test_language_headers(self):
|
|
self.run_main(self.root, "-o", self.out, "--lang", "da", "-q")
|
|
header, _ = read_rows(self.out)
|
|
self.assertEqual(header[:3], ["Navn", "Filer", "Mapper"])
|
|
self.assertEqual(header[7], "WinDirStat Attributter")
|
|
|
|
def test_stdout_matches_file(self):
|
|
self.run_main(self.root, "-o", self.out, "--name", "n", "-q")
|
|
res = subprocess.run([sys.executable, SCRIPT, self.root, "-o", "-", "--name", "n", "-q"],
|
|
capture_output=True, check=True)
|
|
self.assertEqual(res.stdout.decode("utf-8"), open(self.out, encoding="utf-8").read())
|
|
|
|
def test_zip(self):
|
|
self.run_main(self.root, "-o", self.out, "--zip", "-q")
|
|
with zipfile.ZipFile(self.out[:-4] + ".zip") as z:
|
|
self.assertEqual(z.read("out.csv"), open(self.out, "rb").read())
|
|
|
|
def test_output_inside_scanned_tree_is_skipped(self):
|
|
out = os.path.join(self.root, "inside.csv")
|
|
self.run_main(self.root, "-o", out, "--name", "t", "-q")
|
|
names = [r[0] for r in read_rows(out)[1]]
|
|
self.assertNotIn("t\\inside.csv", names)
|
|
self.assertFalse([n for n in names if ".wdsexport-" in n or n.endswith(".part")])
|
|
self.assertFalse([f for f in os.listdir(self.root) if f.startswith(".wdsexport-")])
|
|
|
|
def test_default_name(self):
|
|
self.assertTrue(wdsexport.default_name("/").endswith("-root"))
|
|
self.assertTrue(wdsexport.default_name(self.root).endswith("-tree"))
|
|
|
|
|
|
class ReverseLinesTest(unittest.TestCase):
|
|
def test_small_blocks(self):
|
|
lines = [f"line {i} {'x' * (i % 7)}".encode() for i in range(500)]
|
|
src = io.BytesIO(b"\n".join(lines) + b"\n")
|
|
dst = io.BytesIO()
|
|
wdsexport.copy_reversed_lines(src, dst, block=13)
|
|
self.assertEqual(dst.getvalue(), b"\n".join(reversed(lines)) + b"\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|