Compare commits
3 Commits
a97395370a
...
dev
| Author | SHA1 | Date | |
|---|---|---|---|
| f95c68ee41 | |||
| 2b961baa5b | |||
| 69bb85af01 |
@@ -26,6 +26,7 @@ from typing import Any, ClassVar, Literal
|
||||
|
||||
import click
|
||||
|
||||
from bsv.cli_utils import format_human_byte_size
|
||||
from bsv.repo import default_repository_path
|
||||
from bsv.vfs import (
|
||||
AlreadyExistError,
|
||||
@@ -90,7 +91,7 @@ class BsvPathType(click.ParamType):
|
||||
class AnyPathType(click.ParamType):
|
||||
"""Converter for bsv or fs paths given on the command line."""
|
||||
|
||||
name: ClassVar[str] = "any_path"
|
||||
name: str = "any_path"
|
||||
|
||||
default: Literal["bsv", "fs"]
|
||||
|
||||
@@ -164,7 +165,8 @@ def info(params: RepositoryParams):
|
||||
@click.pass_obj
|
||||
def init(params: RepositoryParams, device_name: str):
|
||||
"""Initialize a bsv repository."""
|
||||
print(f"device_name: {device_name!r}")
|
||||
print(f"Repository path: {params.path!r}")
|
||||
print(f"Device name: {device_name!r}")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@@ -199,23 +201,6 @@ def mkdir(
|
||||
sys.exit(return_code)
|
||||
|
||||
|
||||
SI_PREFIXES = ["", "k", "M", "G", "T", "P", "E", "Z", "Y", "R", "Q"]
|
||||
|
||||
|
||||
def human_byte_size(byte_size: int) -> str:
|
||||
if byte_size == 0:
|
||||
return "0"
|
||||
|
||||
prefix_index = int(math.log2(byte_size) / 10)
|
||||
display_size: float = byte_size / 1024**prefix_index
|
||||
size = (
|
||||
str(math.ceil(display_size))
|
||||
if prefix_index == 0 or display_size >= 10
|
||||
else repr(math.ceil(display_size * 10) / 10)
|
||||
)
|
||||
return f"{size}{SI_PREFIXES[prefix_index]}"
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("files", nargs=-1, type=BsvPathType())
|
||||
@click.option("--filter", flag_value="hidden", default=True, hidden=True)
|
||||
@@ -237,7 +222,7 @@ def ls(
|
||||
if not files:
|
||||
files = (PurePosixPath("/"),)
|
||||
|
||||
filter_md = FileMetadata.is_hidden_files if filter == "hidden" else lambda _: False
|
||||
filter_md = FileMetadata.is_hidden if filter == "hidden" else lambda _: False
|
||||
|
||||
for file_index, file in enumerate(files):
|
||||
if len(files) > 1:
|
||||
@@ -260,7 +245,7 @@ def ls(
|
||||
for name, md in items:
|
||||
mode = str(md.unix_mode)
|
||||
size = (
|
||||
human_byte_size(md.byte_size)
|
||||
format_human_byte_size(md.byte_size)
|
||||
if human_readable
|
||||
else str(md.byte_size)
|
||||
)
|
||||
|
||||
48
src/bsv/cli_utils.py
Normal file
48
src/bsv/cli_utils.py
Normal file
@@ -0,0 +1,48 @@
|
||||
# pybsv - Backup, Synchronization, Versioning.
|
||||
# Copyright (C) 2025 Simon Boyé
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
"""Tools and utilities to build the command-line interface."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
|
||||
BINARY_PREFIXES: Final[list[str]] = [
|
||||
"",
|
||||
"Ki",
|
||||
"Mi",
|
||||
"Gi",
|
||||
"Ti",
|
||||
"Pi",
|
||||
"Ei",
|
||||
"Zi",
|
||||
"Yi",
|
||||
"Ri",
|
||||
"Qi",
|
||||
]
|
||||
|
||||
|
||||
def format_human_byte_size(byte_size: int) -> str:
|
||||
"""Format the given `byte_size` as a human-readable string."""
|
||||
index = min(max((byte_size.bit_length() - 1) // 10, 0), len(BINARY_PREFIXES) - 1)
|
||||
size = byte_size / 1024**index
|
||||
num_digits = len(str(int(size)))
|
||||
decimals = max(0, 3 - num_digits)
|
||||
rounded = round(size, decimals)
|
||||
if rounded == 1024 and index + 1 < len(BINARY_PREFIXES):
|
||||
rounded = 1
|
||||
index += 1
|
||||
return f"{rounded:.16g}{BINARY_PREFIXES[index]}B"
|
||||
272
src/bsv/vfs.py
272
src/bsv/vfs.py
@@ -22,7 +22,7 @@ from functools import total_ordering
|
||||
import os
|
||||
from pathlib import Path, PurePosixPath
|
||||
from stat import S_IFDIR, S_IFLNK, S_IFMT, S_IFREG, S_IMODE, filemode
|
||||
from typing import TYPE_CHECKING, Any, BinaryIO, Literal, Self, cast
|
||||
from typing import TYPE_CHECKING, Any, BinaryIO, Literal, Self
|
||||
|
||||
from typing_extensions import Buffer
|
||||
|
||||
@@ -79,6 +79,119 @@ DEFAULT_DIR_PERMS = Permissions(0o770)
|
||||
DEFAULT_FILE_PERMS = Permissions(0o640)
|
||||
|
||||
|
||||
FileType = Literal["dir", "file", "symlink", "other"]
|
||||
|
||||
_IFMT_MAP: dict[int, FileType] = {
|
||||
S_IFDIR: "dir",
|
||||
S_IFREG: "file",
|
||||
S_IFLNK: "symlink",
|
||||
}
|
||||
|
||||
|
||||
@total_ordering
|
||||
class FileMetadata:
|
||||
"""Metadata associated with vfs files: file type, permissions, etc."""
|
||||
|
||||
path: PurePosixPath
|
||||
type: FileType
|
||||
permissions: Permissions
|
||||
modification_time: datetime
|
||||
byte_size: int
|
||||
_stat: stat_result
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
path: PurePosixPath,
|
||||
*,
|
||||
type: FileType,
|
||||
permissions: Permissions,
|
||||
modification_time: datetime,
|
||||
byte_size: int,
|
||||
):
|
||||
"""Create a `FileMetadata`."""
|
||||
self.path = path
|
||||
self.type = type
|
||||
self.permissions = permissions
|
||||
self.modification_time = modification_time
|
||||
self.byte_size = byte_size
|
||||
|
||||
@classmethod
|
||||
def from_stat(
|
||||
cls,
|
||||
path: PurePosixPath,
|
||||
stat: stat_result,
|
||||
) -> Self:
|
||||
"""Create a `FileMetadata` from a `stat_result`."""
|
||||
return cls(
|
||||
path,
|
||||
type=_IFMT_MAP.get(S_IFMT(stat.st_mode), "other"),
|
||||
permissions=Permissions(S_IMODE(stat.st_mode)),
|
||||
modification_time=datetime.fromtimestamp(stat.st_mtime, UTC),
|
||||
byte_size=stat.st_size,
|
||||
)
|
||||
|
||||
@property
|
||||
def unix_mode(self) -> str:
|
||||
"""Return unix-like mode in the form '-rwxrwxrwx'."""
|
||||
return UNIX_MODE_FILE_TYPE[self.type] + str(self.permissions)
|
||||
|
||||
@property
|
||||
def is_hidden(self) -> bool:
|
||||
"""Return true if the file starts with a '.'."""
|
||||
return self.path.name.startswith(".")
|
||||
|
||||
@property
|
||||
def is_file(self) -> bool:
|
||||
"""Test if this is a file."""
|
||||
return self.type == "file"
|
||||
|
||||
@property
|
||||
def is_dir(self) -> bool:
|
||||
"""Test if this is a directory."""
|
||||
return self.type == "dir"
|
||||
|
||||
@property
|
||||
def is_symlink(self) -> bool:
|
||||
"""Test if this is a symbolic link."""
|
||||
return self.type == "symlink"
|
||||
|
||||
@property
|
||||
def is_other(self) -> bool:
|
||||
"""Test if this is a symbolic link."""
|
||||
return self.type == "other"
|
||||
|
||||
def _as_tuple(
|
||||
self,
|
||||
) -> tuple[PurePosixPath, FileType, Permissions, datetime, int]:
|
||||
return (
|
||||
self.path,
|
||||
self.type,
|
||||
self.permissions,
|
||||
self.modification_time,
|
||||
self.byte_size,
|
||||
)
|
||||
|
||||
def __eq__(self, rhs: Any) -> bool:
|
||||
"""Test if two `Metadata` are the same."""
|
||||
return (
|
||||
self._as_tuple() == rhs._as_tuple()
|
||||
if isinstance(rhs, FileMetadata)
|
||||
else NotImplemented
|
||||
)
|
||||
|
||||
def __lt__(self, rhs: Any) -> bool:
|
||||
"""Compare `rhs.path` with `self.path`."""
|
||||
return self.path < rhs.path if isinstance(rhs, FileMetadata) else NotImplemented
|
||||
|
||||
|
||||
UNIX_MODE_FILE_TYPE = {
|
||||
"dir": "d",
|
||||
"file": "-",
|
||||
"other": "o",
|
||||
"link": "l",
|
||||
}
|
||||
|
||||
|
||||
class VirtualFileSystem:
|
||||
"""Represent a file system, with common file system operations."""
|
||||
|
||||
@@ -95,20 +208,39 @@ class VirtualFileSystem:
|
||||
|
||||
def is_file(self, path: AnyBsvPath) -> bool:
|
||||
"""Test if `path` is a file."""
|
||||
path = self._make_path(path)
|
||||
return self._real_path(path).is_file()
|
||||
return self.metadata(path).is_file
|
||||
|
||||
def is_dir(self, path: AnyBsvPath) -> bool:
|
||||
"""Test if `path` is a directory."""
|
||||
path = self._make_path(path)
|
||||
return self._real_path(path).is_dir()
|
||||
return self.metadata(path).is_dir
|
||||
|
||||
def is_symlink(self, path: AnyBsvPath) -> bool:
|
||||
"""Test if `path` is a symbolic link."""
|
||||
return self.metadata(path).is_symlink
|
||||
|
||||
def is_other(self, path: AnyBsvPath) -> bool:
|
||||
"""Test if `path` is not a file, directory or symbolic link."""
|
||||
return self.metadata(path).is_other
|
||||
|
||||
def metadata(self, path: AnyBsvPath) -> FileMetadata:
|
||||
"""Return the metadata of a given object."""
|
||||
metadata = self.metadata_or_none(path)
|
||||
if metadata is None:
|
||||
msg = f"file '{path}' not found"
|
||||
raise NotFoundError(msg)
|
||||
return metadata
|
||||
|
||||
def metadata_or_none(self, path: AnyBsvPath) -> FileMetadata | None:
|
||||
"""Return the metadata of a given object or `None` if it does not exists."""
|
||||
path = self._make_path(path)
|
||||
return FileMetadata.from_stat(
|
||||
self, path, self._real_path(path).stat(follow_symlinks=False)
|
||||
)
|
||||
try:
|
||||
stat = self._real_path(path).stat(follow_symlinks=False)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
except OSError as err:
|
||||
msg = f"failed to read '{path}' metadata"
|
||||
raise FsError(msg) from err
|
||||
return FileMetadata.from_stat(path, stat)
|
||||
|
||||
def iter_dir(self, path: AnyBsvPath) -> Iterator[FileMetadata]:
|
||||
"""Return the metadata of all items in the directory `path`."""
|
||||
@@ -117,7 +249,7 @@ class VirtualFileSystem:
|
||||
try:
|
||||
for entry in os.scandir(real_path):
|
||||
yield FileMetadata.from_stat(
|
||||
self, path / entry.name, entry.stat(follow_symlinks=False)
|
||||
path / entry.name, entry.stat(follow_symlinks=False)
|
||||
)
|
||||
except OSError as err:
|
||||
msg = f"failed to read directory {path}"
|
||||
@@ -125,31 +257,18 @@ class VirtualFileSystem:
|
||||
|
||||
def read_bytes(self, path: AnyBsvPath) -> bytes:
|
||||
"""Return the content of `path` as `bytes`."""
|
||||
path = self._make_path(path)
|
||||
try:
|
||||
return self._real_path(path).read_bytes()
|
||||
except OSError as err:
|
||||
msg = f"failed to read {path}"
|
||||
raise FsError(msg) from err
|
||||
with self.open_read(path) as stream:
|
||||
return stream.read()
|
||||
|
||||
def write_bytes(self, path: AnyBsvPath, data: Buffer | BinaryIO) -> int:
|
||||
"""Create or replace a file at `path`, setting its content to `data`."""
|
||||
path = self._make_path(path)
|
||||
real_path = self._real_path(path)
|
||||
written = 0
|
||||
|
||||
try:
|
||||
stream = real_path.open("wb")
|
||||
except OSError as err:
|
||||
msg = f"failed to write {path}"
|
||||
raise FsError(msg) from err
|
||||
|
||||
with stream:
|
||||
with self.open_write(path) as sout:
|
||||
if isinstance(data, Buffer):
|
||||
written += stream.write(data)
|
||||
written += sout.write(data)
|
||||
else:
|
||||
while chunk := data.read(65536):
|
||||
written += stream.write(chunk)
|
||||
written += sout.write(chunk)
|
||||
return written
|
||||
|
||||
def open_read(self, path: AnyBsvPath) -> BinaryIO:
|
||||
@@ -227,102 +346,3 @@ class VirtualFileSystem:
|
||||
|
||||
def _real_path(self, path: PurePosixPath) -> Path:
|
||||
return self.path / path.relative_to("/")
|
||||
|
||||
|
||||
FileType = Literal["dir", "file", "link", "other"]
|
||||
|
||||
_IFMT_MAP = {
|
||||
S_IFDIR: "dir",
|
||||
S_IFREG: "file",
|
||||
S_IFLNK: "link",
|
||||
}
|
||||
|
||||
|
||||
@total_ordering
|
||||
class FileMetadata:
|
||||
"""Metadata associated with vfs files: file type, permissions, etc."""
|
||||
|
||||
vfs: VirtualFileSystem
|
||||
path: PurePosixPath
|
||||
type: FileType
|
||||
permissions: Permissions
|
||||
modification_time: datetime
|
||||
byte_size: int
|
||||
_stat: stat_result
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
vfs: VirtualFileSystem,
|
||||
path: PurePosixPath,
|
||||
*,
|
||||
type: FileType,
|
||||
permissions: Permissions,
|
||||
modification_time: datetime,
|
||||
byte_size: int,
|
||||
):
|
||||
"""Create a `FileMetadata`."""
|
||||
self.vfs = vfs
|
||||
self.path = path
|
||||
self.type = type
|
||||
self.permissions = permissions
|
||||
self.modification_time = modification_time
|
||||
self.byte_size = byte_size
|
||||
|
||||
@classmethod
|
||||
def from_stat(
|
||||
cls,
|
||||
vfs: VirtualFileSystem,
|
||||
path: PurePosixPath,
|
||||
stat: stat_result,
|
||||
) -> Self:
|
||||
"""Create a `FileMetadata` from a `stat_result`."""
|
||||
return cls(
|
||||
vfs,
|
||||
path,
|
||||
type=cast("FileType", _IFMT_MAP.get(S_IFMT(stat.st_mode), "other")),
|
||||
permissions=Permissions(S_IMODE(stat.st_mode)),
|
||||
modification_time=datetime.fromtimestamp(stat.st_mtime, UTC),
|
||||
byte_size=stat.st_size,
|
||||
)
|
||||
|
||||
@property
|
||||
def unix_mode(self) -> str:
|
||||
"""Return unix-like mode in the form '-rwxrwxrwx'."""
|
||||
return UNIX_MODE_FILE_TYPE[self.type] + str(self.permissions)
|
||||
|
||||
@property
|
||||
def is_hidden_files(self) -> bool:
|
||||
"""Return true if the file starts with a '.'."""
|
||||
return self.path.name.startswith(".")
|
||||
|
||||
def _as_tuple(
|
||||
self,
|
||||
) -> tuple[VirtualFileSystem, PurePosixPath, FileType, Permissions, datetime, int]:
|
||||
return (
|
||||
self.vfs,
|
||||
self.path,
|
||||
self.type,
|
||||
self.permissions,
|
||||
self.modification_time,
|
||||
self.byte_size,
|
||||
)
|
||||
|
||||
def __eq__(self, rhs: Any) -> bool:
|
||||
"""Test if two `Metadata` are the same."""
|
||||
return (
|
||||
self._as_tuple() == rhs._as_tuple()
|
||||
if isinstance(rhs, FileMetadata)
|
||||
else NotImplemented
|
||||
)
|
||||
|
||||
def __lt__(self, rhs: Any) -> bool:
|
||||
"""Compare `rhs.path` with `self.path`."""
|
||||
return self.path < rhs.path if isinstance(rhs, FileMetadata) else NotImplemented
|
||||
|
||||
|
||||
UNIX_MODE_FILE_TYPE = {
|
||||
"dir": "d",
|
||||
"file": "-",
|
||||
"other": "o",
|
||||
"link": "l",
|
||||
}
|
||||
|
||||
56
tests/test_bsv/test_cli_utils.py
Normal file
56
tests/test_bsv/test_cli_utils.py
Normal file
@@ -0,0 +1,56 @@
|
||||
# pybsv - Backup, Synchronization, Versioning.
|
||||
# Copyright (C) 2025 Simon Boyé
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
"""Tests for cli_utils.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from bsv.cli_utils import format_human_byte_size
|
||||
|
||||
|
||||
def test_format_human_byte_size():
|
||||
assert format_human_byte_size(0) == "0B"
|
||||
assert format_human_byte_size(1) == "1B"
|
||||
assert format_human_byte_size(9) == "9B"
|
||||
assert format_human_byte_size(10) == "10B"
|
||||
assert format_human_byte_size(99) == "99B"
|
||||
assert format_human_byte_size(100) == "100B"
|
||||
assert format_human_byte_size(999) == "999B"
|
||||
assert format_human_byte_size(1000) == "1000B"
|
||||
assert format_human_byte_size(1023) == "1023B"
|
||||
assert format_human_byte_size(2**10) == "1KiB"
|
||||
assert format_human_byte_size(int(1.23456 * 2**10)) == "1.23KiB"
|
||||
assert format_human_byte_size(9 * 2**10) == "9KiB"
|
||||
assert format_human_byte_size(10 * 2**10 - 1) == "10KiB"
|
||||
assert format_human_byte_size(int(98.76543 * 2**10)) == "98.8KiB"
|
||||
assert format_human_byte_size(99 * 2**10 - 1) == "99KiB"
|
||||
assert format_human_byte_size(100 * 2**10 - 1) == "100KiB"
|
||||
assert format_human_byte_size(int(192.8374 * 2**10)) == "193KiB"
|
||||
assert format_human_byte_size(999 * 2**10 - 1) == "999KiB"
|
||||
assert format_human_byte_size(1000 * 2**10 - 1) == "1000KiB"
|
||||
assert format_human_byte_size(2**20 - 1) == "1MiB"
|
||||
assert format_human_byte_size(2**20) == "1MiB"
|
||||
assert format_human_byte_size(2**30) == "1GiB"
|
||||
assert format_human_byte_size(2**40) == "1TiB"
|
||||
assert format_human_byte_size(2**50) == "1PiB"
|
||||
assert format_human_byte_size(2**60) == "1EiB"
|
||||
assert format_human_byte_size(2**70) == "1ZiB"
|
||||
assert format_human_byte_size(2**80) == "1YiB"
|
||||
assert format_human_byte_size(2**90) == "1RiB"
|
||||
assert format_human_byte_size(2**100 - 2**80) == "1QiB"
|
||||
assert format_human_byte_size(2**100) == "1QiB"
|
||||
assert format_human_byte_size(2**110 - 2**90) == "1024QiB"
|
||||
assert format_human_byte_size(2**110) == "1024QiB"
|
||||
assert format_human_byte_size(2**120) == "1048576QiB"
|
||||
@@ -23,7 +23,7 @@ from pathlib import Path, PurePosixPath
|
||||
|
||||
import pytest
|
||||
|
||||
from bsv.vfs import FsError, Permissions, VirtualFileSystem
|
||||
from bsv.vfs import FileMetadata, FsError, Permissions, VirtualFileSystem
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -53,6 +53,158 @@ def test_permissions():
|
||||
assert str(perm1) == "rwxr-x-w-"
|
||||
|
||||
|
||||
########################################################################################
|
||||
# FileMetadata
|
||||
|
||||
|
||||
def test_file_metadata():
|
||||
path = PurePosixPath("/some_dir/some_file")
|
||||
permissions = Permissions(0o1234)
|
||||
mod_time = datetime(2025, 7, 12, 12, 34, 56, tzinfo=UTC)
|
||||
|
||||
file_md = FileMetadata(
|
||||
path,
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
|
||||
assert file_md.path == path
|
||||
assert file_md.type == "file"
|
||||
assert file_md.permissions == permissions
|
||||
assert file_md.modification_time == mod_time
|
||||
assert file_md.byte_size == 123
|
||||
assert file_md.unix_mode == "--w--wxr-T"
|
||||
assert not file_md.is_hidden
|
||||
assert file_md.is_file
|
||||
assert not file_md.is_dir
|
||||
assert not file_md.is_symlink
|
||||
assert not file_md.is_other
|
||||
|
||||
dir_md = FileMetadata(
|
||||
path,
|
||||
type="dir",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
assert dir_md.type == "dir"
|
||||
assert not dir_md.is_file
|
||||
assert dir_md.is_dir
|
||||
assert not dir_md.is_symlink
|
||||
assert not dir_md.is_other
|
||||
|
||||
symlink_md = FileMetadata(
|
||||
path,
|
||||
type="symlink",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
assert symlink_md.type == "symlink"
|
||||
assert not symlink_md.is_file
|
||||
assert not symlink_md.is_dir
|
||||
assert symlink_md.is_symlink
|
||||
assert not symlink_md.is_other
|
||||
|
||||
other_md = FileMetadata(
|
||||
path,
|
||||
type="other",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
assert other_md.type == "other"
|
||||
assert not other_md.is_file
|
||||
assert not other_md.is_dir
|
||||
assert not other_md.is_symlink
|
||||
assert other_md.is_other
|
||||
|
||||
assert FileMetadata(
|
||||
PurePosixPath("/some_dir/.some_file"),
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
).is_hidden
|
||||
|
||||
|
||||
def test_file_metadata_eq():
|
||||
path = PurePosixPath("/some_dir/some_file")
|
||||
permissions = Permissions(0o1234)
|
||||
mod_time = datetime(2025, 7, 12, 12, 34, 56, tzinfo=UTC)
|
||||
|
||||
md = FileMetadata(
|
||||
path,
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
|
||||
assert (
|
||||
FileMetadata(
|
||||
path,
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
== md
|
||||
)
|
||||
assert (
|
||||
FileMetadata(
|
||||
PurePosixPath("/some_dir/some_other_file"),
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
!= md
|
||||
)
|
||||
assert (
|
||||
FileMetadata(
|
||||
path,
|
||||
type="dir",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
!= md
|
||||
)
|
||||
assert (
|
||||
FileMetadata(
|
||||
path,
|
||||
type="file",
|
||||
permissions=Permissions(0o0752),
|
||||
modification_time=mod_time,
|
||||
byte_size=123,
|
||||
)
|
||||
!= md
|
||||
)
|
||||
assert (
|
||||
FileMetadata(
|
||||
path,
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=datetime(2025, 1, 2, 3, 4, 5),
|
||||
byte_size=123,
|
||||
)
|
||||
!= md
|
||||
)
|
||||
assert (
|
||||
FileMetadata(
|
||||
path,
|
||||
type="file",
|
||||
permissions=permissions,
|
||||
modification_time=mod_time,
|
||||
byte_size=124,
|
||||
)
|
||||
!= md
|
||||
)
|
||||
|
||||
|
||||
########################################################################################
|
||||
# mkdir
|
||||
|
||||
@@ -177,18 +329,36 @@ def test_metadata(fs: VirtualFileSystem):
|
||||
assert md.type == "file"
|
||||
assert md.modification_time == file_time
|
||||
assert md.byte_size == len(file_content)
|
||||
assert not md.is_hidden_files
|
||||
assert not md.is_hidden
|
||||
assert fs.metadata("/test_file") == md
|
||||
assert fs.is_file("/test_file")
|
||||
assert not fs.is_dir("/test_file")
|
||||
assert not fs.is_symlink("/test_file")
|
||||
assert not fs.is_other("/test_file")
|
||||
|
||||
fs.set_permissions("/test_file", Permissions(0o644))
|
||||
assert fs.metadata("/test_file") != md
|
||||
|
||||
fs.mkdir("/.test_dir")
|
||||
assert fs.metadata("/.test_dir").type == "dir"
|
||||
assert fs.metadata("/.test_dir").is_hidden_files
|
||||
md = fs.metadata("/.test_dir")
|
||||
assert md.type == "dir"
|
||||
assert fs.metadata("/.test_dir").is_hidden
|
||||
assert not fs.is_file("/.test_dir")
|
||||
assert fs.is_dir("/.test_dir")
|
||||
assert not fs.is_symlink("/.test_dir")
|
||||
assert not fs.is_other("/.test_dir")
|
||||
|
||||
fs.make_link("/test_link", "/link_target")
|
||||
assert fs.metadata("/test_link").type == "link"
|
||||
md = fs.metadata("/test_link")
|
||||
assert md.type == "symlink"
|
||||
assert not fs.is_file("/test_link")
|
||||
assert not fs.is_dir("/test_link")
|
||||
assert fs.is_symlink("/test_link")
|
||||
assert not fs.is_other("/test_link")
|
||||
|
||||
assert fs.metadata_or_none("/does_not_exist") is None
|
||||
with pytest.raises(FsError):
|
||||
fs.metadata("/does_not_exist")
|
||||
|
||||
|
||||
########################################################################################
|
||||
@@ -199,14 +369,14 @@ def test_iter_dir(fs: VirtualFileSystem):
|
||||
expected = [
|
||||
(PurePosixPath("/dir"), "dir"),
|
||||
(PurePosixPath("/file"), "file"),
|
||||
(PurePosixPath("/link"), "link"),
|
||||
(PurePosixPath("/link"), "symlink"),
|
||||
]
|
||||
for path, file_type in expected:
|
||||
if file_type == "dir":
|
||||
fs.mkdir(path)
|
||||
elif file_type == "file":
|
||||
fs.write_bytes(path, b"")
|
||||
elif file_type == "link":
|
||||
elif file_type == "symlink":
|
||||
fs.make_link(path, "/foobar")
|
||||
|
||||
items_metadata = sorted(fs.iter_dir("/"))
|
||||
|
||||
Reference in New Issue
Block a user