diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/support.py b/tests/support.py new file mode 100644 index 0000000..6763d2a --- /dev/null +++ b/tests/support.py @@ -0,0 +1,91 @@ +"""Synthetic message stubs and async helpers, so every test runs offline. + +media.py is duck-typed precisely so these stubs can exist: a test message needs +only the attributes its case is about. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from datetime import UTC, datetime + +_DEFAULT_EXT = {"photo": ".jpg", "video": ".mp4", "audio": ".mp3", + "voice": ".ogg", "document": ".bin"} +_MIME = {"photo": "image/jpeg", "video": "video/mp4", "audio": "audio/mpeg", + "voice": "audio/ogg", "document": "application/octet-stream"} + + +@dataclass +class FakeFile: + name: str | None = None + ext: str | None = None + size: int | None = None + mime_type: str | None = None + + +class FakeMsg: + """A Telethon Message as far as this codebase is concerned.""" + + def __init__(self, id: int, *, kind: str | None = None, + grouped_id: int | None = None, date: datetime | None = None, + name: str | None = None, ext: str | None = None, + size: int | None = 1000, text: str | None = None, + web_preview: object | None = None, sender_id: int | None = None, + sender: object | None = None) -> None: + self.id = id + self.grouped_id = grouped_id + self.date = date or datetime(2026, 1, 1, tzinfo=UTC) + self.message = text + self.sender_id = sender_id + self.sender = sender + self.reply_to = None + self.web_preview = web_preview + # Every media property media_kind() consults, default absent. + self.photo = self.video = self.document = None + self.audio = self.voice = self.gif = self.video_note = None + self.file = None + if kind is not None: + setattr(self, kind, object()) # the property media_kind reads + self.file = FakeFile(name=name, + ext=ext or _DEFAULT_EXT.get(kind, ".bin"), + size=size, + mime_type=_MIME.get(kind)) + + +class FakeClient: + """Serves a fixed ascending message list, honoring offset_id exclusively - + the semantics verified against telethon 1.44.0's _MessagesIter.""" + + def __init__(self, messages: list[FakeMsg]) -> None: + self.messages = messages + self.calls: list[int] = [] + + def iter_messages(self, entity, *, reverse=False, offset_id=0, **kwargs): + assert reverse is True, "the sweep must always be chronological" + self.calls.append(offset_id) + + async def gen(): + for m in self.messages: + if m.id > offset_id: + yield m + + return gen() + + +def collect(agen) -> list: + """Drain an async generator from a synchronous test.""" + + async def drain(): + return [item async for item in agen] + + return asyncio.run(drain()) + + +def run(coro): + return asyncio.run(coro) + + +async def as_agen(items): + for item in items: + yield item diff --git a/tests/test_download_decisions.py b/tests/test_download_decisions.py new file mode 100644 index 0000000..1cf01c6 --- /dev/null +++ b/tests/test_download_decisions.py @@ -0,0 +1,452 @@ +"""Per-file download decisions and the sink loop, against a tmpdir and a fake +fetcher. No network. + +The cases here are the ones where "looks like success" and "is success" can +diverge: dedupe, partial files, the disk guard, and Invariant 2. +""" + +from __future__ import annotations + +import errno +import json +import os +import stat + +import pytest +from telethon.errors import ( + AuthKeyUnregisteredError, + ChannelPrivateError, + FileReferenceExpiredError, + MediaEmptyError, + ServerError, +) +from telethon.errors.common import InvalidBufferError + +from telegram_exporter import downloader, paths +from telegram_exporter.downloader import ( + DOWNLOADED, + FAILED, + SKIPPED, + Config, + download_one, + run_download, + sweep_part_files, + write_title, +) +from telegram_exporter.session import Abort +from telegram_exporter.sidecar import Sidecar +from telegram_exporter.state import State +from telegram_exporter.traversal import Post, build_filters +from tests.support import FakeMsg, as_agen, run + +PAYLOAD = b"x" * 1000 +KIB = 1024 + + +@pytest.fixture(autouse=True) +def no_real_backoff(monkeypatch): + """The retry ladder is 5/10/20 s in production; tests assert the ladder is + walked, not that it sleeps.""" + monkeypatch.setattr(downloader, "backoff", lambda attempt: 0) + + +class FakeFetcher: + def __init__(self, payload: bytes = PAYLOAD, raises=None, refresh_to="same"): + self.payload = payload + self.raises = list(raises or []) # exceptions to raise, in order + self.refresh_to = refresh_to + self.calls = 0 + self.refresh_calls = 0 + + async def fetch(self, msg, fh) -> None: + self.calls += 1 + if self.raises: + raise self.raises.pop(0) + fh.write(self.payload) + + async def refresh(self, message_id): + self.refresh_calls += 1 + return FakeMsg(message_id, kind="photo") if self.refresh_to == "same" \ + else self.refresh_to + + +def cfg(tmp_path, *, min_free=0): + return Config(min_free=min_free, disk_path=tmp_path) + + +def post_dir_for(tmp_path, post_id=1042): + root = paths.export_root(tmp_path, -1001) + target = paths.post_dir(root, post_id) + target.mkdir(parents=True, exist_ok=True) + return root, target + + +# --------------------------------------------------------------------------- # +# The plan's decision table +# --------------------------------------------------------------------------- # + +def test_target_absent_downloads(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher() + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert result.path.read_bytes() == PAYLOAD + assert fetcher.calls == 1 + assert list(target_dir.glob("*.part")) == [] + + +def test_an_existing_non_empty_target_is_skipped_without_a_download(tmp_path): + # Existence, not size equality. fsync before os.replace means a file at the + # target path can only have arrived fully downloaded - and a size-equality + # gate would delete every completed photo on resume, because photo .size + # describes the variant Telethon picked. + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=999_999) + (target_dir / "1042_photo.jpg").write_bytes(b"already here") + fetcher = FakeFetcher() + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == SKIPPED + assert fetcher.calls == 0 + + +def test_a_zero_byte_target_is_unlinked_and_refetched(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + (target_dir / "1042_photo.jpg").touch() + fetcher = FakeFetcher() + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert fetcher.calls == 1 + assert result.path.read_bytes() == PAYLOAD + + +def test_stray_part_files_are_swept(tmp_path): + root, target_dir = post_dir_for(tmp_path) + (target_dir / "1042_photo.jpg.part").write_bytes(b"half") + (root / "9/9_video.mp4.part").parent.mkdir(parents=True) + (root / "9/9_video.mp4.part").write_bytes(b"half") + + assert sweep_part_files(root) == 2 + assert list(root.rglob("*.part")) == [] + + +def test_a_valid_target_wins_over_a_leftover_part_file(tmp_path): + root, target_dir = post_dir_for(tmp_path) + (target_dir / "1042_photo.jpg").write_bytes(PAYLOAD) + (target_dir / "1042_photo.jpg.part").write_bytes(b"half") + sweep_part_files(root) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + + result = run(download_one(FakeFetcher(), target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == SKIPPED + assert (target_dir / "1042_photo.jpg").read_bytes() == PAYLOAD + assert list(target_dir.glob("*.part")) == [] + + +def test_unknown_declared_size_is_accepted_by_the_validator(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="document", size=None) + + result = run(download_one(FakeFetcher(), target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert result.declared_size is None + + +def test_a_hostile_declared_filename_lands_inside_the_post_dir(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="document", name="../../../etc/passwd", + size=len(PAYLOAD)) + + result = run(download_one(FakeFetcher(), target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert result.path.parent == target_dir.resolve() + assert result.path.name == "1042_passwd.bin" + + +def test_the_disk_guard_aborts_before_writing_a_single_byte(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="video", size=10 * KIB) + fetcher = FakeFetcher() + + with pytest.raises(Abort) as excinfo: + run(download_one(fetcher, target_dir, msg, + cfg=cfg(tmp_path, min_free=1 << 62))) + + assert excinfo.value.code == 3 + assert fetcher.calls == 0 + assert list(target_dir.iterdir()) == [] + + +# --------------------------------------------------------------------------- # +# Validator and error taxonomy +# --------------------------------------------------------------------------- # + +def test_a_photo_size_mismatch_is_accepted_as_a_variant(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD) + 431) + + result = run(download_one(FakeFetcher(), target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert result.declared_size != result.size # recorded, not hidden + + +def test_a_document_size_mismatch_retries_then_fails_without_raising(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="document", size=len(PAYLOAD) + 1) + fetcher = FakeFetcher() + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == FAILED + assert fetcher.calls == downloader.MAX_ATTEMPTS + assert list(target_dir.glob("*.part")) == [] + + +def test_an_expired_file_reference_is_refreshed_once_then_succeeds(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher(raises=[FileReferenceExpiredError(request=None)]) + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert fetcher.refresh_calls == 1 + + +def test_a_deleted_message_is_recorded_and_skipped(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher(raises=[FileReferenceExpiredError(request=None)], + refresh_to=None) + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == FAILED + assert result.error == "message deleted" + + +def test_unavailable_media_is_recorded_and_skipped(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher(raises=[MediaEmptyError(request=None)]) + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == FAILED + assert "unavailable" in result.error + + +def test_a_transient_network_error_retries_and_then_succeeds(tmp_path): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + # ConnectionResetError with errno=None is the case that would fall through + # the errno test into "unexpected OS error" if OSError were caught first. + fetcher = FakeFetcher(raises=[ConnectionResetError()]) + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert fetcher.calls == 2 + + +def test_a_zero_byte_download_is_never_accepted(tmp_path): + """download_media returns without writing when the media resolves to an empty + variant. The photo excuse in validate() would otherwise rename a 0-byte file + to the target, where the "zero bytes is junk" dedupe rule could never see it + again - a completed export does not revisit existing targets.""" + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher(payload=b"") + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == FAILED + assert fetcher.calls == downloader.MAX_ATTEMPTS + assert not (target_dir / "1042_photo.jpg").exists() + assert list(target_dir.glob("*.part")) == [] + + +def test_an_unsafe_filename_skips_one_file_instead_of_ending_the_run(tmp_path): + # Defense in depth behind the extension sanitizer: one hostile message must + # never be able to halt an export permanently, which is what happened while + # the ValueError from safe_join escaped and the cursor could not advance. + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher() + + # Simulate a hole in the sanitizer rather than asserting one exists. + with pytest.MonkeyPatch.context() as mp: + mp.setattr(paths, "derive_filename", lambda m: "../escape.jpg") + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == FAILED + assert "unsafe filename" in result.error + assert fetcher.calls == 0 + assert not (target_dir.parent / "escape.jpg").exists() + + +@pytest.mark.parametrize("exc", [ + # A plain -500: RpcCallFailError is only one leaf of ServerError, so catching + # the leaf alone let the parent escape and end the run. + ServerError(request=None, message="INTERNAL_SERVER_ERROR"), + InvalidBufferError(payload=b"\x00\x00\x00\x00"), # corrupt packet on a flaky link +]) +def test_transport_level_errors_retry_rather_than_killing_the_run(tmp_path, exc): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + fetcher = FakeFetcher(raises=[exc]) + + result = run(download_one(fetcher, target_dir, msg, cfg=cfg(tmp_path))) + + assert result.status == DOWNLOADED + assert fetcher.calls == 2 + + +@pytest.mark.parametrize("exc,code", [ + (AuthKeyUnregisteredError(request=None), 4), + (ChannelPrivateError(request=None), 5), + (OSError(errno.ENOSPC, "No space left on device"), 3), +]) +def test_terminal_errors_abort_with_their_own_exit_code(tmp_path, exc, code): + _, target_dir = post_dir_for(tmp_path) + msg = FakeMsg(1042, kind="photo", size=len(PAYLOAD)) + + with pytest.raises(Abort) as excinfo: + run(download_one(FakeFetcher(raises=[exc]), target_dir, msg, + cfg=cfg(tmp_path))) + + assert excinfo.value.code == code + assert list(target_dir.glob("*.part")) == [] + + +# --------------------------------------------------------------------------- # +# The sink loop - Invariant 2 +# --------------------------------------------------------------------------- # + +def drive(tmp_path, posts, *, limit=None, fetcher=None, filters=None): + root = paths.export_root(tmp_path, -1001) + root.mkdir(parents=True, exist_ok=True) + state = State.open(root, chat_id=-1001, chat_title="G", + filters=filters or build_filters()) + with Sidecar(root) as sidecar: + totals = run(run_download(as_agen(posts), fetcher=fetcher or FakeFetcher(), + state=state, sidecar=sidecar, root=root, + cfg=cfg(tmp_path), limit=limit)) + records = [json.loads(line) + for line in (root / "messages.jsonl").read_text().splitlines()] + return root, state, totals, records + + +def album(post_id, ids, *, kind="photo"): + messages = [FakeMsg(i, kind=kind, grouped_id=7, size=len(PAYLOAD)) for i in ids] + return Post(post_id=post_id, messages=messages, max_message_id=max(ids)) + + +def test_a_completed_post_commits_the_cursor_and_writes_the_sidecar(tmp_path): + root, state, totals, records = drive(tmp_path, [album(1042, [1042, 1043])]) + + assert state.cursor_id == 1043 + assert totals.downloaded == 2 + assert totals.posts == 1 + assert [r["message_id"] for r in records] == [1042, 1043] + assert records[0]["files"][0]["path"] == "1042/1042_photo.jpg" + assert (root / "1042" / "1042_photo.jpg").exists() + + +def test_the_cursor_is_never_committed_from_an_abort_path(tmp_path): + # Invariant 2: state.commit is unreachable from every error exit, so the + # cursor can never point past an in-flight post. This is the failure that + # would look exactly like success. + posts = [album(1042, [1042]), album(2000, [2000])] + root = paths.export_root(tmp_path, -1001) + root.mkdir(parents=True, exist_ok=True) + state = State.open(root, chat_id=-1001, chat_title="G", filters=build_filters()) + + class FailSecond(FakeFetcher): + async def fetch(self, msg, fh): + if msg.id == 2000: + raise ChannelPrivateError(request=None) + await super().fetch(msg, fh) + + with Sidecar(root) as sidecar: + with pytest.raises(Abort) as excinfo: + run(run_download(as_agen(posts), fetcher=FailSecond(), state=state, + sidecar=sidecar, root=root, cfg=cfg(tmp_path))) + + assert excinfo.value.code == 5 + assert state.cursor_id == 1042 # the first post, not the second + assert State.read(root)["completed_at"] is None + + +def test_a_second_run_re_downloads_nothing(tmp_path): + posts = [album(1042, [1042, 1043])] + drive(tmp_path, posts) + fetcher = FakeFetcher() + drive(tmp_path, posts, fetcher=fetcher) + + assert fetcher.calls == 0 + + +def test_empty_and_text_only_posts_advance_the_cursor_without_counting(tmp_path): + posts = [Post(post_id=10, messages=[], max_message_id=11), + Post(post_id=12, messages=[FakeMsg(12, text="hi")], max_message_id=12)] + root, state, totals, records = drive( + tmp_path, posts, filters=build_filters(include_text=True)) + + assert state.cursor_id == 12 + assert totals.posts == 0 and totals.downloaded == 0 + assert [r["message_id"] for r in records] == [12] + assert records[0]["files"] == [] # context, not media + assert not (root / "12").exists() # no folder for a text message + + +def test_limit_counts_posts_with_files(tmp_path): + posts = [Post(post_id=10, messages=[], max_message_id=10), + album(11, [11]), album(12, [12]), album(13, [13])] + _, state, totals, _ = drive(tmp_path, posts, limit=2) + + assert totals.posts == 2 + assert state.cursor_id == 12 # stopped after the second, not the third + + +def test_completed_at_is_set_when_the_sweep_is_exhausted(tmp_path): + root, _, _, _ = drive(tmp_path, [album(1042, [1042])]) + assert State.read(root)["completed_at"] is not None + + +def test_the_whole_export_tree_is_private_under_the_process_umask(tmp_path): + """plan.md's "export tree 0700, every file 0600" rested on umask alone, with + nothing observing the result. The sidecar carries other people's message text + and names, so this is the criterion worth asserting rather than assuming.""" + old = os.umask(0o077) + try: + root, _, _, _ = drive(tmp_path, [album(1042, [1042, 1043])]) + write_title(root, "My Group") # cli writes this; drive() does not + finally: + os.umask(old) + + checked = [root, root / "1042", root / "1042" / "1042_photo.jpg", + root / "messages.jsonl", root / ".export-state.json", + root / "title.txt"] + for path in checked: + assert path.exists(), path + assert stat.S_IMODE(path.stat().st_mode) & 0o077 == 0, \ + f"{path} is group/world accessible: {oct(path.stat().st_mode)}" + + +def test_limit_zero_connects_and_does_nothing(tmp_path): + root, state, totals, _ = drive(tmp_path, [album(1042, [1042])], limit=0) + assert totals.posts == 0 + assert state.cursor_id == 0 + assert not (root / "1042").exists() diff --git a/tests/test_estimate.py b/tests/test_estimate.py new file mode 100644 index 0000000..f8ae7af --- /dev/null +++ b/tests/test_estimate.py @@ -0,0 +1,180 @@ +"""The dry-run estimator. + +The property under test is anti-drift: the estimate has to describe exactly what +a real run from the same start would fetch, because that number is the one users +make disk decisions on. +""" + +from __future__ import annotations + +import pytest + +from telegram_exporter import paths +from telegram_exporter.downloader import run_download +from telegram_exporter.estimate import EstimateSink, describe_filters, resume_from, run_estimate +from telegram_exporter.sidecar import Sidecar +from telegram_exporter.state import STATE_NAME, State +from telegram_exporter.traversal import Post, build_filters +from tests.support import FakeMsg, as_agen, run +from tests.test_download_decisions import PAYLOAD, FakeFetcher, cfg + +GIB = 1 << 30 + + +def album(post_id, ids, *, kind="photo", size=len(PAYLOAD)): + return Post(post_id=post_id, + messages=[FakeMsg(i, kind=kind, grouped_id=7, size=size) + for i in ids], + max_message_id=max(ids)) + + +def estimate(tmp_path, posts, *, min_free=0, limit=None, filters=None): + root = paths.export_root(tmp_path, -1001) + sink = EstimateSink() + filters = filters or build_filters() + code = run(run_estimate(as_agen(posts), sink=sink, root=root, title="My Group", + peer_id=-1001, filters=filters, + start_id=resume_from(root, filters), + min_free=min_free, limit=limit)) + return sink, code, root + + +# --------------------------------------------------------------------------- # +# Counting +# --------------------------------------------------------------------------- # + +def test_counts_are_broken_down_by_kind(tmp_path): + posts = [album(1, [1, 2], kind="photo", size=100), + album(3, [3], kind="video", size=5000)] + sink, code, _ = estimate(tmp_path, posts) + + assert sink.posts == 2 and sink.files == 3 + assert sink.by_kind == {"photo": [2, 200], "video": [1, 5000]} + assert sink.total_bytes == 5200 + assert code == 0 + + +def test_unknown_size_files_get_their_own_tally_never_a_zero(tmp_path): + # An unbounded estimate error must be visible, so these are counted on their + # own line rather than coerced into the byte total. + posts = [album(1, [1], kind="document", size=None)] + sink, _, _ = estimate(tmp_path, posts) + + assert sink.unknown_size_files == 1 + assert sink.files == 1 + assert sink.total_bytes == 0 + + +def test_empty_and_text_only_posts_are_not_counted(tmp_path): + posts = [Post(post_id=1, messages=[], max_message_id=2), + Post(post_id=3, messages=[FakeMsg(3, text="hi")], max_message_id=3), + album(4, [4])] + sink, _, _ = estimate(tmp_path, posts) + + assert sink.posts == 1 and sink.files == 1 + + +# --------------------------------------------------------------------------- # +# Verdict +# --------------------------------------------------------------------------- # + +def test_the_verdict_is_short_and_exit_two_when_it_will_not_fit(tmp_path, capsys): + _, code, _ = estimate(tmp_path, [album(1, [1])], min_free=1 << 62) + assert code == 2 + out = capsys.readouterr().out + assert "SHORT by" in out + # A SHORT verdict is advisory: the per-file check_disk is the enforcement, + # and the run stops partway with a resumable cursor rather than corrupting. + assert "advisory" in out + + +def test_the_verdict_fits_when_there_is_room(tmp_path, capsys): + _, code, _ = estimate(tmp_path, [album(1, [1])], min_free=0) + assert code == 0 + assert "FITS" in capsys.readouterr().out + + +def test_photo_totals_are_labelled_approximate(tmp_path, capsys): + estimate(tmp_path, [album(1, [1], kind="photo"), album(2, [2], kind="video")]) + out = capsys.readouterr().out + photo_line = next(line for line in out.splitlines() if "photos" in line) + video_line = next(line for line in out.splitlines() if "videos" in line) + assert "approximate" in photo_line + assert "approximate" not in video_line + + +# --------------------------------------------------------------------------- # +# The cursor fix, and no side effects +# --------------------------------------------------------------------------- # + +def test_dry_run_writes_nothing_at_all(tmp_path): + before = sorted(p.name for p in tmp_path.iterdir()) + _, _, root = estimate(tmp_path, [album(1, [1])]) + + assert not root.exists() # not even the export root + assert sorted(p.name for p in tmp_path.iterdir()) == before + + +def test_dry_run_starts_where_the_real_run_would(tmp_path): + # Without reading the cursor, `--dry-run && run` fails closed on every + # resumed export: a mostly-downloaded group would be re-measured in full, + # exit 2, and refuse work that fits comfortably. + root = paths.export_root(tmp_path, -1001) + root.mkdir(parents=True) + filters = build_filters(types="photo") + State.open(root, chat_id=-1001, chat_title="G", filters=filters).commit(1042) + + assert resume_from(root, filters) == 1042 + + +def test_a_filter_change_makes_dry_run_measure_from_zero(tmp_path): + # The real run would refuse (exit 7), so reusing its cursor would describe a + # run that cannot happen. + root = paths.export_root(tmp_path, -1001) + root.mkdir(parents=True) + State.open(root, chat_id=-1001, chat_title="G", + filters=build_filters(types="photo")).commit(1042) + + assert resume_from(root, build_filters(types="video")) == 0 + + +def test_resume_from_is_zero_with_no_prior_state(tmp_path): + assert resume_from(paths.export_root(tmp_path, -1001), build_filters()) == 0 + assert not (tmp_path / STATE_NAME).exists() + + +# --------------------------------------------------------------------------- # +# Anti-drift: the estimate and the real run describe the same work +# --------------------------------------------------------------------------- # + +@pytest.mark.parametrize("limit", [None, 2]) +def test_dry_run_and_real_run_agree_on_posts_and_files(tmp_path, limit): + posts = [album(1, [1, 2]), album(3, [3]), album(4, [4, 5, 6]), + Post(post_id=7, messages=[], max_message_id=7), album(8, [8])] + + sink, _, _ = estimate(tmp_path, posts, limit=limit) + + root = paths.export_root(tmp_path / "real", -1001) + root.mkdir(parents=True) + state = State.open(root, chat_id=-1001, chat_title="G", filters=build_filters()) + with Sidecar(root) as sidecar: + totals = run(run_download(as_agen(posts), fetcher=FakeFetcher(), state=state, + sidecar=sidecar, root=root, + cfg=cfg(tmp_path), limit=limit)) + + assert sink.posts == totals.posts + assert sink.files == totals.downloaded + # Bytes agree exactly here because the fake payload matches the declared + # size; against the live API only photo sizes are approximate. + assert sink.total_bytes == totals.bytes_downloaded + + +def test_filters_are_described_for_the_report_header(): + described = describe_filters(build_filters(types="video,photo", + since="2026-01-01", + max_size="100MB", + include_text=True)) + assert "types=photo,video" in described + assert "since=2026-01-01" in described + assert "max-size" in described and "include-text" in described + assert describe_filters(build_filters()) == "(none)" diff --git a/tests/test_lock.py b/tests/test_lock.py new file mode 100644 index 0000000..983f778 --- /dev/null +++ b/tests/test_lock.py @@ -0,0 +1,103 @@ +"""Single-instance lock under real process contention. + +flock is tested with a second process because that is the only thing that +exercises it: two acquisitions inside one process would succeed on most +platforms and prove nothing. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from telegram_exporter.cli import exclusive +from telegram_exporter.session import Abort + +REPO_SRC = str(Path(__file__).resolve().parents[1] / "src") + +HOLDER = """ +import sys, time +from pathlib import Path +from telegram_exporter.cli import exclusive +with exclusive(Path(sys.argv[1])): + print("HELD", flush=True) + time.sleep(60) +""" + +CONTENDER = """ +import sys +from pathlib import Path +from telegram_exporter.cli import exclusive +from telegram_exporter.session import Abort +try: + with exclusive(Path(sys.argv[1])): + sys.exit(0) +except Abort as a: + print(a.reason, file=sys.stderr) + sys.exit(a.code) +""" + + +def _env(): + env = dict(os.environ) + env["PYTHONPATH"] = REPO_SRC + os.pathsep + env.get("PYTHONPATH", "") + return env + + +@pytest.fixture +def holder(tmp_path): + lock = tmp_path / ".export.lock" + proc = subprocess.Popen([sys.executable, "-c", HOLDER, str(lock)], + stdout=subprocess.PIPE, text=True, env=_env()) + assert proc.stdout.readline().strip() == "HELD", "holder failed to acquire" + try: + yield lock + finally: + proc.kill() + proc.wait(timeout=10) + + +def test_a_second_acquire_exits_three(holder): + result = subprocess.run([sys.executable, "-c", CONTENDER, str(holder)], + capture_output=True, text=True, env=_env(), timeout=30) + assert result.returncode == 3 + assert "busy" in result.stderr + + +def test_contention_names_the_holder(holder): + with pytest.raises(Abort) as excinfo: + with exclusive(holder): + pass + assert excinfo.value.code == 3 + for field in ("pid=", "host=", "started="): + assert field in excinfo.value.reason + + +def test_contention_touches_no_file(holder, tmp_path): + before = sorted(p.name for p in tmp_path.iterdir()) + with pytest.raises(Abort): + with exclusive(holder): + pass + assert sorted(p.name for p in tmp_path.iterdir()) == before + + +def test_the_lock_is_released_when_the_holder_dies(tmp_path): + lock = tmp_path / ".export.lock" + proc = subprocess.Popen([sys.executable, "-c", HOLDER, str(lock)], + stdout=subprocess.PIPE, text=True, env=_env()) + assert proc.stdout.readline().strip() == "HELD" + proc.kill() # kill -9: flock releases on process death, + proc.wait(timeout=10) # so there is no stale-lock cleanup to get wrong + with exclusive(lock): + pass + + +def test_different_export_roots_do_not_contend(holder, tmp_path): + # Exporting two different groups concurrently is legitimate, so the lock is + # per-root rather than global. + with exclusive(tmp_path / "other" / ".export.lock"): + pass diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..99b5a92 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,64 @@ +"""Logging policy and the credential surface. + +telethon's request logging carries api_id, phone_number and phone_code_hash, so +its floor is enforced by this test rather than by a checkbox. +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import pytest + +from telegram_exporter.cli import build_parser, configure_logging + +SRC = Path(__file__).resolve().parents[1] / "src" / "telegram_exporter" +SOURCES = sorted(SRC.glob("*.py")) + + +@pytest.mark.parametrize("verbose", [False, True]) +def test_the_telethon_logger_never_drops_below_warning(verbose): + configure_logging(verbose=verbose) + assert logging.getLogger("telethon").getEffectiveLevel() >= logging.WARNING + assert logging.getLogger("asyncio").getEffectiveLevel() >= logging.WARNING + + +def test_verbose_affects_only_our_own_logger(): + configure_logging(verbose=True) + assert logging.getLogger("telegram_exporter").level == logging.DEBUG + configure_logging(verbose=False) + assert logging.getLogger("telegram_exporter").level == logging.INFO + + +def test_there_is_no_flag_that_unmutes_telethon(): + options = [action.option_strings for action in build_parser()._actions] + flat = [opt for group in options for opt in group] + assert not any("telethon" in opt or "debug" in opt for opt in flat) + + +def test_the_2fa_password_is_reachable_only_through_getpass(): + """No flag, no env var, no file - it is needed once per session lifetime, and + prompting costs nothing.""" + joined = "\n".join(p.read_text() for p in SOURCES) + assert joined.count("getpass(") == 1 + for forbidden in ["TG_PASSWORD", "TG_2FA", "TG_PASS", "--password", "--2fa"]: + assert forbidden not in joined + + +def test_no_credential_is_read_from_anywhere_but_the_documented_sources(): + joined = "\n".join(p.read_text() for p in SOURCES) + env_reads = {"TG_API_ID", "TG_API_HASH", "TG_PHONE", "XDG_STATE_HOME"} + found = {token for token in env_reads if token in joined} + assert found == env_reads # and nothing else + # The app does not read .env; the shell already handles that. + assert "dotenv" not in joined + + +def test_no_credential_appears_in_the_captured_log_output(caplog): + configure_logging(verbose=True) + log = logging.getLogger("telegram_exporter.test") + with caplog.at_level(logging.DEBUG): + log.debug("group %r (id %s)", "My Group", -1001234567890) + assert "api_hash" not in caplog.text + assert "+1555" not in caplog.text diff --git a/tests/test_no_concurrency.py b/tests/test_no_concurrency.py new file mode 100644 index 0000000..a4ac641 --- /dev/null +++ b/tests/test_no_concurrency.py @@ -0,0 +1,62 @@ +"""Concurrency 1, enforced by an AST scan rather than by discipline. + +Telethon does not parallelize transfers, so parallelism buys nothing and costs +flood-wait escalation: waits climb from ~7 s to 4 h+ once Telegram decides an +account is abusive. The temptation arrives precisely when an export feels slow, +which is why this is a test and not a comment. + +There is deliberately no --workers flag for this test to have to police. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import pytest + +SRC = Path(__file__).resolve().parents[1] / "src" / "telegram_exporter" + +FORBIDDEN = { + "gather", "create_task", "ensure_future", "as_completed", "wait_for_all", + "to_thread", "run_in_executor", "TaskGroup", + "ThreadPoolExecutor", "ProcessPoolExecutor", "Thread", "Process", "Pool", +} + + +def _called_names(tree: ast.AST): + for node in ast.walk(tree): + if isinstance(node, ast.Call): + func = node.func + if isinstance(func, ast.Attribute): + yield func.attr, node.lineno + elif isinstance(func, ast.Name): + yield func.id, node.lineno + elif isinstance(node, ast.ImportFrom): + for alias in node.names: + yield alias.name, node.lineno + elif isinstance(node, ast.AsyncWith): + pass + + +@pytest.mark.parametrize("path", sorted(SRC.glob("*.py")), ids=lambda p: p.name) +def test_no_parallelism_primitive_is_used(path): + tree = ast.parse(path.read_text(), filename=str(path)) + hits = [(name, line) for name, line in _called_names(tree) if name in FORBIDDEN] + assert not hits, f"{path.name} uses a parallelism primitive: {hits}" + + +def test_the_scan_would_actually_catch_something(): + # A guard that cannot fail is worse than no guard: it makes passing the + # default. This is the same shape of mistake the album-adjacency test had. + tree = ast.parse("import asyncio\n" + "async def f():\n" + " await asyncio.gather(g(), h())\n") + assert any(name in FORBIDDEN for name, _ in _called_names(tree)) + + +def test_there_is_no_workers_flag(): + from telegram_exporter.cli import build_parser + flat = [opt for action in build_parser()._actions for opt in action.option_strings] + assert not any("worker" in opt or "concurren" in opt or "parallel" in opt + for opt in flat) diff --git a/tests/test_paths.py b/tests/test_paths.py new file mode 100644 index 0000000..2fb86c7 --- /dev/null +++ b/tests/test_paths.py @@ -0,0 +1,150 @@ +"""Path safety. Every row of the plan's test table, plus the invariant that no +sanitized name can escape its post directory.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from telegram_exporter.paths import ( + MAX_NAME_BYTES, + derive_filename, + export_root, + post_dir, + safe_join, + sanitize, +) +from tests.support import FakeMsg + + +@pytest.mark.parametrize("raw,expected", [ + ("../../../etc/passwd", "passwd"), + ("a/b/c.jpg", "c.jpg"), + ("..", "FALLBACK"), + (".", "FALLBACK"), + ("", "FALLBACK"), + ("...", "FALLBACK"), + ("\x00evil.jpg", "evil.jpg"), + ("....jpg", "jpg"), + ("-rf.jpg", "rf.jpg"), + ("--force me.jpg", "force me.jpg"), + ("a\\b\\c.jpg", "c.jpg"), + ("we:ird*na?me.jpg", "we_ird_na_me.jpg"), + ("tab\tsep.jpg", "tabsep.jpg"), # control chars are removed, not spaced + ("sp aced.jpg", "sp aced.jpg"), # whitespace runs collapse + ("/etc/shadow", "shadow"), +]) +def test_sanitize_table(raw, expected): + assert sanitize(raw, fallback="FALLBACK") == expected + + +def test_truncation_is_byte_counted_and_keeps_the_extension(): + # 300 CJK characters is ~900 bytes: a character-counted truncation would + # produce a name ext4 rejects with OSError: File name too long. + name = "あ" * 300 + ".jpg" + out = sanitize(name, fallback="x") + assert len(out.encode()) <= MAX_NAME_BYTES + assert out.endswith(".jpg") + assert len(out) > 10 # not collapsed to the fallback + + +def test_truncation_of_a_pathological_extension_still_yields_a_usable_name(): + out = sanitize("stem." + "e" * 400, fallback="fb") + assert 0 < len(out.encode()) <= MAX_NAME_BYTES + + +def test_no_sanitized_name_escapes_its_post_dir(tmp_path): + hostile = ["../../../etc/passwd", "..", "/etc/shadow", "a/../../b", + "\x00../x", "....//..//etc/passwd"] + base = post_dir(export_root(tmp_path, -1001), 1042) + for raw in hostile: + joined = safe_join(base, sanitize(raw, fallback="fb")) + assert joined.is_relative_to(base.resolve()) + + +def test_safe_join_raises_rather_than_repairing(): + base = Path("/tmp/exports/g-1001/1042") + for escape in ["../x", "../../etc/passwd", "/etc/passwd", ".."]: + with pytest.raises(ValueError): + safe_join(base, escape) + + +def test_export_root_is_the_chat_id_never_the_title(tmp_path): + assert export_root(tmp_path, -1001234567890).name == "g-1001234567890" + + +@pytest.mark.parametrize("bad", ["..", "../../etc", "", "1042; rm -rf", None, 1.5, True]) +def test_int_only_path_components(tmp_path, bad): + # A group title cannot reach a path component: there is no code path from + # title to export_root, and anything non-int is refused by type. + with pytest.raises(TypeError): + export_root(tmp_path, bad) + with pytest.raises(TypeError): + post_dir(tmp_path, bad) + + +def test_derive_filename_is_keyed_on_message_id(): + msg = FakeMsg(1043, kind="photo") + assert derive_filename(msg) == "1043_photo.jpg" + + +def test_hostile_declared_filename_becomes_a_leaf(): + msg = FakeMsg(7, kind="document", name="../../../etc/passwd", ext=".bin") + assert derive_filename(msg) == "7_passwd.bin" + + +def test_two_media_in_one_post_get_distinct_names(): + a, b = FakeMsg(1042, kind="photo", grouped_id=9), FakeMsg(1043, kind="photo", + grouped_id=9) + assert derive_filename(a) != derive_filename(b) + + +def test_filename_is_identical_across_filter_sets(): + # derive_filename takes no filter argument, which is the point: the same + # message addresses the same file whether --types photo or --types video was + # used, so dedupe survives a filter change. + msg = FakeMsg(1044, kind="video", name="clip.mp4") + assert derive_filename(msg) == derive_filename(msg) == "1044_clip.mp4" + + +@pytest.mark.parametrize("name,expected", [ + ("holiday.\x00jpg", "7_holiday.jpg"), # NUL: used to escape as ValueError + # The ESC byte is removed; the remaining "[31m" is inert text, not an escape. + ("clip.\x1b[31mmp4", "7_clip.[31mmp4"), + ("doc.j:pg", "7_doc.j_pg"), # reserved character + ("shot.jp g", "7_shot.jp g"), + ("archive.tar.gz", "7_archive.tar.gz"), # a real double extension survives + ("weird.", "7_weird.bin"), # empty extension falls back +]) +def test_the_extension_is_sanitized_like_the_stem(name, expected): + """The extension comes from the same attacker-supplied filename attribute as + the stem. Sanitizing only the stem left a hole: a NUL in the extension made + Path.resolve raise inside safe_join, and that ValueError escaped the download + loop - so the cursor never advanced past the message and every later run died + on it.""" + assert derive_filename(FakeMsg(7, kind="document", name=name)) == expected + + +def test_no_control_character_survives_anywhere_in_a_filename(): + for raw in ["a\x00.b\x00c", "\x1b].jpg", "x.\ny", "n.\r\tz"]: + out = derive_filename(FakeMsg(3, kind="document", name=raw)) + assert not any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in out), out + + +def test_a_whole_filename_stays_inside_ext4s_255_byte_component_limit(): + # The stem is capped at 200 UTF-8 bytes, so the "{id}_" prefix and the + # extension have to be byte-counted too or the total overflows. + worst = FakeMsg(9_999_999_999_999_999, + kind="document", + name="あ" * 300 + "." + "ん" * 16) + assert len(derive_filename(worst).encode()) <= 255 + + +def test_missing_name_falls_back_to_the_media_kind(): + assert derive_filename(FakeMsg(5, kind="voice")) == "5_voice.ogg" + + +def test_fallbacks_never_include_a_timestamp_or_anything_nondeterministic(): + msg = FakeMsg(11, kind="document", name=" ", ext=".pdf") + assert derive_filename(msg) == derive_filename(msg) == "11_document.pdf" diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..aad270e --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,249 @@ +"""Session file privacy, the git-safety refusal, and the flood-wait primitives. + +Phase 2's success criteria name all three; none of them belongs in the logging or +lock test files, and every one is a silent-failure mode if it regresses. +""" + +from __future__ import annotations + +import os +import re +import stat +import subprocess +from pathlib import Path + +import pytest +from telethon.errors import FloodWaitError + +from telegram_exporter import session as session_mod +from telegram_exporter.session import ( + Abort, + _flood_sleep, + aiter_with_flood_retry, + assert_not_committable, + assert_session_private, + default_session_path, + load_api_credentials, + prepare_session_path, + with_flood_retry, +) +from tests.support import FakeMsg, collect, run + +CLI_SRC = Path(__file__).resolve().parents[1] / "src" / "telegram_exporter" / "cli.py" + + +@pytest.fixture +def no_sleep(monkeypatch): + """Assert the wait *happens* without waiting for it.""" + slept = [] + + async def fake_sleep(seconds): + slept.append(seconds) + + monkeypatch.setattr(session_mod.asyncio, "sleep", fake_sleep) + return slept + + +# --------------------------------------------------------------------------- # +# Session file privacy +# --------------------------------------------------------------------------- # + +def test_the_session_suffix_is_added_so_the_right_file_is_protected(tmp_path): + # Telethon appends '.session' when absent, so chmod-ing the un-suffixed path + # passed to --session would protect nothing. + created = prepare_session_path(tmp_path / "mine") + assert created.name == "mine.session" + assert created.exists() + + +def test_a_new_session_file_is_created_private(tmp_path): + created = prepare_session_path(tmp_path / "s.session") + assert stat.S_IMODE(created.stat().st_mode) & 0o077 == 0 + + +def test_an_existing_loose_session_file_is_tightened(tmp_path): + loose = tmp_path / "s.session" + loose.touch() + os.chmod(loose, 0o644) + + prepare_session_path(loose) + + assert stat.S_IMODE(loose.stat().st_mode) == 0o600 + + +def test_every_sqlite_sibling_is_checked_not_just_the_session(tmp_path): + # -journal, -wal and -shm hold the same secrets as the database itself. + created = prepare_session_path(tmp_path / "s.session") + sibling = tmp_path / "s.session-journal" + sibling.touch() + os.chmod(sibling, 0o644) + + with pytest.raises(SystemExit, match="insecure mode"): + assert_session_private(created) + + +def test_a_private_session_passes_the_assertion(tmp_path): + created = prepare_session_path(tmp_path / "s.session") + (tmp_path / "s.session-journal").touch(mode=0o600) + assert_session_private(created) + + +def test_the_default_session_path_is_outside_any_repo(): + # The safe location is the default, so the git check only bites a + # deliberately chosen path. + assert default_session_path().is_absolute() + assert "tg-export" in str(default_session_path()) + + +def test_umask_is_the_first_statement_of_main(): + # It has to precede anything that can create a file: sqlite writes the auth + # key *during* login, so a post-hoc chmod is provably too late. + body = CLI_SRC.read_text().split("def main()")[1] + first = next(line.strip() for line in body.splitlines()[1:] if line.strip()) + assert first.startswith("os.umask(0o077)") + + +# --------------------------------------------------------------------------- # +# Git safety - robust to --out / --session pointing anywhere +# --------------------------------------------------------------------------- # + +def _git_repo(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + (repo / ".gitignore").write_text("ignored/\n*.session\n") + return repo + + +def test_a_committable_path_inside_a_repo_is_refused(tmp_path): + repo = _git_repo(tmp_path) + with pytest.raises(SystemExit, match="refusing"): + assert_not_committable(repo / "exports", is_dir=True) + + +def test_an_ignored_path_inside_a_repo_is_allowed(tmp_path): + repo = _git_repo(tmp_path) + (repo / "ignored").mkdir() + assert_not_committable(repo / "ignored") + assert_not_committable(repo / "creds.session") + + +def test_a_directory_pattern_matches_before_the_directory_exists(tmp_path): + # git check-ignore returns "not ignored" for `exports` when it cannot see + # that it is a directory, so without the trailing-slash retry the export root + # would be refused on the first run and accepted on every run after it. + repo = _git_repo(tmp_path) + assert not (repo / "ignored").exists() + assert_not_committable(repo / "ignored", is_dir=True) + assert_not_committable(repo / "ignored" / "g-1001", is_dir=True) + + +def test_the_directory_retry_never_excuses_a_file(tmp_path): + # The retry asks git to read the path as a directory, which would be unsound + # for a session file: a `.gitignore` line of `creds.session/` must not excuse + # a *file* named creds.session, which git would happily commit. + repo = tmp_path / "repo" + repo.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + (repo / ".gitignore").write_text("creds.session/\n") + + assert_not_committable(repo / "creds.session", is_dir=True) # a dir: ignored + with pytest.raises(SystemExit): + assert_not_committable(repo / "creds.session") # a file: refused + + +def test_a_path_outside_any_repo_is_allowed(tmp_path): + assert_not_committable(tmp_path / "anywhere" / "exports") + + +def test_a_nonexistent_path_is_still_checked(tmp_path): + # An export root does not exist on a first run, which is exactly when this + # check matters; running git from a nonexistent cwd would raise instead. + repo = _git_repo(tmp_path) + with pytest.raises(SystemExit): + assert_not_committable(repo / "deep" / "not" / "created" / "yet") + + +# --------------------------------------------------------------------------- # +# Credentials +# --------------------------------------------------------------------------- # + +def test_missing_credentials_produce_an_actionable_error(monkeypatch): + monkeypatch.delenv("TG_API_ID", raising=False) + monkeypatch.delenv("TG_API_HASH", raising=False) + with pytest.raises(Abort) as excinfo: + load_api_credentials() + assert "my.telegram.org" in excinfo.value.reason + + +def test_a_non_numeric_api_id_is_named(monkeypatch): + monkeypatch.setenv("TG_API_ID", "abc") + monkeypatch.setenv("TG_API_HASH", "hash") + with pytest.raises(Abort, match="must be an integer"): + load_api_credentials() + + +# --------------------------------------------------------------------------- # +# Flood-wait primitives +# --------------------------------------------------------------------------- # + +def test_a_flood_wait_is_slept_through_once_by_default(no_sleep): + calls = [] + + async def flaky(): + calls.append(1) + if len(calls) == 1: + raise FloodWaitError(request=None, capture=30) + return "ok" + + assert run(with_flood_retry(flaky)) == "ok" + assert len(calls) == 2 + assert no_sleep and no_sleep[0] >= 30 # the wait plus jitter, never a tight retry + + +def test_a_wait_beyond_the_explicit_ceiling_exits_six(no_sleep): + with pytest.raises(Abort) as excinfo: + run(_flood_sleep(FloodWaitError(request=None, capture=7200), 3600)) + assert excinfo.value.code == 6 + assert not no_sleep # aborted instead of sleeping + + +def test_with_no_ceiling_even_a_four_hour_wait_is_slept(no_sleep): + run(_flood_sleep(FloodWaitError(request=None, capture=4 * 3600), None)) + assert no_sleep[0] >= 4 * 3600 + + +def test_the_flood_log_names_the_computed_wake_time(no_sleep, caplog): + # The mitigation for the no-ceiling default: an unattended four-hour sleep + # has to read as a sleep rather than as a hang. + with caplog.at_level("WARNING"): + run(_flood_sleep(FloodWaitError(request=None, capture=90), None)) + assert "sleeping until" in caplog.text + assert re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", caplog.text) + + +def test_a_mid_sweep_flood_wait_resumes_from_the_last_yielded_id(no_sleep): + """The single-value wrapper cannot cover `async for`: Telethon raises on the + *next page fetch*, deep inside the loop. Rebuilding from the last yielded id + is what keeps the sweep gap-free.""" + messages = [FakeMsg(i, kind="photo") for i in (10, 11, 12, 13)] + attempts = [] + + def make_agen(since): + attempts.append(since) + + async def gen(): + for m in messages: + if m.id <= since: + continue + if m.id == 12 and len(attempts) == 1: + raise FloodWaitError(request=None, capture=5) + yield m + + return gen() + + got = collect(aiter_with_flood_retry(make_agen, start_id=0)) + + assert [m.id for m in got] == [10, 11, 12, 13] # no gap, no replay + assert attempts == [0, 11] # rebuilt from the last yielded + assert len(no_sleep) == 1 diff --git a/tests/test_sidecar.py b/tests/test_sidecar.py new file mode 100644 index 0000000..c2a113c --- /dev/null +++ b/tests/test_sidecar.py @@ -0,0 +1,121 @@ +"""messages.jsonl: partial-line repair, rotation, and union semantics. + +Not in the plan's original file list, but phase 5's success criteria name both +behaviours ("a truncated final line is repaired at startup", "--reset-state +rotates the sidecar"), and neither belongs in the download-decision table. +""" + +from __future__ import annotations + +import json + +from telegram_exporter.sidecar import SIDECAR_NAME, Sidecar +from telegram_exporter.traversal import Post +from tests.support import FakeMsg + + +def lines(root): + return (root / SIDECAR_NAME).read_text().splitlines() + + +def test_a_partial_trailing_line_is_truncated_at_startup(tmp_path): + # A host crash mid-write otherwise leaves a fragment that breaks every JSONL + # consumer downstream. + path = tmp_path / SIDECAR_NAME + path.write_text('{"message_id": 1}\n{"message_id": 2, "fi') + + with Sidecar(tmp_path): + pass + + assert lines(tmp_path) == ['{"message_id": 1}'] + + +def test_a_complete_final_line_without_a_newline_is_kept(tmp_path): + path = tmp_path / SIDECAR_NAME + path.write_text('{"message_id": 1}\n{"message_id": 2}') + + with Sidecar(tmp_path): + pass + + assert [json.loads(line)["message_id"] for line in lines(tmp_path)] == [1, 2] + + +def test_an_intact_file_is_left_alone(tmp_path): + path = tmp_path / SIDECAR_NAME + path.write_text('{"message_id": 1}\n') + + with Sidecar(tmp_path): + pass + + assert lines(tmp_path) == ['{"message_id": 1}'] + + +def test_an_empty_or_absent_file_is_not_a_repair_case(tmp_path): + with Sidecar(tmp_path): + pass + assert lines(tmp_path) == [] + + +def test_rotation_preserves_the_old_records_under_a_timestamp(tmp_path): + (tmp_path / SIDECAR_NAME).write_text('{"message_id": 1}\n') + sidecar = Sidecar(tmp_path) + + rotated = sidecar.rotate() + + assert rotated is not None and rotated.exists() + assert rotated.name.startswith("messages-") and rotated.suffix == ".jsonl" + assert not (tmp_path / SIDECAR_NAME).exists() + # Rotation, not deletion: the current sidecar describes exactly one filter + # regime while the previous one stays inspectable. + assert json.loads(rotated.read_text())["message_id"] == 1 + + +def test_rotating_a_missing_sidecar_is_a_no_op(tmp_path): + assert Sidecar(tmp_path).rotate() is None + + +def test_records_are_append_only_events_so_consumers_union_them(tmp_path): + """Two runs under different filters both record message 1043. Reading only + the last record would report a narrower file list than what is on disk.""" + msg_photo = FakeMsg(1043, kind="photo", grouped_id=8) + post = Post(post_id=1042, messages=[msg_photo], max_message_id=1043) + + for _ in range(2): + with Sidecar(tmp_path) as sidecar: + sidecar.append(post, []) + sidecar.fsync() + + records = [json.loads(line) for line in lines(tmp_path)] + assert [r["message_id"] for r in records] == [1043, 1043] + assert all(r["post_id"] == 1042 and r["grouped_id"] == 8 for r in records) + + +def test_a_text_only_message_gets_the_same_shape_with_no_files(tmp_path): + post = Post(post_id=12, messages=[FakeMsg(12, text="hello")], max_message_id=12) + with Sidecar(tmp_path) as sidecar: + sidecar.append(post, []) + sidecar.fsync() + + record = json.loads(lines(tmp_path)[0]) + assert record["files"] == [] and record["caption"] == "hello" + assert record["errors"] == [] + + +def test_sender_name_is_only_taken_from_an_already_cached_sender(tmp_path): + # Never an extra get_entity call: on a 100k-message sweep that is 100k extra + # RPCs and a flood ban, in exchange for a display string. + class Sender: + first_name, last_name, username, title = "Alice", None, "alice", None + + with_sender = FakeMsg(1, kind="photo", sender=Sender(), sender_id=777) + without = FakeMsg(2, kind="photo", sender=None, sender_id=778) + post = Post(post_id=1, messages=[with_sender, without], max_message_id=2) + + with Sidecar(tmp_path) as sidecar: + sidecar.append(post, []) + sidecar.fsync() + + records = [json.loads(line) for line in lines(tmp_path)] + assert records[0]["sender_name"] == "Alice" + assert records[1]["sender_name"] is None + assert records[1]["sender_id"] == 778 diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 0000000..39caaca --- /dev/null +++ b/tests/test_state.py @@ -0,0 +1,105 @@ +"""The resume cursor: Invariant 3 (filters stored verbatim, mismatch refused with +a named diff) and atomic saves.""" + +from __future__ import annotations + +import json + +import pytest + +from telegram_exporter.session import Abort +from telegram_exporter.state import STATE_NAME, State +from telegram_exporter.traversal import build_filters + +CHAT = -1001234567890 + + +def open_state(root, *, filters=None, chat_id=CHAT, reset=False): + return State.open(root, chat_id=chat_id, chat_title="My Group", + filters=filters or build_filters(), reset=reset) + + +def test_a_new_state_file_starts_at_zero(tmp_path): + state = open_state(tmp_path) + assert state.cursor_id == 0 + assert (tmp_path / STATE_NAME).exists() + assert json.loads((tmp_path / STATE_NAME).read_text())["chat_id"] == CHAT + + +def test_commit_persists_and_leaves_no_temp_file(tmp_path): + state = open_state(tmp_path) + state.commit(1042) + assert json.loads((tmp_path / STATE_NAME).read_text())["cursor_id"] == 1042 + assert list(tmp_path.glob("*.tmp")) == [] + assert open_state(tmp_path).cursor_id == 1042 + + +def test_a_cursor_regression_is_ignored(tmp_path): + state = open_state(tmp_path) + state.commit(1042) + state.commit(7) + assert state.cursor_id == 1042 + + +def test_completed_at_is_set_only_on_clean_exhaustion(tmp_path): + state = open_state(tmp_path) + assert state.data["completed_at"] is None + state.mark_completed() + assert state.data["completed_at"] is not None + # A new run clears it: the field answers "did *this* export finish?". + assert open_state(tmp_path).data["completed_at"] is None + + +def test_a_filter_change_is_refused_and_names_what_changed(tmp_path): + open_state(tmp_path, filters=build_filters(types="photo,video", + max_size="100MB")).commit(1042) + with pytest.raises(Abort) as excinfo: + open_state(tmp_path, filters=build_filters(types="video")) + assert excinfo.value.code == 7 + reason = excinfo.value.reason + assert "filter mismatch" in reason + assert "types" in reason and "max_size" in reason + assert "--reset-state" in reason + + +def test_toggling_include_text_is_a_filter_change(tmp_path): + # It changes what a post *contains*, so running without it and then with it + # against the same cursor would silently skip every text record. + open_state(tmp_path, filters=build_filters()).commit(50) + with pytest.raises(Abort) as excinfo: + open_state(tmp_path, filters=build_filters(include_text=True)) + assert excinfo.value.code == 7 + assert "include_text" in excinfo.value.reason + + +def test_a_different_chat_against_an_existing_state_file_is_refused(tmp_path): + open_state(tmp_path) + with pytest.raises(Abort) as excinfo: + open_state(tmp_path, chat_id=-100999) + assert excinfo.value.code == 7 + assert "chat mismatch" in excinfo.value.reason + + +def test_reset_state_starts_over_without_complaining(tmp_path): + open_state(tmp_path, filters=build_filters(types="photo")).commit(1042) + fresh = open_state(tmp_path, filters=build_filters(types="video"), reset=True) + assert fresh.cursor_id == 0 + assert fresh.data["filters"]["types"] == ["video"] + + +def test_limit_is_not_part_of_the_stored_filter_set(tmp_path): + # A `--limit 20` run is a resumable partial export; if --limit were stored, + # a limited test run would poison every later full run. + assert "limit" not in open_state(tmp_path).data["filters"] + + +def test_a_corrupt_state_file_is_refused_rather_than_ignored(tmp_path): + (tmp_path / STATE_NAME).write_text("{not json") + with pytest.raises(Abort) as excinfo: + open_state(tmp_path) + assert excinfo.value.code == 7 + + +def test_read_is_side_effect_free_for_dry_run(tmp_path): + assert State.read(tmp_path) is None + assert not (tmp_path / STATE_NAME).exists() diff --git a/tests/test_traversal.py b/tests/test_traversal.py new file mode 100644 index 0000000..54f90e1 --- /dev/null +++ b/tests/test_traversal.py @@ -0,0 +1,251 @@ +"""Album grouping, filter-invariant identity, and the album-split tripwire. + +All synthetic: no client, no network. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from telegram_exporter.media import media_size +from telegram_exporter.traversal import ( + AlbumSplitError, + Filters, + SweepOrderError, + TraversalError, + build_filters, + iter_posts, + keep, + parse_date, + parse_size, +) +from tests.support import FakeClient, FakeMsg, collect + +NO_FILTERS = Filters() + + +def sweep(messages, *, filters=NO_FILTERS, after_id=0): + return collect(iter_posts(FakeClient(messages), object(), + after_id=after_id, filters=filters)) + + +# --------------------------------------------------------------------------- # +# Grouping +# --------------------------------------------------------------------------- # + +def test_a_three_photo_album_is_one_post(): + posts = sweep([FakeMsg(i, kind="photo", grouped_id=7) for i in (10, 11, 12)]) + assert len(posts) == 1 + assert posts[0].post_id == 10 + assert [m.id for m in posts[0].messages] == [10, 11, 12] + assert posts[0].max_message_id == 12 + + +def test_two_adjacent_albums_do_not_merge(): + posts = sweep([FakeMsg(10, kind="photo", grouped_id=7), + FakeMsg(11, kind="photo", grouped_id=7), + FakeMsg(12, kind="photo", grouped_id=8), + FakeMsg(13, kind="photo", grouped_id=8)]) + assert [p.post_id for p in posts] == [10, 12] + + +def test_a_standalone_message_between_albums_gets_its_own_post(): + posts = sweep([FakeMsg(10, kind="photo", grouped_id=7), + FakeMsg(11, kind="video"), + FakeMsg(12, kind="photo", grouped_id=8)]) + assert [p.post_id for p in posts] == [10, 11, 12] + + +def test_an_album_at_the_end_of_history_is_flushed(): + posts = sweep([FakeMsg(10, kind="video"), + FakeMsg(11, kind="photo", grouped_id=9), + FakeMsg(12, kind="photo", grouped_id=9)]) + assert [p.post_id for p in posts] == [10, 11] + assert len(posts[-1].messages) == 2 + + +def test_after_id_is_an_exclusive_lower_bound(): + msgs = [FakeMsg(i, kind="photo") for i in (10, 11, 12)] + assert [p.post_id for p in sweep(msgs, after_id=11)] == [12] + + +# --------------------------------------------------------------------------- # +# Invariant 1 - identity is filter-invariant +# --------------------------------------------------------------------------- # + +def test_identity_survives_a_filter_dropping_members(): + album = [FakeMsg(10, kind="photo", grouped_id=7), + FakeMsg(11, kind="video", grouped_id=7), + FakeMsg(12, kind="photo", grouped_id=7)] + photos = sweep(album, filters=build_filters(types="photo"))[0] + videos = sweep(album, filters=build_filters(types="video"))[0] + + assert photos.post_id == videos.post_id == 10 + assert photos.max_message_id == videos.max_message_id == 12 + assert [m.id for m in photos.messages] == [10, 12] + assert [m.id for m in videos.messages] == [11] + + +def test_a_fully_filtered_group_still_yields_a_post_so_the_cursor_advances(): + # Skipping empty posts stalls the cursor: a --types video run over a + # photo-heavy tail would advance only at videos, so every resume re-sweeps + # tens of thousands of messages to download nothing. + posts = sweep([FakeMsg(10, kind="photo", grouped_id=7), + FakeMsg(11, kind="photo", grouped_id=7)], + filters=build_filters(types="video")) + assert len(posts) == 1 + assert posts[0].messages == [] + assert posts[0].max_message_id == 11 + + +# --------------------------------------------------------------------------- # +# The tripwire and the ordering assertion +# --------------------------------------------------------------------------- # + +def test_a_reopened_grouped_id_raises_album_split_error(): + with pytest.raises(AlbumSplitError, match="grouped_id 7 reopened"): + sweep([FakeMsg(10, kind="photo", grouped_id=7), + FakeMsg(11, kind="photo", grouped_id=8), + FakeMsg(12, kind="photo", grouped_id=9), + FakeMsg(13, kind="photo", grouped_id=7)]) # A B C A + + +def test_the_tripwire_catches_an_immediately_reopened_album(): + with pytest.raises(AlbumSplitError): + sweep([FakeMsg(10, kind="photo", grouped_id=7), + FakeMsg(11, kind="video"), + FakeMsg(12, kind="photo", grouped_id=7)]) # A B A + + +def test_a_non_ascending_stream_is_refused(): + class Descending(FakeClient): + def iter_messages(self, entity, **kwargs): + async def gen(): + for m in self.messages: + yield m + return gen() + + with pytest.raises(SweepOrderError, match="not strictly ascending"): + collect(iter_posts(Descending([FakeMsg(12, kind="photo"), + FakeMsg(10, kind="photo")]), + object(), filters=NO_FILTERS)) + + +def test_the_album_buffer_is_bounded(): + with pytest.raises(SweepOrderError, match="album buffer exceeded"): + sweep([FakeMsg(i, kind="photo", grouped_id=7) for i in range(100, 130)]) + + +def test_the_sweep_tripwires_survive_python_dash_o(): + # `python -O` strips bare asserts. These two guard the same silent corruption + # as AlbumSplitError beside them, so they are real exceptions. + assert issubclass(SweepOrderError, TraversalError) + assert issubclass(AlbumSplitError, TraversalError) + assert not issubclass(SweepOrderError, AssertionError) + + +# --------------------------------------------------------------------------- # +# Filters +# --------------------------------------------------------------------------- # + +def test_text_only_messages_are_dropped_by_default(): + posts = sweep([FakeMsg(10, text="hello"), FakeMsg(11, kind="photo")]) + assert [p.post_id for p in posts] == [10, 11] + assert posts[0].messages == [] # yielded, but empty + assert len(posts[1].messages) == 1 + + +def test_include_text_keeps_text_messages_without_media(): + posts = sweep([FakeMsg(10, text="hello")], + filters=build_filters(include_text=True)) + assert len(posts[0].messages) == 1 + assert posts[0].messages[0].file is None # nothing to download + + +def test_a_link_preview_is_not_media(): + # Telethon's Message.photo also returns the web preview's photo, so without + # the web_preview guard every message containing a URL would look like an + # attachment. + msg = FakeMsg(10, kind="photo", text="look", web_preview=object()) + assert keep(msg, NO_FILTERS) is False + assert keep(msg, build_filters(include_text=True)) is True + + +def test_filters_compose(): + filters = build_filters(types="video", since="2026-01-01", max_size="50MB") + inside = FakeMsg(1, kind="video", size=1_000_000, + date=datetime(2026, 6, 1, tzinfo=UTC)) + too_old = FakeMsg(2, kind="video", size=1_000_000, + date=datetime(2025, 6, 1, tzinfo=UTC)) + too_big = FakeMsg(3, kind="video", size=99_000_000, + date=datetime(2026, 6, 1, tzinfo=UTC)) + wrong_kind = FakeMsg(4, kind="photo", size=1_000, + date=datetime(2026, 6, 1, tzinfo=UTC)) + assert keep(inside, filters) is True + assert not any(keep(m, filters) for m in (too_old, too_big, wrong_kind)) + + +def test_unknown_size_is_included_by_max_size_never_silently_dropped(): + msg = FakeMsg(1, kind="document", size=None) + assert media_size(msg) is None + assert keep(msg, build_filters(max_size="10MB")) is True + + +def test_media_size_is_none_rather_than_zero_for_all_three_consumers(): + # The estimator, the validator and the disk guard each have their own rule + # for None; what matters here is that the helper never raises and never + # invents a 0. + assert media_size(FakeMsg(1, kind="document", size=None)) is None + assert media_size(FakeMsg(2, text="hi")) is None + assert media_size(FakeMsg(3, kind="photo", size=17)) == 17 + + +def test_unknown_types_are_rejected_at_parse_time(): + with pytest.raises(ValueError, match="unknown media types"): + build_filters(types="photo,sticker") + + +@pytest.mark.parametrize("spec", [" ", ",", " , ", ",,"]) +def test_a_token_less_types_value_is_refused(spec): + """`--types " "` (easy to produce from an unset shell variable) used to build + an empty kind set: a filter that drops every media kind while serializing as + "no filter". A run would sweep all history, download nothing, commit an + end-of-history cursor and set completed_at - and a later correct run would see + matching filters, resume from that cursor and report success having fetched + nothing. Silent total data loss, so it is refused at the boundary.""" + with pytest.raises(ValueError, match="names no media kind"): + build_filters(types=spec) + + +def test_an_empty_kind_set_could_not_masquerade_as_no_filter(): + # Second line of defense, in case a caller bypasses build_filters. + assert Filters(types=frozenset()).to_state()["types"] == [] + assert Filters(types=None).to_state()["types"] is None + + +@pytest.mark.parametrize("text,expected", [ + ("100MB", 100_000_000), ("2GiB", 2 * 2**30), ("1500000", 1_500_000), + ("50 MB", 50_000_000), ("1KiB", 1024), ("1.5GB", 1_500_000_000), +]) +def test_parse_size(text, expected): + assert parse_size(text) == expected + + +def test_parse_size_rejects_nonsense(): + with pytest.raises(ValueError): + parse_size("plenty") + + +def test_dates_are_utc_aware_so_comparison_cannot_raise(): + # msg.date is UTC-aware; a naive bound would raise TypeError mid-sweep. + assert parse_date("2026-01-01").tzinfo is not None + assert parse_date("2026-01-01T00:00:00+07:00").tzinfo is not None + + +def test_stored_filters_are_verbatim_and_comparable(): + stored = build_filters(types="video,photo", max_size="100MB", + include_text=True).to_state() + assert stored == {"types": ["photo", "video"], "since": None, "until": None, + "max_size": 100_000_000, "include_text": True}