143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
"""Path-compatible access to a registered in-memory repository."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path, PosixPath
|
|
from typing import ClassVar
|
|
|
|
from typing_extensions import Buffer
|
|
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class MemoryPath(PosixPath):
|
|
"""Expose a ``Path`` API over a registered ``MemoryRepoExec`` instance."""
|
|
|
|
_repositories: ClassVar[dict[Path, MemoryRepoExec]] = {}
|
|
|
|
@classmethod
|
|
def from_repo(cls, repository: MemoryRepoExec) -> "MemoryPath":
|
|
"""Register one in-memory repository and return its root path."""
|
|
|
|
root = Path(str(repository.root))
|
|
cls._repositories[root] = repository
|
|
return cls(str(root))
|
|
|
|
@property
|
|
def repository(self) -> MemoryRepoExec:
|
|
"""Return the registered repository which owns this lexical path."""
|
|
|
|
path = Path(str(self))
|
|
candidates = [
|
|
(root, repository)
|
|
for root, repository in self._repositories.items()
|
|
if path == root or root in path.parents
|
|
]
|
|
if not candidates:
|
|
raise ValueError(f"No in-memory repository owns {self}.")
|
|
return max(candidates, key=lambda item: len(item[0].parts))[1]
|
|
|
|
def read_text(self, encoding: str | None = None, errors: str | None = None) -> str:
|
|
"""Read text from the registered in-memory repository."""
|
|
|
|
del errors
|
|
return self.repository.read_text(self, encoding=encoding or "utf-8")
|
|
|
|
def write_text(
|
|
self,
|
|
data: str,
|
|
encoding: str | None = None,
|
|
errors: str | None = None,
|
|
newline: str | None = None,
|
|
) -> int:
|
|
"""Write text to the registered in-memory repository."""
|
|
|
|
del errors
|
|
content = data if newline is None else data.replace("\n", newline)
|
|
self.repository.write_text(self, content, encoding=encoding or "utf-8")
|
|
return len(content)
|
|
|
|
def read_bytes(self) -> bytes:
|
|
"""Read bytes from the registered in-memory repository."""
|
|
|
|
return self.repository.read_bytes(self)
|
|
|
|
def write_bytes(self, data: Buffer) -> int:
|
|
"""Write bytes to the registered in-memory repository."""
|
|
|
|
content = bytes(data)
|
|
self.repository.write_bytes(self, content)
|
|
return len(content)
|
|
|
|
def exists(self, *, follow_symlinks: bool = True) -> bool:
|
|
"""Report lexical existence without consulting the host filesystem."""
|
|
|
|
del follow_symlinks
|
|
return self.repository.exists(self)
|
|
|
|
def is_file(self) -> bool:
|
|
"""Report whether this lexical path has in-memory file content."""
|
|
|
|
return self.repository.is_file(self)
|
|
|
|
def is_dir(self) -> bool:
|
|
"""Report whether this lexical path has in-memory directory state."""
|
|
|
|
return self.repository.is_dir(self)
|
|
|
|
def mkdir(
|
|
self,
|
|
mode: int = 0o777,
|
|
parents: bool = False,
|
|
exist_ok: bool = False,
|
|
) -> None:
|
|
"""Create lexical directories without touching the host filesystem."""
|
|
|
|
del mode
|
|
self.repository.mkdir(self, parents=parents, exist_ok=exist_ok)
|
|
|
|
def unlink(self, missing_ok: bool = False) -> None:
|
|
"""Remove one lexical in-memory file."""
|
|
|
|
self.repository.delete(self, missing_ok=missing_ok)
|
|
|
|
def rmdir(self) -> None:
|
|
"""Remove one lexical in-memory directory tree."""
|
|
|
|
self.repository.delete(self)
|
|
|
|
def replace(self, target: object) -> "MemoryPath":
|
|
"""Move one lexical in-memory file to its replacement target."""
|
|
|
|
replacement = Path(str(target))
|
|
self.repository.replace(self, replacement)
|
|
return MemoryPath(str(replacement))
|
|
|
|
def rename(self, target: object) -> "MemoryPath":
|
|
"""Move one lexical in-memory file to its renamed target."""
|
|
|
|
renamed = Path(str(target))
|
|
self.repository.move(self, renamed)
|
|
return MemoryPath(str(renamed))
|
|
|
|
def touch(self, mode: int = 0o666, exist_ok: bool = True) -> None:
|
|
"""Create an empty lexical file when absent."""
|
|
|
|
del mode
|
|
if self.exists() and not exist_ok:
|
|
raise FileExistsError(self)
|
|
if not self.exists():
|
|
self.write_bytes(b"")
|
|
|
|
def iterdir(self):
|
|
"""Yield direct in-memory children in deterministic order."""
|
|
|
|
yield from (MemoryPath(str(path)) for path in self.repository.list_directory(self))
|
|
|
|
def resolve(self, strict: bool = False) -> "MemoryPath":
|
|
"""Return the lexical normalized path without asking the host filesystem."""
|
|
|
|
del strict
|
|
path = Path(str(self))
|
|
normalized = path if path.is_absolute() else self.repository.root / path
|
|
return MemoryPath(str(normalized))
|