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))