- 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>
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
import subprocess
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncEngine
|
|
|
|
from app.access.service import AccessService
|
|
from app.config import AccessSettings, GitSettings, RuntimeSettings, Settings, TelegramSettings
|
|
from app.db.migrate import upgrade
|
|
from app.db.session import SessionMaker, make_engine, make_sessionmaker
|
|
from app.repo.git import UploadsRepo
|
|
from app.texts import Texts, load_texts
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
ADMIN = 1001
|
|
OUTSIDER = 2002
|
|
|
|
|
|
@pytest.fixture
|
|
def texts() -> Texts:
|
|
return load_texts(ROOT / "config")
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(tmp_path: Path) -> Settings:
|
|
return Settings(
|
|
telegram=TelegramSettings(bot_token="42:TEST-TOKEN"),
|
|
access=AccessSettings(admin_ids=frozenset({ADMIN})),
|
|
runtime=RuntimeSettings(config_dir=ROOT / "config", heartbeat_file=tmp_path / "hb"),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def engine() -> AsyncIterator[AsyncEngine]:
|
|
"""Fresh in-memory SQLite with all migrations applied — the same path the bot takes at startup."""
|
|
eng = make_engine("sqlite+aiosqlite://")
|
|
await upgrade(eng)
|
|
yield eng
|
|
await eng.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def sessions(engine: AsyncEngine) -> SessionMaker:
|
|
return make_sessionmaker(engine)
|
|
|
|
|
|
@pytest.fixture
|
|
async def access(sessions: SessionMaker, settings: Settings) -> AccessService:
|
|
service = AccessService(sessions, settings.access)
|
|
await service.bootstrap_admins()
|
|
return service
|
|
|
|
|
|
def git(cwd: Path, *args: str) -> str:
|
|
env = {
|
|
"GIT_AUTHOR_NAME": "Тест",
|
|
"GIT_AUTHOR_EMAIL": "t@t",
|
|
"GIT_COMMITTER_NAME": "Тест",
|
|
"GIT_COMMITTER_EMAIL": "t@t",
|
|
"PATH": "/usr/bin:/bin",
|
|
"HOME": str(cwd),
|
|
}
|
|
return subprocess.run(["git", *args], cwd=cwd, env=env, check=True, capture_output=True, text=True).stdout
|
|
|
|
|
|
@pytest.fixture
|
|
def remote(tmp_path: Path) -> Path:
|
|
"""A bare 'SourceCraft' repo: services/hello, a file with a space and Cyrillic, a binary, 25 files in one dir."""
|
|
work = tmp_path / "seed"
|
|
work.mkdir()
|
|
git(work, "init", "-q", "-b", "main")
|
|
(work / "services" / "hello").mkdir(parents=True)
|
|
(work / "services" / "hello" / "main.py").write_text("print('hello')\n", encoding="utf-8")
|
|
(work / "README.md").write_text("# uploads\n", encoding="utf-8")
|
|
git(work, "add", "-A")
|
|
git(work, "commit", "-q", "-m", "init")
|
|
(work / "docs").mkdir()
|
|
(work / "docs" / "мои заметки.md").write_text("привет\n", encoding="utf-8")
|
|
(work / "docs" / "logo.bin").write_bytes(b"\x00\x01\x02binary")
|
|
(work / "many").mkdir()
|
|
for i in range(25):
|
|
(work / "many" / f"f{i:02}.txt").write_text(str(i), encoding="utf-8")
|
|
git(work, "add", "-A")
|
|
git(work, "commit", "-q", "-m", "docs and many")
|
|
bare = tmp_path / "remote.git"
|
|
git(tmp_path, "clone", "-q", "--bare", str(work), str(bare))
|
|
return bare
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(remote: Path, tmp_path: Path) -> UploadsRepo:
|
|
return UploadsRepo(GitSettings(uploads_url=str(remote), clone_dir=tmp_path / "clone", fetch_interval=0))
|