(source_fd: int, size: int, offset: int, source_path: str)
| 118 | |
| 119 | |
| 120 | def _safe_pread(source_fd: int, size: int, offset: int, source_path: str) -> bytes: |
| 121 | try: |
| 122 | # pread keeps read position explicit and thread-safe for random access. |
| 123 | return os.pread(source_fd, size, offset) |
| 124 | except PermissionError as error: |
| 125 | raise BlockExtractionError( |
| 126 | f"Cannot read source {source_path}: {error}", |
| 127 | f"Permission denied: cannot read {source_path} (run as root).", |
| 128 | ) from error |
| 129 | except OSError as error: |
| 130 | raise BlockExtractionError( |
| 131 | f"Cannot read source {source_path}: {error}", |
| 132 | _get_read_error_message(source_path, error), |
| 133 | ) from error |
| 134 | |
| 135 | |
| 136 | def _validate_range(offset: int, length: int, chunk_size: int) -> None: |
no test coverage detected