ElebobaandClaude Opus 5.5
dbb17bd774
- UploadsRepo: локальный клон, fetch не чаще fetch_interval, ls-tree -z (пробелы/кириллица), cat-file, последний коммит по пути; git без shell, ssh с -F /dev/null (без конфига хоста) - «📝 Коммит»: папки/файлы с размерами, ⬆️ Вверх, страницы по 20, карточка файла с последним коммитом, «👁 Показать» (<pre> или документом); в кнопках только индексы - ключ для контейнера: GIT_SSH_KEY_B64 → временный файл 0600 (смонтированный файл хоста контейнерному uid не прочитать); volume /data для клона - тесты на настоящем bare-репозитории во временной папке Refs #8 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.repo.git import GitError, UploadsRepo
|
|
from tests.conftest import git
|
|
|
|
|
|
async def test_clone_and_list_root_dirs_first(repo: UploadsRepo) -> None:
|
|
await repo.refresh()
|
|
|
|
entries = await repo.list_dir()
|
|
assert [(e.name, e.kind) for e in entries] == [
|
|
("docs", "dir"),
|
|
("many", "dir"),
|
|
("services", "dir"),
|
|
("README.md", "file"),
|
|
]
|
|
assert entries[-1].size == len("# uploads\n")
|
|
|
|
|
|
async def test_unicode_and_spaces_in_names(repo: UploadsRepo) -> None:
|
|
await repo.refresh()
|
|
|
|
names = [e.path for e in await repo.list_dir("docs")]
|
|
assert names == ["docs/logo.bin", "docs/мои заметки.md"]
|
|
assert (await repo.read("docs/мои заметки.md")).decode() == "привет\n"
|
|
|
|
|
|
async def test_last_commit_per_path(repo: UploadsRepo) -> None:
|
|
await repo.refresh()
|
|
|
|
hello = await repo.last_commit("services/hello/main.py")
|
|
docs = await repo.last_commit("docs/logo.bin")
|
|
assert hello is not None and hello.subject == "init" and hello.author == "Тест"
|
|
assert docs is not None and docs.subject == "docs and many" and len(docs.sha) == 40
|
|
|
|
|
|
async def test_missing_dir_is_an_error(repo: UploadsRepo) -> None:
|
|
await repo.refresh()
|
|
with pytest.raises(GitError, match="no such directory"):
|
|
await repo.list_dir("nope")
|
|
|
|
|
|
async def test_refresh_sees_new_commits_and_respects_interval(repo: UploadsRepo, remote: Path, tmp_path: Path) -> None:
|
|
await repo.refresh()
|
|
before = await repo.head()
|
|
|
|
other = tmp_path / "other"
|
|
git(tmp_path, "clone", "-q", str(remote), str(other))
|
|
(other / "new.txt").write_text("x", encoding="utf-8")
|
|
git(other, "add", "-A")
|
|
git(other, "commit", "-q", "-m", "new")
|
|
git(other, "push", "-q", "origin", "main")
|
|
|
|
await repo.refresh()
|
|
assert await repo.head() != before
|
|
assert "new.txt" in [e.name for e in await repo.list_dir()]
|
|
|
|
|
|
async def test_fetch_is_throttled(remote: Path, tmp_path: Path) -> None:
|
|
from app.config import GitSettings
|
|
|
|
throttled = UploadsRepo(GitSettings(uploads_url=str(remote), clone_dir=tmp_path / "c2", fetch_interval=3600))
|
|
await throttled.refresh()
|
|
head = await throttled.head()
|
|
other = tmp_path / "o2"
|
|
git(tmp_path, "clone", "-q", str(remote), str(other))
|
|
(other / "x.txt").write_text("x", encoding="utf-8")
|
|
git(other, "add", "-A")
|
|
git(other, "commit", "-q", "-m", "x")
|
|
git(other, "push", "-q", "origin", "main")
|
|
|
|
await throttled.refresh()
|
|
assert await throttled.head() == head # within the interval: no fetch
|
|
await throttled.refresh(force=True)
|
|
assert await throttled.head() != head
|
|
|
|
|
|
async def test_unreachable_remote(tmp_path: Path) -> None:
|
|
from app.config import GitSettings
|
|
|
|
broken = UploadsRepo(GitSettings(uploads_url=str(tmp_path / "missing.git"), clone_dir=tmp_path / "c3"))
|
|
with pytest.raises(GitError, match="clone"):
|
|
await broken.refresh()
|