TheLovinator/discord-rss-bot
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add support for extensions
Some checks failed
Test and build Docker image / docker (push) Failing after 3s
Details

Browse source
This commit is contained in:
Joakim Hellsén 2026-07-20 06:02:25 +02:00
commit 793f67db42
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
39 changed files with 3670 additions and 1309 deletions
Download patch file Download diff file Expand all files Collapse all files

4
tests/test_blacklist.py
View file

@ -11,9 +11,9 @@ from reader import Feed
from reader import Reader
from reader import make_reader
from discord_rss_bot.filter.blacklist import entry_should_be_skipped
from discord_rss_bot.filter.blacklist import feed_has_blacklist_tags
from discord_rss_bot.filter.evaluator import entry_should_be_skipped
from discord_rss_bot.filter.evaluator import evaluate_entry_filters
from discord_rss_bot.filter.evaluator import feed_has_blacklist_tags
from discord_rss_bot.filter.evaluator import get_entry_fields
from discord_rss_bot.filter.evaluator import get_filter_values_from_reader

6
tests/test_custom_filter.py
View file

@ -5,9 +5,9 @@ import tempfile
from pathlib import Path
from typing import TYPE_CHECKING
from discord_rss_bot.custom_filters import encode_url
from discord_rss_bot.custom_filters import entry_is_blacklisted
from discord_rss_bot.custom_filters import entry_is_whitelisted
from discord_rss_bot.filter.evaluator import encode_url
from discord_rss_bot.filter.evaluator import entry_is_blacklisted
from discord_rss_bot.filter.evaluator import entry_is_whitelisted
from discord_rss_bot.settings import get_reader
if TYPE_CHECKING:

4
tests/test_custom_message.py
View file

@ -394,7 +394,6 @@ def test_save_embed_serializes_embed_and_writes_feed_tag() -> None:
thumbnail_url="https://example.com/thumb.png",
footer_text="Footer",
footer_icon_url="https://example.com/footer.png",
show_steam_game_icon_in_thumbnail=True,
)
save_embed(reader=reader, feed=feed, embed=embed)
@ -405,7 +404,6 @@ def test_save_embed_serializes_embed_and_writes_feed_tag() -> None:
parsed = typing.cast("dict[str, str]", __import__("json").loads(call_args[2]))
assert parsed["title"] == "Title"
assert parsed["footer_icon_url"] == "https://example.com/footer.png"
assert parsed["show_steam_game_icon_in_thumbnail"] is True
def test_get_embed_returns_default_embed_when_tag_is_empty() -> None:
@ -461,13 +459,11 @@ def test_get_embed_data_coerces_values_to_strings() -> None:
"thumbnail_url": 8,
"footer_text": 9,
"footer_icon_url": 10,
"show_steam_game_icon_in_thumbnail": "true",
},
)
assert embed.title == "1"
assert embed.footer_icon_url == "10"
assert embed.show_steam_game_icon_in_thumbnail is True
class TestNormalizeMessageUsername:

1137
tests/test_extensions.py Normal file
View file

@ -0,0 +1,1137 @@
"""Tests for the feed extension system.
Tests cover the ABC base class, plugin discovery, per-feed storage, the
``run_extensions`` integration function, and the built-in example plugin.
"""
from __future__ import annotations
import json
import os
import tempfile
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler
from http.server import ThreadingHTTPServer
from pathlib import Path
from threading import Thread
from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import pytest
from reader import Reader as ReaderType
from reader import make_reader
from discord_rss_bot.custom_message import CustomEmbed
from discord_rss_bot.custom_message import get_embed
from discord_rss_bot.custom_message import replace_tags_in_embed
from discord_rss_bot.custom_message import replace_tags_in_text_message
from discord_rss_bot.custom_message import save_embed
from discord_rss_bot.extensions import FeedExtension
from discord_rss_bot.extensions import auto_enable_extensions_for_feed
from discord_rss_bot.extensions import discover_plugins
from discord_rss_bot.extensions import registry_clear
from discord_rss_bot.extensions import run_extensions
from discord_rss_bot.extensions.base import FeedExtension as FeedExtensionABC
from discord_rss_bot.extensions.hoyolab import HoyolabExtension
from discord_rss_bot.extensions.jwplayer_thumbnail import _SLUG_CACHE
from discord_rss_bot.extensions.jwplayer_thumbnail import JWPlayerThumbnailExtension
from discord_rss_bot.extensions.steam import SteamExtension
from discord_rss_bot.extensions.storage import get_enabled_extensions_for_feed
from discord_rss_bot.extensions.storage import set_enabled_extensions_for_feed
from discord_rss_bot.extensions.wordpress import WordPressExtension
from discord_rss_bot.extensions.youtube import YouTubeExtension
if TYPE_CHECKING:
from collections.abc import Iterator
from reader import Entry
from reader import Reader
from discord_rss_bot.feeds import JsonValue
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _clear_registry() -> Iterator[None]:
"""Clear the extension registry before and after each test.
Yields:
Control back to the test body, then cleans up.
"""
registry_clear()
yield
registry_clear()
@pytest.fixture
def temp_extensions_dir() -> Iterator[str]:
"""Create a temporary directory to act as the extensions directory.
Yields:
The path to the temporary extensions directory.
"""
tmpdir: str = tempfile.mkdtemp(prefix="ext-test-")
old_env: str | None = os.environ.pop("EXTENSIONS_DIR", None)
os.environ["EXTENSIONS_DIR"] = tmpdir
yield tmpdir
if old_env is not None:
os.environ["EXTENSIONS_DIR"] = old_env
else:
os.environ.pop("EXTENSIONS_DIR", None)
@pytest.fixture
def mock_reader() -> MagicMock:
"""A mock Reader with tag storage support.
Returns:
A MagicMock configured with set_tag/get_tag side effects.
"""
reader: MagicMock = MagicMock()
reader.tags = {} # type: ignore[valid-type]
def _resolve_feed_key(feed_or_url: str | SimpleNamespace) -> str:
"""Extract the feed URL string from either a string or SimpleNamespace.
Returns:
The URL as a string.
"""
if isinstance(feed_or_url, str):
return feed_or_url
url = getattr(feed_or_url, "url", None)
if url is not None:
return str(url)
return str(feed_or_url)
def set_tag(feed_or_url: str | SimpleNamespace, key: str, value: JsonValue) -> None:
feed_key: str = _resolve_feed_key(feed_or_url)
if feed_key not in reader.tags:
reader.tags[feed_key] = {}
reader.tags[feed_key][key] = value
def get_tag(feed_or_url: str | SimpleNamespace, key: str, default: JsonValue = None) -> JsonValue:
feed_key: str = _resolve_feed_key(feed_or_url)
feed_tags = reader.tags.get(feed_key, {})
return feed_tags.get(key, default)
reader.set_tag.side_effect = set_tag
reader.get_tag.side_effect = get_tag
return reader
@pytest.fixture
def mock_feed() -> SimpleNamespace:
"""A feed object with all attributes required by tag replacement.
Returns:
A SimpleNamespace with feed-like attributes.
"""
return SimpleNamespace(
added=None,
author="Feed Author",
authors_str="Feed Author",
last_exception=None,
last_updated=None,
link="https://example.com/feed",
subtitle="",
title="Example Feed",
updated=None,
updates_enabled=True,
url="https://example.com/feed.xml",
user_title="",
version="atom10",
)
@pytest.fixture
def mock_entry(mock_feed: SimpleNamespace) -> SimpleNamespace:
"""An entry object with all attributes required by tag replacement.
Returns:
A SimpleNamespace with entry-like attributes.
"""
return SimpleNamespace(
added=None,
author="Entry Author",
authors_str="Entry Author",
content=[SimpleNamespace(value="<p>Hello world</p>")],
feed=mock_feed,
feed_url=mock_feed.url,
id="entry-1",
important=False,
link="https://example.com/entry-1",
published=None,
read=False,
read_modified=None,
summary="",
title="Test Entry",
updated=None,
)
# ---------------------------------------------------------------------------
# Tests: base.py
# ---------------------------------------------------------------------------
def test_feed_extension_abc_cannot_be_instantiated() -> None:
"""FeedExtension should be abstract and not instantiable directly."""
with pytest.raises(TypeError):
FeedExtensionABC() # type: ignore[abstract]
def test_feed_extension_subclass_can_be_instantiated() -> None:
"""A concrete subclass with ``process_entry`` should work."""
class ConcreteExtension(FeedExtensionABC):
name = "test_concrete"
def process_entry(self, entry: Entry, reader: Reader) -> dict[str, str]: # ruff:ignore[unused-method-argument]
return {"hello": "world"}
instance = ConcreteExtension()
assert instance.name == "test_concrete"
def test_feed_extension_name_defaults_to_empty_string() -> None:
"""The ``name`` class variable defaults to ``""``."""
class UnnamedExtension(FeedExtensionABC):
def process_entry(self, entry: Entry, reader: Reader) -> dict[str, str]: # ruff:ignore[unused-method-argument]
return {}
assert not UnnamedExtension.name, "Expected name to be empty"
# ---------------------------------------------------------------------------
# Tests: discovery.py
# ---------------------------------------------------------------------------
BUILT_IN_EXTENSIONS: frozenset[str] = frozenset({"steam", "youtube", "hoyolab", "jwplayer_thumbnail", "wordpress"})
def _assert_only_built_in_extensions(registry: dict[str, type[FeedExtension]]) -> None:
"""Assert that *registry* contains exactly the built-in extensions."""
registered: set[str] = set(registry.keys())
unexpected: set[str] = registered - BUILT_IN_EXTENSIONS
missing: frozenset[str] = BUILT_IN_EXTENSIONS - registered
assert not unexpected, f"Registry contains unexpected extensions: {unexpected}"
assert not missing, f"Registry missing built-in extensions: {missing}"
def test_discover_plugins_empty_directory(temp_extensions_dir: str) -> None:
"""An empty external directory still has built-in extensions."""
registry: dict[str, type[FeedExtension]] = discover_plugins(force=True)
_assert_only_built_in_extensions(registry)
def test_discover_plugins_missing_directory() -> None:
"""If the external extensions directory doesn't exist, built-ins still load."""
old_env: str | None = os.environ.pop("EXTENSIONS_DIR", None)
try:
os.environ["EXTENSIONS_DIR"] = str(Path(tempfile.gettempdir()) / "nonexistent-extensions-dir-12345")
registry: dict[str, type[FeedExtension]] = discover_plugins(force=True)
_assert_only_built_in_extensions(registry)
finally:
if old_env is not None:
os.environ["EXTENSIONS_DIR"] = old_env
else:
os.environ.pop("EXTENSIONS_DIR", None)
def test_discover_plugins_imports_plugin(temp_extensions_dir: str) -> None:
"""A valid plugin file should be discovered and registered alongside built-ins."""
plugin_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class TestPlugin(FeedExtension):
name = "test_plugin"
description = "A test plugin."
def process_entry(self, entry, reader):
return {"test_var": "hello"}
"""
plugin_path: Path = Path(temp_extensions_dir) / "test_plugin.py"
plugin_path.write_text(plugin_code, encoding="utf-8")
registry: dict[str, type[FeedExtension]] = discover_plugins(force=True)
assert "test_plugin" in registry
assert registry["test_plugin"].name == "test_plugin"
assert registry["test_plugin"].description == "A test plugin."
# Built-in extensions should also be present.
for name in BUILT_IN_EXTENSIONS:
assert name in registry, f"Missing built-in extension {name}"
def test_discover_plugins_skips_broken_plugin(temp_extensions_dir: str) -> None:
"""A plugin that raises during import should be skipped, not crash."""
plugin_path: Path = Path(temp_extensions_dir) / "broken.py"
plugin_path.write_text("raise SyntaxError('bad syntax'", encoding="utf-8")
# Should not raise.
registry: dict[str, type[FeedExtension]] = discover_plugins(force=True)
assert "broken" not in registry
# Built-in extensions should still load.
_assert_only_built_in_extensions(registry)
def test_discover_plugins_skips_init_py(temp_extensions_dir: str) -> None:
"""__init__.py files in the extensions directory should be ignored."""
init_path: Path = Path(temp_extensions_dir) / "__init__.py"
init_path.write_text("# package init", encoding="utf-8")
registry: dict[str, type[FeedExtension]] = discover_plugins(force=True)
_assert_only_built_in_extensions(registry)
def test_discover_plugins_deduplicates_by_name(temp_extensions_dir: str) -> None:
"""If two plugins define the same name, the last one wins."""
plugin_a: str = """
from discord_rss_bot.extensions.base import FeedExtension
class PluginA(FeedExtension):
name = "dup_name"
def process_entry(self, entry, reader):
return {"from": "a"}
"""
plugin_b: str = """
from discord_rss_bot.extensions.base import FeedExtension
class PluginB(FeedExtension):
name = "dup_name"
def process_entry(self, entry, reader):
return {"from": "b"}
"""
(Path(temp_extensions_dir) / "a.py").write_text(plugin_a)
(Path(temp_extensions_dir) / "b.py").write_text(plugin_b)
registry: dict[str, type[FeedExtension]] = discover_plugins(force=True)
assert "dup_name" in registry
# The last one alphabetically (b.py) should win.
assert registry["dup_name"].__name__ == "PluginB"
# ---------------------------------------------------------------------------
# Tests: storage.py
# ---------------------------------------------------------------------------
def test_get_enabled_extensions_empty_when_no_tag(mock_reader: MagicMock, mock_feed: SimpleNamespace) -> None:
"""If no extensions tag is set, an empty list is returned."""
enabled: list[str] = get_enabled_extensions_for_feed(mock_reader, mock_feed.url)
assert enabled == []
def test_set_and_get_enabled_extensions(mock_reader: MagicMock, mock_feed: SimpleNamespace) -> None:
"""Setting enabled extensions and retrieving them should round-trip."""
names: list[str] = ["jwplayer_thumbnail", "encode_links"]
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, names)
enabled: list[str] = get_enabled_extensions_for_feed(mock_reader, mock_feed.url)
assert enabled == names
def test_set_enabled_extensions_clears_list(mock_reader: MagicMock, mock_feed: SimpleNamespace) -> None:
"""Setting an empty list should clear the enabled extensions."""
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, ["some_plugin"])
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, [])
enabled: list[str] = get_enabled_extensions_for_feed(mock_reader, mock_feed.url)
assert enabled == []
def test_get_enabled_extensions_handles_string_tag(mock_reader: MagicMock, mock_feed: SimpleNamespace) -> None:
"""If the tag is stored as a JSON string, it should still be parsed."""
def json_string_tag(feed_url: str, key: str, default: JsonValue = None) -> str:
return json.dumps(["plugin_a", "plugin_b"])
mock_reader.get_tag.side_effect = json_string_tag
enabled: list[str] = get_enabled_extensions_for_feed(mock_reader, mock_feed.url)
assert enabled == ["plugin_a", "plugin_b"]
def test_get_enabled_extensions_handles_empty_string_tag(mock_reader: MagicMock, mock_feed: SimpleNamespace) -> None:
"""An empty string tag should return an empty list."""
def empty_string_tag(feed_url: str, key: str, default: JsonValue = None) -> str:
return ""
mock_reader.get_tag.side_effect = empty_string_tag
enabled: list[str] = get_enabled_extensions_for_feed(mock_reader, mock_feed.url)
assert enabled == []
# ---------------------------------------------------------------------------
# Tests: run_extensions (integration)
# ---------------------------------------------------------------------------
def test_run_extensions_empty_when_none_enabled(
mock_reader: MagicMock,
mock_entry: SimpleNamespace,
mock_feed: SimpleNamespace,
) -> None:
"""If no extensions are enabled for the feed, the result is empty."""
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, [])
result: dict[str, str] = run_extensions(mock_entry, mock_reader) # type: ignore[arg-type]
assert result == {}
def test_run_extensions_with_missing_plugin(
mock_reader: MagicMock,
mock_entry: SimpleNamespace,
mock_feed: SimpleNamespace,
) -> None:
"""Enabled extension that doesn't exist in the registry is skipped."""
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, ["nonexistent_plugin"])
result: dict[str, str] = run_extensions(mock_entry, mock_reader) # type: ignore[arg-type]
assert result == {}
def test_run_extensions_with_registered_plugin(
mock_reader: MagicMock,
mock_entry: SimpleNamespace,
mock_feed: SimpleNamespace,
temp_extensions_dir: str,
) -> None:
"""A registered and enabled extension should produce variables."""
# Register a plugin via discovery.
plugin_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class TestPlugin(FeedExtension):
name = "test_plugin"
def process_entry(self, entry, reader):
return {"custom_var": "hello_from_plugin"}
"""
(Path(temp_extensions_dir) / "my_plugin.py").write_text(plugin_code)
discover_plugins(force=True)
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, ["test_plugin"])
result: dict[str, str] = run_extensions(mock_entry, mock_reader) # type: ignore[arg-type]
assert result == {"custom_var": "hello_from_plugin"}
def test_run_extensions_continues_after_plugin_error(
mock_reader: MagicMock,
mock_entry: SimpleNamespace,
mock_feed: SimpleNamespace,
temp_extensions_dir: str,
) -> None:
"""If one plugin raises, others should still run and produce results."""
# Register two plugins: one that raises and one that works.
good_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class GoodPlugin(FeedExtension):
name = "good_plugin"
def process_entry(self, entry, reader):
return {"good_var": "ok"}
"""
bad_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class BadPlugin(FeedExtension):
name = "bad_plugin"
def process_entry(self, entry, reader):
raise RuntimeError("This plugin failed!")
"""
(Path(temp_extensions_dir) / "good.py").write_text(good_code)
(Path(temp_extensions_dir) / "bad.py").write_text(bad_code)
discover_plugins(force=True)
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, ["bad_plugin", "good_plugin"])
result: dict[str, str] = run_extensions(mock_entry, mock_reader) # type: ignore[arg-type]
assert result == {"good_var": "ok"}
# ---------------------------------------------------------------------------
# Tests: tag replacement integration (custom_message.py)
# ---------------------------------------------------------------------------
def test_replace_tags_in_embed_uses_extension_variables(
mock_reader: MagicMock,
mock_feed: SimpleNamespace,
mock_entry: SimpleNamespace,
temp_extensions_dir: str,
) -> None:
"""Extension variables should be available in embed tag replacement."""
# Register a test plugin.
plugin_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class EmbedVarPlugin(FeedExtension):
name = "embed_var"
def process_entry(self, entry, reader):
return {"custom_thumbnail": "https://example.com/thumb.jpg"}
"""
(Path(temp_extensions_dir) / "embed_var.py").write_text(plugin_code)
discover_plugins(force=True)
# Enable the plugin for this feed.
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, ["embed_var"])
# Create an embed that uses the extension variable.
embed: CustomEmbed = get_embed(mock_reader, mock_feed) # type: ignore[arg-type]
embed.image_url = "{{custom_thumbnail}}"
save_embed(mock_reader, mock_feed, embed) # type: ignore[arg-type]
# Run replacement.
result: CustomEmbed = replace_tags_in_embed(mock_feed, mock_entry, mock_reader) # type: ignore[arg-type]
assert "https://example.com/thumb.jpg" in result.image_url
def test_replace_tags_in_text_message_uses_extension_variables(
mock_reader: MagicMock,
mock_feed: SimpleNamespace,
mock_entry: SimpleNamespace,
temp_extensions_dir: str,
) -> None:
"""Extension variables should be available in text message tag replacement."""
# Register a test plugin.
plugin_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class TextVarPlugin(FeedExtension):
name = "text_var"
def process_entry(self, entry, reader):
return {"custom_text": "hello_from_extension"}
"""
(Path(temp_extensions_dir) / "text_var.py").write_text(plugin_code)
discover_plugins(force=True)
# Enable the plugin for this feed.
set_enabled_extensions_for_feed(mock_reader, mock_feed.url, ["text_var"])
# Set a custom message that uses the extension variable.
mock_reader.set_tag(mock_feed.url, "custom_message", "{{custom_text}}")
# Run replacement.
result: str = replace_tags_in_text_message(mock_entry, mock_reader) # type: ignore[arg-type]
assert "hello_from_extension" in result
# ---------------------------------------------------------------------------
# Tests: JWPlayer example plugin
# ---------------------------------------------------------------------------
def test_jwplayer_thumbnail_extension_extracts_image() -> None:
"""The JWPlayer thumbnail extension should extract the image URL."""
ext = JWPlayerThumbnailExtension()
raw_html: str = """
<div id="player_01"></div>
<script type="text/javascript">
jwplayer("player_01").setup({
file: "https://example.com/video.mp4",
image: "https://example.com/thumbnail.jpg",
});
</script>
"""
entry: SimpleNamespace = SimpleNamespace(
id="test",
content=[SimpleNamespace(value=raw_html)],
summary="",
feed=SimpleNamespace(url="https://example.com/feed.xml"),
)
result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type]
assert result.get("jwplayer_thumbnail") == "https://example.com/thumbnail.jpg"
assert result.get("jwplayer_file") == "https://example.com/video.mp4"
def test_jwplayer_thumbnail_extension_returns_empty_without_content() -> None:
"""If the entry has no content, the extension should return an empty dict."""
ext = JWPlayerThumbnailExtension()
entry: SimpleNamespace = SimpleNamespace(
id="test",
content=[],
summary="",
link="https://example.com/video",
feed=SimpleNamespace(url="https://example.com/feed.xml"),
)
result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type]
assert result == {}
def test_jwplayer_thumbnail_extension_returns_empty_without_match() -> None:
"""If the HTML has no JWPlayer pattern, the extension should return empty."""
ext = JWPlayerThumbnailExtension()
raw_html: str = "<p>No player here.</p>"
entry: SimpleNamespace = SimpleNamespace(
id="test",
content=[SimpleNamespace(value=raw_html)],
summary="",
link="https://example.com/video",
feed=SimpleNamespace(url="https://example.com/feed.xml"),
)
result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type]
assert result == {}
def test_jwplayer_thumbnail_extension_matches_hentaigasm_format() -> None:
"""The extension should extract URLs from the actual hentaigasm.com feed format."""
ext = JWPlayerThumbnailExtension()
# This is the actual HTML structure from the main hentaigasm feed.
raw_html: str = """
<p style="text-align: center;"><strong>HENTAIGASM EXCLUSIVE!</strong></p>
<div style="width:620px; height:349px">
<script src="https://content.jwplatform.com/libraries/SAHhwvZq.js"></script>
<script>jwplayer.key="zTEbSn/eAplL0RLXT030FzOcek6qXmtrxju6Jg=="</script>
<div id="player_01"></div>
<script type="text/javascript">
.jwplayer("player_01").setup({
file: "https://hgasm2.com/Test Video 1 Subbed.mp4",
width: "620",
height: "349",
skin: "seven",
preload: "none",
autostart: "false",
image: "https://hgasm1.com/thumbnail/Test Video 1 Subbed.jpg",
advertising: {}
});
</script>
</div>
<a class="btn btn-pink" href="https://hgasm3.com/Test Video 1 Subbed.mp4" download>DOWNLOAD</a>
"""
entry: SimpleNamespace = SimpleNamespace(
id="test-hentai",
content=[SimpleNamespace(value=raw_html)],
summary="",
feed=SimpleNamespace(url="https://hentaigasm.com/feed/"),
)
result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type]
assert result.get("jwplayer_thumbnail") == "https://hgasm1.com/thumbnail/Test%20Video%201%20Subbed.jpg"
assert result.get("jwplayer_file") == "https://hgasm2.com/Test%20Video%201%20Subbed.mp4"
# ---------------------------------------------------------------------------
# Tests: auto_enable_url_patterns
# ---------------------------------------------------------------------------
def test_auto_enable_by_url_pattern_matches(
mock_reader: MagicMock,
temp_extensions_dir: str,
) -> None:
"""An extension with matching auto_enable_url_patterns should be auto-enabled.
Check that ``auto_enable_extensions_for_feed`` activates the plugin.
"""
# Register a test plugin with a URL pattern.
plugin_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class AutoPlugin(FeedExtension):
name = "auto_test"
auto_enable_url_patterns = [r"example\\.com/feeds"]
def process_entry(self, entry, reader):
return {"auto_var": "enabled"}
"""
plugin_path: Path = Path(temp_extensions_dir) / "auto_plugin.py"
plugin_path.write_text(plugin_code, encoding="utf-8")
discover_plugins(force=True)
mock_feed_url: str = "https://example.com/feeds/test.xml"
result: list[str] = auto_enable_extensions_for_feed(mock_reader, mock_feed_url)
assert "auto_test" in result
def test_auto_enable_by_url_pattern_does_not_match(
mock_reader: MagicMock,
temp_extensions_dir: str,
) -> None:
"""An extension whose pattern does not match should NOT be auto-enabled."""
# Register the same plugin as the matching test, so this test actually
# validates that a non-matching URL does not trigger auto-enable.
plugin_code: str = """
from discord_rss_bot.extensions.base import FeedExtension
class AutoPlugin(FeedExtension):
name = "auto_test"
auto_enable_url_patterns = [r"example\\.com/feeds"]
def process_entry(self, entry, reader):
return {"auto_var": "enabled"}
"""
plugin_path: Path = Path(temp_extensions_dir) / "auto_plugin.py"
plugin_path.write_text(plugin_code, encoding="utf-8")
discover_plugins(force=True)
mock_feed_url: str = "https://other-site.com/feed.xml"
result: list[str] = auto_enable_extensions_for_feed(mock_reader, mock_feed_url)
assert "auto_test" not in result
def test_auto_enable_does_not_duplicate_explicitly_enabled(
mock_reader: MagicMock,
) -> None:
"""If an extension is already explicitly enabled, auto-enable should not add it again."""
mock_feed_url: str = "https://example.com/feeds/test.xml"
# Explicitly enable the extension first.
set_enabled_extensions_for_feed(mock_reader, mock_feed_url, ["auto_test"])
result: list[str] = auto_enable_extensions_for_feed(mock_reader, mock_feed_url)
# Should contain auto_test only once.
assert result == ["auto_test"]
def test_matches_feed_url_classmethod() -> None:
"""The ``matches_feed_url`` classmethod should work correctly."""
assert FeedExtension.matches_feed_url("http://example.com") is False, "Base class has no patterns"
def test_steam_extension_has_auto_enable_patterns() -> None:
"""The built-in Steam extension should declare URL patterns."""
assert len(SteamExtension.auto_enable_url_patterns) > 0
assert SteamExtension.matches_feed_url("https://store.steampowered.com/feeds/news/app/570/")
assert SteamExtension.matches_feed_url("https://steamcommunity.com/games/570/")
assert not SteamExtension.matches_feed_url("https://example.com/feed.xml")
def test_youtube_extension_has_auto_enable_patterns() -> None:
"""The built-in YouTube extension should declare URL patterns."""
assert len(YouTubeExtension.auto_enable_url_patterns) > 0
assert YouTubeExtension.matches_feed_url("https://www.youtube.com/feeds/videos.xml?channel_id=123")
assert not YouTubeExtension.matches_feed_url("https://example.com/feed.xml")
def test_hoyolab_extension_has_auto_enable_patterns() -> None:
"""The built-in Hoyolab extension should declare URL patterns."""
assert len(HoyolabExtension.auto_enable_url_patterns) > 0
assert HoyolabExtension.matches_feed_url("https://feeds.c3kay.de/hoyolab.xml")
assert not HoyolabExtension.matches_feed_url("https://example.com/feed.xml")
# ---------------------------------------------------------------------------
# Integration test with a real WordPress comment RSS feed (the original issue)
# ---------------------------------------------------------------------------
_COMMENTS_FEED_XML: str = r"""<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0"
xmlns:content="http://purl.org/rss/1.0/modules/content/"
xmlns:dc="http://purl.org/dc/elements/1.1/"
xmlns:atom="http://www.w3.org/2005/Atom"
xmlns:sy="http://purl.org/rss/1.0/modules/syndication/">
<channel>
<title>Comments on: Test Article</title>
<link>https://example.com/test-article/</link>
<description></description>
<lastBuildDate>Sat, 18 Jul 2026 05:11:01 +0000</lastBuildDate>
<sy:updatePeriod>hourly</sy:updatePeriod>
<sy:updateFrequency>1</sy:updateFrequency>
<generator>https://wordpress.org/?v=5.8.2</generator>
<item>
<title>By: Anonymous</title>
<link>https://example.com/test-article/comment-page-1/#comment-820928</link>
<dc:creator><![CDATA[Anonymous]]></dc:creator>
<pubDate>Sat, 18 Jul 2026 05:11:01 +0000</pubDate>
<guid isPermaLink="false">https://example.com/?p=5633#comment-820928</guid>
<description><![CDATA[Am I the only one that saw the &quot;Bookmark us&quot; message?]]></description>
<content:encoded><![CDATA[<p>Am I the only one that saw &#8220;Bookmark us&#8221; message?</p>]]></content:encoded>
</item>
<item>
<title>By: Unknown</title>
<link>https://example.com/test-article/comment-page-1/#comment-820900</link>
<dc:creator><![CDATA[Unknown]]></dc:creator>
<pubDate>Fri, 17 Jul 2026 14:29:42 +0000</pubDate>
<guid isPermaLink="false">https://example.com/?p=5633#comment-820900</guid>
<description><![CDATA[Please uncensor the video]]></description>
<content:encoded><![CDATA[<p>Please uncensor the video</p>]]></content:encoded>
</item>
<item>
<title>By: mid</title>
<link>https://example.com/test-article/comment-page-1/#comment-820887</link>
<dc:creator><![CDATA[mid]]></dc:creator>
<pubDate>Fri, 17 Jul 2026 06:20:56 +0000</pubDate>
<guid isPermaLink="false">https://example.com/?p=5633#comment-820887</guid>
<description><![CDATA[that&#039;s not uncensored]]></description>
<content:encoded><![CDATA[<p>that&#8217;s not uncensored</p>]]></content:encoded>
</item>
</channel>
</rss>
"""
class _CommentsFeedHandler(BaseHTTPRequestHandler):
"""Serves the WordPress comment RSS XML on any request."""
def do_GET(self) -> None:
self.send_response(200)
self.send_header("Content-Type", "application/xml")
self.end_headers()
self.wfile.write(_COMMENTS_FEED_XML.encode("utf-8"))
def log_message(self, _format: str, *args: str | int) -> None:
pass
@contextmanager
def _serve_feed() -> Iterator[str]:
"""Start a local HTTP server serving the comments feed XML.
Yields:
The URL of the feed.
"""
with ThreadingHTTPServer(("127.0.0.1", 0), _CommentsFeedHandler) as server:
server_thread = Thread(target=server.serve_forever, daemon=True)
server_thread.start()
try:
yield f"http://127.0.0.1:{server.server_port}/feed.xml"
finally:
server.shutdown()
server_thread.join()
@pytest.mark.slow
def test_wordpress_comments_feed_pipeline() -> None:
"""End-to-end test using the WordPress comment RSS feed from the original issue.
Validates that:
1. The feed XML parses correctly via the reader library
2. Entry content and summary are populated as expected
3. The jwplayer_thumbnail extension gracefully returns empty (no script tags)
4. Tag replacement does not crash with comment feed data
"""
with (
tempfile.TemporaryDirectory() as tmpdir,
_serve_feed() as feed_url,
):
db_path: Path = Path(tmpdir) / "test.sqlite"
reader: ReaderType = make_reader(url=str(db_path))
try:
# Add and update the feed.
reader.add_feed(feed_url)
reader.update_feed(feed_url)
feed = reader.get_feed(feed_url)
entries: list[Entry] = list(reader.get_entries(feed=feed_url))
# The comments feed should have 3 entries.
assert len(entries) == 3, f"Expected 3 comments, got {len(entries)}"
for entry in entries:
# Each entry should have a title (comment author).
assert entry.title is not None
assert entry.title.startswith("By:")
# Each entry should have a link back to the comment.
assert entry.link is not None
assert "comment-page" in entry.link
# Summary should be plain text (from <description>),
# content should be HTML (from <content:encoded>).
assert entry.summary is not None
assert "<p>" not in entry.summary
if entry.content:
assert "<p>" in entry.content[0].value
# The jwplayer_thumbnail extension should return empty
# because this feed has no JWPlayer script blocks.
ext = JWPlayerThumbnailExtension()
result: dict[str, str] = ext.process_entry(entry, reader) # type: ignore[arg-type]
assert result == {}, f"JWPlayer extension should return empty for comment feed, got {result}"
# Tag replacement should not crash.
replaced_text: str = replace_tags_in_text_message(entry, reader)
assert isinstance(replaced_text, str)
# Embed replacement should not crash.
replaced_embed = replace_tags_in_embed(feed, entry, reader)
assert replaced_embed is not None
# Verify content:encoded was parsed into the content field.
first_entry = entries[0]
assert first_entry.content is not None
raw_content: str = first_entry.content[0].value
assert "Bookmark us" in raw_content or "bookmark" in raw_content.lower()
assert "<p>" in raw_content
finally:
reader.close()
# ---------------------------------------------------------------------------
# Main feed XML (hentaigasm-style, with JWPlayer video entries)
# ---------------------------------------------------------------------------
_MAIN_FEED_XML: str = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0"
xmlns:content="http://purl.org/rss/1.0/modules/content/"
xmlns:dc="http://purl.org/dc/elements/1.1/"
xmlns:atom="http://www.w3.org/2005/Atom"
xmlns:sy="http://purl.org/rss/1.0/modules/syndication/">
<channel>
<title>Hentaigasm - Test Feed</title>
<link>https://example.com</link>
<description></description>
<lastBuildDate>Sun, 19 Jul 2026 04:20:40 +0000</lastBuildDate>
<item>
<title>Test Video 1 Subbed</title>
<link>https://example.com/test-video-1/</link>
<dc:creator><![CDATA[admin]]></dc:creator>
<pubDate>Sun, 19 Jul 2026 04:00:49 +0000</pubDate>
<guid isPermaLink="false">https://example.com/?p=5501</guid>
<description><![CDATA[]]></description>
<content:encoded><![CDATA[
<p style="text-align: center;"><strong>HENTAIGASM EXCLUSIVE!</strong></p>
<div style="width:620px; height:349px">
<script src="https://content.jwplatform.com/libraries/SAHhwvZq.js"></script>
<script>jwplayer.key="zTEbSn/eAplL0RLXT030FzOcek6qXmtrxju6Jg=="</script>
<div id="player_01"></div>
<script type="text/javascript">
jwplayer("player_01").setup({
file: "https://cdn.example.com/Test Video 1 Subbed.mp4",
width: "620",
height: "349",
skin: "seven",
preload: "none",
autostart: "false",
image: "https://cdn.example.com/thumbnail/Test Video 1 Subbed.jpg",
advertising: {}
});
</script>
</div>
]]></content:encoded>
</item>
</channel>
</rss>
"""
class _MainFeedHandler(BaseHTTPRequestHandler):
"""Serves the main feed XML with JWPlayer video entries."""
def do_GET(self) -> None:
self.send_response(200)
self.send_header("Content-Type", "application/xml")
self.end_headers()
self.wfile.write(_MAIN_FEED_XML.encode("utf-8"))
def log_message(self, _format: str, *args: str | int) -> None:
pass
@contextmanager
def _serve_main_feed() -> Iterator[str]:
"""Start a local HTTP server serving the main feed XML with JWPlayer content.
Yields:
The URL of the feed.
"""
with ThreadingHTTPServer(("127.0.0.1", 0), _MainFeedHandler) as server:
server_thread = Thread(target=server.serve_forever, daemon=True)
server_thread.start()
try:
yield f"http://127.0.0.1:{server.server_port}/feed.xml"
finally:
server.shutdown()
server_thread.join()
@pytest.mark.slow
def test_hentaigasm_main_feed_jwplayer_extraction() -> None:
"""End-to-end test using a hentaigasm-style main feed with JWPlayer content.
This test confirms that feedparser strips ``<script>`` tags from
``<content:encoded>``, so the JWPlayer ``image:`` and ``file:``
properties are NOT available in ``entry.content``.
The extension falls back to fetching the HTML page at ``entry.link``,
but in the test environment no real server is available, so the
extension returns empty. A separate test validates the HTTP fetch
fallback with a local server.
"""
with (
tempfile.TemporaryDirectory() as tmpdir,
_serve_main_feed() as feed_url,
):
db_path: Path = Path(tmpdir) / "test.sqlite"
reader: ReaderType = make_reader(url=str(db_path))
try:
reader.add_feed(feed_url)
reader.update_feed(feed_url)
entries = list(reader.get_entries(feed=feed_url))
# Should have 1 video entry.
assert len(entries) == 1, f"Expected 1 entry, got {len(entries)}"
entry = entries[0]
assert entry.title is not None
# CONFIRMATION: feedparser strips <script> tags entirely.
# The content <p> and <div> survive, but all <script> blocks
# (including the jwplayer setup with image:/file:) are gone.
assert entry.content is not None
assert len(entry.content) > 0
raw_content: str = entry.content[0].value
assert "jwplayer" not in raw_content, (
f"BUG: script content survived parsing! Content: {raw_content[:300]!r}"
)
assert "image:" not in raw_content
assert "file:" not in raw_content
assert "player_01" in raw_content # non-script div survives
# Auto-enable is URL-based — the test feed URL (127.0.0.1) doesn't
# match "hentaigasm\.com", so we manually enable for this test.
set_enabled_extensions_for_feed(reader, entry.feed.url, ["jwplayer_thumbnail"])
enabled: list[str] = get_enabled_extensions_for_feed(reader, entry.feed.url)
assert "jwplayer_thumbnail" in enabled
finally:
reader.close()
@pytest.mark.slow
def test_jwplayer_thumbnail_wordpress_batch_fallback() -> None:
"""Verify the batch WordPress API fallback works.
Serves a fake WordPress REST API response (a list of posts) so the
extension extracts JWPlayer URLs from the cached batch data.
"""
_SLUG_CACHE.clear()
wp_json: str = json.dumps([
{
"slug": "test-slug",
"content": {
"rendered": (
'<p>Test</p><div id="player_01"></div>'
'<script>jwplayer("player_01").setup({'
'file: "https://cdn.example.com/video.mp4", '
'image: "https://cdn.example.com/thumb.jpg"'
"});</script>"
),
},
},
{
"slug": "other-post",
"content": {"rendered": "<p>No player here.</p>"},
},
])
class _APIHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(wp_json.encode("utf-8"))
def log_message(self, _format: str, *args: str | int) -> None:
pass
with ThreadingHTTPServer(("127.0.0.1", 0), _APIHandler) as server:
server_thread = Thread(target=server.serve_forever, daemon=True)
server_thread.start()
port: int = server.server_port
try:
entry = SimpleNamespace(
id="test-wp-batch",
content=[],
summary="",
link=f"http://127.0.0.1:{port}/test-slug/",
feed=SimpleNamespace(url=f"http://127.0.0.1:{port}/feed/"),
)
ext = JWPlayerThumbnailExtension()
result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type]
assert result.get("jwplayer_thumbnail") == "https://cdn.example.com/thumb.jpg", (
f"Batch WP fallback should extract thumbnail, got {result}"
)
assert result.get("jwplayer_file") == "https://cdn.example.com/video.mp4", (
f"Batch WP fallback should extract file, got {result}"
)
# Second entry from the same site uses cache — no API call.
entry2 = SimpleNamespace(
id="other-post",
content=[],
summary="",
link=f"http://127.0.0.1:{port}/other-post/",
feed=SimpleNamespace(url=f"http://127.0.0.1:{port}/feed/"),
)
result2: dict[str, str] = ext.process_entry(entry2, MagicMock()) # type: ignore[arg-type]
assert result2 == {}, "Entry without player should return empty"
finally:
server.shutdown()
server_thread.join()
_SLUG_CACHE.clear()
# ---------------------------------------------------------------------------
# Tests: WordPress extension
# ---------------------------------------------------------------------------
def test_wordpress_extension_uses_shared_batch_cache() -> None:
"""The WordPress extension should use the shared batch cache."""
_SLUG_CACHE.clear()
content_html: str = (
"<p>Test</p>"
"<script>jwplayer().setup({file: 'https://cdn.example.com/v.mp4', image: 'https://cdn.example.com/thumb.jpg'});</script>"
)
# Pre-populate the shared cache with the new richer format.
_SLUG_CACHE["https://example.com"] = {
"test-slug": {
"content": content_html,
"excerpt": "<p>Test excerpt</p>",
"title": "Test Post",
},
}
ext = WordPressExtension()
entry = SimpleNamespace(
id="test",
link="https://example.com/test-slug/",
feed=SimpleNamespace(url="https://example.com/feed/"),
)
result = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type]
assert result.get("wp_jwplayer_thumbnail") == "https://cdn.example.com/thumb.jpg"
assert result.get("wp_jwplayer_file") == "https://cdn.example.com/v.mp4"
assert result.get("wp_content_raw") == content_html
assert result.get("wp_content") is not None
assert "Test" in result.get("wp_content", "")
assert result.get("wp_excerpt_raw") == "<p>Test excerpt</p>"
assert result.get("wp_excerpt") is not None
assert "Test excerpt" in result.get("wp_excerpt", "")
# No spaces in URL values.
for key, val in result.items():
if key.startswith("wp_jwplayer"):
assert " " not in val, f"URL should be encoded, got spaces: {val}"
_SLUG_CACHE.clear()
def test_wordpress_extension_provides_correct_variables() -> None:
"""The WordPress extension should declare the correct variables."""
expected: set[str] = {
"wp_content",
"wp_content_raw",
"wp_excerpt",
"wp_excerpt_raw",
"wp_jwplayer_file",
"wp_jwplayer_thumbnail",
}
assert set(WordPressExtension.provides_variables) == expected

211
tests/test_feeds.py
View file

@ -22,6 +22,8 @@ from reader import StorageError
from reader import make_reader
from discord_rss_bot import feeds
from discord_rss_bot.extensions.steam import extract_app_id
from discord_rss_bot.extensions.youtube import is_youtube_feed_url
from discord_rss_bot.feeds import JsonObject
from discord_rss_bot.feeds import capture_full_page_screenshot
from discord_rss_bot.feeds import create_feed
@ -31,13 +33,11 @@ from discord_rss_bot.feeds import extract_domain
from discord_rss_bot.feeds import get_entry_delivery_mode
from discord_rss_bot.feeds import get_screenshot_layout
from discord_rss_bot.feeds import get_webhook_url
from discord_rss_bot.feeds import is_youtube_feed
from discord_rss_bot.feeds import screenshot_filename_for_entry
from discord_rss_bot.feeds import send_discord_quest_notification
from discord_rss_bot.feeds import send_entry_to_discord
from discord_rss_bot.feeds import send_to_discord
from discord_rss_bot.feeds import set_entry_as_read
from discord_rss_bot.feeds import should_send_embed_check
from discord_rss_bot.feeds import truncate_webhook_message
@ -119,57 +119,15 @@ def test_truncate_webhook_message_long_message():
def test_is_youtube_feed():
"""Test the is_youtube_feed function."""
"""Test the is_youtube_feed_url function."""
# YouTube feed URLs
assert is_youtube_feed("https://www.youtube.com/feeds/videos.xml?channel_id=123456") is True
assert is_youtube_feed("https://www.youtube.com/feeds/videos.xml?user=username") is True
assert is_youtube_feed_url("https://www.youtube.com/feeds/videos.xml?channel_id=123456")
assert is_youtube_feed_url("https://www.youtube.com/feeds/videos.xml?user=username")
# Non-YouTube feed URLs
assert is_youtube_feed("https://www.example.com/feed.xml") is False
assert is_youtube_feed("https://www.youtube.com/watch?v=123456") is False
assert is_youtube_feed("https://www.reddit.com/r/Python/.rss") is False
@patch("discord_rss_bot.feeds.logger")
def test_should_send_embed_check_youtube_feeds(mock_logger: MagicMock) -> None:
"""Test should_send_embed_check returns False for YouTube feeds regardless of settings."""
# Create mocks
mock_reader = MagicMock()
mock_entry = MagicMock()
# Configure a YouTube feed
mock_entry.feed.url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456"
# Set reader to return True for should_send_embed (would normally create an embed)
mock_reader.get_tag.return_value = True
# Result should be False, overriding the feed settings
result = should_send_embed_check(mock_reader, mock_entry)
assert result is False, "YouTube feeds should never use embeds"
# Function should not even call get_tag for YouTube feeds
mock_reader.get_tag.assert_not_called()
@patch("discord_rss_bot.feeds.logger")
def test_should_send_embed_check_normal_feeds(mock_logger: MagicMock) -> None:
"""Test should_send_embed_check returns feed settings for non-YouTube feeds."""
# Create mocks
mock_reader = MagicMock()
mock_entry = MagicMock()
# Configure a normal feed
mock_entry.feed.url = "https://www.example.com/feed.xml"
# Test with should_send_embed set to True
mock_reader.get_tag.return_value = True
result = should_send_embed_check(mock_reader, mock_entry)
assert result is True, "Normal feeds should use embeds when enabled"
# Test with should_send_embed set to False
mock_reader.get_tag.return_value = False
result = should_send_embed_check(mock_reader, mock_entry)
assert result is False, "Normal feeds should not use embeds when disabled"
assert not is_youtube_feed_url("https://www.example.com/feed.xml")
assert not is_youtube_feed_url("https://www.youtube.com/watch?v=123456")
assert not is_youtube_feed_url("https://www.reddit.com/r/Python/.rss")
def test_get_entry_delivery_mode_prefers_delivery_mode_tag() -> None:
@ -204,11 +162,7 @@ def test_get_entry_delivery_mode_falls_back_to_legacy_embed_flag() -> None:
@patch("discord_rss_bot.feeds.execute_webhook")
@patch("discord_rss_bot.feeds.create_text_webhook")
@patch("discord_rss_bot.feeds.create_hoyolab_webhook")
@patch("discord_rss_bot.feeds.fetch_hoyolab_post")
def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook(
mock_fetch_hoyolab_post: MagicMock,
mock_create_hoyolab_webhook: MagicMock,
mock_create_text_webhook: MagicMock,
mock_execute_webhook: MagicMock,
) -> None:
@ -230,8 +184,6 @@ def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook(
result = send_entry_to_discord(entry, reader)
assert result is None
mock_fetch_hoyolab_post.assert_not_called()
mock_create_hoyolab_webhook.assert_not_called()
mock_create_text_webhook.assert_called_once_with(
"https://discord.test/webhook",
entry,
@ -243,11 +195,7 @@ def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook(
@patch("discord_rss_bot.feeds.execute_webhook")
@patch("discord_rss_bot.feeds.create_screenshot_webhook")
@patch("discord_rss_bot.feeds.create_hoyolab_webhook")
@patch("discord_rss_bot.feeds.fetch_hoyolab_post")
def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook(
mock_fetch_hoyolab_post: MagicMock,
mock_create_hoyolab_webhook: MagicMock,
mock_create_screenshot_webhook: MagicMock,
mock_execute_webhook: MagicMock,
) -> None:
@ -269,8 +217,6 @@ def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook(
result = send_entry_to_discord(entry, reader)
assert result is None
mock_fetch_hoyolab_post.assert_not_called()
mock_create_hoyolab_webhook.assert_not_called()
mock_create_screenshot_webhook.assert_called_once_with(
"https://discord.test/webhook",
entry,
@ -281,14 +227,11 @@ def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook(
@patch("discord_rss_bot.feeds.execute_webhook")
@patch("discord_rss_bot.feeds.create_embed_webhook")
@patch("discord_rss_bot.feeds.create_hoyolab_webhook")
@patch("discord_rss_bot.feeds.fetch_hoyolab_post")
def test_send_entry_to_discord_hoyolab_embed_mode_uses_hoyolab_webhook(
mock_fetch_hoyolab_post: MagicMock,
mock_create_hoyolab_webhook: MagicMock,
def test_send_entry_to_discord_hoyolab_embed_mode_uses_embed_webhook(
mock_create_embed_webhook: MagicMock,
mock_execute_webhook: MagicMock,
) -> None:
"""Embed mode should use the standard embed pipeline, not a Hoyolab bypass."""
entry = MagicMock()
entry.id = "entry-3"
entry.feed.url = "https://feeds.c3kay.de/hoyolab.xml"
@ -301,21 +244,14 @@ def test_send_entry_to_discord_hoyolab_embed_mode_uses_hoyolab_webhook(
"delivery_mode": "embed",
}.get(key, default)
mock_fetch_hoyolab_post.return_value = {"post": {"subject": "News"}}
hoyolab_webhook = MagicMock()
mock_create_hoyolab_webhook.return_value = hoyolab_webhook
embed_webhook = MagicMock()
mock_create_embed_webhook.return_value = embed_webhook
result = send_entry_to_discord(entry, reader)
assert result is None
mock_fetch_hoyolab_post.assert_called_once_with("38588239")
mock_create_hoyolab_webhook.assert_called_once_with(
"https://discord.test/webhook",
entry,
{"post": {"subject": "News"}},
)
mock_create_embed_webhook.assert_not_called()
mock_execute_webhook.assert_called_once_with(hoyolab_webhook, entry, reader=reader)
mock_create_embed_webhook.assert_called_once()
mock_execute_webhook.assert_called_once_with(embed_webhook, entry, reader=reader)
def test_get_screenshot_layout_prefers_mobile_tag() -> None:
@ -391,7 +327,7 @@ def test_get_feed_media_gallery_image_limit_defaults_to_first_image() -> None:
],
)
def test_extract_steam_app_id_from_url(url: str, expected_app_id: str | None) -> None:
assert feeds.extract_steam_app_id_from_url(url) == expected_app_id
assert extract_app_id(url) == expected_app_id
@pytest.mark.parametrize(
@ -942,102 +878,6 @@ def test_create_embed_webhook_can_use_steam_game_icon_thumbnail(
entry.summary = ""
entry.content = []
entry.feed.url = "https://store.steampowered.com/feeds/news/app/570/?cc=us&l=english"
mock_replace_tags_in_embed.return_value = feeds.CustomEmbed(
description="Steam news",
thumbnail_url="https://example.com/custom-thumb.jpg",
show_steam_game_icon_in_thumbnail=True,
)
with patch("discord_rss_bot.feeds.Path.is_file", return_value=False):
webhook = feeds.create_embed_webhook("https://discord.com/api/webhooks/123/abc", entry, reader)
assert "components" not in webhook.json
embeds = webhook.json.get("embeds")
assert isinstance(embeds, list)
assert isinstance(embeds[0], dict)
assert embeds[0]["thumbnail"] == {
"url": "https://cdn.cloudflare.steamstatic.com/steam/apps/570/capsule_sm_120.jpg",
}
assert webhook.files == []
mock_fetch_ttvdrops_campaign_media_items.assert_not_called()
@patch("discord_rss_bot.feeds.fetch_ttvdrops_campaign_media_items", return_value=[])
@patch("discord_rss_bot.feeds.replace_tags_in_embed")
def test_create_embed_webhook_prefers_local_steam_game_icon_thumbnail(
mock_replace_tags_in_embed: MagicMock,
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
local_icon_bytes = b"local-steam-icon"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0,
"webhook_text_length_limit": 4000,
}.get(key, default)
entry = MagicMock()
entry.id = "entry-steam-local-1"
entry.title = "Dota 2 patch notes"
entry.link = "https://steamcommunity.com/games/570/announcements/detail/1234567890"
entry.summary = ""
entry.content = []
entry.feed.url = "https://store.steampowered.com/feeds/news/app/570/?cc=us&l=english"
mock_replace_tags_in_embed.return_value = feeds.CustomEmbed(
description="Steam news",
thumbnail_url="https://example.com/custom-thumb.jpg",
show_steam_game_icon_in_thumbnail=True,
)
with (
patch("discord_rss_bot.feeds.Path.is_file", return_value=True),
patch("discord_rss_bot.feeds.Path.read_bytes", return_value=local_icon_bytes),
):
webhook = feeds.create_embed_webhook("https://discord.com/api/webhooks/123/abc", entry, reader)
embeds = webhook.json.get("embeds")
assert isinstance(embeds, list)
assert isinstance(embeds[0], dict)
assert len(webhook.files) == 1
uploaded_icon = webhook.files[0]
assert uploaded_icon.content == local_icon_bytes
assert uploaded_icon.filename.startswith("steam-app-570-")
assert uploaded_icon.filename.endswith(".png")
assert embeds[0]["thumbnail"] == {"url": f"attachment://{uploaded_icon.filename}"}
mock_fetch_ttvdrops_campaign_media_items.assert_not_called()
@patch("discord_rss_bot.feeds.fetch_ttvdrops_campaign_media_items", return_value=[])
@patch("discord_rss_bot.feeds.replace_tags_in_embed")
def test_create_embed_webhook_does_not_inject_steam_thumbnail_when_app_id_is_missing(
mock_replace_tags_in_embed: MagicMock,
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0,
"webhook_text_length_limit": 4000,
}.get(key, default)
entry = MagicMock()
entry.id = "entry-steam-2"
entry.title = "Steam group post"
entry.link = "https://steamcommunity.com/groups/example/announcements/detail/1234567890"
entry.summary = ""
entry.content = []
entry.feed.url = "https://steamcommunity.com/groups/example/rss/"
mock_replace_tags_in_embed.return_value = feeds.CustomEmbed(
description="Steam group news",
thumbnail_url="https://example.com/custom-thumb.jpg",
show_steam_game_icon_in_thumbnail=True,
)
webhook = feeds.create_embed_webhook("https://discord.com/api/webhooks/123/abc", entry, reader)
assert "components" not in webhook.json
embeds = webhook.json.get("embeds")
assert isinstance(embeds, list)
assert isinstance(embeds[0], dict)
assert "thumbnail" not in embeds[0]
mock_fetch_ttvdrops_campaign_media_items.assert_not_called()
@patch("discord_rss_bot.feeds.fetch_ttvdrops_campaign_media_items", return_value=[])
@ -1338,10 +1178,11 @@ def test_send_entry_to_discord_youtube_feed(
mock_entry.feed.url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456"
mock_entry.feed_url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456"
# Mock the tags
# Mock the tags — with no delivery_mode tag and should_send_embed=True,
# the entry delivery mode will be "embed", which calls create_embed_webhook.
mock_reader.get_tag.side_effect = lambda feed, tag, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.com/api/webhooks/123/abc",
"should_send_embed": True, # This should be ignored for YouTube feeds
"should_send_embed": True,
}.get(tag, default)
# Mock custom message
@ -1355,23 +1196,19 @@ def test_send_entry_to_discord_youtube_feed(
# Call the function
send_entry_to_discord(mock_entry, mock_reader)
# Assertions
mock_create_embed.assert_not_called()
mock_discord_webhook.assert_called_once()
# Check webhook was created with the right message
webhook_call_kwargs = mock_discord_webhook.call_args[1]
assert "content" in webhook_call_kwargs, "Webhook should have content"
assert webhook_call_kwargs["url"] == "https://discord.com/api/webhooks/123/abc"
# Assertions — YouTube feeds now follow standard delivery mode.
# With should_send_embed=True, embed mode is used.
mock_create_embed.assert_called_once()
mock_discord_webhook.assert_not_called()
# Verify execute_webhook was called
mock_execute_webhook.assert_called_once_with(mock_webhook, mock_entry, reader=mock_reader)
mock_execute_webhook.assert_called_once()
def test_extract_domain_youtube_feed() -> None:
"""Test extract_domain for YouTube feeds."""
url: str = "https://www.youtube.com/feeds/videos.xml?channel_id=123456"
assert extract_domain(url) == "YouTube", "YouTube feeds should return 'YouTube' as the domain."
assert extract_domain(url) == "YouTube"
def test_extract_domain_reddit_feed() -> None:

248
tests/test_hoyolab_api.py
View file

@ -1,248 +0,0 @@
from __future__ import annotations
import json
import typing
from types import SimpleNamespace
from unittest.mock import MagicMock
from unittest.mock import patch
import pytest
import requests
from discord_rss_bot.hoyolab_api import create_hoyolab_webhook
from discord_rss_bot.hoyolab_api import extract_post_id_from_hoyolab_url
from discord_rss_bot.hoyolab_api import fetch_hoyolab_post
from discord_rss_bot.hoyolab_api import is_c3kay_feed
if typing.TYPE_CHECKING:
from reader import Entry
class TestExtractPostIdFromHoyolabUrl:
def test_extract_post_id_from_article_url(self) -> None:
"""Test extracting post ID from a direct article URL."""
test_cases: list[str] = [
"https://www.hoyolab.com/article/38588239",
"http://hoyolab.com/article/12345",
"https://www.hoyolab.com/article/987654321/comments",
]
expected_ids: list[str] = ["38588239", "12345", "987654321"]
for url, expected_id in zip(test_cases, expected_ids, strict=False):
assert extract_post_id_from_hoyolab_url(url) == expected_id
def test_url_without_post_id(self) -> None:
"""Test with a URL that doesn't have a post ID."""
test_cases: list[str] = [
"https://www.hoyolab.com/community",
]
for url in test_cases:
assert extract_post_id_from_hoyolab_url(url) is None
def test_edge_cases(self) -> None:
"""Test edge cases like None, empty string, and malformed URLs."""
test_cases: list[str | None] = [
None,
"",
"not_a_url",
"http:/", # Malformed URL
]
for url in test_cases:
assert extract_post_id_from_hoyolab_url(url) is None # type: ignore
def make_entry(link: str | None = "https://www.hoyolab.com/article/38588239") -> SimpleNamespace:
feed: SimpleNamespace = SimpleNamespace(url="https://feeds.c3kay.de/hoyolab.xml")
return SimpleNamespace(
id="entry-123",
link=link,
feed=feed,
)
class TestIsC3KayFeed:
def test_true_for_c3kay_feed(self) -> None:
assert is_c3kay_feed("https://feeds.c3kay.de/rss") is True
def test_false_for_non_c3kay_feed(self) -> None:
assert is_c3kay_feed("https://example.com/rss") is False
class TestFetchHoyolabPost:
@patch("discord_rss_bot.hoyolab_api.requests.get")
def test_returns_none_for_empty_post_id(self, mock_get: MagicMock) -> None:
assert fetch_hoyolab_post("") is None
mock_get.assert_not_called()
@patch("discord_rss_bot.hoyolab_api.requests.get")
def test_returns_post_data_for_success_response(self, mock_get: MagicMock) -> None:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"retcode": 0,
"data": {
"post": {
"post_id": "38588239",
"subject": "Event",
},
},
}
mock_get.return_value = mock_response
result = fetch_hoyolab_post("38588239")
assert result == {"post_id": "38588239", "subject": "Event"}
assert mock_get.call_args.args[0].endswith("post_id=38588239")
@patch("discord_rss_bot.hoyolab_api.logger")
@patch("discord_rss_bot.hoyolab_api.requests.get")
def test_returns_none_and_logs_warning_for_non_success_payload(
self,
mock_get: MagicMock,
mock_logger: MagicMock,
) -> None:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = "bad payload"
mock_response.json.return_value = {
"retcode": -1,
"data": {},
}
mock_get.return_value = mock_response
result = fetch_hoyolab_post("38588239")
assert result is None
mock_logger.warning.assert_called_once()
@patch("discord_rss_bot.hoyolab_api.logger")
@patch("discord_rss_bot.hoyolab_api.requests.get")
def test_returns_none_and_logs_exception_on_request_error(
self,
mock_get: MagicMock,
mock_logger: MagicMock,
) -> None:
mock_get.side_effect = requests.RequestException("network issue")
result = fetch_hoyolab_post("38588239")
assert result is None
mock_logger.exception.assert_called_once()
class TestCreateHoyolabWebhook:
@patch("discord_rss_bot.hoyolab_api.requests.get")
@patch("discord_rss_bot.hoyolab_api.DiscordEmbed")
@patch("discord_rss_bot.hoyolab_api.DiscordWebhook")
def test_builds_embed_webhook_with_full_post_data(
self,
mock_webhook_cls: MagicMock,
mock_embed_cls: MagicMock,
mock_requests_get: MagicMock,
) -> None:
webhook_instance = MagicMock()
embed_instance = MagicMock()
mock_webhook_cls.return_value = webhook_instance
mock_embed_cls.return_value = embed_instance
video_response = MagicMock()
video_response.ok = True
video_response.content = b"video-bytes"
mock_requests_get.return_value = video_response
post_data = {
"post": {
"subject": "Update 4.0",
"content": json.dumps({"describe": "Patch notes"}),
"desc": "fallback description",
"structured_content": json.dumps(
[{"insert": {"video": "https://www.youtube.com/embed/abc123_XY"}}],
),
"event_start_date": "1712000000",
"event_end_date": "1712600000",
"created_at": "1711000000",
},
"image_list": [{"url": "https://img.example.com/hero.jpg", "height": 1080, "width": 1920}],
"video": {"url": "https://cdn.example.com/video.mp4"},
"game": {"color": "#11AAFF"},
"user": {"nickname": "Paimon", "avatar_url": "https://img.example.com/avatar.jpg"},
"classification": {"name": "Official"},
}
entry = make_entry(link=None)
entry = typing.cast("Entry", entry)
webhook = create_hoyolab_webhook("https://discord.test/webhook", entry, post_data) # type: ignore
assert webhook is webhook_instance
mock_webhook_cls.assert_called_once_with(url="https://discord.test/webhook", rate_limit_retry=True)
embed_instance.set_title.assert_called_once_with("Update 4.0")
embed_instance.set_url.assert_called_once_with("https://feeds.c3kay.de/hoyolab.xml")
embed_instance.set_image.assert_called_once_with(
url="https://img.example.com/hero.jpg",
height=1080,
width=1920,
)
embed_instance.set_color.assert_called_once_with("11AAFF")
embed_instance.set_footer.assert_called_once_with(text="Official")
embed_instance.add_embed_field.assert_any_call(name="Start", value="<t:1712000000:R>")
embed_instance.add_embed_field.assert_any_call(name="End", value="<t:1712600000:R>")
embed_instance.set_timestamp.assert_called_once_with(timestamp="1711000000")
webhook_instance.add_file.assert_called_once_with(file=b"video-bytes", filename="entry-123.mp4")
webhook_instance.add_embed.assert_called_once_with(embed_instance)
assert webhook_instance.content == "https://www.youtube.com/watch?v=abc123_XY"
webhook_instance.remove_embeds.assert_called_once()
@patch("discord_rss_bot.hoyolab_api.requests.get")
@patch("discord_rss_bot.hoyolab_api.DiscordEmbed")
@patch("discord_rss_bot.hoyolab_api.DiscordWebhook")
def test_handles_invalid_structured_content_without_removing_embeds(
self,
mock_webhook_cls: MagicMock,
mock_embed_cls: MagicMock,
mock_requests_get: MagicMock,
) -> None:
webhook_instance = MagicMock()
embed_instance = MagicMock()
mock_webhook_cls.return_value = webhook_instance
mock_embed_cls.return_value = embed_instance
mock_requests_get.return_value = MagicMock(ok=False)
post_data = {
"post": {
"subject": "News",
"content": "{}",
"structured_content": "not-json",
},
}
entry = make_entry()
entry = typing.cast("Entry", entry)
webhook = create_hoyolab_webhook("https://discord.test/webhook", entry, post_data) # type: ignore
assert webhook is webhook_instance
webhook_instance.remove_embeds.assert_not_called()
def test_extract_post_id_with_querystring() -> None:
url = "https://www.hoyolab.com/article/38588239?utm_source=feed"
assert extract_post_id_from_hoyolab_url(url) == "38588239"
def test_extract_post_id_non_string_input_returns_none() -> None:
assert extract_post_id_from_hoyolab_url(None) is None # type: ignore[arg-type]
@pytest.mark.parametrize(
("url", "expected"),
[
("https://feeds.c3kay.de/rss", True),
("https://www.hoyolab.com/feed", False),
],
)
def test_is_c3kay_feed_parametrized(*, url: str, expected: bool) -> None:
assert is_c3kay_feed(url) is expected

19
tests/test_main.py
View file

@ -32,6 +32,8 @@ if TYPE_CHECKING:
from reader import Entry
from reader import Reader
from discord_rss_bot.feeds import JsonValue
client: TestClient = TestClient(app)
webhook_name: str = "Hello, I am a webhook!"
webhook_url: str = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"
@ -318,7 +320,7 @@ def test_feed_page_shows_steam_thumbnail_hint_for_steam_feeds() -> None:
assert feed_url == self.feed.url
return self.feed
def get_tag(self, _resource: object, key: str, default: TestTagValue = None) -> TestTagValue:
def get_tag(self, _resource: str | DummyFeed, key: str, default: TestTagValue = None) -> TestTagValue:
return {
"webhooks": [],
"delivery_mode": "embed",
@ -2645,7 +2647,7 @@ def test_export_opml_with_stub_reader() -> None:
filename="reader-feeds-2026-07-18-12-00-00.opml",
)
def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object:
def get_tag(self, _resource: tuple[()], _key: str, default: TestTagValue = None) -> TestTagValue:
return default
def get_feeds(self) -> list:
@ -2672,7 +2674,7 @@ class StubImportConfirmReader:
self.added_urls: list[str] = []
self.tags: dict[tuple[str, str], str] = {}
def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object:
def get_tag(self, _resource: tuple[()], _key: str, default: TestTagValue = None) -> TestTagValue:
"""Stub get_tag.
Returns:
@ -2705,7 +2707,7 @@ class StubImportConfirmReaderWithExisting:
self.added_urls: list[str] = []
self.tags: dict[tuple[str, str], str] = {}
def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object:
def get_tag(self, _resource: tuple[()], _key: str, default: TestTagValue = None) -> TestTagValue:
"""Stub get_tag.
Returns:
@ -2857,7 +2859,6 @@ def test_post_embed_saves_all_fields() -> None:
"thumbnail_url": "https://example.com/thumb.png",
"footer_text": "Footer Text",
"footer_icon_url": "https://example.com/footer.png",
"show_steam_game_icon_in_thumbnail": "true",
},
follow_redirects=False,
)
@ -2880,7 +2881,6 @@ def test_post_embed_saves_all_fields() -> None:
assert saved["thumbnail_url"] == "https://example.com/thumb.png"
assert saved["footer_text"] == "Footer Text"
assert saved["footer_icon_url"] == "https://example.com/footer.png"
assert saved["show_steam_game_icon_in_thumbnail"] is True
finally:
app.dependency_overrides = {}
@ -2958,7 +2958,6 @@ def test_post_embed_allows_clearing_all_fields() -> None:
"thumbnail_url": "https://old.example.com/thumb.png",
"footer_text": "Old Footer",
"footer_icon_url": "https://old.example.com/footer.png",
"show_steam_game_icon_in_thumbnail": True,
})
stub = _make_stub_reader_for_embed(stored_embed=existing)
app.dependency_overrides[get_reader_dependency] = lambda: stub
@ -2998,7 +2997,6 @@ def test_post_embed_allows_clearing_all_fields() -> None:
assert not saved["thumbnail_url"]
assert not saved["footer_text"]
assert not saved["footer_icon_url"]
assert saved["show_steam_game_icon_in_thumbnail"] is False
finally:
app.dependency_overrides = {}
@ -3016,7 +3014,6 @@ def test_post_embed_untouched_fields_retain_values() -> None:
"thumbnail_url": "",
"footer_text": "Old Footer",
"footer_icon_url": "",
"show_steam_game_icon_in_thumbnail": False,
})
stub = _make_stub_reader_for_embed(stored_embed=existing)
app.dependency_overrides[get_reader_dependency] = lambda: stub
@ -3056,7 +3053,6 @@ def test_post_embed_untouched_fields_retain_values() -> None:
assert saved["author_name"] == "Author"
assert saved["author_url"] == "https://a.example.com"
assert saved["footer_text"] == "Old Footer"
assert saved["show_steam_game_icon_in_thumbnail"] is False
finally:
app.dependency_overrides = {}
@ -3094,7 +3090,6 @@ def test_post_embed_saves_empty_description_when_no_prior_embed_exists() -> None
saved: dict[str, str] = json.loads(json_arg)
assert saved["title"] == "Just a Title"
assert not saved["description"], f"Expected empty description, got {saved['description']!r}"
assert saved["show_steam_game_icon_in_thumbnail"] is False
finally:
app.dependency_overrides = {}
@ -3412,7 +3407,7 @@ class _StubReaderForMass:
return self._webhook_url
return default
def set_tag(self, resource: str, key: str, value: object) -> None: # pyright: ignore[reportArgumentType]
def set_tag(self, resource: str, key: str, value: JsonValue) -> None:
"""Record tag set calls."""
self.set_tag_calls.append((resource, key, value))

2
tests/test_settings.py
View file

@ -39,7 +39,7 @@ class _AutodiscoverHandler(BaseHTTPRequestHandler):
b'type="application/rss+xml" title="Example"></head></html>',
)
def log_message(self, _format: str, *_args: object) -> None:
def log_message(self, _format: str, *_args: str | int) -> None:
"""Suppress HTTP request logging during tests."""

4
tests/test_whitelist.py
View file

@ -11,8 +11,8 @@ from reader import make_reader
from discord_rss_bot.filter.evaluator import evaluate_entry_filters
from discord_rss_bot.filter.evaluator import get_filter_values_from_reader
from discord_rss_bot.filter.whitelist import has_white_tags
from discord_rss_bot.filter.whitelist import should_be_sent
from discord_rss_bot.filter.evaluator import has_white_tags
from discord_rss_bot.filter.evaluator import should_be_sent
if TYPE_CHECKING:
from collections.abc import Iterable