refactor(render): этап 3 — слияние близнецов вокруг _prepare_feed_posts (#30) #39
+159
-146
@@ -14,6 +14,7 @@ import logging
|
||||
import asyncio
|
||||
import re
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
from feedgen.feed import FeedGenerator
|
||||
@@ -28,6 +29,23 @@ Config = get_settings()
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PreparedFeed:
|
||||
"""Output of _prepare_feed_posts — everything both formatters need after
|
||||
fetch -> enrich -> render -> filter -> sanitize."""
|
||||
channel_username: str
|
||||
channel_title: str # used by RSS metadata only
|
||||
posts: list[dict] # rendered, filtered, sorted, SANITIZED
|
||||
|
||||
|
||||
class ChannelNotFound(Exception):
|
||||
"""Channel could not be resolved to a username; formatter -> create_error_feed."""
|
||||
def __init__(self, channel_identifier):
|
||||
# The prepared identifier (str or int), rendered into the error feed.
|
||||
self.channel_identifier = channel_identifier
|
||||
super().__init__(str(channel_identifier))
|
||||
|
||||
def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]:
|
||||
"""
|
||||
Create media groups based on time difference between messages
|
||||
@@ -309,18 +327,122 @@ def _render_pipeline(messages: list[Message],
|
||||
return posts
|
||||
|
||||
|
||||
async def _prepare_feed_posts(channel: str | int,
|
||||
client: Client,
|
||||
*,
|
||||
limit: int,
|
||||
exclude_flags: str | None,
|
||||
exclude_text: str | None,
|
||||
merge_seconds: int,
|
||||
history_limit: int,
|
||||
enrich_replies: bool,
|
||||
log_prefix: str) -> PreparedFeed:
|
||||
"""Single code path feeding both feeds: validate -> resolve chat -> fetch history ->
|
||||
(optionally) enrich replies -> render/filter/sort/sanitize in one worker thread. The
|
||||
RSS/HTML formatters used to duplicate this ~80% verbatim; path differences are now
|
||||
EXPLICIT parameters (history_limit: RSS over-fetches limit*2; enrich_replies: HTML-only;
|
||||
log_prefix 'rss'|'html' keeps today's paired log-line names).
|
||||
|
||||
Raises ChannelNotFound (formatter -> create_error_feed), re-raises errors.FloodWait
|
||||
for both the get_chat and the get_history path (§3.9; api_server -> HTTP 429), and
|
||||
wraps any other resolution/history error in ValueError with unified text (§3.10).
|
||||
"""
|
||||
# 1) Validate limit.
|
||||
if limit < 1:
|
||||
raise ValueError(f"limit must be positive, got {limit}")
|
||||
if limit > 200:
|
||||
raise ValueError(f"limit cannot exceed 200, got {limit}")
|
||||
|
||||
post_parser = PostParser(client=client)
|
||||
|
||||
# 2) Resolve the chat. NOTE: keep `from tg_cache import ...` INSIDE this function —
|
||||
# feed tests monkeypatch tg_cache and rely on late name resolution.
|
||||
channel_info_start_time = time.time()
|
||||
try:
|
||||
channel = post_parser.channel_name_prepare(channel)
|
||||
from tg_cache import cached_get_chat
|
||||
channel_info = await cached_get_chat(post_parser.client, channel)
|
||||
channel_title = channel_info.title or f"Telegram: {channel}"
|
||||
channel_username = channel_info.username or (str(channel_info.id) if channel_info.id and str(channel_info.id).startswith('-100') else None)
|
||||
if not channel_username:
|
||||
# Prepared channel (which could be int) is carried into the error feed.
|
||||
logger.warning(f"Could not get username for channel {channel}, using identifier for error feed.")
|
||||
raise ChannelNotFound(channel)
|
||||
except (errors.UsernameInvalid, errors.UsernameNotOccupied) as e:
|
||||
logger.warning(f"Channel not found error for {channel}: {str(e)}")
|
||||
raise ChannelNotFound(channel) from e
|
||||
except errors.FloodWait:
|
||||
# Let FloodWait bubble up to api_server.py, which returns HTTP 429 with Retry-After.
|
||||
raise
|
||||
except ChannelNotFound:
|
||||
# The no-username branch above — not a resolution failure to wrap in ValueError.
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error during get_chat for channel '{channel}' (type: {type(channel)}): {str(e)}", exc_info=True)
|
||||
raise ValueError(f"Failed to get chat info for {channel}: {str(e)}") from e
|
||||
|
||||
channel_info_elapsed = time.time() - channel_info_start_time
|
||||
logger.debug(f"{log_prefix}_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds")
|
||||
|
||||
# 3) Fetch history.
|
||||
messages_start_time = time.time()
|
||||
try:
|
||||
from tg_cache import cached_get_chat_history
|
||||
messages = await cached_get_chat_history(post_parser.client, channel, limit=history_limit)
|
||||
except errors.FloodWait:
|
||||
# §3.9: FloodWait from history propagates -> HTTP 429 (previously wrapped -> 400).
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error during get_chat_history for channel '{channel}' (type: {type(channel)}): {str(e)}", exc_info=True)
|
||||
raise ValueError(f"Failed to get chat history for {channel}: {str(e)}") from e
|
||||
|
||||
messages_elapsed = time.time() - messages_start_time
|
||||
logger.debug(f"{log_prefix}_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds")
|
||||
|
||||
# 4) Optional reply enrichment (HTML-only today — deliberate, keeps RSS polling cheap).
|
||||
if enrich_replies:
|
||||
enrichment_start_time = time.time()
|
||||
messages = await _reply_enrichment(client, messages)
|
||||
enrichment_elapsed = time.time() - enrichment_start_time
|
||||
logger.debug(f"{log_prefix}_reply_enrichment_timing: channel {channel}, replies enriched in {enrichment_elapsed:.3f} seconds")
|
||||
|
||||
# 5) Process messages into groups and render them. The whole grouping/trimming/
|
||||
# rendering pipeline is CPU-heavy (deepcopy + per-message rendering) and contains no
|
||||
# await, so run it in ONE worker thread to keep the event loop responsive.
|
||||
processing_start_time = time.time()
|
||||
try:
|
||||
posts = await asyncio.to_thread(
|
||||
_render_pipeline, messages, post_parser, limit,
|
||||
exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'],
|
||||
channel,
|
||||
)
|
||||
finally:
|
||||
# Persist media file-ids collected during rendering with a single bulk upsert —
|
||||
# in a finally so a partial render still records what it collected (the flush is
|
||||
# best-effort and swallows its own errors, so it cannot mask a render exception).
|
||||
await post_parser._flush_pending_media_ids()
|
||||
|
||||
processing_elapsed = time.time() - processing_start_time
|
||||
logger.debug(f"{log_prefix}_messages_processing_timing: channel {channel}, {len(posts)} posts processed in {processing_elapsed:.3f} seconds")
|
||||
|
||||
return PreparedFeed(channel_username=channel_username, channel_title=channel_title, posts=posts)
|
||||
|
||||
|
||||
async def generate_channel_rss(channel: str | int,
|
||||
client: Client,
|
||||
limit: int = 20,
|
||||
client: Client,
|
||||
limit: int = 20,
|
||||
exclude_flags: str | None = None,
|
||||
exclude_text: str | None = None,
|
||||
merge_seconds: int = 5
|
||||
) -> str:
|
||||
"""
|
||||
Generate RSS feed for channel using actual messages
|
||||
Generate RSS feed for channel using actual messages.
|
||||
|
||||
Thin formatter over the shared _prepare_feed_posts: it only turns the prepared,
|
||||
sanitized posts into a feedgen RSS document. RSS over-fetches (history_limit=limit*2)
|
||||
and skips reply enrichment (enrich_replies=False).
|
||||
Args:
|
||||
channel: Telegram channel name
|
||||
post_parser: Optional PostParser instance. If not provided, will create new one
|
||||
client: Telegram client instance
|
||||
limit: Maximum number of posts to include in the RSS feed
|
||||
exclude_flags: Flags to exclude from the RSS feed
|
||||
@@ -329,44 +451,22 @@ async def generate_channel_rss(channel: str | int,
|
||||
RSS feed as string in XML format
|
||||
"""
|
||||
total_start_time = time.time()
|
||||
|
||||
if limit < 1:
|
||||
raise ValueError(f"limit must be positive, got {limit}")
|
||||
if limit > 200:
|
||||
raise ValueError(f"limit cannot exceed 200, got {limit}")
|
||||
base_url = Config['pyrogram_bridge_url']
|
||||
|
||||
try:
|
||||
post_parser = PostParser(client=client)
|
||||
|
||||
prepared = await _prepare_feed_posts(
|
||||
channel, client,
|
||||
limit=limit, exclude_flags=exclude_flags, exclude_text=exclude_text,
|
||||
merge_seconds=merge_seconds, history_limit=limit * 2,
|
||||
enrich_replies=False, log_prefix="rss",
|
||||
)
|
||||
|
||||
channel_username = prepared.channel_username
|
||||
channel_title = prepared.channel_title
|
||||
final_posts = prepared.posts
|
||||
|
||||
fg = FeedGenerator()
|
||||
fg.load_extension('dc')
|
||||
base_url = Config['pyrogram_bridge_url']
|
||||
|
||||
channel_info_start_time = time.time()
|
||||
try:
|
||||
channel = post_parser.channel_name_prepare(channel)
|
||||
from tg_cache import cached_get_chat
|
||||
channel_info = await cached_get_chat(post_parser.client, channel)
|
||||
channel_title = channel_info.title or f"Telegram: {channel}"
|
||||
channel_username = channel_info.username or (str(channel_info.id) if channel_info.id and str(channel_info.id).startswith('-100') else None)
|
||||
if not channel_username:
|
||||
# Use prepared channel (which could be int) for error feed if username fails
|
||||
logger.warning(f"Could not get username for channel {channel}, using identifier for error feed.")
|
||||
return create_error_feed(str(channel), base_url) # Ensure channel is string for error feed
|
||||
except (errors.UsernameInvalid, errors.UsernameNotOccupied) as e:
|
||||
logger.warning(f"Channel not found error for {channel}: {str(e)}")
|
||||
return create_error_feed(channel, base_url)
|
||||
except errors.FloodWait:
|
||||
# Let FloodWait bubble up to api_server.py, which returns HTTP 429 with Retry-After header
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error during get_chat for channel '{channel}' (type: {type(channel)}): {str(e)}", exc_info=True) # Log error specifically for get_chat
|
||||
# Re-raise the original exception to be caught by the outer handler if needed,
|
||||
# but add specific logging here.
|
||||
raise ValueError(f"Failed to get chat info for {channel}: {str(e)}") from e # Raise a more specific error perhaps
|
||||
|
||||
channel_info_elapsed = time.time() - channel_info_start_time
|
||||
logger.debug(f"rss_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds")
|
||||
|
||||
# Set feed metadata
|
||||
main_name = f"{channel_title} (@{channel_username})"
|
||||
@@ -375,41 +475,7 @@ async def generate_channel_rss(channel: str | int,
|
||||
fg.description(f'Telegram channel {channel_username} RSS feed')
|
||||
fg.language('ru')
|
||||
fg.id(f"{base_url}/rss/{channel_username}") # Use username for feed ID consistency
|
||||
|
||||
# Collect messages
|
||||
messages_start_time = time.time()
|
||||
try:
|
||||
from tg_cache import cached_get_chat_history
|
||||
|
||||
messages = await cached_get_chat_history(post_parser.client, channel, limit=limit*2)
|
||||
except Exception as e:
|
||||
logger.error(f"Error during get_chat_history for channel '{channel}' (type: {type(channel)}): {str(e)}", exc_info=True) # Log error specifically for get_chat_history
|
||||
raise ValueError(f"Failed to get chat history for {channel}: {str(e)}") from e # Raise a more specific error
|
||||
|
||||
messages_elapsed = time.time() - messages_start_time
|
||||
logger.debug(f"rss_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds")
|
||||
|
||||
# Process messages into groups and render them.
|
||||
# The whole grouping/trimming/rendering pipeline is CPU-heavy (deepcopy +
|
||||
# per-message rendering) and contains no await, so run it in ONE worker
|
||||
# thread to keep the event loop responsive.
|
||||
processing_start_time = time.time()
|
||||
try:
|
||||
final_posts = await asyncio.to_thread(
|
||||
_render_pipeline, messages, post_parser, limit,
|
||||
exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'],
|
||||
channel,
|
||||
)
|
||||
finally:
|
||||
# Persist media file-ids collected during rendering with a single bulk
|
||||
# upsert — in a finally so a partial render still records what it collected
|
||||
# (the flush is best-effort and swallows its own errors, so it cannot mask a
|
||||
# render exception).
|
||||
await post_parser._flush_pending_media_ids()
|
||||
|
||||
processing_elapsed = time.time() - processing_start_time
|
||||
logger.debug(f"rss_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds")
|
||||
|
||||
# Generate feed entries
|
||||
feed_gen_start_time = time.time()
|
||||
|
||||
@@ -459,7 +525,9 @@ async def generate_channel_rss(channel: str | int,
|
||||
total_elapsed = time.time() - total_start_time
|
||||
logger.debug(f"rss_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds")
|
||||
return rss_feed
|
||||
|
||||
|
||||
except ChannelNotFound as e:
|
||||
return create_error_feed(str(e.channel_identifier), base_url)
|
||||
except Exception as e:
|
||||
logger.error(f"generate_channel_rss: channel {channel}, error {str(e)}")
|
||||
raise
|
||||
@@ -521,89 +589,32 @@ async def generate_channel_html(channel: str | int,
|
||||
merge_seconds: int = 5
|
||||
) -> str:
|
||||
"""
|
||||
Generate HTML feed for channel using actual messages
|
||||
Generate HTML feed for channel using actual messages.
|
||||
|
||||
Thin formatter over the shared _prepare_feed_posts: it only joins the prepared,
|
||||
sanitized posts with the <hr> divider. HTML fetches exactly `limit` messages and
|
||||
enables reply enrichment (enrich_replies=True).
|
||||
Args:
|
||||
channel: Telegram channel name
|
||||
post_parser: Optional PostParser instance. If not provided, will create new one
|
||||
client: Telegram client instance
|
||||
limit: Maximum number of posts to include in the RSS feed
|
||||
exclude_flags: Flags to exclude from the RSS feed
|
||||
limit: Maximum number of posts to include in the HTML feed
|
||||
exclude_flags: Flags to exclude from the HTML feed
|
||||
exclude_text: Text to exclude from posts
|
||||
Returns:
|
||||
HTML feed as string
|
||||
"""
|
||||
total_start_time = time.time()
|
||||
|
||||
if limit < 1:
|
||||
raise ValueError(f"limit must be positive, got {limit}")
|
||||
if limit > 200:
|
||||
raise ValueError(f"limit cannot exceed 200, got {limit}")
|
||||
base_url = Config['pyrogram_bridge_url']
|
||||
|
||||
try:
|
||||
post_parser = PostParser(client=client)
|
||||
|
||||
base_url = Config['pyrogram_bridge_url']
|
||||
|
||||
channel_info_start_time = time.time()
|
||||
try:
|
||||
channel = post_parser.channel_name_prepare(channel)
|
||||
logger.debug(f"Prepared channel identifier for HTML: {channel} (type: {type(channel)})") # Log prepared channel
|
||||
from tg_cache import cached_get_chat
|
||||
channel_info = await cached_get_chat(post_parser.client, channel)
|
||||
channel_username = channel_info.username or (str(channel_info.id) if channel_info.id and str(channel_info.id).startswith('-100') else None)
|
||||
if not channel_username:
|
||||
logger.warning(f"Could not get username for channel {channel} in HTML generation, returning error feed structure (as string). NOTE: This should ideally return HTML error page.")
|
||||
# For HTML, returning an error feed string might not be ideal. Consider returning a dedicated HTML error page.
|
||||
return create_error_feed(str(channel), base_url) # Ensure channel is string for error feed
|
||||
except (errors.UsernameInvalid, errors.UsernameNotOccupied) as e:
|
||||
logger.warning(f"Channel not found error for {channel} in HTML generation: {str(e)}")
|
||||
# Consider returning a dedicated HTML error page.
|
||||
return create_error_feed(channel, base_url)
|
||||
except errors.FloodWait:
|
||||
# Let FloodWait bubble up to api_server.py, which returns HTTP 429 with Retry-After header
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error during get_chat for channel '{channel}' (type: {type(channel)}) in HTML generation: {str(e)}", exc_info=True)
|
||||
raise ValueError(f"Failed to get chat info for {channel} in HTML generation: {str(e)}") from e
|
||||
prepared = await _prepare_feed_posts(
|
||||
channel, client,
|
||||
limit=limit, exclude_flags=exclude_flags, exclude_text=exclude_text,
|
||||
merge_seconds=merge_seconds, history_limit=limit,
|
||||
enrich_replies=True, log_prefix="html",
|
||||
)
|
||||
|
||||
channel_info_elapsed = time.time() - channel_info_start_time
|
||||
logger.debug(f"html_channel_info_timing: channel {channel}, retrieved in {channel_info_elapsed:.3f} seconds")
|
||||
|
||||
# Collect messages
|
||||
messages_start_time = time.time()
|
||||
try:
|
||||
from tg_cache import cached_get_chat_history
|
||||
|
||||
messages = await cached_get_chat_history(post_parser.client, channel, limit=limit)
|
||||
except Exception as e:
|
||||
logger.error(f"Error during get_chat_history for channel '{channel}' (type: {type(channel)}) in HTML generation: {str(e)}", exc_info=True)
|
||||
raise ValueError(f"Failed to get chat history for {channel} in HTML generation: {str(e)}") from e
|
||||
|
||||
messages_elapsed = time.time() - messages_start_time
|
||||
logger.debug(f"html_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds")
|
||||
|
||||
# Enrich messages with reply to messages
|
||||
enrichment_start_time = time.time()
|
||||
messages = await _reply_enrichment(client, messages)
|
||||
enrichment_elapsed = time.time() - enrichment_start_time
|
||||
logger.debug(f"html_reply_enrichment_timing: channel {channel}, replies enriched in {enrichment_elapsed:.3f} seconds")
|
||||
|
||||
# Process messages into groups and render them in ONE worker thread
|
||||
# (CPU-heavy, no await) to keep the event loop responsive.
|
||||
processing_start_time = time.time()
|
||||
try:
|
||||
final_posts = await asyncio.to_thread(
|
||||
_render_pipeline, messages, post_parser, limit,
|
||||
exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'],
|
||||
channel,
|
||||
)
|
||||
finally:
|
||||
# Persist media file-ids collected during rendering with a single bulk
|
||||
# upsert — in a finally so a partial render still records what it collected.
|
||||
await post_parser._flush_pending_media_ids()
|
||||
|
||||
processing_elapsed = time.time() - processing_start_time
|
||||
logger.debug(f"html_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds")
|
||||
final_posts = prepared.posts
|
||||
|
||||
# Generate HTML content.
|
||||
html_gen_start_time = time.time()
|
||||
@@ -620,7 +631,9 @@ async def generate_channel_html(channel: str | int,
|
||||
logger.debug(f"html_total_generation_timing: channel {channel}, total time {total_elapsed:.3f} seconds")
|
||||
|
||||
return html
|
||||
|
||||
|
||||
except ChannelNotFound as e:
|
||||
return create_error_feed(str(e.channel_identifier), base_url)
|
||||
except Exception as e:
|
||||
logger.error(f"html_generation_error: channel {channel}, error {str(e)}")
|
||||
raise
|
||||
|
||||
@@ -0,0 +1,289 @@
|
||||
# flake8: noqa
|
||||
# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
|
||||
# pylint: disable=redefined-outer-name, line-too-long
|
||||
"""
|
||||
Stage 3 (render-pipeline refactor epic, issue #30/#34) — reply-enrichment lock.
|
||||
|
||||
Stage 3 merges generate_channel_rss/generate_channel_html around the shared
|
||||
_prepare_feed_posts and moves _reply_enrichment (the live client.get_messages
|
||||
reply-target fetch) INTO that shared path (enrich_replies=True for HTML,
|
||||
False for RSS).
|
||||
|
||||
The stage-0 golden oracle CANNOT see this move: its harness replays the corpus
|
||||
through monkeypatched tg_cache but leaves client.get_messages a no-op (the client
|
||||
is a bare SimpleNamespace, so get_messages raises AttributeError and enrichment
|
||||
silently no-ops). So enrichment is locked here instead, exactly as mandated on #30:
|
||||
|
||||
(a) get_messages is BATCHED by chat_id (one call per chat, not one per reply);
|
||||
(b) the fetched reply target is set onto the right source message by
|
||||
(chat_id, message.id);
|
||||
(c) the enriched reply block actually renders in the HTML feed.
|
||||
|
||||
Plus: RSS deliberately does NOT enrich (keeps polling cheap) — locked too.
|
||||
"""
|
||||
from types import SimpleNamespace
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from pyrogram import errors
|
||||
|
||||
import rss_generator as rss_module
|
||||
from rss_generator import _reply_enrichment, generate_channel_html, generate_channel_rss
|
||||
|
||||
|
||||
class _Str(str):
|
||||
"""Stand-in for Pyrogram's Str: .html returns the raw string unchanged, so a
|
||||
reply-target text reaches the pre-sanitize body like real entity text would."""
|
||||
@property
|
||||
def html(self):
|
||||
return str(self)
|
||||
|
||||
|
||||
def _reply_target(mid, text):
|
||||
"""A resolved reply-to-message as _format_reply_info consumes it."""
|
||||
return SimpleNamespace(id=mid, text=text, caption=None, sender_chat=None)
|
||||
|
||||
|
||||
def make_message(mid, chat_id=-1001234567890, username="testchan", text="post",
|
||||
reply_to_message_id=None, reply_to_message=None, date=None):
|
||||
m = SimpleNamespace()
|
||||
m.id = mid
|
||||
m.date = date or datetime(2024, 1, 1, 12, 0, mid % 60, tzinfo=timezone.utc)
|
||||
m.text = _Str(text) if text is not None else None
|
||||
m.caption = None
|
||||
m.media = None
|
||||
m.web_page = None
|
||||
m.poll = None
|
||||
m.service = None
|
||||
m.forward_origin = None
|
||||
m.reply_to_message = reply_to_message
|
||||
m.reply_to_message_id = reply_to_message_id
|
||||
m.sender_chat = None
|
||||
m.from_user = None
|
||||
m.reactions = None
|
||||
m.views = 100
|
||||
m.media_group_id = None
|
||||
m.show_caption_above_media = False
|
||||
m.chat = SimpleNamespace(id=chat_id, username=username)
|
||||
for attr in ("photo", "video", "document", "audio", "voice",
|
||||
"video_note", "animation", "sticker"):
|
||||
setattr(m, attr, None)
|
||||
return m
|
||||
|
||||
|
||||
class RecordingClient:
|
||||
"""Fake Telegram client: records each get_messages(chat_id, ids) call and returns,
|
||||
for every requested id, a 'full' message carrying a resolved .reply_to_message."""
|
||||
def __init__(self, target_text_for=lambda chat_id, mid: f"TARGET_{chat_id}_{mid}"):
|
||||
self.calls = [] # list[(chat_id, list[ids])]
|
||||
self._target_text_for = target_text_for
|
||||
|
||||
async def get_messages(self, chat_id, ids):
|
||||
self.calls.append((chat_id, list(ids)))
|
||||
out = []
|
||||
for mid in ids:
|
||||
full = SimpleNamespace(
|
||||
id=mid,
|
||||
empty=False,
|
||||
reply_to_message=_reply_target(mid + 5000, self._target_text_for(chat_id, mid)),
|
||||
)
|
||||
out.append(full)
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# (a) batching by chat_id + (b) target set onto the right source message.
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.asyncio
|
||||
async def test_reply_enrichment_batches_by_chat_and_sets_target():
|
||||
CHAT_A, CHAT_B = -100111, -100222
|
||||
messages = [
|
||||
make_message(10, chat_id=CHAT_A, reply_to_message_id=1),
|
||||
make_message(11, chat_id=CHAT_A, reply_to_message_id=2),
|
||||
make_message(12, chat_id=CHAT_A, reply_to_message_id=3),
|
||||
make_message(20, chat_id=CHAT_B, reply_to_message_id=4),
|
||||
make_message(21, chat_id=CHAT_B, reply_to_message_id=5),
|
||||
make_message(30, chat_id=CHAT_A, reply_to_message_id=None), # no reply -> not fetched
|
||||
]
|
||||
client = RecordingClient()
|
||||
|
||||
result = await _reply_enrichment(client, messages)
|
||||
|
||||
# (a) BATCHED: exactly one call per chat_id (2), NOT one per reply (5).
|
||||
assert len(client.calls) == 2, f"expected 2 batched calls, got {client.calls}"
|
||||
calls_by_chat = dict(client.calls)
|
||||
assert calls_by_chat[CHAT_A] == [10, 11, 12], "chat A ids not batched into one call"
|
||||
assert calls_by_chat[CHAT_B] == [20, 21], "chat B ids not batched into one call"
|
||||
|
||||
# (b) each fetched target is set onto the SOURCE message keyed by (chat_id, id).
|
||||
by_id = {m.id: m for m in result}
|
||||
for mid, chat in [(10, CHAT_A), (11, CHAT_A), (12, CHAT_A), (20, CHAT_B), (21, CHAT_B)]:
|
||||
assert by_id[mid].reply_to_message is not None, f"message {mid} not enriched"
|
||||
assert by_id[mid].reply_to_message.text == f"TARGET_{chat}_{mid}", \
|
||||
f"message {mid} got the wrong reply target"
|
||||
# The reply-less message is untouched.
|
||||
assert by_id[30].reply_to_message is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reply_enrichment_no_replies_makes_no_calls():
|
||||
messages = [make_message(1), make_message(2)] # no reply_to_message_id
|
||||
client = RecordingClient()
|
||||
await _reply_enrichment(client, messages)
|
||||
assert client.calls == [], "get_messages must not be called when nothing needs enrichment"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# (c) the enriched reply block renders in the HTML feed — AND enrichment now runs
|
||||
# inside the shared _prepare_feed_posts path (enrich_replies=True for HTML).
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _patch_feed_source(monkeypatch, messages):
|
||||
async def fake_get_chat(client, channel):
|
||||
return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
|
||||
|
||||
async def fake_get_history(client, channel, limit=20):
|
||||
return messages
|
||||
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_html_feed_renders_enriched_reply_block(monkeypatch):
|
||||
MARKER = "ENRICHED_REPLY_MARKER_XYZ"
|
||||
# A shallow message: it has a reply_to_message_id but NO resolved reply_to_message
|
||||
# yet — enrichment must fetch and fill it before render.
|
||||
msg = make_message(77, chat_id=-1001234567890, reply_to_message_id=70,
|
||||
reply_to_message=None, text="body text")
|
||||
_patch_feed_source(monkeypatch, [msg])
|
||||
|
||||
client = RecordingClient(target_text_for=lambda chat_id, mid: MARKER)
|
||||
html = await generate_channel_html("testchan", client=client, limit=5)
|
||||
|
||||
# Enrichment ran inside _prepare_feed_posts and was BATCHED (one call for the chat).
|
||||
assert client.calls == [(-1001234567890, [77])], f"enrichment not run/batched: {client.calls}"
|
||||
# The resolved reply target renders as a reply block carrying the marker text.
|
||||
assert '<div class="message-reply">' in html, "reply block not rendered"
|
||||
assert MARKER in html, "resolved reply-target text missing from the rendered feed"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rss_feed_does_not_enrich_replies(monkeypatch):
|
||||
# RSS deliberately skips enrichment (enrich_replies=False) to keep polling cheap.
|
||||
msg = make_message(88, chat_id=-1001234567890, reply_to_message_id=80,
|
||||
reply_to_message=None, text="body text")
|
||||
_patch_feed_source(monkeypatch, [msg])
|
||||
|
||||
client = RecordingClient()
|
||||
await generate_channel_rss("testchan", client=client, limit=5)
|
||||
assert client.calls == [], "RSS must NOT call get_messages (enrich_replies=False)"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# History over-fetch fork: RSS over-fetches limit*2 (headroom for grouping/trim),
|
||||
# HTML fetches exactly limit. This is a REAL behavioral fork that the golden oracle
|
||||
# CANNOT see (its fake_get_chat_history ignores the limit kwarg and returns the whole
|
||||
# corpus, so both paths trim to GOLDEN_LIMIT identically). A refactor that accidentally
|
||||
# unified the two would pass every golden test — so lock the forwarded limit here.
|
||||
# --------------------------------------------------------------------------- #
|
||||
@pytest.mark.asyncio
|
||||
async def test_rss_overfetches_history_html_does_not(monkeypatch):
|
||||
N = 7
|
||||
seen = {}
|
||||
|
||||
async def fake_get_chat(client, channel):
|
||||
return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
|
||||
|
||||
async def spy_get_history(client, channel, limit=20):
|
||||
seen.setdefault("limits", []).append(limit)
|
||||
return []
|
||||
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat_history", spy_get_history, raising=False)
|
||||
|
||||
await generate_channel_rss("testchan", client=SimpleNamespace(), limit=N)
|
||||
assert seen["limits"] == [2 * N], f"RSS must over-fetch history limit*2, got {seen['limits']}"
|
||||
|
||||
seen["limits"] = []
|
||||
await generate_channel_html("testchan", client=SimpleNamespace(), limit=N)
|
||||
assert seen["limits"] == [N], f"HTML must fetch history limit (no over-fetch), got {seen['limits']}"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Shared error handling in _prepare_feed_posts, verified through BOTH formatters.
|
||||
# The golden oracle cannot see these paths (§3.9/§3.10 are error branches), so they
|
||||
# are locked here. FEED_FUNCS lets each case assert RSS and HTML behave identically.
|
||||
# --------------------------------------------------------------------------- #
|
||||
FEED_FUNCS = [generate_channel_rss, generate_channel_html]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("feed_func", FEED_FUNCS)
|
||||
async def test_channel_not_found_returns_error_feed(monkeypatch, feed_func):
|
||||
async def raise_not_found(client, channel):
|
||||
raise errors.UsernameNotOccupied("no such user")
|
||||
|
||||
async def fake_get_history(client, channel, limit=20):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat", raise_not_found, raising=False)
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
|
||||
|
||||
out = await feed_func("ghostchan", client=SimpleNamespace(), limit=5)
|
||||
# ChannelNotFound -> create_error_feed (RSS-XML error feed) in both paths.
|
||||
assert "does not exist" in out, f"{feed_func.__name__} did not return the error feed"
|
||||
assert "ghostchan" in out
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("feed_func", FEED_FUNCS)
|
||||
async def test_floodwait_from_get_chat_propagates(monkeypatch, feed_func):
|
||||
async def raise_flood(client, channel):
|
||||
raise errors.FloodWait(value=11)
|
||||
|
||||
async def fake_get_history(client, channel, limit=20):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat", raise_flood, raising=False)
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
|
||||
|
||||
# FloodWait must reach api_server unwrapped (mapped there to HTTP 429), NOT ValueError.
|
||||
with pytest.raises(errors.FloodWait):
|
||||
await feed_func("floodchan", client=SimpleNamespace(), limit=5)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("feed_func", FEED_FUNCS)
|
||||
async def test_floodwait_from_history_propagates(monkeypatch, feed_func):
|
||||
# Registry §3.9 (the fix this stage lands): FloodWait raised while fetching HISTORY
|
||||
# must propagate -> HTTP 429. Before stage 3 it fell into `except Exception` and was
|
||||
# wrapped in ValueError -> HTTP 400. The golden cannot see this; locked here.
|
||||
async def fake_get_chat(client, channel):
|
||||
return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
|
||||
|
||||
async def raise_flood_history(client, channel, limit=20):
|
||||
raise errors.FloodWait(value=13)
|
||||
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat_history", raise_flood_history, raising=False)
|
||||
|
||||
with pytest.raises(errors.FloodWait):
|
||||
await feed_func("testchan", client=SimpleNamespace(), limit=5)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("feed_func", FEED_FUNCS)
|
||||
async def test_other_history_error_becomes_valueerror(monkeypatch, feed_func):
|
||||
# Any NON-FloodWait history failure is still wrapped in ValueError (api_server -> 400).
|
||||
async def fake_get_chat(client, channel):
|
||||
return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
|
||||
|
||||
async def raise_runtime(client, channel, limit=20):
|
||||
raise RuntimeError("history backend exploded")
|
||||
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
|
||||
monkeypatch.setattr("tg_cache.cached_get_chat_history", raise_runtime, raising=False)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
await feed_func("testchan", client=SimpleNamespace(), limit=5)
|
||||
Reference in New Issue
Block a user