220 lines
8.5 KiB
Python
220 lines
8.5 KiB
Python
"""Хранилище файлов вложений (bug-and-attachments, hub-api «Хранилище файлов вложений»)."""
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import stat
|
||
import time
|
||
|
||
import pytest
|
||
|
||
from sdlc_hub.core import db, files, settings
|
||
|
||
ABC = hashlib.sha256(b"abc").hexdigest() # ba7816bf…
|
||
DAY = 86400
|
||
|
||
|
||
@pytest.fixture
|
||
def root(tmp_path):
|
||
return tmp_path / "files"
|
||
|
||
|
||
@pytest.fixture
|
||
def conn(open_db):
|
||
"""База в памяти с настройками по умолчанию; закрывается после теста."""
|
||
c = open_db()
|
||
settings.ensure_defaults(c)
|
||
return c
|
||
|
||
|
||
def test_put_creates_closed_dir_and_file(root):
|
||
assert files.put(root, b"abc") == (ABC, 3)
|
||
assert ABC.startswith("ba7816bf")
|
||
assert (root / ABC).read_bytes() == b"abc"
|
||
assert stat.S_IMODE(os.stat(root).st_mode) == 0o700
|
||
assert [p.name for p in root.iterdir()] == [ABC] # временных файлов не осталось
|
||
|
||
|
||
def test_put_again_keeps_one_copy_and_touches_mtime(root):
|
||
files.put(root, b"abc")
|
||
old = time.time() - 40 * DAY
|
||
os.utime(root / ABC, (old, old))
|
||
assert files.put(root, b"abc") == (ABC, 3)
|
||
assert [p.name for p in root.iterdir()] == [ABC]
|
||
assert (root / ABC).stat().st_mtime > time.time() - 60 # свежая постановка не теряет файл в уборке
|
||
|
||
|
||
def test_put_rewrites_truncated_copy(root):
|
||
files.put(root, b"abc")
|
||
(root / ABC).write_bytes(b"ab") # оборванная копия
|
||
files.put(root, b"abc")
|
||
assert (root / ABC).read_bytes() == b"abc"
|
||
|
||
|
||
def test_put_limits(root):
|
||
with pytest.raises(files.EmptyFile):
|
||
files.put(root, b"")
|
||
with pytest.raises(files.TooLarge):
|
||
files.put(root, b"\0" * (files.MAX_SIZE + 1))
|
||
assert not root.exists() or list(root.iterdir()) == []
|
||
sha, size = files.put(root, b"\0" * files.MAX_SIZE) # ровно 10 485 760 байт — принимается
|
||
assert files.MAX_SIZE == 10_485_760 and size == files.MAX_SIZE and files.size(root, sha) == files.MAX_SIZE
|
||
|
||
|
||
def test_put_failure_leaves_no_temp(root, monkeypatch):
|
||
def broken(*args):
|
||
raise OSError("диск полон")
|
||
monkeypatch.setattr(files.os, "replace", broken)
|
||
with pytest.raises(OSError, match="диск полон"):
|
||
files.put(root, b"abc")
|
||
assert list(root.iterdir()) == []
|
||
|
||
|
||
def test_put_after_dir_removed(root):
|
||
files.put(root, b"abc")
|
||
(root / ABC).unlink()
|
||
root.rmdir()
|
||
assert files.put(root, b"abc") == (ABC, 3) and files.exists(root, ABC)
|
||
assert stat.S_IMODE(os.stat(root).st_mode) == 0o700
|
||
|
||
|
||
def test_lookup(root):
|
||
files.put(root, b"abc")
|
||
assert files.exists(root, ABC) and files.size(root, ABC) == 3 and files.read(root, ABC) == b"abc"
|
||
assert files.path(root, ABC) == root / ABC
|
||
missing = "0" * 64
|
||
assert not files.exists(root, missing) and files.size(root, missing) is None and files.read(root, missing) is None
|
||
|
||
|
||
@pytest.mark.parametrize("bad", [ABC.upper(), ABC[:63], ABC + "0", "..", "../hub.sqlite", ABC + "\n", "", None, 5])
|
||
def test_bad_sha_never_builds_path(root, bad):
|
||
"""APFS не различает регистр: ABC.upper() нашёл бы файл ABC, если бы путь строился без проверки."""
|
||
files.put(root, b"abc")
|
||
assert files.is_sha(bad) is False
|
||
assert not files.exists(root, bad) and files.size(root, bad) is None and files.read(root, bad) is None
|
||
with pytest.raises(ValueError):
|
||
files.path(root, bad)
|
||
|
||
|
||
def _file(root, content, age_days):
|
||
sha, _ = files.put(root, content)
|
||
t = time.time() - age_days * DAY
|
||
os.utime(root / sha, (t, t))
|
||
return sha
|
||
|
||
|
||
def _command(conn, sha, status):
|
||
if conn.execute("SELECT 1 FROM batch WHERE id = 1").fetchone() is None:
|
||
conn.execute("INSERT INTO batch(id, title, author, created_at) VALUES (1, 'x', 'claude', ?)", (db.now(),))
|
||
conn.execute("INSERT INTO command(batch_id, seq, type, params, status) VALUES (1, 0, 'ensure_attachment', ?, ?)",
|
||
(json.dumps({"key": "A-1", "file": sha, "filename": "a.xlsx"}), status))
|
||
|
||
|
||
def test_sweep_spec_scenario(root, conn):
|
||
"""Спека hub-api «Уборка старых файлов»: A старше 30 дней без ссылок, B старше 30 дней в queued, C вчерашний."""
|
||
a = _file(root, b"A", 31)
|
||
b = _file(root, b"B", 31)
|
||
c = _file(root, b"C", 1)
|
||
_command(conn, b, "queued")
|
||
assert files.sweep(conn, root) == [a]
|
||
assert sorted(p.name for p in root.iterdir()) == sorted([b, c])
|
||
|
||
|
||
def test_sweep_keeps_claimed_and_shared_file(root, conn):
|
||
shared = _file(root, b"S", 31)
|
||
claimed = _file(root, b"K", 31)
|
||
gone = _file(root, b"G", 31)
|
||
_command(conn, shared, "done")
|
||
_command(conn, shared, "queued") # тот же файл под другим именем в другой команде
|
||
_command(conn, claimed, "claimed")
|
||
for status in ("done", "cancelled", "failed", "skipped", "confirmed_by_state"):
|
||
_command(conn, gone, status)
|
||
assert files.sweep(conn, root) == [gone]
|
||
assert files.exists(root, shared) and files.exists(root, claimed)
|
||
|
||
|
||
def test_sweep_removes_stale_temp_and_handles_missing_dir(root, conn, tmp_path):
|
||
assert files.sweep(conn, tmp_path / "нет") == []
|
||
files.put(root, b"abc")
|
||
tmp = root / f".{ABC}.x1y2.tmp" # брошен сбоем посреди записи
|
||
tmp.write_bytes(b"ab")
|
||
t = time.time() - 31 * DAY
|
||
os.utime(tmp, (t, t))
|
||
assert files.sweep(conn, root) == [tmp.name]
|
||
assert files.exists(root, ABC)
|
||
|
||
|
||
def test_sweep_uses_now(root, conn):
|
||
sha = _file(root, b"N", 0)
|
||
assert files.sweep(conn, root) == []
|
||
assert files.sweep(conn, root, now=time.time() + 31 * DAY) == [sha]
|
||
|
||
|
||
def test_put_rewrites_corrupted_copy_of_same_size(root):
|
||
"""Финальное ревью M-2: испорченная копия того же размера лечится повторной загрузкой."""
|
||
files.put(root, b"abc")
|
||
(root / ABC).write_bytes(b"xyz") # тот же размер, другое содержимое
|
||
assert files.put(root, b"abc") == (ABC, 3)
|
||
assert (root / ABC).read_bytes() == b"abc"
|
||
|
||
|
||
# --- allow_empty, touch, sweep_downloads (#83) ---------------------------------
|
||
|
||
def test_put_allow_empty(root):
|
||
assert files.put(root, b"", allow_empty=True) == (hashlib.sha256(b"").hexdigest(), 0)
|
||
with pytest.raises(files.EmptyFile):
|
||
files.put(root, b"")
|
||
|
||
|
||
def test_touch_updates_mtime(root):
|
||
sha, _ = files.put(root, b"abc")
|
||
old = time.time() - 5 * DAY
|
||
os.utime(root / sha, (old, old))
|
||
files.touch(root, sha)
|
||
assert (root / sha).stat().st_mtime > time.time() - 60
|
||
files.touch(root, "0" * 64) # файла нет — не ошибка
|
||
files.touch(root, "../x") # негодный sha — не ошибка
|
||
|
||
|
||
def test_sweep_downloads(tmp_path):
|
||
root = tmp_path / "downloads"
|
||
old, fresh = root / "31337", root / "31338"
|
||
for d in (old, fresh):
|
||
d.mkdir(parents=True)
|
||
(d / "a.png").write_bytes(b"x")
|
||
(root / "loose.txt").write_text("x")
|
||
t = time.time() - 31 * DAY
|
||
os.utime(old / "a.png", (t, t))
|
||
os.utime(old, (t, t))
|
||
os.utime(root / "loose.txt", (t, t))
|
||
assert files.sweep_downloads(root) == ["31337"]
|
||
assert not old.exists() and fresh.exists() and (root / "loose.txt").exists()
|
||
assert files.sweep_downloads(tmp_path / "нет") == []
|
||
|
||
|
||
@pytest.mark.parametrize("func,attr", [(files.size, "stat"), (files.read, "read_bytes")])
|
||
def test_unreadable_file_is_none_with_warning(root, monkeypatch, caplog, func, attr):
|
||
from pathlib import Path
|
||
sha = files.put(root, b"abc")[0]
|
||
|
||
def denied(self, *args, **kwargs):
|
||
raise PermissionError(13, "Permission denied")
|
||
monkeypatch.setattr(Path, attr, denied)
|
||
with caplog.at_level("WARNING"):
|
||
assert func(root, sha) is None
|
||
assert sha in caplog.text and "не читается" in caplog.text
|
||
|
||
|
||
@pytest.mark.parametrize("func", [files.size, files.read])
|
||
def test_directory_instead_of_file_is_none(root, caplog, func):
|
||
(root / ABC).mkdir(parents=True)
|
||
with caplog.at_level("WARNING"):
|
||
assert func(root, ABC) is None
|
||
|
||
|
||
@pytest.mark.parametrize("func", [files.size, files.read])
|
||
def test_missing_file_is_none_without_log(root, caplog, func):
|
||
root.mkdir()
|
||
with caplog.at_level("WARNING"):
|
||
assert func(root, ABC) is None
|
||
assert caplog.text == ""
|