File: roundtrip.py

package info (click to toggle)
zchunk 1.5.2%2Bds1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 4,704 kB
  • sloc: ansic: 13,244; python: 457; sh: 260; makefile: 13
file content (127 lines) | stat: -rw-r--r-- 3,998 bytes parent folder | download | duplicates (2)
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
"""A very simple test for the command-line zchunk tools."""

from __future__ import annotations

import dataclasses
import os
import pathlib
import subprocess  # noqa: S404
import sys
import tempfile
import typing

from pychunk import common


if typing.TYPE_CHECKING:
    from typing import Final


@dataclasses.dataclass(frozen=True)
class Config(common.Config):
    """Runtime configuration."""

    tempd: pathlib.Path

    uncompressed: pathlib.Path
    recompressed: pathlib.Path


def parse_args(dirname: str) -> Config:
    """Parse the command-line arguments, deduce some things."""
    parser: Final = common.base_parser("roundtrip")

    args: Final = parser.parse_args()
    bindir: Final = pathlib.Path(args.bindir).absolute()
    if not bindir.is_dir():
        sys.exit(f"Not a directory: {bindir}")
    zck: Final = bindir / "zck"
    if not zck.is_file() or not os.access(zck, os.X_OK):
        sys.exit(f"Not an executable file: {zck}")

    tempd: Final = pathlib.Path(dirname).absolute()
    return Config(
        tempd=tempd,
        bindir=bindir,
        env=common.get_runenv(),
        orig=pathlib.Path(args.filename).absolute(),
        compressed=tempd / "words.txt.zck",
        uncompressed=tempd / "un/words.txt",
        recompressed=tempd / "re/words.txt.zck",
    )


def do_uncompress(cfg: Config, orig_size: int) -> None:
    """Uncompress and compare."""
    # OK, so unzck's behavior is... weird.
    cfg.uncompressed.parent.mkdir(mode=0o755)

    print(f"Extracting {cfg.compressed} to {cfg.uncompressed}")
    if cfg.uncompressed.exists():
        sys.exit(f"Did not expect {cfg.uncompressed} to exist")
    subprocess.check_call(
        [cfg.bindir / "unzck", "--", cfg.compressed],
        shell=False,
        env=cfg.env,
        cwd=cfg.uncompressed.parent,
    )
    if not cfg.uncompressed.is_file():
        subprocess.check_call(["ls", "-lt", "--", cfg.tempd], shell=False)
        sys.exit(f"unzck did not create the {cfg.uncompressed} file")

    new_size: Final = cfg.uncompressed.stat().st_size
    print(f"Uncompressed size {new_size}")
    if new_size != orig_size:
        sys.exit(f"Uncompressed size {new_size} != original size {orig_size}")

    print(f"Comparing {cfg.orig} to {cfg.uncompressed}")
    subprocess.check_call(["cmp", "--", cfg.orig, cfg.uncompressed], shell=False, env=cfg.env)


def do_recompress(cfg: Config, comp_size: int) -> None:
    """Recompress the file and compare."""
    # OK, so zck's behavior is also weird...
    cfg.recompressed.parent.mkdir(mode=0o755)

    print(f"Recompressing {cfg.uncompressed} to {cfg.recompressed}")
    if cfg.recompressed.exists():
        sys.exit(f"Did not expect {cfg.recompressed} to exist")
    subprocess.check_call(
        [cfg.bindir / "zck", "--", cfg.uncompressed],
        shell=False,
        env=cfg.env,
        cwd=cfg.recompressed.parent,
    )
    if not cfg.recompressed.is_file():
        sys.exit(f"zck did not create the {cfg.recompressed} file")

    new_size: Final = cfg.recompressed.stat().st_size
    print(f"Recompressed size {new_size}")
    if new_size != comp_size:
        sys.exit(f"Recompressed size {new_size} != compressed size {comp_size}")

    print(f"Comparing {cfg.compressed} to {cfg.recompressed}")
    subprocess.check_call(
        ["cmp", "--", cfg.compressed, cfg.recompressed],
        shell=False,
        env=cfg.env,
    )


def main() -> None:
    """Create a temporary directory, compress a file, analyze it."""
    with tempfile.TemporaryDirectory() as dirname:
        print(f"Using temporary directory {dirname}")
        cfg: Final = parse_args(dirname)
        orig_size: Final = cfg.orig.stat().st_size
        print(f"{cfg.orig} is {orig_size} bytes long")

        comp_size: Final = common.do_compress(cfg, orig_size)
        common.read_chunks(cfg, orig_size, comp_size)
        do_uncompress(cfg, orig_size)
        do_recompress(cfg, comp_size)
        print("Seems fine!")


if __name__ == "__main__":
    main()