test: cover path safety, resume, flood waits and the concurrency rule

180 tests, all offline against synthetic message stubs: no credentials,
no network, no live group.

The cases that matter are the ones where 'looks like success' and 'is
success' diverge - a cursor ahead of what is on disk, a split album, a
zero-byte file renamed to its target, an empty filter set serialized as
no filter at all.

Two guards are tested for their ability to fail rather than assumed:
the AST scan that forbids parallelism primitives is checked against a
sample that must trip it, and lock contention is proven with a real
second process, since two acquisitions in one process would succeed and
prove nothing.
This commit is contained in:
tiennm99 committed 2026-08-12 21:30:00 +07:00
1 parent 275ec47576
commit 5bf0992fe6
12 files changed
+1828

No files matched your search

View File
Whitespace-only changes.
+91
View File
@@ -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
+452
View File
@@ -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()
+180
View File
@@ -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)"
+103
View File
@@ -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
+64
View File
@@ -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
+62
View File
@@ -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)
+150
View File
@@ -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"
+249
View File
@@ -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
+121
View File
@@ -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
+105
View File
@@ -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()
+251
View File
@@ -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}