software/kisstdlib/./test/test_fs.py

A set of modules for the Python programming language designed for system programming with conceptually and algebraically simple primitives.

Files

Raw Source

Contents

# Copyright (c) 2025 Jan Malakhovski <oxij@oxij.org>
#
# This file is a part of `kisstdlib` project.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

"""Testing `kisstdlib.fs` module."""

import os

from kisstdlib.failure import *
from kisstdlib.fs import *


def test_iter_subtree(tmp_path: str) -> None:
    os.chdir(tmp_path)

    def touch(path: str) -> None:
        with open(path, "wb") as f:
            f.close()

    os.makedirs("example")
    touch("example/a")
    touch("example/aa")
    touch("example/b")
    touch("example/c")
    touch("example.dot")
    touch("example*not")

    w = list(map(lambda x: os.path.relpath(x[0], tmp_path), iter_subtree(".")))
    assert w == [
        ".",
        "example*not",
        "example.dot",
        "example",
        "example/a",
        "example/aa",
        "example/b",
        "example/c",
    ]


def atomic1(defer: bool, tmp_path: str) -> None:
    os.chdir(tmp_path)

    sync: DeferredSync[str] | bool = True
    if defer:
        sync = DeferredSync(True)

    def check(es: list[list[str]], ed: list[list[str]] | None = None) -> None:
        if isinstance(sync, DeferredSync):
            gs: list[list[str]] = []
            sync.copy().sync(True, gs)
            assert es == gs

        if ed is not None:
            gd = list(describe_forest(["."], hash_length=8))
            assert ed == gd

    atomic_write(b"test a", "test.a", sync=sync)
    atomic_write(b"test b", "test.b", sync=sync)
    atomic_write(b"test c", "test.c", sync=sync)
    check(
        [
            ["fsync", "test.a.part"],
            ["fsync", "test.b.part"],
            ["fsync", "test.c.part"],
            ["rename", "test.a.part", "test.a"],
            ["rename", "test.b.part", "test.b"],
            ["rename", "test.c.part", "test.c"],
            ["fsync_win", "test.a"],
            ["fsync_win", "test.b"],
            ["fsync_win", "test.c"],
            ["fsync_dir", "."],
        ]
    )
    atomic_rename("test.a", "a/test.a", False, sync=sync)
    check(
        [
            ["fsync", "test.a.part"],
            ["fsync", "test.b.part"],
            ["fsync", "test.c.part"],
            ["rename", "test.a.part", "test.a"],
            ["rename", "test.b.part", "test.b"],
            ["rename", "test.c.part", "test.c"],
            ["barrier"],
            ["fsync_win", "test.a"],
            ["fsync_win", "test.b"],
            ["fsync_win", "test.c"],
            ["fsync_dir", "."],
            ["rename", "test.a", "a/test.a"],
            ["fsync_win", "a/test.a"],
            ["fsync_dir", "a"],
            ["fsync_dir", "."],
        ]
    )
    atomic_rename("test.b", "a/test.b", False, sync=sync)
    check(
        [
            ["fsync", "test.a.part"],
            ["fsync", "test.b.part"],
            ["fsync", "test.c.part"],
            ["rename", "test.a.part", "test.a"],
            ["rename", "test.b.part", "test.b"],
            ["rename", "test.c.part", "test.c"],
            ["barrier"],
            ["fsync_win", "test.a"],
            ["fsync_win", "test.b"],
            ["fsync_win", "test.c"],
            ["fsync_dir", "."],
            ["rename", "test.a", "a/test.a"],
            ["rename", "test.b", "a/test.b"],
            ["fsync_win", "a/test.a"],
            ["fsync_win", "a/test.b"],
            ["fsync_dir", "a"],
            ["fsync_dir", "."],
        ]
    )
    atomic_rename("test.c", "b/test.c", False, sync=sync)
    check(
        [
            ["fsync", "test.a.part"],
            ["fsync", "test.b.part"],
            ["fsync", "test.c.part"],
            ["rename", "test.a.part", "test.a"],
            ["rename", "test.b.part", "test.b"],
            ["rename", "test.c.part", "test.c"],
            ["barrier"],
            ["fsync_win", "test.a"],
            ["fsync_win", "test.b"],
            ["fsync_win", "test.c"],
            ["fsync_dir", "."],
            ["rename", "test.a", "a/test.a"],
            ["rename", "test.b", "a/test.b"],
            ["rename", "test.c", "b/test.c"],
            ["fsync_win", "a/test.a"],
            ["fsync_win", "a/test.b"],
            ["fsync_win", "b/test.c"],
            ["fsync_dir", "a"],
            ["fsync_dir", "b"],
            ["fsync_dir", "."],
        ]
    )
    atomic_rename("a/test.b", "test.b", False, sync=sync)
    check(
        [
            ["fsync", "test.a.part"],
            ["fsync", "test.b.part"],
            ["fsync", "test.c.part"],
            ["rename", "test.a.part", "test.a"],
            ["rename", "test.b.part", "test.b"],
            ["rename", "test.c.part", "test.c"],
            ["barrier"],
            ["fsync_win", "test.a"],
            ["fsync_win", "test.b"],
            ["fsync_win", "test.c"],
            ["fsync_dir", "."],
            ["rename", "test.a", "a/test.a"],
            ["rename", "test.b", "a/test.b"],
            ["rename", "test.c", "b/test.c"],
            ["barrier"],
            ["fsync_win", "a/test.a"],
            ["fsync_win", "a/test.b"],
            ["fsync_win", "b/test.c"],
            ["fsync_dir", "a"],
            ["fsync_dir", "b"],
            ["fsync_dir", "."],
            ["rename", "a/test.b", "test.b"],
            ["fsync_win", "test.b"],
            ["fsync_dir", "."],
            ["fsync_dir", "a"],
        ]
    )
    if isinstance(sync, DeferredSync):
        sync.flush()
    check(
        [],
        [
            ["./", "dir"],
            ["a/", "dir"],
            ["a/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
            ["b/", "dir"],
            ["b/test.c", "reg", "size", "6", "sha256", "806b51ad"],
            ["test.b", "reg", "size", "6", "sha256", "6346935e"],
        ],
    )

    atomic_link("a/test.a", "test.a", True, sync=sync)
    atomic_symlink("b/test.c", "test.c", True, sync=sync)
    check(
        [
            ["fsync", "test.a.part"],
            ["fsync", "test.c.part"],
            ["replace", "test.a.part", "test.a"],
            ["replace", "test.c.part", "test.c"],
            ["fsync_win", "test.a"],
            ["fsync_win", "test.c"],
            ["fsync_dir", "."],
        ]
    )
    if isinstance(sync, DeferredSync):
        sync.flush()
    check([])

    atomic_copy2("test.a", "x/test.a", True, sync=sync)
    atomic_copy2("test.c", "x/test.c", True, sync=sync)
    atomic_link("test.c", "x/test.c.lnk", True, follow_symlinks=False, sync=sync)
    atomic_copy2("test.c", "x/test.c.sym", True, follow_symlinks=False, sync=sync)
    check(
        [
            ["fsync", "x/test.a.part"],
            ["fsync", "x/test.c.lnk.part"],
            ["fsync", "x/test.c.part"],
            ["fsync", "x/test.c.sym.part"],
            ["replace", "x/test.a.part", "x/test.a"],
            ["replace", "x/test.c.part", "x/test.c"],
            ["replace", "x/test.c.lnk.part", "x/test.c.lnk"],
            ["replace", "x/test.c.sym.part", "x/test.c.sym"],
            ["fsync_win", "x/test.a"],
            ["fsync_win", "x/test.c"],
            ["fsync_win", "x/test.c.lnk"],
            ["fsync_win", "x/test.c.sym"],
            ["fsync_dir", "x"],
        ]
    )
    if isinstance(sync, DeferredSync):
        sync.flush()
    check(
        [],
        [
            ["./", "dir"],
            ["a/", "dir"],
            ["a/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
            ["b/", "dir"],
            ["b/test.c", "reg", "size", "6", "sha256", "806b51ad"],
            ["test.a", "ref", "==>", "a/test.a"],
            ["test.b", "reg", "size", "6", "sha256", "6346935e"],
            ["test.c", "sym", "->", "b/test.c"],
            ["x/", "dir"],
            ["x/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
            ["x/test.c", "reg", "size", "6", "sha256", "806b51ad"],
            ["x/test.c.lnk", "ref", "==>", "test.c"],
            ["x/test.c.sym", "sym", "->", "b/test.c"],
        ],
    )

    atomic_copy2("test.a", "y/test.a", True, sync=sync)
    atomic_copy2("test.c", "y/test.c", True, sync=sync)
    atomic_link("test.c", "y/test.c.lnk", True, follow_symlinks=False, sync=sync)
    atomic_copy2("test.c", "y/test.c.sym", True, follow_symlinks=False, sync=sync)
    atomic_move("test.c", "z/test.c", True, sync=sync)
    check(
        [
            ["fsync", "y/test.a.part"],
            ["fsync", "y/test.c.lnk.part"],
            ["fsync", "y/test.c.part"],
            ["fsync", "y/test.c.sym.part"],
            ["fsync", "z/test.c.part"],
            ["replace", "y/test.a.part", "y/test.a"],
            ["replace", "y/test.c.part", "y/test.c"],
            ["replace", "y/test.c.lnk.part", "y/test.c.lnk"],
            ["replace", "y/test.c.sym.part", "y/test.c.sym"],
            ["replace", "z/test.c.part", "z/test.c"],
            ["fsync_win", "y/test.a"],
            ["fsync_win", "y/test.c"],
            ["fsync_win", "y/test.c.lnk"],
            ["fsync_win", "y/test.c.sym"],
            ["fsync_win", "z/test.c"],
            ["fsync_dir", "y"],
            ["fsync_dir", "z"],
            ["unlink", "test.c"],
            ["fsync_dir", "."],
        ]
    )
    if isinstance(sync, DeferredSync):
        sync.flush()
    check(
        [],
        [
            ["./", "dir"],
            ["a/", "dir"],
            ["a/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
            ["b/", "dir"],
            ["b/test.c", "reg", "size", "6", "sha256", "806b51ad"],
            ["test.a", "ref", "==>", "a/test.a"],
            ["test.b", "reg", "size", "6", "sha256", "6346935e"],
            ["x/", "dir"],
            ["x/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
            ["x/test.c", "reg", "size", "6", "sha256", "806b51ad"],
            ["x/test.c.lnk", "sym", "->", "b/test.c"],
            ["x/test.c.sym", "sym", "->", "b/test.c"],
            ["y/", "dir"],
            ["y/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
            ["y/test.c", "reg", "size", "6", "sha256", "806b51ad"],
            ["y/test.c.lnk", "ref", "==>", "x/test.c.lnk"],
            ["y/test.c.sym", "sym", "->", "b/test.c"],
            ["z/", "dir"],
            ["z/test.c", "ref", "==>", "y/test.c.lnk"],
        ],
    )

    atomic_symlink("/home", "y/home")

    # to test `describe_forest` with multiple paths
    w = list(describe_forest(["x", "y"], hash_length=8))
    # print(repr(w))
    assert w == [
        ["0/", "dir"],
        ["0/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
        ["0/test.c", "reg", "size", "6", "sha256", "806b51ad"],
        ["0/test.c.lnk", "sym", "->", "b/test.c"],
        ["0/test.c.sym", "sym", "->", "b/test.c"],
        ["1/", "dir"],
        ["1/home", "sym", "->", "/home"],
        ["1/test.a", "reg", "size", "6", "sha256", "1136b2eb"],
        ["1/test.c", "reg", "size", "6", "sha256", "806b51ad"],
        ["1/test.c.lnk", "ref", "==>", "0/test.c.lnk"],
        ["1/test.c.sym", "sym", "->", "b/test.c"],
    ]


def test_without_DeferredSync(tmp_path: str) -> None:
    return atomic1(False, tmp_path)


def test_with_DeferredSync(tmp_path: str) -> None:
    return atomic1(True, tmp_path)