Coverage for src/debputy/filesystem_scan.py: 64%
1362 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-22 10:58 +0000
1import atexit
2import contextlib
3import dataclasses
4import errno
5import io
6import operator
7import os
8import shutil
9import stat
10import subprocess
11import tempfile
12import time
13import typing
14from abc import ABC
15from contextlib import suppress
16from typing import (
17 Optional,
18 cast,
19 Any,
20 ContextManager,
21 TextIO,
22 BinaryIO,
23 Generic,
24 TypeVar,
25 overload,
26 Literal,
27 Never,
28)
29from collections.abc import Iterable, Iterator, Mapping, Callable
30from weakref import ref, ReferenceType
32from debputy.exceptions import (
33 PureVirtualPathError,
34 DebputyFSIsROError,
35 DebputyMetadataAccessError,
36 TestPathWithNonExistentFSPathError,
37 SymlinkLoopError,
38)
39from debputy.intermediate_manifest import PathType
40from debputy.manifest_parser.base_types import (
41 ROOT_DEFINITION,
42 StaticFileSystemOwner,
43 StaticFileSystemGroup,
44)
45from debputy.plugin.api.spec import (
46 VirtualPath,
47 PathDef,
48 PathMetadataReference,
49 PMT,
50)
51from debputy.types import VP
52from debputy.util import (
53 generated_content_dir,
54 _error,
55 escape_shell,
56 assume_not_none,
57 _normalize_path,
58 _debug_log,
59)
61BY_BASENAME = operator.attrgetter("name")
63FSP = TypeVar("FSP", bound="OSFSOverlayBase", covariant=True)
64FSC = TypeVar("FSC", bound="OSFSOverlayBase", covariant=True)
67BinaryOpenMode = Literal[
68 "rb",
69 "r+b",
70 "wb",
71 "w+b",
72 "xb",
73 "ab",
74]
75TextOpenMode = Literal[
76 "r",
77 "r+",
78 "rt",
79 "r+t",
80 "w",
81 "w+",
82 "wt",
83 "w+t",
84 "x",
85 "xt",
86 "a",
87 "at",
88]
89OpenMode = Literal[BinaryOpenMode, TextOpenMode]
92class AlwaysEmptyReadOnlyMetadataReference(PathMetadataReference[PMT]):
93 __slots__ = ("_metadata_type", "_owning_plugin", "_current_plugin")
95 def __init__(
96 self,
97 owning_plugin: str,
98 current_plugin: str,
99 metadata_type: type[PMT],
100 ) -> None:
101 self._owning_plugin = owning_plugin
102 self._current_plugin = current_plugin
103 self._metadata_type = metadata_type
105 @property
106 def is_present(self) -> bool:
107 return False
109 @property
110 def can_read(self) -> bool:
111 return self._owning_plugin == self._current_plugin
113 @property
114 def can_write(self) -> bool:
115 return False
117 @property
118 def value(self) -> PMT | None:
119 if self.can_read: 119 ↛ 121line 119 didn't jump to line 121 because the condition on line 119 was always true
120 return None
121 raise DebputyMetadataAccessError(
122 f"Cannot read the metadata {self._metadata_type.__name__} owned by"
123 f" {self._owning_plugin} as the metadata has not been made"
124 f" readable to the plugin {self._current_plugin}."
125 )
127 @value.setter
128 def value(self, new_value: PMT) -> None:
129 if self._is_owner:
130 raise DebputyFSIsROError(
131 f"Cannot set the metadata {self._metadata_type.__name__} as the path is read-only"
132 )
133 raise DebputyMetadataAccessError(
134 f"Cannot set the metadata {self._metadata_type.__name__} owned by"
135 f" {self._owning_plugin} as the metadata has not been made"
136 f" read-write to the plugin {self._current_plugin}."
137 )
139 @property
140 def _is_owner(self) -> bool:
141 return self._owning_plugin == self._current_plugin
144@dataclasses.dataclass(slots=True)
145class PathMetadataValue(Generic[PMT]):
146 owning_plugin: str
147 metadata_type: type[PMT]
148 value: PMT | None = None
150 def can_read_value(self, current_plugin: str) -> bool:
151 return self.owning_plugin == current_plugin
153 def can_write_value(self, current_plugin: str) -> bool:
154 return self.owning_plugin == current_plugin
157class PathMetadataReferenceImplementation(PathMetadataReference[PMT]):
158 __slots__ = ("_owning_path", "_current_plugin", "_path_metadata_value")
160 def __init__(
161 self,
162 owning_path: "VirtualPathBase",
163 current_plugin: str,
164 path_metadata_value: PathMetadataValue[PMT],
165 ) -> None:
166 self._owning_path = owning_path
167 self._current_plugin = current_plugin
168 self._path_metadata_value = path_metadata_value
170 @property
171 def is_present(self) -> bool:
172 if not self.can_read: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 return False
174 return self._path_metadata_value.value is not None
176 @property
177 def can_read(self) -> bool:
178 return self._path_metadata_value.can_read_value(self._current_plugin)
180 @property
181 def can_write(self) -> bool:
182 if not self._path_metadata_value.can_write_value(self._current_plugin): 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 return False
184 owning_path = self._owning_path
185 return owning_path.is_read_write and not owning_path.is_detached
187 @property
188 def value(self) -> PMT | None:
189 if self.can_read: 189 ↛ 191line 189 didn't jump to line 191 because the condition on line 189 was always true
190 return self._path_metadata_value.value
191 raise DebputyMetadataAccessError(
192 f"Cannot read the metadata {self._metadata_type_name} owned by"
193 f" {self._owning_plugin} as the metadata has not been made"
194 f" readable to the plugin {self._current_plugin}."
195 )
197 @value.setter
198 def value(self, new_value: PMT) -> None:
199 if not self.can_write: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 m = "set" if new_value is not None else "delete"
201 raise DebputyMetadataAccessError(
202 f"Cannot {m} the metadata {self._metadata_type_name} owned by"
203 f" {self._owning_plugin} as the metadata has not been made"
204 f" read-write to the plugin {self._current_plugin}."
205 )
206 owning_path = self._owning_path
207 if not owning_path.is_read_write: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 raise DebputyFSIsROError(
209 f"Cannot set the metadata {self._metadata_type_name} as the path is read-only"
210 )
211 if owning_path.is_detached: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true
212 raise TypeError(
213 f"Cannot set the metadata {self._metadata_type_name} as the path is detached"
214 )
215 self._path_metadata_value.value = new_value
217 @property
218 def _is_owner(self) -> bool:
219 return self._owning_plugin == self._current_plugin
221 @property
222 def _owning_plugin(self) -> str:
223 return self._path_metadata_value.owning_plugin
225 @property
226 def _metadata_type_name(self) -> str:
227 return self._path_metadata_value.metadata_type.__name__
230def _cp_a(source: str, dest: str) -> None:
231 cmd = ["cp", "-a", source, dest]
232 try:
233 subprocess.check_call(cmd)
234 except subprocess.CalledProcessError:
235 full_command = escape_shell(*cmd)
236 _error(
237 f"The attempt to make an internal copy of {escape_shell(source)} failed. Please review the output of cp"
238 f" above to understand what went wrong. The full command was: {full_command}"
239 )
242def _split_path(path: str) -> tuple[bool, bool, list[str]]:
243 must_be_dir = True if path.endswith("/") else False
244 absolute = False
245 if path.startswith("/"):
246 absolute = True
247 path = "." + path
248 path_parts = path.rstrip("/").split("/")
249 if must_be_dir:
250 path_parts.append(".")
251 return absolute, must_be_dir, path_parts
254def _root(path: VP) -> "VirtualPathBase":
255 current = path
256 while not current.is_root_dir():
257 parent = current.parent_dir
258 assert parent is not None # type hint
259 current = parent
260 assert isinstance(current, VirtualPathBase)
261 return current
264def _check_fs_path_is_file(
265 fs_path: str,
266 unlink_on_error: Optional["VirtualPath"] = None,
267) -> None:
268 had_issue = False
269 try:
270 # FIXME: Check mode, and use the Virtual Path to cache the result as a side-effect
271 st = os.lstat(fs_path)
272 except FileNotFoundError:
273 had_issue = True
274 else:
275 if not stat.S_ISREG(st.st_mode) or st.st_nlink > 1: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 had_issue = True
277 if not had_issue: 277 ↛ 280line 277 didn't jump to line 280 because the condition on line 277 was always true
278 return
280 if unlink_on_error:
281 with suppress(FileNotFoundError):
282 os.unlink(fs_path)
283 raise TypeError(
284 "The provided FS backing file was deleted, replaced with a non-file entry or it was hard"
285 " linked to another file. The entry has been disconnected."
286 )
289class CurrentPluginContextManager:
290 __slots__ = ("_plugin_names",)
292 def __init__(self, initial_plugin_name: str) -> None:
293 self._plugin_names = [initial_plugin_name]
295 @property
296 def current_plugin_name(self) -> str:
297 return self._plugin_names[-1]
299 @contextlib.contextmanager
300 def change_plugin_context(self, new_plugin_name: str) -> Iterator[str]:
301 self._plugin_names.append(new_plugin_name)
302 yield new_plugin_name
303 self._plugin_names.pop()
306class VirtualPathBase(VirtualPath, ABC):
307 __slots__ = ()
309 def stat(self) -> os.stat_result:
310 # TODO: Remove
311 """Attempt to do stat of the underlying path (if it exists)
313 *Avoid* using `stat()` whenever possible where a more specialized attribute exist. The
314 `stat()` call returns the data from the file system and often, `debputy` does *not* track
315 its state in the file system. As an example, if you want to know the file system mode of
316 a path, please use the `mode` attribute instead.
318 This never follow symlinks (it behaves like `os.lstat`). It will raise an error
319 if the path is not backed by a file system object (that is, `has_fs_path` is False).
321 :return: The stat result or an error.
322 """
323 raise NotImplementedError()
325 @property
326 def size(self) -> int:
327 """Resolve the file size (`st_size`)
329 This may be using `stat()` and therefore `fs_path`.
331 :return: The size of the file in bytes
332 """
333 return self.stat().st_size
335 def _orphan_safe_path(self) -> str:
336 return self.path
338 def _rw_check(self) -> None:
339 if not self.is_read_write:
340 raise DebputyFSIsROError(
341 f'Attempt to write to "{self._orphan_safe_path()}" failed:'
342 " Debputy Virtual File system is R/O."
343 )
345 @property
346 def is_detached(self) -> bool:
347 raise NotImplementedError
349 def is_root_dir(self) -> bool:
350 # The root directory is never detachable in the current setup
351 return not self.is_detached and self.parent_dir is None
353 def lookup(self, path: str) -> Optional["VirtualPathBase"]:
354 match, missing = self.attempt_lookup(path)
355 if missing:
356 return None
357 return match
359 def attempt_lookup(self, path: str) -> tuple["VirtualPathBase", list[str]]:
360 if self.is_detached: 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 raise ValueError(
362 f'Cannot perform lookup via "{self._orphan_safe_path()}": The path is detached'
363 )
364 absolute, must_be_dir, path_parts = _split_path(path)
365 current = _root(self) if absolute else self
366 path_parts.reverse()
367 link_expansions = set()
368 while path_parts:
369 dir_part = path_parts.pop()
370 if dir_part == ".":
371 continue
372 if dir_part == "..":
373 if current.is_root_dir(): 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true
374 raise ValueError(f'The path "{path}" escapes the root dir')
375 p = current.parent_dir
376 assert p is not None # type hint
377 current = cast("VirtualPathBase", p)
378 continue
379 try:
380 current = cast("VirtualPathBase", current[dir_part])
381 except KeyError:
382 path_parts.append(dir_part)
383 path_parts.reverse()
384 if must_be_dir:
385 path_parts.pop()
386 return current, path_parts
387 if current.is_symlink and path_parts:
388 if current.path in link_expansions:
389 # This is our loop detection for now. It might have some false positives where you
390 # could safely resolve the same symlink twice. However, given that this use-case is
391 # basically non-existent in practice for packaging, we just stop here for now.
392 raise SymlinkLoopError(
393 f'The path "{path}" traversed the symlink "{current.path}" multiple'
394 " times. Currently, traversing the same symlink twice is considered"
395 " a loop by `debputy` even if the path would eventually resolve."
396 " Consider filing a feature request if you have a benign case that"
397 " triggers this error."
398 )
399 link_expansions.add(current.path)
400 link_target = current.readlink()
401 link_absolute, _, link_path_parts = _split_path(link_target)
402 if link_absolute:
403 current = _root(current)
404 else:
405 current = cast(
406 "VirtualPathBase", assume_not_none(current.parent_dir)
407 )
408 link_path_parts.reverse()
409 path_parts.extend(link_path_parts)
410 return current, []
412 def mkdirs(self, path: str) -> "VirtualPath":
413 current: VirtualPath
414 current, missing_parts = self.attempt_lookup(
415 f"{path}/" if not path.endswith("/") else path
416 )
417 if not current.is_dir: 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 raise ValueError(
419 f'mkdirs of "{path}" failed: This would require {current.path} to not exist OR be'
420 " a directory. However, that path exist AND is a not directory."
421 )
422 for missing_part in missing_parts:
423 assert missing_part not in (".", "..")
424 current = current.mkdir(missing_part)
425 return current
427 def prune_if_empty_dir(self) -> None:
428 """Remove this and all (now) empty parent directories
430 Same as: `rmdir --ignore-fail-on-non-empty --parents`
432 This operation may cause the path (and any of its parent directories) to become "detached"
433 and therefore unsafe to use in further operations.
434 """
435 self._rw_check()
437 if not self.is_dir: 437 ↛ 438line 437 didn't jump to line 438 because the condition on line 437 was never true
438 raise TypeError(f"{self._orphan_safe_path()} is not a directory")
439 # No-op for the root directory. There is never a case where you want to delete this directory
440 # (and even if you could, debputy will need it for technical reasons, so the root dir stays)
441 if any(self.iterdir()) or self.is_root_dir():
442 return
443 parent_dir = self.parent_dir
445 # Recursive does not matter; we already know the directory is empty.
446 self.unlink()
448 if parent_dir: 448 ↛ exitline 448 didn't return from function 'prune_if_empty_dir' because the condition on line 448 was always true
449 typing.cast(VirtualPathBase, parent_dir).prune_if_empty_dir()
451 def _current_plugin(self) -> str:
452 if self.is_detached: 452 ↛ 453line 452 didn't jump to line 453 because the condition on line 452 was never true
453 raise TypeError("Cannot resolve the current plugin; path is detached")
454 return _root(self)._current_plugin()
456 @overload
457 def open_child( 457 ↛ exitline 457 didn't return from function 'open_child' because
458 self,
459 name: str,
460 mode: TextOpenMode = "r",
461 buffering: int = -1,
462 ) -> TextIO: ...
464 @overload
465 def open_child( 465 ↛ exitline 465 didn't return from function 'open_child' because
466 self,
467 name: str,
468 mode: BinaryOpenMode,
469 buffering: int = -1,
470 ) -> BinaryIO: ...
472 @contextlib.contextmanager
473 def open_child(
474 self,
475 name: str,
476 mode: BinaryOpenMode | TextOpenMode = "r",
477 buffering: int = -1,
478 ):
479 """Open a child path of the current directory in a given mode. Usually used with a context manager
481 The path is opened according to the `mode` parameter similar to the built-in `open` in Python.
482 The following symbols are accepted with the same meaning as Python's open:
483 * `r`
484 * `w`
485 * `x`
486 * `a`
487 * `+`
488 * `b`
489 * `t`
491 Like Python's `open`, this can create a new file provided the file system is in read-write mode.
492 Though unlike Python's open, symlinks are not followed and cannot be opened. Any newly created
493 file will start with (os.stat) mode of 0o0644. The (os.stat) mode of existing paths are left
494 as-is.
496 :param name: The name of the child path to open. Must be a basename.
497 :param mode: The mode to open the file with such as `r` or `w`. See Python's `open` for more
498 examples.
499 :param buffering: Same as open(..., buffering=...) where supported. Notably during
500 testing, the content may be purely in memory and use a BytesIO/StringIO
501 (which does not accept that parameter, but then it is buffered in a different way)
502 :return: The file handle.
503 """
504 existing = self.get(name)
505 if "r" in mode: 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true
506 if existing is None:
507 raise ValueError(
508 f"Path {self.path}/{name} does not exist and mode had `r`"
509 )
510 if "+" not in mode:
511 # open(byte_io="b" in mode) matches no typed signature overload.
512 if "b" in mode:
513 with existing.open(byte_io=True, buffering=buffering) as fd:
514 yield fd
515 else:
516 with existing.open(byte_io=False, buffering=buffering) as fd:
517 yield fd
519 encoding = None if "b" in mode else "utf-8"
521 if existing: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 if "x" in mode:
523 raise ValueError(
524 f"Path {existing.path} already exists and mode had `x`"
525 )
526 with (
527 existing.replace_fs_path_content() as fs_path,
528 open(fs_path, mode, encoding=encoding) as fd,
529 ):
530 yield fd
531 else:
532 assert "r" not in mode
533 # unlink_if_exists=False as a precaution (the "already exists" should not end up here).
534 with (
535 self.add_file(name, mode=0o0644, unlink_if_exists=False) as new_path,
536 open(new_path.fs_path, mode, encoding=encoding) as fd,
537 ):
538 yield fd
541class InMemoryVirtualPathBase(VirtualPathBase, ABC):
542 __slots__ = (
543 "_basename",
544 "_parent_dir",
545 "_path_cache",
546 "_parent_path_cache",
547 "_last_known_parent_path",
548 "_mode",
549 "_owner",
550 "_group",
551 "_mtime",
552 "_stat_cache",
553 "_metadata",
554 "__weakref__",
555 )
557 def __init__(
558 self,
559 basename: str,
560 parent: Optional["InMemoryVirtualPathBase"],
561 initial_mode: int | None = None,
562 mtime: float | None = None,
563 stat_cache: os.stat_result | None = None,
564 ) -> None:
565 self._basename = basename
566 self._path_cache: str | None = None
567 self._parent_path_cache: str | None = None
568 self._last_known_parent_path: str | None = None
569 self._mode = initial_mode
570 self._mtime = mtime
571 self._stat_cache = stat_cache
572 self._metadata: dict[tuple[str, type[Any]], PathMetadataValue[Any]] = {}
573 # The `_owner` and `_group` is directly access outside the class via `tar_owner_info`
574 self._owner = ROOT_DEFINITION
575 self._group = ROOT_DEFINITION
577 # The self._parent_dir = None is to create `_parent_dir` because the parent_dir setter calls
578 # is_orphaned, which assumes self._parent_dir is an attribute.
579 self._parent_dir: ReferenceType["InMemoryVirtualPathBase"] | None = None
580 if parent is not None:
581 self.parent_dir = parent
583 @property
584 def name(self) -> str:
585 return self._basename
587 @name.setter
588 def name(self, new_name: str) -> None:
589 self._rw_check()
590 if new_name == self._basename: 590 ↛ 591line 590 didn't jump to line 591 because the condition on line 590 was never true
591 return
592 if self.is_detached: 592 ↛ 593line 592 didn't jump to line 593 because the condition on line 592 was never true
593 self._basename = new_name
594 return
595 self._rw_check()
596 parent = self.parent_dir
597 # This little parent_dir dance ensures the parent dir detects the rename properly
598 self.parent_dir = None
599 self._basename = new_name
600 self.parent_dir = parent
602 # Overridden for new return type
603 def iterdir(self) -> Iterable["InMemoryVirtualPathBase"]:
604 raise NotImplementedError
606 def all_paths(self) -> Iterable["InMemoryVirtualPathBase"]:
607 yield self
608 if not self.is_dir:
609 return
610 by_basename = BY_BASENAME
611 stack = sorted(self.iterdir(), key=by_basename, reverse=True)
612 while stack:
613 current = stack.pop()
614 yield current
615 if current.is_dir and not current.is_detached:
616 stack.extend(sorted(current.iterdir(), key=by_basename, reverse=True))
618 def walk(
619 self,
620 ) -> Iterable[tuple["InMemoryVirtualPathBase", list["InMemoryVirtualPathBase"]]]:
621 # FIXME: can this be more "os.walk"-like without making it harder to implement?
622 if not self.is_dir: 622 ↛ 623line 622 didn't jump to line 623 because the condition on line 622 was never true
623 yield self, []
624 return
625 by_basename = BY_BASENAME
626 stack = [self]
627 while stack:
628 current = stack.pop()
629 children = sorted(current.iterdir(), key=by_basename)
630 assert not children or current.is_dir
631 yield current, children
632 # Removing the directory counts as discarding the children.
633 if not current.is_detached: 633 ↛ 627line 633 didn't jump to line 627 because the condition on line 633 was always true
634 stack.extend(reversed(children))
636 def _orphan_safe_path(self) -> str:
637 if not self.is_detached or self._last_known_parent_path is not None: 637 ↛ 639line 637 didn't jump to line 639 because the condition on line 637 was always true
638 return self.path
639 return f"<orphaned>/{self.name}"
641 @property
642 def is_detached(self) -> bool:
643 parent = self._parent_dir
644 if parent is None:
645 return True
646 resolved_parent = parent()
647 if resolved_parent is None: 647 ↛ 648line 647 didn't jump to line 648 because the condition on line 647 was never true
648 return True
649 return resolved_parent.is_detached
651 # The __getitem__ behaves like __getitem__ from Dict but __iter__ would ideally work like a Sequence.
652 # However, that does not feel compatible, so lets force people to use .children instead for the Sequence
653 # behavior to avoid surprises for now.
654 # (Maybe it is a non-issue, but it is easier to add the API later than to remove it once we have committed
655 # to using it)
656 __iter__ = None
658 # Overridden for new return type
659 def __getitem__(self, key: object) -> "InMemoryVirtualPathBase":
660 raise NotImplementedError
662 # Overridden for new return type
663 def get(self, key: str) -> "InMemoryVirtualPathBase | None":
664 return typing.cast("InMemoryVirtualPathBase | None", super().get(key))
666 def _add_child(self, child: "InMemoryVirtualPathBase") -> None:
667 raise TypeError(
668 f"{self._orphan_safe_path()!r} is not a directory (or did not implement this method)"
669 )
671 def _remove_child(self, child: "InMemoryVirtualPathBase") -> None:
672 raise TypeError(
673 f"{self._orphan_safe_path()!r} is not a directory (or did not implement this method)"
674 )
676 @property
677 def path(self) -> str:
678 parent_path = self.parent_dir_path
679 if (
680 self._parent_path_cache is not None
681 and self._parent_path_cache == parent_path
682 ):
683 return assume_not_none(self._path_cache)
684 if parent_path is None: 684 ↛ 685line 684 didn't jump to line 685 because the condition on line 684 was never true
685 raise ReferenceError(
686 f"The path {self.name} is detached! {self.__class__.__name__}"
687 )
688 self._parent_path_cache = parent_path
689 ret = os.path.join(parent_path, self.name)
690 self._path_cache = ret
691 return ret
693 @property
694 def parent_dir(self) -> Optional["InMemoryVirtualPathBase"]:
695 p_ref = self._parent_dir
696 p = p_ref() if p_ref is not None else None
697 if p is None: 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true
698 raise ReferenceError(
699 f"The path {self.name} is detached! {self.__class__.__name__}"
700 )
701 return p
703 @parent_dir.setter
704 def parent_dir(self, new_parent: Optional["InMemoryVirtualPathBase"]) -> None:
705 self._rw_check()
706 if new_parent is not None:
707 if not new_parent.is_dir: 707 ↛ 708line 707 didn't jump to line 708 because the condition on line 707 was never true
708 raise ValueError(
709 f"The parent {new_parent._orphan_safe_path()!r} must be a directory"
710 )
711 new_parent._rw_check()
712 old_parent = None
713 self._last_known_parent_path = None
714 if not self.is_detached:
715 old_parent = self.parent_dir
716 assume_not_none(old_parent)._remove_child(self)
717 if new_parent is not None:
718 self._parent_dir = ref(new_parent)
719 new_parent._add_child(self)
720 else:
721 if old_parent is not None and not old_parent.is_detached: 721 ↛ 723line 721 didn't jump to line 723 because the condition on line 721 was always true
722 self._last_known_parent_path = old_parent.path
723 self._parent_dir = None
724 self._parent_path_cache = None
726 @property
727 def parent_dir_path(self) -> str | None:
728 if self.is_detached: 728 ↛ 729line 728 didn't jump to line 729 because the condition on line 728 was never true
729 return self._last_known_parent_path
730 return assume_not_none(self.parent_dir).path
732 def chown(
733 self,
734 owner: StaticFileSystemOwner | None,
735 group: StaticFileSystemGroup | None,
736 ) -> None:
737 """Change the owner/group of this path
739 :param owner: The desired owner definition for this path. If None, then no change of owner is performed.
740 :param group: The desired group definition for this path. If None, then no change of group is performed.
741 """
742 self._rw_check()
744 if owner is not None:
745 self._owner = owner.ownership_definition
746 if group is not None:
747 self._group = group.ownership_definition
749 def stat(self) -> os.stat_result:
750 st = self._stat_cache
751 if st is None:
752 st = self._uncached_stat()
753 self._stat_cache = st
754 return st
756 def _uncached_stat(self) -> os.stat_result:
757 raise NotImplementedError
759 @property
760 def mode(self) -> int:
761 current_mode = self._mode
762 if current_mode is None: 762 ↛ 763line 762 didn't jump to line 763 because the condition on line 762 was never true
763 current_mode = stat.S_IMODE(self.stat().st_mode)
764 self._mode = current_mode
765 return current_mode
767 @mode.setter
768 def mode(self, new_mode: int) -> None:
769 self._rw_check()
770 min_bit = 0o700 if self.is_dir else 0o400
771 if (new_mode & min_bit) != min_bit: 771 ↛ 772line 771 didn't jump to line 772 because the condition on line 771 was never true
772 omode = oct(new_mode)[2:]
773 omin = oct(min_bit)[2:]
774 raise ValueError(
775 f'Attempt to set mode of path "{self._orphan_safe_path()}" to {omode} rejected;'
776 f" Minimum requirements are {omin} (read-bit and, for dirs, exec bit for user)."
777 " There are no paths that do not need these requirements met and they can cause"
778 " problems during build or on the final system."
779 )
780 self._mode = new_mode
782 def _ensure_min_mode(self) -> None:
783 min_bit = 0o700 if self.is_dir else 0o600
784 if self.has_fs_path and (self.mode & 0o600) != 0o600: 784 ↛ 785line 784 didn't jump to line 785 because the condition on line 784 was never true
785 try:
786 fs_path = self.fs_path
787 except TestPathWithNonExistentFSPathError:
788 pass
789 else:
790 st = os.stat(fs_path)
791 new_fs_mode = stat.S_IMODE(st.st_mode) | min_bit
792 _debug_log(
793 f"Applying chmod {oct(min_bit)[2:]} {fs_path} ({self.path}) to avoid problems down the line"
794 )
795 os.chmod(fs_path, new_fs_mode)
796 self.mode |= min_bit
798 def _resolve_initial_mtime(self) -> float:
799 raise NotImplementedError
801 @property
802 def mtime(self) -> float:
803 mtime = self._mtime
804 if mtime is None:
805 mtime = self._resolve_initial_mtime()
806 self._mtime = mtime
807 return mtime
809 @mtime.setter
810 def mtime(self, new_mtime: float) -> None:
811 self._rw_check()
812 self._mtime = new_mtime
814 @property
815 def _can_replace_inline(self) -> bool:
816 return False
818 @contextlib.contextmanager
819 def add_file(
820 self,
821 name: str,
822 *,
823 unlink_if_exists: bool = True,
824 use_fs_path_mode: bool = False,
825 mode: int = 0o0644,
826 mtime: float | None = None,
827 # Special-case parameters that are not exposed in the API
828 fs_basename_matters: bool = False,
829 subdir_key: str | None = None,
830 ) -> Iterator["InMemoryVirtualPathBase"]:
831 if "/" in name or name in {".", ".."}: 831 ↛ 832line 831 didn't jump to line 832 because the condition on line 831 was never true
832 raise ValueError(f'Invalid file name: "{name}"')
833 if not self.is_dir: 833 ↛ 834line 833 didn't jump to line 834 because the condition on line 833 was never true
834 raise TypeError(
835 f"Cannot create {self._orphan_safe_path()}/{name}:"
836 f" {self._orphan_safe_path()} is not a directory"
837 )
838 self._rw_check()
839 existing = self.get(name)
840 if existing is not None: 840 ↛ 841line 840 didn't jump to line 841 because the condition on line 840 was never true
841 if not unlink_if_exists:
842 raise ValueError(
843 f'The path "{self._orphan_safe_path()}" already contains a file called "{name}"'
844 f" and exist_ok was False"
845 )
846 existing.unlink(recursive=False)
848 if fs_basename_matters and subdir_key is None: 848 ↛ 849line 848 didn't jump to line 849 because the condition on line 848 was never true
849 raise ValueError(
850 "When fs_basename_matters is True, a subdir_key must be provided"
851 )
853 directory = generated_content_dir(subdir_key=subdir_key)
855 if fs_basename_matters: 855 ↛ 856line 855 didn't jump to line 856 because the condition on line 855 was never true
856 fs_path = os.path.join(directory, name)
857 with open(fs_path, "xb") as _:
858 # Ensure that the fs_path exists
859 pass
860 child = FSBackedFilePath(
861 name,
862 self,
863 fs_path,
864 replaceable_inline=True,
865 mtime=mtime,
866 )
867 yield child
868 else:
869 with tempfile.NamedTemporaryFile(
870 dir=directory, suffix=f"__{name}", delete=False
871 ) as fd:
872 fs_path = fd.name
873 child = FSBackedFilePath(
874 name,
875 self,
876 fs_path,
877 replaceable_inline=True,
878 mtime=mtime,
879 )
880 fd.close()
881 yield child
883 if use_fs_path_mode: 883 ↛ 885line 883 didn't jump to line 885 because the condition on line 883 was never true
884 # Ensure the caller can see the current mode
885 os.chmod(fs_path, mode)
886 _check_fs_path_is_file(fs_path, unlink_on_error=child)
887 child._reset_caches()
888 if not use_fs_path_mode: 888 ↛ exitline 888 didn't return from function 'add_file' because the condition on line 888 was always true
889 child.mode = mode
891 def insert_file_from_fs_path(
892 self,
893 name: str,
894 fs_path: str,
895 *,
896 exist_ok: bool = True,
897 use_fs_path_mode: bool = False,
898 mode: int = 0o0644,
899 require_copy_on_write: bool = True,
900 follow_symlinks: bool = True,
901 reference_path: VirtualPath | None = None,
902 ) -> "InMemoryVirtualPathBase":
903 if "/" in name or name in {".", ".."}: 903 ↛ 904line 903 didn't jump to line 904 because the condition on line 903 was never true
904 raise ValueError(f'Invalid file name: "{name}"')
905 if not self.is_dir: 905 ↛ 906line 905 didn't jump to line 906 because the condition on line 905 was never true
906 raise TypeError(
907 f"Cannot create {self._orphan_safe_path()}/{name}:"
908 f" {self._orphan_safe_path()} is not a directory"
909 )
910 self._rw_check()
911 if name in self and not exist_ok: 911 ↛ 912line 911 didn't jump to line 912 because the condition on line 911 was never true
912 raise ValueError(
913 f'The path "{self._orphan_safe_path()}" already contains a file called "{name}"'
914 f" and exist_ok was False"
915 )
916 new_fs_path = fs_path
917 if follow_symlinks:
918 if reference_path is not None: 918 ↛ 919line 918 didn't jump to line 919 because the condition on line 918 was never true
919 raise ValueError(
920 "The reference_path cannot be used with follow_symlinks"
921 )
922 new_fs_path = os.path.realpath(new_fs_path, strict=True)
924 fmode: int | None = mode
925 if use_fs_path_mode:
926 fmode = None
928 st = None
929 if reference_path is None:
930 st = os.lstat(new_fs_path)
931 if stat.S_ISDIR(st.st_mode): 931 ↛ 932line 931 didn't jump to line 932 because the condition on line 931 was never true
932 raise ValueError(
933 f'The provided path "{fs_path}" is a directory. However, this'
934 " method does not support directories"
935 )
937 if not stat.S_ISREG(st.st_mode): 937 ↛ 938line 937 didn't jump to line 938 because the condition on line 937 was never true
938 if follow_symlinks:
939 raise ValueError(
940 f"The resolved fs_path ({new_fs_path}) was not a file."
941 )
942 raise ValueError(f"The provided fs_path ({fs_path}) was not a file.")
943 return FSBackedFilePath(
944 name,
945 self,
946 new_fs_path,
947 initial_mode=fmode,
948 stat_cache=st,
949 replaceable_inline=not require_copy_on_write,
950 reference_path=reference_path,
951 )
953 def add_symlink(
954 self,
955 link_name: str,
956 link_target: str,
957 *,
958 reference_path: VirtualPath | None = None,
959 ) -> "InMemoryVirtualPathBase":
960 if "/" in link_name or link_name in {".", ".."}: 960 ↛ 961line 960 didn't jump to line 961 because the condition on line 960 was never true
961 raise ValueError(
962 f'Invalid file name: "{link_name}" (it must be a valid basename)'
963 )
964 if not self.is_dir: 964 ↛ 965line 964 didn't jump to line 965 because the condition on line 964 was never true
965 raise TypeError(
966 f"Cannot create {self._orphan_safe_path()}/{link_name}:"
967 f" {self._orphan_safe_path()} is not a directory"
968 )
969 self._rw_check()
971 existing = self.get(link_name)
972 if existing: 972 ↛ 974line 972 didn't jump to line 974 because the condition on line 972 was never true
973 # Emulate ln -sf with attempts a non-recursive unlink first.
974 existing.unlink(recursive=False)
976 return SymlinkVirtualPath(
977 link_name,
978 self,
979 link_target,
980 reference_path=reference_path,
981 )
983 def mkdir(
984 self,
985 name: str,
986 *,
987 reference_path: VirtualPath | None = None,
988 ) -> "InMemoryVirtualPathBase":
989 if "/" in name or name in {".", ".."}: 989 ↛ 990line 989 didn't jump to line 990 because the condition on line 989 was never true
990 raise ValueError(
991 f'Invalid file name: "{name}" (it must be a valid basename)'
992 )
993 if not self.is_dir: 993 ↛ 994line 993 didn't jump to line 994 because the condition on line 993 was never true
994 raise TypeError(
995 f"Cannot create {self._orphan_safe_path()}/{name}:"
996 f" {self._orphan_safe_path()} is not a directory"
997 )
998 if reference_path is not None and not reference_path.is_dir: 998 ↛ 999line 998 didn't jump to line 999 because the condition on line 998 was never true
999 raise ValueError(
1000 f'The provided fs_path "{reference_path.fs_path}" exist but it is not a directory!'
1001 )
1002 self._rw_check()
1004 existing = self.get(name)
1005 if existing: 1005 ↛ 1006line 1005 didn't jump to line 1006 because the condition on line 1005 was never true
1006 raise ValueError(f"Path {existing.path} already exist")
1007 return VirtualDirectoryFSPath(name, self, reference_path=reference_path)
1009 def mkdirs(self, path: str) -> "InMemoryVirtualPathBase":
1010 return cast("InMemoryVirtualPathBase", super().mkdirs(path))
1012 @property
1013 def is_read_write(self) -> bool:
1014 """When true, the file system entry may be mutated
1016 :return: Whether file system mutations are permitted.
1017 """
1018 if self.is_detached:
1019 return True
1020 return assume_not_none(self.parent_dir).is_read_write
1022 def unlink(self, *, recursive: bool = False) -> None:
1023 """Unlink a file or a directory
1025 This operation will detach the path from the file system (causing "is_detached" to return True).
1027 Note that the root directory cannot be deleted.
1029 :param recursive: If True, then non-empty directories will be unlinked as well removing everything inside them
1030 as well. When False, an error is raised if the path is a non-empty directory
1031 """
1032 if self.is_detached: 1032 ↛ 1033line 1032 didn't jump to line 1033 because the condition on line 1032 was never true
1033 return
1034 if not recursive and self.is_dir and any(self.iterdir()): 1034 ↛ 1035line 1034 didn't jump to line 1035 because the condition on line 1034 was never true
1035 raise ValueError(
1036 f'Refusing to unlink "{self.path}": The directory was not empty and recursive was False'
1037 )
1038 # The .parent_dir setter does a _rw_check() for us.
1039 self.parent_dir = None
1041 def _reset_caches(self) -> None:
1042 self._mtime = None
1043 self._stat_cache = None
1045 def metadata(
1046 self,
1047 metadata_type: type[PMT],
1048 *,
1049 owning_plugin: str | None = None,
1050 ) -> PathMetadataReference[PMT]:
1051 current_plugin = self._current_plugin()
1052 if owning_plugin is None: 1052 ↛ 1054line 1052 didn't jump to line 1054 because the condition on line 1052 was always true
1053 owning_plugin = current_plugin
1054 metadata_key = (owning_plugin, metadata_type)
1055 metadata_value = self._metadata.get(metadata_key)
1056 if metadata_value is None:
1057 if self.is_detached: 1057 ↛ 1058line 1057 didn't jump to line 1058 because the condition on line 1057 was never true
1058 raise TypeError(
1059 f"Cannot access the metadata {metadata_type.__name__}: The path is detached."
1060 )
1061 if not self.is_read_write:
1062 return AlwaysEmptyReadOnlyMetadataReference(
1063 owning_plugin,
1064 current_plugin,
1065 metadata_type,
1066 )
1067 metadata_value = PathMetadataValue(owning_plugin, metadata_type)
1068 self._metadata[metadata_key] = metadata_value
1069 return PathMetadataReferenceImplementation(
1070 self,
1071 current_plugin,
1072 metadata_value,
1073 )
1075 @contextlib.contextmanager
1076 def replace_fs_path_content(
1077 self,
1078 *,
1079 use_fs_path_mode: bool = False,
1080 ) -> Iterator[str]:
1081 if not self.is_file: 1081 ↛ 1082line 1081 didn't jump to line 1082 because the condition on line 1081 was never true
1082 raise TypeError(
1083 f'Cannot replace contents of "{self._orphan_safe_path()}" as it is not a file'
1084 )
1085 self._rw_check()
1086 fs_path = self.fs_path
1087 if not self._can_replace_inline: 1087 ↛ 1099line 1087 didn't jump to line 1099 because the condition on line 1087 was always true
1088 fs_path = self.fs_path
1089 directory = generated_content_dir()
1090 with tempfile.NamedTemporaryFile(
1091 dir=directory, suffix=f"__{self.name}", delete=False
1092 ) as new_path_fd:
1093 new_path_fd.close()
1094 _cp_a(fs_path, new_path_fd.name)
1095 fs_path = new_path_fd.name
1096 self._replaced_path(fs_path)
1097 assert self.fs_path == fs_path
1099 current_mtime = self._mtime
1100 if current_mtime is not None:
1101 os.utime(fs_path, (current_mtime, current_mtime))
1103 current_mode = self.mode
1104 yield fs_path
1105 _check_fs_path_is_file(fs_path, unlink_on_error=self)
1106 if not use_fs_path_mode: 1106 ↛ 1108line 1106 didn't jump to line 1108 because the condition on line 1106 was always true
1107 os.chmod(fs_path, current_mode)
1108 self._reset_caches()
1110 def _replaced_path(self, new_fs_path: str) -> None:
1111 raise NotImplementedError
1114class InMemoryVirtualPath(InMemoryVirtualPathBase, ABC):
1115 __slots__ = ("_children",)
1117 def __init__(
1118 self,
1119 basename: str,
1120 parent: Optional["InMemoryVirtualPathBase"],
1121 children: dict[str, "InMemoryVirtualPathBase"] | None = None,
1122 initial_mode: int | None = None,
1123 mtime: float | None = None,
1124 stat_cache: os.stat_result | None = None,
1125 ) -> None:
1126 super().__init__(
1127 basename,
1128 parent,
1129 initial_mode,
1130 mtime,
1131 stat_cache,
1132 )
1133 self._children = children
1135 def __repr__(self) -> str:
1136 return (
1137 f"{self.__class__.__name__}({self._orphan_safe_path()!r},"
1138 f" is_file={self.is_file},"
1139 f" is_dir={self.is_dir},"
1140 f" is_symlink={self.is_symlink},"
1141 f" has_fs_path={self.has_fs_path},"
1142 f" children_len={len(self._children) if self._children else 0})"
1143 )
1145 def iterdir(self) -> Iterable["InMemoryVirtualPathBase"]:
1146 if self._children is not None:
1147 yield from self._children.values()
1149 def __getitem__(self, key) -> "InMemoryVirtualPathBase":
1150 if self._children is None:
1151 raise KeyError(
1152 f"{key} (note: {self._orphan_safe_path()!r} has no children)"
1153 )
1154 if isinstance(key, InMemoryVirtualPathBase): 1154 ↛ 1155line 1154 didn't jump to line 1155 because the condition on line 1154 was never true
1155 key = key.name
1156 return self._children[key]
1158 def __delitem__(self, key) -> None:
1159 self._rw_check()
1160 children = self._children
1161 if children is None: 1161 ↛ 1162line 1161 didn't jump to line 1162 because the condition on line 1161 was never true
1162 raise KeyError(key)
1163 del children[key]
1165 def _remove_child(self, child: "InMemoryVirtualPathBase") -> None:
1166 children = assume_not_none(self._children)
1167 del children[child.name]
1169 def _add_child(self, child: "InMemoryVirtualPathBase") -> None:
1170 self._rw_check()
1171 if not self.is_dir: 1171 ↛ 1172line 1171 didn't jump to line 1172 because the condition on line 1171 was never true
1172 raise TypeError(f"{self._orphan_safe_path()!r} is not a directory")
1173 if self._children is None:
1174 self._children = {}
1176 conflict_child = self.get(child.name)
1177 if conflict_child is not None: 1177 ↛ 1178line 1177 didn't jump to line 1178 because the condition on line 1177 was never true
1178 conflict_child.unlink(recursive=True)
1179 self._children[child.name] = child
1181 def _uncached_stat(self) -> os.stat_result:
1182 return os.lstat(self.fs_path)
1184 def _resolve_initial_mtime(self) -> float:
1185 return self.stat().st_mtime
1188class InMemoryOverlayFSDirectory(InMemoryVirtualPathBase):
1189 __slots__ = (
1190 "_underlying_directory",
1191 "_virtual_children",
1192 "_deleted_children",
1193 "_has_underlying_read_children",
1194 )
1196 def __init__(
1197 self,
1198 underlying_directory: VirtualPath,
1199 basename: str,
1200 parent: Optional["InMemoryVirtualPathBase"],
1201 initial_mode: int | None = None,
1202 mtime: float | None = None,
1203 stat_cache: os.stat_result | None = None,
1204 ) -> None:
1205 super().__init__(
1206 basename,
1207 parent,
1208 initial_mode,
1209 mtime,
1210 stat_cache,
1211 )
1212 self._underlying_directory: VirtualPath = underlying_directory
1213 self._virtual_children: dict[str, InMemoryVirtualPathBase] = {}
1214 self._deleted_children = set[str]()
1215 self._has_underlying_read_children = False
1217 def __getitem__(self, key: object) -> "InMemoryVirtualPathBase":
1218 if not isinstance(key, str) or key in self._deleted_children:
1219 raise KeyError(key)
1220 if c := self._virtual_children.get(key):
1221 return c
1222 if not self._ensure_read_underlying_children():
1223 raise KeyError(key)
1224 return self._virtual_children[key]
1226 def _ensure_read_underlying_children(self) -> bool:
1227 if self._has_underlying_read_children:
1228 return False
1229 for real_child in self._underlying_directory.iterdir():
1230 basename = real_child.name
1231 if basename in self._virtual_children or basename in self._deleted_children:
1232 continue
1233 if real_child.is_dir:
1234 child = InMemoryOverlayFSDirectory(
1235 real_child,
1236 basename,
1237 self,
1238 )
1239 elif real_child.is_symlink:
1240 child = SymlinkVirtualPath(
1241 basename,
1242 self,
1243 real_child.readlink(),
1244 reference_path=real_child,
1245 )
1246 else:
1247 assert real_child.is_file and real_child.has_fs_path
1248 child = FSBackedFilePath(
1249 basename,
1250 self,
1251 real_child.fs_path,
1252 reference_path=real_child,
1253 )
1254 self._virtual_children[child.name] = child
1255 return True
1257 def _add_child(self, child: "InMemoryVirtualPathBase") -> None:
1258 self._virtual_children[child.name] = child
1260 def _remove_child(self, child: "InMemoryVirtualPathBase") -> None:
1261 if child.name in self._deleted_children:
1262 raise KeyError(child.name)
1263 try:
1264 del self._virtual_children[child.name]
1265 deleted = True
1266 except KeyError:
1267 deleted = False
1268 if not deleted and child.name not in self._underlying_directory:
1269 raise KeyError(child.name)
1270 self._deleted_children.add(child.name)
1272 def _uncached_stat(self) -> os.stat_result:
1273 try:
1274 return self._underlying_directory.stat() # type: ignore
1275 except AttributeError:
1276 raise PureVirtualPathError("The stat method has not been implemented")
1278 def _resolve_initial_mtime(self) -> float:
1279 return self._underlying_directory.mtime
1282class InMemoryOverlayFSRootDirectory(InMemoryOverlayFSDirectory):
1283 __slots__ = ("_fs_path", "_fs_read_write", "_plugin_context")
1285 def __init__(self, underlying_directory: VirtualPath) -> None:
1286 self._fs_read_write = True
1287 super().__init__(
1288 underlying_directory,
1289 ".",
1290 None,
1291 initial_mode=0o755,
1292 )
1293 self._plugin_context = CurrentPluginContextManager("debputy")
1295 def is_root_dir(self) -> bool:
1296 return True
1298 @property
1299 def is_detached(self) -> bool:
1300 return False
1302 def unlink(self, *, recursive: bool = False) -> None:
1303 # There is never a case where you want to delete this directory (and even if you could,
1304 # debputy will need it for technical reasons, so the root dir stays)
1305 raise TypeError("Cannot delete the root directory")
1307 def _current_plugin(self) -> str:
1308 return self._plugin_context.current_plugin_name
1310 @contextlib.contextmanager
1311 def change_plugin_context(self, new_plugin: str) -> Iterator[str]:
1312 with self._plugin_context.change_plugin_context(new_plugin) as r:
1313 yield r
1316class VirtualFSPathBase(InMemoryVirtualPath, ABC):
1317 __slots__ = ()
1319 def __init__(
1320 self,
1321 basename: str,
1322 parent: Optional["InMemoryVirtualPathBase"],
1323 children: dict[str, "InMemoryVirtualPathBase"] | None = None,
1324 initial_mode: int | None = None,
1325 mtime: float | None = None,
1326 stat_cache: os.stat_result | None = None,
1327 ) -> None:
1328 super().__init__(
1329 basename,
1330 parent,
1331 children,
1332 initial_mode=initial_mode,
1333 mtime=mtime,
1334 stat_cache=stat_cache,
1335 )
1337 def _resolve_initial_mtime(self) -> float:
1338 return time.time()
1340 @property
1341 def has_fs_path(self) -> bool:
1342 return False
1344 def stat(self) -> os.stat_result:
1345 if not self.has_fs_path:
1346 raise PureVirtualPathError(
1347 "stat() is only applicable to paths backed by the file system. The path"
1348 f" {self._orphan_safe_path()!r} is purely virtual"
1349 )
1350 return super().stat()
1352 @property
1353 def fs_path(self) -> str:
1354 if not self.has_fs_path:
1355 raise PureVirtualPathError(
1356 "fs_path is only applicable to paths backed by the file system. The path"
1357 f" {self._orphan_safe_path()!r} is purely virtual"
1358 )
1359 return self.fs_path
1362class InMemoryVirtualRootDir(InMemoryVirtualPath):
1363 __slots__ = ("_fs_path", "_fs_read_write", "_plugin_context")
1365 def __init__(self, fs_path: str | None = None) -> None:
1366 self._fs_path = fs_path
1367 self._fs_read_write = True
1368 super().__init__(
1369 ".",
1370 None,
1371 children={},
1372 initial_mode=0o755,
1373 )
1374 self._plugin_context = CurrentPluginContextManager("debputy")
1376 @property
1377 def is_detached(self) -> bool:
1378 return False
1380 def _orphan_safe_path(self) -> str:
1381 return self.name
1383 @property
1384 def path(self) -> str:
1385 return self.name
1387 @property
1388 def parent_dir(self) -> Optional["InMemoryVirtualPathBase"]:
1389 return None
1391 @parent_dir.setter
1392 def parent_dir(self, new_parent: InMemoryVirtualPathBase | None) -> None:
1393 if new_parent is not None:
1394 raise ValueError("The root directory cannot become a non-root directory")
1396 @property
1397 def parent_dir_path(self) -> str | None:
1398 return None
1400 @property
1401 def is_dir(self) -> bool:
1402 return True
1404 @property
1405 def is_file(self) -> bool:
1406 return False
1408 @property
1409 def is_symlink(self) -> bool:
1410 return False
1412 def readlink(self) -> str:
1413 raise TypeError(f'"{self._orphan_safe_path()!r}" is a directory; not a symlink')
1415 @property
1416 def has_fs_path(self) -> bool:
1417 return self._fs_path is not None
1419 def stat(self) -> os.stat_result:
1420 if not self.has_fs_path:
1421 raise PureVirtualPathError(
1422 "stat() is only applicable to paths backed by the file system. The path"
1423 f" {self._orphan_safe_path()!r} is purely virtual"
1424 )
1425 return os.stat(self.fs_path)
1427 @property
1428 def fs_path(self) -> str:
1429 if not self.has_fs_path: 1429 ↛ 1430line 1429 didn't jump to line 1430 because the condition on line 1429 was never true
1430 raise PureVirtualPathError(
1431 "fs_path is only applicable to paths backed by the file system. The path"
1432 f" {self._orphan_safe_path()!r} is purely virtual"
1433 )
1434 return assume_not_none(self._fs_path)
1436 @property
1437 def is_read_write(self) -> bool:
1438 return self._fs_read_write
1440 @is_read_write.setter
1441 def is_read_write(self, new_value: bool) -> None:
1442 self._fs_read_write = new_value
1444 def unlink(self, *, recursive: bool = False) -> None:
1445 # There is never a case where you want to delete this directory (and even if you could,
1446 # debputy will need it for technical reasons, so the root dir stays)
1447 raise TypeError("Cannot delete the root directory")
1449 def _current_plugin(self) -> str:
1450 return self._plugin_context.current_plugin_name
1452 @contextlib.contextmanager
1453 def change_plugin_context(self, new_plugin: str) -> Iterator[str]:
1454 with self._plugin_context.change_plugin_context(new_plugin) as r:
1455 yield r
1458class VirtualPathWithReference(VirtualFSPathBase, ABC):
1459 __slots__ = ("_reference_path",)
1461 def __init__(
1462 self,
1463 basename: str,
1464 parent: InMemoryVirtualPathBase,
1465 *,
1466 default_mode: int,
1467 reference_path: VirtualPath | None = None,
1468 ) -> None:
1469 super().__init__(
1470 basename,
1471 parent=parent,
1472 initial_mode=reference_path.mode if reference_path else default_mode,
1473 )
1474 self._reference_path = reference_path
1476 @property
1477 def has_fs_path(self) -> bool:
1478 ref_path = self._reference_path
1479 return ref_path is not None and ref_path.has_fs_path
1481 def _resolve_initial_mtime(self) -> float:
1482 ref_path = self._reference_path
1483 if ref_path: 1483 ↛ 1485line 1483 didn't jump to line 1485 because the condition on line 1483 was always true
1484 return ref_path.mtime
1485 return super()._resolve_initial_mtime()
1487 @property
1488 def fs_path(self) -> str:
1489 ref_path = self._reference_path
1490 if ref_path is not None and ( 1490 ↛ 1494line 1490 didn't jump to line 1494 because the condition on line 1490 was always true
1491 not super().has_fs_path or super().fs_path == ref_path.fs_path
1492 ):
1493 return ref_path.fs_path
1494 return super().fs_path
1496 def stat(self) -> os.stat_result:
1497 ref_path = self._reference_path
1498 if ref_path is not None and (
1499 not super().has_fs_path or super().fs_path == ref_path.fs_path
1500 ):
1501 return typing.cast(VirtualPathBase, ref_path).stat()
1502 return super().stat()
1504 @overload
1505 def open( 1505 ↛ exitline 1505 didn't return from function 'open' because
1506 self,
1507 *,
1508 byte_io: Literal[False] = False,
1509 buffering: int = -1,
1510 ) -> TextIO: ...
1512 @overload
1513 def open( 1513 ↛ exitline 1513 didn't return from function 'open' because
1514 self,
1515 *,
1516 byte_io: Literal[True],
1517 buffering: Literal[0] = ...,
1518 ) -> io.FileIO: ...
1520 @overload
1521 def open( 1521 ↛ exitline 1521 didn't return from function 'open' because
1522 self,
1523 *,
1524 byte_io: Literal[True],
1525 buffering: int = -1,
1526 ) -> io.BufferedReader: ...
1528 def open(self, *, byte_io=False, buffering=-1):
1529 reference_path = self._reference_path
1530 if reference_path is not None and reference_path.fs_path == self.fs_path:
1531 return reference_path.open(byte_io=byte_io, buffering=buffering)
1532 return super().open(byte_io=byte_io, buffering=buffering)
1535class VirtualDirectoryFSPath(VirtualPathWithReference):
1536 __slots__ = ("_reference_path",)
1538 def __init__(
1539 self,
1540 basename: str,
1541 parent: InMemoryVirtualPathBase,
1542 *,
1543 reference_path: VirtualPath | None = None,
1544 ) -> None:
1545 super().__init__(
1546 basename,
1547 parent,
1548 reference_path=reference_path,
1549 default_mode=0o755,
1550 )
1551 self._reference_path = reference_path
1552 assert reference_path is None or reference_path.is_dir
1553 self._ensure_min_mode()
1555 @property
1556 def is_dir(self) -> bool:
1557 return True
1559 @property
1560 def is_file(self) -> bool:
1561 return False
1563 @property
1564 def is_symlink(self) -> bool:
1565 return False
1567 def readlink(self) -> str:
1568 raise TypeError(f'"{self._orphan_safe_path()!r}" is a directory; not a symlink')
1571class SymlinkVirtualPath(VirtualPathWithReference):
1572 __slots__ = ("_link_target",)
1574 def __init__(
1575 self,
1576 basename: str,
1577 parent_dir: InMemoryVirtualPathBase,
1578 link_target: str,
1579 *,
1580 reference_path: VirtualPath | None = None,
1581 ) -> None:
1582 super().__init__(
1583 basename,
1584 parent=parent_dir,
1585 default_mode=_SYMLINK_MODE,
1586 reference_path=reference_path,
1587 )
1588 self._link_target = link_target
1590 @property
1591 def is_dir(self) -> bool:
1592 return False
1594 @property
1595 def is_file(self) -> bool:
1596 return False
1598 @property
1599 def is_symlink(self) -> bool:
1600 return True
1602 def readlink(self) -> str:
1603 return self._link_target
1605 @property
1606 def size(self) -> int:
1607 return len(self.readlink())
1610class FSBackedFilePath(VirtualPathWithReference):
1611 __slots__ = ("_fs_path", "_replaceable_inline")
1613 def __init__(
1614 self,
1615 basename: str,
1616 parent_dir: InMemoryVirtualPathBase,
1617 fs_path: str,
1618 *,
1619 replaceable_inline: bool = False,
1620 initial_mode: int | None = None,
1621 mtime: float | None = None,
1622 stat_cache: os.stat_result | None = None,
1623 reference_path: VirtualPath | None = None,
1624 ) -> None:
1625 super().__init__(
1626 basename,
1627 parent_dir,
1628 default_mode=0o644,
1629 reference_path=reference_path,
1630 )
1631 self._fs_path = fs_path
1632 self._replaceable_inline = replaceable_inline
1633 if initial_mode is not None:
1634 self.mode = initial_mode
1635 if mtime is not None:
1636 self._mtime = mtime
1637 self._stat_cache = stat_cache
1638 assert (
1639 not replaceable_inline or "debputy/scratch-dir/" in fs_path
1640 ), f"{fs_path} should not be inline-replaceable -- {self.path}"
1641 self._ensure_min_mode()
1643 @property
1644 def is_dir(self) -> bool:
1645 return False
1647 @property
1648 def is_file(self) -> bool:
1649 return True
1651 @property
1652 def is_symlink(self) -> bool:
1653 return False
1655 def readlink(self) -> str:
1656 raise TypeError(f'"{self._orphan_safe_path()!r}" is a file; not a symlink')
1658 @property
1659 def has_fs_path(self) -> bool:
1660 return True
1662 @property
1663 def fs_path(self) -> str:
1664 return self._fs_path
1666 @property
1667 def _can_replace_inline(self) -> bool:
1668 return self._replaceable_inline
1670 def _replaced_path(self, new_fs_path: str) -> None:
1671 self._fs_path = new_fs_path
1672 self._reference_path = None
1673 self._replaceable_inline = True
1676_SYMLINK_MODE = 0o777
1679class VirtualTestPath(InMemoryVirtualPath):
1680 __slots__ = (
1681 "_path_type",
1682 "_has_fs_path",
1683 "_fs_path",
1684 "_link_target",
1685 "_content",
1686 "_materialized_content",
1687 )
1689 def __init__(
1690 self,
1691 basename: str,
1692 parent_dir: InMemoryVirtualPathBase | None,
1693 mode: int | None = None,
1694 mtime: float | None = None,
1695 is_dir: bool = False,
1696 has_fs_path: bool | None = False,
1697 fs_path: str | None = None,
1698 link_target: str | None = None,
1699 content: str | None = None,
1700 materialized_content: str | None = None,
1701 ) -> None:
1702 if is_dir:
1703 self._path_type = PathType.DIRECTORY
1704 elif link_target is not None:
1705 self._path_type = PathType.SYMLINK
1706 if mode is not None and mode != _SYMLINK_MODE: 1706 ↛ 1707line 1706 didn't jump to line 1707 because the condition on line 1706 was never true
1707 raise ValueError(
1708 f'Please do not assign a mode to symlinks. Triggered for "{basename}".'
1709 )
1710 assert mode is None or mode == _SYMLINK_MODE
1711 else:
1712 self._path_type = PathType.FILE
1714 if mode is not None:
1715 initial_mode = mode
1716 else:
1717 initial_mode = 0o755 if is_dir else 0o644
1719 self._link_target = link_target
1720 if has_fs_path is None:
1721 has_fs_path = bool(fs_path)
1722 self._has_fs_path = has_fs_path
1723 self._fs_path = fs_path
1724 self._materialized_content = materialized_content
1725 super().__init__(
1726 basename,
1727 parent=parent_dir,
1728 initial_mode=initial_mode,
1729 mtime=mtime,
1730 )
1731 self._content = content
1733 @property
1734 def is_dir(self) -> bool:
1735 return self._path_type == PathType.DIRECTORY
1737 @property
1738 def is_file(self) -> bool:
1739 return self._path_type == PathType.FILE
1741 @property
1742 def is_symlink(self) -> bool:
1743 return self._path_type == PathType.SYMLINK
1745 def readlink(self) -> str:
1746 if not self.is_symlink: 1746 ↛ 1747line 1746 didn't jump to line 1747 because the condition on line 1746 was never true
1747 raise TypeError(f"readlink is only valid for symlinks ({self.path!r})")
1748 link_target = self._link_target
1749 assert link_target is not None
1750 return link_target
1752 def _resolve_initial_mtime(self) -> float:
1753 return time.time()
1755 @property
1756 def has_fs_path(self) -> bool:
1757 return self._has_fs_path
1759 def stat(self) -> os.stat_result:
1760 if self.has_fs_path:
1761 path = self.fs_path
1762 if path is None: 1762 ↛ 1763line 1762 didn't jump to line 1763 because the condition on line 1762 was never true
1763 raise PureVirtualPathError(
1764 f"The test wants a real stat of {self._orphan_safe_path()!r}, which this mock path"
1765 " cannot provide!"
1766 )
1767 try:
1768 return os.stat(path)
1769 except FileNotFoundError as e:
1770 raise PureVirtualPathError(
1771 f"The test wants a real stat of {self._orphan_safe_path()!r}, which this mock path"
1772 " cannot provide! (An fs_path was provided, but it did not exist)"
1773 ) from e
1775 raise PureVirtualPathError(
1776 "stat() is only applicable to paths backed by the file system. The path"
1777 f" {self._orphan_safe_path()!r} is purely virtual"
1778 )
1780 @property
1781 def size(self) -> int:
1782 if self._content is not None:
1783 return len(self._content.encode("utf-8"))
1784 if self.is_symlink:
1785 return len(self.readlink())
1786 if not self.has_fs_path or self.fs_path is None:
1787 return 0
1788 return self.stat().st_size
1790 @property
1791 def fs_path(self) -> str:
1792 if self.has_fs_path:
1793 if self._fs_path is None and self._materialized_content is not None:
1794 with tempfile.NamedTemporaryFile(
1795 mode="w+t",
1796 encoding="utf-8",
1797 suffix=f"__{self.name}",
1798 delete=False,
1799 ) as fd:
1800 filepath = fd.name
1801 fd.write(self._materialized_content)
1802 self._fs_path = filepath
1803 atexit.register(lambda: os.unlink(filepath))
1805 path = self._fs_path
1806 if path is None: 1806 ↛ 1807line 1806 didn't jump to line 1807 because the condition on line 1806 was never true
1807 raise PureVirtualPathError(
1808 f"The test wants a real file system entry of {self._orphan_safe_path()!r}, which this "
1809 " mock path cannot provide!"
1810 )
1811 return path
1812 raise PureVirtualPathError(
1813 "fs_path is only applicable to paths backed by the file system. The path"
1814 f" {self._orphan_safe_path()!r} is purely virtual"
1815 )
1817 def replace_fs_path_content(
1818 self,
1819 *,
1820 use_fs_path_mode: bool = False,
1821 ) -> ContextManager[str]:
1822 if self._content is not None: 1822 ↛ 1823line 1822 didn't jump to line 1823 because the condition on line 1822 was never true
1823 raise TypeError(
1824 f"The `replace_fs_path_content()` method was called on {self.path}. Said path was"
1825 " created with `content` but for this method to work, the path should have been"
1826 " created with `materialized_content`"
1827 )
1828 return super().replace_fs_path_content(use_fs_path_mode=use_fs_path_mode)
1830 @overload
1831 def open_child( 1831 ↛ exitline 1831 didn't return from function 'open_child' because
1832 self,
1833 name: str,
1834 mode: TextOpenMode = "r",
1835 buffering: int = -1,
1836 ) -> TextIO: ...
1838 @overload
1839 def open_child( 1839 ↛ exitline 1839 didn't return from function 'open_child' because
1840 self,
1841 name: str,
1842 mode: BinaryOpenMode,
1843 buffering: int = -1,
1844 ) -> BinaryIO: ...
1846 @contextlib.contextmanager
1847 def open_child(self, name, mode="r", buffering=-1):
1848 existing = self.get(name)
1849 if existing or "r" in mode:
1850 with super().open_child(name, mode, buffering=buffering) as fd:
1851 yield fd
1852 return
1853 if "b" in mode:
1854 fd = io.BytesIO(b"")
1855 yield fd
1856 content = fd.getvalue().decode("utf-8")
1857 else:
1858 fd = io.StringIO("")
1859 yield fd
1860 content = fd.getvalue()
1861 VirtualTestPath(
1862 name,
1863 self,
1864 mode=0o644,
1865 content=content,
1866 has_fs_path=True,
1867 )
1869 @overload
1870 def open( 1870 ↛ exitline 1870 didn't return from function 'open' because
1871 self,
1872 *,
1873 byte_io: Literal[False] = False,
1874 buffering: int = -1,
1875 ) -> TextIO: ...
1877 @overload
1878 def open( 1878 ↛ exitline 1878 didn't return from function 'open' because
1879 self,
1880 *,
1881 byte_io: Literal[True],
1882 buffering: Literal[0] = ...,
1883 ) -> io.FileIO: ...
1885 @overload
1886 def open( 1886 ↛ exitline 1886 didn't return from function 'open' because
1887 self,
1888 *,
1889 byte_io: Literal[True],
1890 buffering: int = -1,
1891 ) -> io.BufferedReader: ...
1893 def open(self, *, byte_io=False, buffering=-1):
1894 if self._content is None:
1895 try:
1896 return super().open(byte_io=byte_io, buffering=buffering)
1897 except FileNotFoundError as e:
1898 raise TestPathWithNonExistentFSPathError(
1899 f"The test path {self.path} had an fs_path {self._fs_path}, which does not"
1900 " exist. This exception can only occur in the testsuite. Either have the"
1901 " test provide content for the path (`virtual_path_def(..., content=...) or,"
1902 " if that is too painful in general, have the code accept this error as a "
1903 " test only-case and provide a default."
1904 ) from e
1906 if byte_io:
1907 return io.BytesIO(self._content.encode("utf-8"))
1908 return io.StringIO(self._content)
1910 def _replaced_path(self, new_fs_path: str) -> None:
1911 self._fs_path = new_fs_path
1914class OSFSOverlayBase(VirtualPathBase, Generic[FSP]):
1915 __slots__ = (
1916 "_path",
1917 "_fs_path",
1918 "_parent",
1919 "__weakref__",
1920 )
1922 def __init__(
1923 self,
1924 path: str,
1925 fs_path: str,
1926 parent: FSP | None,
1927 ) -> None:
1928 self._path: str = path
1929 prefix = "/" if fs_path.startswith("/") else ""
1930 self._fs_path: str = prefix + _normalize_path(fs_path, with_prefix=False)
1931 self._parent: ReferenceType[FSP] | None = (
1932 ref(parent) if parent is not None else None
1933 )
1935 @property
1936 def name(self) -> str:
1937 return os.path.basename(self._path)
1939 @property
1940 def path(self) -> str:
1941 return self._path
1943 @property
1944 def parent_dir(self) -> Optional["FSP"]:
1945 parent = self._parent
1946 if parent is None:
1947 return None
1948 resolved = parent()
1949 if resolved is None:
1950 raise RuntimeError("Parent was garbage collected!")
1951 return resolved
1953 @property
1954 def fs_path(self) -> str:
1955 return self._fs_path
1957 def stat(self) -> os.stat_result:
1958 return os.lstat(self.fs_path)
1960 @property
1961 def is_dir(self) -> bool:
1962 # The root path can have a non-existent fs_path (such as d/tmp not always existing)
1963 try:
1964 return stat.S_ISDIR(self.stat().st_mode)
1965 except FileNotFoundError:
1966 return False
1968 @property
1969 def is_file(self) -> bool:
1970 # The root path can have a non-existent fs_path (such as d/tmp not always existing)
1971 try:
1972 return stat.S_ISREG(self.stat().st_mode)
1973 except FileNotFoundError:
1974 return False
1976 @property
1977 def is_symlink(self) -> bool:
1978 # The root path can have a non-existent fs_path (such as d/tmp not always existing)
1979 try:
1980 return stat.S_ISLNK(self.stat().st_mode)
1981 except FileNotFoundError:
1982 return False
1984 @property
1985 def has_fs_path(self) -> bool:
1986 return True
1988 @overload
1989 def open( 1989 ↛ exitline 1989 didn't return from function 'open' because
1990 self,
1991 *,
1992 byte_io: Literal[False] = False,
1993 buffering: int = -1,
1994 ) -> TextIO: ...
1996 @overload
1997 def open( 1997 ↛ exitline 1997 didn't return from function 'open' because
1998 self,
1999 *,
2000 byte_io: Literal[True],
2001 buffering: Literal[0] = ...,
2002 ) -> io.FileIO: ...
2004 @overload
2005 def open( 2005 ↛ exitline 2005 didn't return from function 'open' because
2006 self,
2007 *,
2008 byte_io: Literal[True],
2009 buffering: int = -1,
2010 ) -> io.BufferedReader: ...
2012 def open(self, *, byte_io=False, buffering=-1):
2013 # Allow symlinks for open here, because we can let the OS resolve the symlink reliably in this
2014 # case.
2015 if not self.is_file and not self.is_symlink:
2016 raise TypeError(
2017 f"Cannot open {self.path} for reading: It is not a file nor a symlink"
2018 )
2020 if byte_io:
2021 return open(self.fs_path, "rb", buffering=buffering)
2022 return open(self.fs_path, encoding="utf-8", buffering=buffering)
2024 def metadata(
2025 self,
2026 metadata_type: type[PMT],
2027 *,
2028 owning_plugin: str | None = None,
2029 ) -> PathMetadataReference[PMT]:
2030 current_plugin = self._current_plugin()
2031 if owning_plugin is None:
2032 owning_plugin = current_plugin
2033 return AlwaysEmptyReadOnlyMetadataReference(
2034 owning_plugin,
2035 current_plugin,
2036 metadata_type,
2037 )
2039 def all_paths(self) -> Iterable["OSFSControlPath"]:
2040 yield cast("OSFSControlPath", self)
2041 if not self.is_dir:
2042 return
2043 stack = list(self.iterdir())
2044 stack.reverse()
2045 while stack:
2046 current = cast("OSFSControlPath", stack.pop())
2047 yield current
2048 if current.is_dir:
2049 stack.extend(reversed(list(current.iterdir())))
2051 def _resolve_children(
2052 self,
2053 new_child: Callable[[str, str, FSP], FSC],
2054 ) -> Mapping[str, FSC]:
2055 if not self.is_dir:
2056 return {}
2057 dir_path = self.path
2058 dir_fs_path = self.fs_path
2059 children = {}
2060 for name in sorted(os.listdir(dir_fs_path), key=os.path.basename):
2061 child_path = os.path.join(dir_path, name) if dir_path != "." else name
2062 child_fs_path = (
2063 os.path.join(dir_fs_path, name) if dir_fs_path != "." else name
2064 )
2065 children[name] = new_child(
2066 child_path,
2067 child_fs_path,
2068 cast("FSP", self),
2069 )
2070 return children
2073class OSFSROOverlay(OSFSOverlayBase["OSFSROOverlay"]):
2074 __slots__ = (
2075 "_stat_cache",
2076 "_readlink_cache",
2077 "_children",
2078 "_stat_failed_cache",
2079 )
2081 def __init__(
2082 self,
2083 path: str,
2084 fs_path: str,
2085 parent: Optional["OSFSROOverlay"],
2086 ) -> None:
2087 super().__init__(path, fs_path, parent=parent)
2088 self._stat_cache: os.stat_result | None = None
2089 self._readlink_cache: str | None = None
2090 self._stat_failed_cache = False
2091 self._children: Mapping[str, OSFSROOverlay] | None = None
2093 @classmethod
2094 def create_root_dir(cls, path: str, fs_path: str) -> "OSFSROOverlay":
2095 return OSFSROOverlay(path, fs_path, None)
2097 def iterdir(self) -> Iterable["OSFSROOverlay"]:
2098 if not self.is_dir:
2099 return
2100 if self._children is None:
2101 self._ensure_children_are_resolved()
2102 yield from assume_not_none(self._children).values()
2104 def lookup(self, path: str) -> Optional["OSFSROOverlay"]:
2105 if not self.is_dir:
2106 return None
2107 if self._children is None:
2108 self._ensure_children_are_resolved()
2110 absolute, _, path_parts = _split_path(path)
2111 current = cast("OSFSROOverlay", _root(self)) if absolute else self
2112 for no, dir_part in enumerate(path_parts):
2113 if dir_part == ".":
2114 continue
2115 if dir_part == "..":
2116 if current.is_root_dir():
2117 raise ValueError(f'The path "{path}" escapes the root dir')
2118 p = current.parent_dir
2119 assert p is not None # Type hint
2120 current = cast("OSFSROOverlay", p)
2121 continue
2122 try:
2123 current = cast("OSFSROOverlay", current[dir_part])
2124 except KeyError:
2125 return None
2126 return current
2128 def _ensure_children_are_resolved(self) -> None:
2129 if not self.is_dir or self._children:
2130 return
2131 self._children = self._resolve_children(
2132 lambda n, fsp, p: OSFSROOverlay(n, fsp, p)
2133 )
2135 @property
2136 def is_detached(self) -> bool:
2137 return False
2139 def __getitem__(self, key) -> "VirtualPath":
2140 if not self.is_dir: 2140 ↛ 2142line 2140 didn't jump to line 2142 because the condition on line 2140 was always true
2141 raise KeyError(key)
2142 if self._children is None:
2143 self._ensure_children_are_resolved()
2144 if isinstance(key, InMemoryVirtualPathBase):
2145 key = key.name
2146 return assume_not_none(self._children)[key]
2148 def __delitem__(self, key) -> Never:
2149 self._error_ro_fs()
2151 @property
2152 def is_read_write(self) -> bool:
2153 return False
2155 def _rw_check(self) -> Never:
2156 self._error_ro_fs()
2158 def _error_ro_fs(self) -> Never:
2159 raise DebputyFSIsROError(
2160 f'Attempt to write to "{self.path}" failed:'
2161 " Debputy Virtual File system is R/O."
2162 )
2164 def stat(self) -> os.stat_result:
2165 if self._stat_failed_cache: 2165 ↛ 2166line 2165 didn't jump to line 2166 because the condition on line 2165 was never true
2166 raise FileNotFoundError(
2167 errno.ENOENT, os.strerror(errno.ENOENT), self.fs_path
2168 )
2170 if self._stat_cache is None: 2170 ↛ 2176line 2170 didn't jump to line 2176 because the condition on line 2170 was always true
2171 try:
2172 self._stat_cache = os.lstat(self.fs_path)
2173 except FileNotFoundError:
2174 self._stat_failed_cache = True
2175 raise
2176 return self._stat_cache
2178 @property
2179 def mode(self) -> int:
2180 return stat.S_IMODE(self.stat().st_mode)
2182 @mode.setter
2183 def mode(self, _unused: int) -> Never:
2184 self._error_ro_fs()
2186 @property
2187 def mtime(self) -> float:
2188 return self.stat().st_mtime
2190 @mtime.setter
2191 def mtime(self, new_mtime: float) -> Never:
2192 self._error_ro_fs()
2194 def readlink(self) -> str:
2195 if not self.is_symlink:
2196 raise TypeError(f"readlink is only valid for symlinks ({self.path!r})")
2197 if self._readlink_cache is None:
2198 self._readlink_cache = os.readlink(self.fs_path)
2199 return self._readlink_cache
2201 def chown(
2202 self,
2203 owner: StaticFileSystemOwner | None,
2204 group: StaticFileSystemGroup | None,
2205 ) -> Never:
2206 self._error_ro_fs()
2208 def mkdir(self, name: str) -> Never:
2209 self._error_ro_fs()
2211 def add_file(
2212 self,
2213 name: str,
2214 *,
2215 unlink_if_exists: bool = True,
2216 use_fs_path_mode: bool = False,
2217 mode: int = 0o0644,
2218 mtime: float | None = None,
2219 ) -> Never:
2220 self._error_ro_fs()
2222 def add_symlink(self, link_name: str, link_target: str) -> Never:
2223 self._error_ro_fs()
2225 def unlink(self, *, recursive: bool = False) -> Never:
2226 self._error_ro_fs()
2229class OSFSROOverlayRootDir(OSFSROOverlay):
2230 __slots__ = ("_plugin_context",)
2232 def __init__(self, path: str, fs_path: str) -> None:
2233 super().__init__(path, fs_path, None)
2234 self._plugin_context = CurrentPluginContextManager("debputy")
2236 def _current_plugin(self) -> str:
2237 return self._plugin_context.current_plugin_name
2239 @contextlib.contextmanager
2240 def change_plugin_context(self, new_plugin: str) -> Iterator[str]:
2241 with self._plugin_context.change_plugin_context(new_plugin) as r:
2242 yield r
2245class OSFSControlPath(OSFSOverlayBase["OSFSControlPath"]):
2247 def iterdir(self) -> Iterable["OSFSControlPath"]:
2248 if not self.is_dir:
2249 return
2250 yield from self._resolve_children(
2251 lambda n, fsp, p: OSFSControlPath(n, fsp, p)
2252 ).values()
2254 def lookup(self, path: str) -> Optional["OSFSControlPath"]:
2255 if not self.is_dir:
2256 return None
2258 absolute, _, path_parts = _split_path(path)
2259 current = cast("OSFSControlPath", _root(self)) if absolute else self
2260 for no, dir_part in enumerate(path_parts):
2261 if dir_part == ".":
2262 continue
2263 if dir_part == "..":
2264 if current.is_root_dir():
2265 raise ValueError(f'The path "{path}" escapes the root dir')
2266 p = current.parent_dir
2267 assert p is not None # type hint
2268 current = cast("OSFSControlPath", p)
2269 continue
2270 try:
2271 current = cast("OSFSControlPath", current[dir_part])
2272 except KeyError:
2273 return None
2274 return current
2276 @property
2277 def is_detached(self) -> bool:
2278 try:
2279 self.stat()
2280 except FileNotFoundError:
2281 return True
2282 else:
2283 return False
2285 def __getitem__(self, key) -> "VirtualPath":
2286 if not self.is_dir:
2287 raise KeyError(key)
2288 children = self._resolve_children(lambda n, fsp, p: OSFSControlPath(n, fsp, p))
2289 if isinstance(key, InMemoryVirtualPathBase):
2290 key = key.name
2291 return children[key]
2293 def __delitem__(self, key) -> None:
2294 self[key].unlink()
2296 @property
2297 def is_read_write(self) -> bool:
2298 return True
2300 @property
2301 def mode(self) -> int:
2302 return stat.S_IMODE(self.stat().st_mode)
2304 @mode.setter
2305 def mode(self, new_mode: int) -> None:
2306 os.chmod(self.fs_path, new_mode)
2308 @property
2309 def mtime(self) -> float:
2310 return self.stat().st_mtime
2312 @mtime.setter
2313 def mtime(self, new_mtime: float) -> None:
2314 os.utime(self.fs_path, (new_mtime, new_mtime))
2316 def readlink(self) -> Never:
2317 if not self.is_symlink:
2318 raise TypeError(f"readlink is only valid for symlinks ({self.path!r})")
2319 assert False
2321 def chown(
2322 self,
2323 owner: StaticFileSystemOwner | None,
2324 group: StaticFileSystemGroup | None,
2325 ) -> None:
2326 raise ValueError(
2327 "No need to chown paths in the control.tar: They are always root:root"
2328 )
2330 def mkdir(self, name: str) -> Never:
2331 raise TypeError("The control.tar never contains subdirectories.")
2333 @contextlib.contextmanager
2334 def add_file(
2335 self,
2336 name: str,
2337 *,
2338 unlink_if_exists: bool = True,
2339 use_fs_path_mode: bool = False,
2340 mode: int = 0o0644,
2341 mtime: float | None = None,
2342 ) -> Iterator["VirtualPath"]:
2343 if "/" in name or name in {".", ".."}:
2344 raise ValueError(f'Invalid file name: "{name}"')
2345 if not self.is_dir:
2346 raise TypeError(
2347 f"Cannot create {self._orphan_safe_path()}/{name}:"
2348 f" {self._orphan_safe_path()} is not a directory"
2349 )
2350 self._rw_check()
2351 existing = self.get(name)
2352 if existing is not None:
2353 if not unlink_if_exists:
2354 raise ValueError(
2355 f'The path "{self._orphan_safe_path()}" already contains a file called "{name}"'
2356 f" and exist_ok was False"
2357 )
2358 assert existing.is_file
2360 fs_path = os.path.join(self.fs_path, name)
2361 # This truncates the existing file if any, so we do not have to unlink the previous entry.
2362 with open(fs_path, "wb") as fd:
2363 # Ensure that the fs_path exists and default mode is reasonable
2364 os.chmod(fd.fileno(), mode)
2365 child = OSFSControlPath(
2366 name,
2367 fs_path,
2368 self,
2369 )
2370 yield child
2371 _check_fs_path_is_file(fs_path, unlink_on_error=child)
2372 child.mode = mode
2374 @contextlib.contextmanager
2375 def replace_fs_path_content(
2376 self,
2377 *,
2378 use_fs_path_mode: bool = False,
2379 ) -> Iterator[str]:
2380 if not self.is_file:
2381 raise TypeError(
2382 f'Cannot replace contents of "{self._orphan_safe_path()}" as it is not a file'
2383 )
2384 restore_mode = self.mode if use_fs_path_mode else None
2385 yield self.fs_path
2386 _check_fs_path_is_file(self.fs_path, self)
2387 if restore_mode is not None:
2388 self.mode = restore_mode
2390 def add_symlink(self, link_name: str, link_target: str) -> Never:
2391 raise TypeError("The control.tar never contains symlinks.")
2393 def unlink(self, *, recursive: bool = False) -> None:
2394 if self._parent is None:
2395 return
2396 # By virtue of the control FS only containing paths, we can assume `recursive` never
2397 # matters and that `os.unlink` will be sufficient.
2398 assert self.is_file
2399 os.unlink(self.fs_path)
2402class FSControlRootDir(OSFSControlPath):
2404 @classmethod
2405 def create_root_dir(cls, fs_path: str) -> "FSControlRootDir":
2406 return FSControlRootDir(".", fs_path, None)
2408 def insert_file_from_fs_path(
2409 self,
2410 name: str,
2411 fs_path: str,
2412 *,
2413 exist_ok: bool = True,
2414 use_fs_path_mode: bool = False,
2415 mode: int = 0o0644,
2416 # Ignored, but accepted for compat with FSPath's variant of this function.
2417 # - This is used by install_or_generate_conffiles.
2418 reference_path: VirtualPath | None = None, # noqa
2419 ) -> "OSFSControlPath":
2420 if "/" in name or name in {".", ".."}:
2421 raise ValueError(f'Invalid file name: "{name}"')
2422 if not self.is_dir:
2423 raise TypeError(
2424 f"Cannot create {self._orphan_safe_path()}/{name}:"
2425 f" {self._orphan_safe_path()} is not a directory"
2426 )
2427 self._rw_check()
2428 if name in self and not exist_ok:
2429 raise ValueError(
2430 f'The path "{self._orphan_safe_path()}" already contains a file called "{name}"'
2431 f" and exist_ok was False"
2432 )
2434 target_path = os.path.join(self.fs_path, name)
2435 if use_fs_path_mode:
2436 shutil.copymode(
2437 fs_path,
2438 target_path,
2439 follow_symlinks=True,
2440 )
2441 else:
2442 shutil.copyfile(
2443 fs_path,
2444 target_path,
2445 follow_symlinks=True,
2446 )
2447 os.chmod(target_path, mode)
2448 return cast("OSFSControlPath", self[name])
2451def as_path_def(pd: str | PathDef) -> PathDef:
2452 return PathDef(pd) if isinstance(pd, str) else pd
2455def as_path_defs(paths: Iterable[str | PathDef]) -> Iterable[PathDef]:
2456 yield from (as_path_def(p) for p in paths)
2459def build_virtual_fs(
2460 paths: Iterable[str | PathDef],
2461 read_write_fs: bool = False,
2462) -> "InMemoryVirtualPathBase":
2463 root_dir: InMemoryVirtualRootDir | None = None
2464 directories: dict[str, InMemoryVirtualPathBase] = {}
2465 non_directories = set()
2467 def _ensure_parent_dirs(p: str) -> None:
2468 current = p.rstrip("/")
2469 missing_dirs = []
2470 while True:
2471 current = os.path.dirname(current)
2472 if current in directories:
2473 break
2474 if current in non_directories: 2474 ↛ 2475line 2474 didn't jump to line 2475 because the condition on line 2474 was never true
2475 raise ValueError(
2476 f'Conflicting definition for "{current}". The path "{p}" wants it as a directory,'
2477 ' but it is defined as a non-directory. (Ensure dirs end with "/")'
2478 )
2479 missing_dirs.append(current)
2480 for dir_path in reversed(missing_dirs):
2481 parent_dir = directories[os.path.dirname(dir_path)]
2482 d = VirtualTestPath(os.path.basename(dir_path), parent_dir, is_dir=True)
2483 directories[dir_path] = d
2485 for path_def in as_path_defs(paths):
2486 path = path_def.path_name
2487 if path in directories or path in non_directories: 2487 ↛ 2488line 2487 didn't jump to line 2488 because the condition on line 2487 was never true
2488 raise ValueError(
2489 f'Duplicate definition of "{path}". Can be false positive if input is not in'
2490 ' "correct order" (ensure directories occur before their children)'
2491 )
2492 if root_dir is None:
2493 root_fs_path = None
2494 if path in (".", "./", "/"):
2495 root_fs_path = path_def.fs_path
2496 root_dir = InMemoryVirtualRootDir(fs_path=root_fs_path)
2497 directories["."] = root_dir
2499 if path not in (".", "./", "/") and not path.startswith("./"):
2500 path = "./" + path
2501 if path not in (".", "./", "/"):
2502 _ensure_parent_dirs(path)
2503 if path in (".", "./"):
2504 assert "." in directories
2505 continue
2506 is_dir = False
2507 if path.endswith("/"):
2508 path = path[:-1]
2509 is_dir = True
2510 directory = directories[os.path.dirname(path)]
2511 assert not is_dir or not bool(
2512 path_def.link_target
2513 ), f"is_dir={is_dir} vs. link_target={path_def.link_target}"
2514 fs_path = VirtualTestPath(
2515 os.path.basename(path),
2516 directory,
2517 is_dir=is_dir,
2518 mode=path_def.mode,
2519 mtime=path_def.mtime,
2520 has_fs_path=path_def.has_fs_path,
2521 fs_path=path_def.fs_path,
2522 link_target=path_def.link_target,
2523 content=path_def.content,
2524 materialized_content=path_def.materialized_content,
2525 )
2526 assert not fs_path.is_detached
2527 if fs_path.is_dir:
2528 directories[fs_path.path] = fs_path
2529 else:
2530 non_directories.add(fs_path.path)
2532 if root_dir is None:
2533 root_dir = InMemoryVirtualRootDir()
2535 root_dir.is_read_write = read_write_fs
2536 return root_dir