"""Unit: phase 106 task 03 — mtime-preserving archive unpack (D2). Pins the date source of the upload path: a zip member's DOS ``date_time`` and a tar member's ``mtime`` survive the unpack as the extracted file's atime+mtime, so the importer (task 04) reads the archive's ORIGINAL file dates — the owner's "file metadata (hopefully) preserved in the tar or zip archive process" made real. Regular files only: directories/symlinks/hardlinks are untouched (they carry the extraction-time values, never the member's). The safety behavior (zip-slip, absolute members, link targets, the extraction cap) is unchanged and pinned by ``tests/unit/test_archive_upload.py`` — this file adds only the date pins. """ from __future__ import annotations import os import tarfile import zipfile from datetime import UTC, datetime from io import BytesIO from pathlib import Path from app.rag.archive_upload import unpack_archive #: The old member timestamp both fixtures carry: 2020-01-02 03:04:06 #: UTC — the zip as the DOS tuple ``(2020, 1, 2, 3, 4, 6)``, the tar #: as the epoch seconds. OLD_MTIME = datetime(2020, 1, 2, 3, 4, 6, tzinfo=UTC) OLD_EPOCH = OLD_MTIME.timestamp() # 1577934246.0 #: mtime granularity tolerance (the task pin: ±1 s). _TOLERANCE_S = 1.0 _PAYLOAD = b"# Old note\ncontent from 2020\n" def _zip_with_old_mtime(path: Path) -> None: """A two-member zip (a directory + one file), both carrying the old fixed ``date_time`` tuple.""" with zipfile.ZipFile(path, "w") as zf: dir_info = zipfile.ZipInfo("docs/", date_time=(2020, 1, 2, 3, 4, 6)) dir_info.external_attr = (0o40755 << 16) zf.writestr(dir_info, b"") file_info = zipfile.ZipInfo("docs/note.md", date_time=(2020, 1, 2, 3, 4, 6)) file_info.external_attr = (0o100644 << 16) zf.writestr(file_info, _PAYLOAD) def _tar_with_old_mtime(path: Path) -> None: """A two-member tar (a directory + one file), both carrying the old epoch ``mtime``.""" with tarfile.open(path, "w") as tf: dir_ti = tarfile.TarInfo("docs/") dir_ti.type = tarfile.DIRTYPE dir_ti.mode = 0o755 dir_ti.mtime = OLD_EPOCH tf.addfile(dir_ti) file_ti = tarfile.TarInfo("docs/note.md") file_ti.size = len(_PAYLOAD) file_ti.mode = 0o644 file_ti.mtime = OLD_EPOCH tf.addfile(file_ti, BytesIO(_PAYLOAD)) def _extract(archive: Path, tmp_path: Path) -> tuple[Path, os.stat_result]: """Unpack ``archive``; return (file path, pre-read stat). The stat happens BEFORE the content read (a read refreshes atime under relatime — the atime pin needs the unpacked value); the content is asserted byte-identical (the mtime work moved no bytes).""" target = tmp_path / "out" unpack_archive(archive, target, max_extract_bytes=1 << 20) dest = target / "docs" / "note.md" st = dest.stat() # before the read — reading would refresh atime assert dest.read_bytes() == _PAYLOAD return dest, st def test_zip_member_mtime_is_restored(tmp_path: Path) -> None: """A zip member's DOS ``date_time`` lands as the extracted file's mtime (and atime — ``os.utime(ns=(t, t))`` sets both), within the mtime-granularity tolerance.""" archive = tmp_path / "old.zip" _zip_with_old_mtime(archive) _dest, st = _extract(archive, tmp_path) assert abs(st.st_mtime - OLD_EPOCH) <= _TOLERANCE_S assert abs(st.st_atime - OLD_EPOCH) <= _TOLERANCE_S def test_tar_member_mtime_is_restored(tmp_path: Path) -> None: """A tar member's epoch ``mtime`` lands as the extracted file's mtime (and atime), within the mtime-granularity tolerance.""" archive = tmp_path / "old.tar" _tar_with_old_mtime(archive) _dest, st = _extract(archive, tmp_path) assert abs(st.st_mtime - OLD_EPOCH) <= _TOLERANCE_S assert abs(st.st_atime - OLD_EPOCH) <= _TOLERANCE_S def test_zip_directory_member_mtime_is_not_restored(tmp_path: Path) -> None: """Regular files ONLY: a zip directory member carrying the old ``date_time`` keeps the EXTRACTION-time mtime (≈ now, long after the old 2020 value) — directories are never indexed, so their dates don't matter; the pin is that the unpacker doesn't utime them (the task contract).""" archive = tmp_path / "old.zip" _zip_with_old_mtime(archive) target = tmp_path / "out" unpack_archive(archive, target, max_extract_bytes=1 << 20) assert (target / "docs").stat().st_mtime > OLD_EPOCH + _TOLERANCE_S def test_tar_directory_member_mtime_is_not_restored(tmp_path: Path) -> None: """The tar twin of the zip directory pin: a directory member with the old ``mtime`` is not utime'd (regular files only).""" archive = tmp_path / "old.tar" _tar_with_old_mtime(archive) target = tmp_path / "out" unpack_archive(archive, target, max_extract_bytes=1 << 20) assert (target / "docs").stat().st_mtime > OLD_EPOCH + _TOLERANCE_S