ElebobaandClaude Opus 5.5
0f27f405a7
- make demo-repo: локальный bare-репозиторий загрузок из infra/uploads-template - make run: бот на этой машине (SQLite, настройки из .env) - fix: при относительном clone_dir клон уезжал в data/data/uploads (git clone запускается из родительской папки) — путь теперь абсолютный; тест на это - data/ в .gitignore Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
97 lines
3.3 KiB
Python
97 lines
3.3 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()
|
|
|
|
|
|
async def test_relative_clone_dir(remote: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.config import GitSettings
|
|
|
|
monkeypatch.chdir(tmp_path)
|
|
relative = UploadsRepo(GitSettings(uploads_url=str(remote), clone_dir=Path("data/uploads")))
|
|
await relative.refresh()
|
|
|
|
assert (tmp_path / "data" / "uploads" / ".git").is_dir()
|
|
assert "README.md" in [e.name for e in await relative.list_dir()]
|