(path: pathlib.Path, *, expected_size: int)
| 185 | |
| 186 | |
| 187 | def sha256_file(path: pathlib.Path, *, expected_size: int) -> str: |
| 188 | status = regular_status(path, ceiling=MAX_CANDIDATE_BYTES, label="candidate object") |
| 189 | if status.st_size != expected_size: |
| 190 | raise ContractError(f"candidate object size changed: {path}") |
| 191 | digest = hashlib.sha256() |
| 192 | flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) |
| 193 | descriptor = os.open(path, flags) |
| 194 | try: |
| 195 | opened = os.fstat(descriptor) |
| 196 | if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != ( |
| 197 | status.st_dev, |
| 198 | status.st_ino, |
| 199 | ): |
| 200 | raise ContractError(f"candidate object changed while being opened: {path}") |
| 201 | with os.fdopen(descriptor, "rb", closefd=False) as handle: |
| 202 | for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| 203 | digest.update(chunk) |
| 204 | finally: |
| 205 | os.close(descriptor) |
| 206 | return digest.hexdigest() |
| 207 | |
| 208 | |
| 209 | def validate_candidate_row(row: dict[str, str], *, target: str, variant: str) -> None: |
no test coverage detected