Compare commits

...

3 commits

Author SHA1 Message Date
c55610affa
Improve tests
All checks were successful
Test and build Docker image / docker (push) Successful in 1m40s
2026-04-09 21:38:33 +02:00
cba35edb19
Fix bug in healthcheck 2026-04-09 19:14:12 +02:00
e9d219676f
Add pytest-cov to development dependencies 2026-04-09 19:05:57 +02:00
13 changed files with 881 additions and 75 deletions

View file

@ -61,7 +61,7 @@ jobs:
run: uv sync --all-extras --all-groups run: uv sync --all-extras --all-groups
- name: Run tests - name: Run tests
run: uv run pytest run: uv run pytest -o addopts='-n 5 --dist loadfile -m ""'
- id: tags - id: tags
name: Compute image tags name: Compute image tags

View file

@ -38,14 +38,16 @@ repos:
# An extremely fast Python linter and formatter. # An extremely fast Python linter and formatter.
- repo: https://github.com/astral-sh/ruff-pre-commit - repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.15.5 rev: v0.15.10
hooks: hooks:
- id: ruff-format - id: ruff-format
- id: ruff types_or: [ python, pyi, jupyter, pyproject ]
- id: ruff-check
types_or: [ python, pyi, jupyter, pyproject ]
args: ["--fix", "--exit-non-zero-on-fix"] args: ["--fix", "--exit-non-zero-on-fix"]
# Static checker for GitHub Actions workflow files. # Static checker for GitHub Actions workflow files.
- repo: https://github.com/rhysd/actionlint - repo: https://github.com/rhysd/actionlint
rev: v1.7.11 rev: v1.7.12
hooks: hooks:
- id: actionlint - id: actionlint

View file

@ -256,6 +256,7 @@ def replace_tags_in_embed(feed: Feed, entry: Entry, reader: Reader) -> CustomEmb
content = format_entry_html_for_discord(content) content = format_entry_html_for_discord(content)
feed_added: str = feed.added.strftime("%Y-%m-%d %H:%M:%S") if feed.added else "Never" feed_added: str = feed.added.strftime("%Y-%m-%d %H:%M:%S") if feed.added else "Never"
feed_last_exception: str = feed.last_exception.value_str if feed.last_exception else ""
feed_last_updated: str = feed.last_updated.strftime("%Y-%m-%d %H:%M:%S") if feed.last_updated else "Never" feed_last_updated: str = feed.last_updated.strftime("%Y-%m-%d %H:%M:%S") if feed.last_updated else "Never"
feed_updated: str = feed.updated.strftime("%Y-%m-%d %H:%M:%S") if feed.updated else "Never" feed_updated: str = feed.updated.strftime("%Y-%m-%d %H:%M:%S") if feed.updated else "Never"
entry_added: str = entry.added.strftime("%Y-%m-%d %H:%M:%S") if entry.added else "Never" entry_added: str = entry.added.strftime("%Y-%m-%d %H:%M:%S") if entry.added else "Never"
@ -269,14 +270,6 @@ def replace_tags_in_embed(feed: Feed, entry: Entry, reader: Reader) -> CustomEmb
embed.author_name = embed.title embed.author_name = embed.title
embed.title = "" embed.title = ""
feed_added: str = feed.added.strftime("%Y-%m-%d %H:%M:%S") if feed.added else "Never"
feed_last_exception: str = feed.last_exception.value_str if feed.last_exception else ""
feed_last_updated: str = feed.last_updated.strftime("%Y-%m-%d %H:%M:%S") if feed.last_updated else "Never"
feed_updated: str = feed.updated.strftime("%Y-%m-%d %H:%M:%S") if feed.updated else "Never"
entry_published: str = entry.published.strftime("%Y-%m-%d %H:%M:%S") if entry.published else "Never"
entry_read_modified: str = entry.read_modified.strftime("%Y-%m-%d %H:%M:%S") if entry.read_modified else "Never"
entry_updated: str = entry.updated.strftime("%Y-%m-%d %H:%M:%S") if entry.updated else "Never"
list_of_replacements: list[dict[str, str]] = [ list_of_replacements: list[dict[str, str]] = [
{"{{feed_author}}": feed.author or ""}, {"{{feed_author}}": feed.author or ""},
{"{{feed_added}}": feed_added or ""}, {"{{feed_added}}": feed_added or ""},

View file

@ -16,6 +16,7 @@ def healthcheck() -> None:
r: requests.Response = requests.get(url="http://localhost:5000", timeout=5) r: requests.Response = requests.get(url="http://localhost:5000", timeout=5)
if r.ok: if r.ok:
sys.exit(0) sys.exit(0)
sys.exit(1)
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f"Healthcheck failed: {e}", file=sys.stderr) # noqa: T201 print(f"Healthcheck failed: {e}", file=sys.stderr) # noqa: T201
sys.exit(1) sys.exit(1)

View file

@ -22,14 +22,20 @@ dependencies = [
] ]
[dependency-groups] [dependency-groups]
dev = ["djlint", "pytest", "pytest-randomly", "pytest-xdist"] dev = [
"djlint",
"identify",
"pytest",
"pytest-cov",
"pytest-randomly",
"pytest-xdist",
]
[build-system] [build-system]
requires = ["poetry-core>=1.0.0"] requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api" build-backend = "poetry.core.masonry.api"
[tool.ruff] [tool.ruff]
preview = true preview = true
unsafe-fixes = true unsafe-fixes = true
fix = true fix = true
@ -87,7 +93,11 @@ lint.ignore = [
"tests/*" = ["S101", "D103", "PLR2004"] "tests/*" = ["S101", "D103", "PLR2004"]
[tool.pytest.ini_options] [tool.pytest.ini_options]
addopts = "-n 5 --dist loadfile" addopts = "-n 5 --dist loadfile -m \"not integration and not slow\""
markers = [
"integration: tests that exercise external integrations or end-to-end flows",
"slow: tests that are intentionally slower than the default fast test suite",
]
filterwarnings = [ filterwarnings = [
"ignore::bs4.GuessedAtParserWarning", "ignore::bs4.GuessedAtParserWarning",
"ignore:functools\\.partial will be a method descriptor in future Python versions; wrap it in staticmethod\\(\\) if you want to preserve the old behavior:FutureWarning", "ignore:functools\\.partial will be a method descriptor in future Python versions; wrap it in staticmethod\\(\\) if you want to preserve the old behavior:FutureWarning",

View file

@ -0,0 +1,107 @@
from __future__ import annotations
import os
import sys
from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
import tests.conftest as hooks
if TYPE_CHECKING:
from pathlib import Path
import pytest
def test_pytest_addoption_registers_real_git_backup_flag() -> None:
"""The hook should register the opt-in flag for real git-backup tests."""
parser: MagicMock = MagicMock()
hooks.pytest_addoption(parser)
parser.addoption.assert_called_once_with(
"--run-real-git-backup-tests",
action="store_true",
default=False,
help="Run tests that push git backup state to a real repository.",
)
def test_pytest_sessionstart_initializes_worker_data_dir(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""The hook should set worker-scoped state and silence bs4 locator warnings."""
monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw3")
monkeypatch.setattr(hooks.tempfile, "gettempdir", lambda: str(tmp_path))
filterwarnings_mock: MagicMock = MagicMock()
monkeypatch.setattr(hooks.warnings, "filterwarnings", filterwarnings_mock)
hooks.pytest_sessionstart(session=MagicMock())
expected_dir: Path = tmp_path / "discord-rss-bot-tests" / "gw3"
assert expected_dir.exists(), f"Expected worker dir to exist: {expected_dir}"
assert os.environ.get("DISCORD_RSS_BOT_DATA_DIR") == str(expected_dir)
filterwarnings_mock.assert_any_call("ignore", category=hooks.MarkupResemblesLocatorWarning)
def test_pytest_sessionstart_refreshes_preloaded_settings_and_main_modules(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""Preloaded modules should be re-pointed to worker-local storage and refreshed."""
monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw9")
monkeypatch.setattr(hooks.tempfile, "gettempdir", lambda: str(tmp_path))
get_reader: MagicMock = MagicMock()
get_reader.cache_clear = MagicMock() # type: ignore[attr-defined]
close: MagicMock = MagicMock()
settings_module = SimpleNamespace(data_dir="stale", get_reader=get_reader)
main_module = SimpleNamespace(reader=SimpleNamespace(close=close))
monkeypatch.setitem(sys.modules, "discord_rss_bot.settings", settings_module)
monkeypatch.setitem(sys.modules, "discord_rss_bot.main", main_module)
hooks.pytest_sessionstart(session=MagicMock())
expected_dir: Path = tmp_path / "discord-rss-bot-tests" / "gw9"
assert settings_module.data_dir == str(expected_dir)
get_reader.cache_clear.assert_called_once() # type: ignore[attr-defined]
close.assert_called_once()
get_reader.assert_called_once()
def test_pytest_sessionstart_suppresses_reader_close_exceptions(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""Reader close failures should not prevent rebuilding the main reader."""
monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw11")
monkeypatch.setattr(hooks.tempfile, "gettempdir", lambda: str(tmp_path))
get_reader: MagicMock = MagicMock()
settings_module = SimpleNamespace(data_dir="stale", get_reader=get_reader)
failing_reader = SimpleNamespace(close=MagicMock(side_effect=RuntimeError("close failed")))
main_module = SimpleNamespace(reader=failing_reader)
monkeypatch.setitem(sys.modules, "discord_rss_bot.settings", settings_module)
monkeypatch.setitem(sys.modules, "discord_rss_bot.main", main_module)
hooks.pytest_sessionstart(session=MagicMock())
get_reader.assert_called_once()
def test_pytest_collection_modifyitems_noops_when_real_git_backup_tests_enabled() -> None:
"""When the flag is enabled, collection hook should return immediately."""
config: MagicMock = MagicMock()
config.getoption.return_value = True
items: list[MagicMock] = [MagicMock()]
hooks.pytest_collection_modifyitems(config=config, items=items)
config.getoption.assert_called_once_with("--run-real-git-backup-tests")

View file

@ -9,11 +9,18 @@ import pytest
from discord_rss_bot.custom_message import CustomEmbed from discord_rss_bot.custom_message import CustomEmbed
from discord_rss_bot.custom_message import format_entry_html_for_discord from discord_rss_bot.custom_message import format_entry_html_for_discord
from discord_rss_bot.custom_message import get_custom_message
from discord_rss_bot.custom_message import get_embed
from discord_rss_bot.custom_message import get_embed_data
from discord_rss_bot.custom_message import get_first_image
from discord_rss_bot.custom_message import replace_tags_in_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 replace_tags_in_text_message
from discord_rss_bot.custom_message import save_embed
from discord_rss_bot.custom_message import try_to_replace
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
from reader import Entry from reader import Entry
from reader.types import Feed
# https://docs.discord.com/developers/reference#message-formatting # https://docs.discord.com/developers/reference#message-formatting
TIMESTAMP_FORMATS: tuple[str, ...] = ( TIMESTAMP_FORMATS: tuple[str, ...] = (
@ -138,3 +145,217 @@ def test_replace_tags_in_embed_preserves_timestamp_tags(
for timestamp_tag in TIMESTAMP_FORMATS: for timestamp_tag in TIMESTAMP_FORMATS:
assert timestamp_tag in embed.description assert timestamp_tag in embed.description
def test_try_to_replace_returns_original_message_when_replace_fails() -> None:
rendered = try_to_replace(typing.cast("str", None), "{{tag}}", "value")
assert rendered is None
@patch("discord_rss_bot.custom_message.get_custom_message")
def test_replace_tags_in_text_message_uses_last_content_item_and_unescapes_newline(
mock_get_custom_message: MagicMock,
) -> None:
mock_reader = MagicMock()
mock_get_custom_message.return_value = "{{entry_content}}\\n{{entry_content_raw}}"
entry_ns: SimpleNamespace = make_entry("<p>Summary</p>")
entry_ns.content = [SimpleNamespace(value="<p>First content</p>"), SimpleNamespace(value="<p>Last content</p>")]
entry: Entry = typing.cast("Entry", entry_ns)
rendered: str = replace_tags_in_text_message(entry, reader=mock_reader)
assert "Last content" in rendered
assert "<p>First content</p>" in rendered
assert "\\n" not in rendered
assert "\n" in rendered
@patch("discord_rss_bot.custom_message.get_custom_message")
def test_replace_tags_in_text_message_skips_non_string_replacement_values(
mock_get_custom_message: MagicMock,
) -> None:
mock_reader = MagicMock()
mock_get_custom_message.return_value = "{{entry_id}}"
entry_ns: SimpleNamespace = make_entry("<p>Summary</p>")
entry_ns.id = 123
entry: Entry = typing.cast("Entry", entry_ns)
rendered: str = replace_tags_in_text_message(entry, reader=mock_reader)
assert rendered == "{{entry_id}}"
def test_get_first_image_prefers_content_image_over_summary_image() -> None:
summary = '<p><img src="https://example.com/from-summary.jpg" /></p>'
content = '<p><img src="https://example.com/from-content.jpg" /></p>'
image = get_first_image(summary, content)
assert image == "https://example.com/from-content.jpg"
def test_get_first_image_uses_summary_when_content_image_is_invalid() -> None:
summary = '<p><img src="https://example.com/from-summary.jpg" /></p>'
content = '<p><img src="javascript:alert(1)" /></p>'
image = get_first_image(summary, content)
assert image == "https://example.com/from-summary.jpg"
def test_get_first_image_returns_empty_when_images_have_no_src() -> None:
summary = "<p></p>"
content = '<p><img alt="missing source" /></p>'
image = get_first_image(summary, content)
assert not image
def test_get_first_image_returns_empty_when_summary_image_url_is_invalid() -> None:
summary = '<p><img src="javascript:alert(1)" /></p>'
image = get_first_image(summary, content=None)
assert not image
def test_get_first_image_returns_empty_when_summary_image_has_no_src() -> None:
summary = '<p><img alt="missing source" /></p>'
image = get_first_image(summary, content=None)
assert not image
@patch("discord_rss_bot.custom_message.get_embed")
def test_replace_tags_in_embed_moves_title_to_author_name_when_required(
mock_get_embed: MagicMock,
) -> None:
mock_reader = MagicMock()
mock_get_embed.return_value = CustomEmbed(
title="{{entry_title}}",
author_name="",
author_url="https://example.com/author",
)
entry_ns: SimpleNamespace = make_entry("<p>Summary</p>")
entry: Entry = typing.cast("Entry", entry_ns)
embed: CustomEmbed = replace_tags_in_embed(entry_ns.feed, entry, reader=mock_reader)
assert not embed.title
assert embed.author_name == "Entry Title"
@patch("discord_rss_bot.custom_message.get_embed")
def test_replace_tags_in_embed_uses_last_content_item(
mock_get_embed: MagicMock,
) -> None:
mock_reader = MagicMock()
mock_get_embed.return_value = CustomEmbed(description="{{entry_content}}")
entry_ns: SimpleNamespace = make_entry("<p>Summary</p>")
entry_ns.content = [SimpleNamespace(value="<p>Old content</p>"), SimpleNamespace(value="<p>New content</p>")]
entry: Entry = typing.cast("Entry", entry_ns)
embed: CustomEmbed = replace_tags_in_embed(entry_ns.feed, entry, reader=mock_reader)
assert "New content" in embed.description
def test_get_custom_message_returns_empty_string_on_value_error() -> None:
reader = MagicMock()
feed = make_feed()
reader.get_tag.side_effect = ValueError
feed = typing.cast("Feed", feed)
custom_message = get_custom_message(reader=reader, feed=feed)
assert not custom_message
def test_save_embed_serializes_embed_and_writes_feed_tag() -> None:
reader = MagicMock()
feed = make_feed()
feed = typing.cast("Feed", feed)
embed = CustomEmbed(
title="Title",
description="Description",
color="#123456",
author_name="Author",
author_url="https://example.com/author",
author_icon_url="https://example.com/author.png",
image_url="https://example.com/image.png",
thumbnail_url="https://example.com/thumb.png",
footer_text="Footer",
footer_icon_url="https://example.com/footer.png",
)
save_embed(reader=reader, feed=feed, embed=embed)
reader.set_tag.assert_called_once()
call_args = reader.set_tag.call_args.args
assert call_args[1] == "embed"
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"
def test_get_embed_returns_default_embed_when_tag_is_empty() -> None:
reader = MagicMock()
feed = make_feed()
feed = typing.cast("Feed", feed)
reader.get_tag.return_value = ""
embed = get_embed(reader=reader, feed=feed)
assert embed.color == "#469ad9"
def test_get_embed_reads_embed_from_dict_tag() -> None:
reader = MagicMock()
feed = make_feed()
feed = typing.cast("Feed", feed)
reader.get_tag.return_value = {
"title": "Dict title",
"description": "Dict description",
"color": 123,
}
embed = get_embed(reader=reader, feed=feed)
assert embed.title == "Dict title"
assert embed.description == "Dict description"
assert embed.color == "123"
def test_get_embed_reads_embed_from_json_string() -> None:
reader = MagicMock()
feed = make_feed()
feed = typing.cast("Feed", feed)
reader.get_tag.return_value = '{"title": "Json title", "footer_text": "Json footer"}'
embed = get_embed(reader=reader, feed=feed)
assert embed.title == "Json title"
assert embed.footer_text == "Json footer"
def test_get_embed_data_coerces_values_to_strings() -> None:
embed = get_embed_data(
{
"title": 1,
"description": 2,
"color": 3,
"author_name": 4,
"author_url": 5,
"author_icon_url": 6,
"image_url": 7,
"thumbnail_url": 8,
"footer_text": 9,
"footer_icon_url": 10,
},
)
assert embed.title == "1"
assert embed.footer_icon_url == "10"

View file

@ -8,20 +8,33 @@ from unittest.mock import MagicMock
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from reader import EntryNotFoundError
from reader import Feed from reader import Feed
from reader import FeedNotFoundError
from reader import Reader from reader import Reader
from reader import StorageError
from reader import make_reader from reader import make_reader
from discord_rss_bot import feeds
from discord_rss_bot.feeds import execute_webhook
from discord_rss_bot.feeds import extract_domain from discord_rss_bot.feeds import extract_domain
from discord_rss_bot.feeds import get_webhook_url
from discord_rss_bot.feeds import is_youtube_feed from discord_rss_bot.feeds import is_youtube_feed
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_entry_to_discord
from discord_rss_bot.feeds import send_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 should_send_embed_check
from discord_rss_bot.feeds import truncate_webhook_message from discord_rss_bot.feeds import truncate_webhook_message
def test_send_to_discord() -> None: def test_send_to_discord() -> None:
"""Test sending to Discord.""" """Test sending to Discord."""
# Skip early if no webhook URL is configured to avoid a real network request.
webhook_url: str | None = os.environ.get("TEST_WEBHOOK_URL")
if not webhook_url:
pytest.skip("No webhook URL provided.")
with tempfile.TemporaryDirectory() as temp_dir: with tempfile.TemporaryDirectory() as temp_dir:
# Create the temp directory. # Create the temp directory.
Path.mkdir(Path(temp_dir), exist_ok=True) Path.mkdir(Path(temp_dir), exist_ok=True)
@ -41,13 +54,6 @@ def test_send_to_discord() -> None:
feed: Feed = reader.get_feed("https://www.reddit.com/r/Python/.rss") feed: Feed = reader.get_feed("https://www.reddit.com/r/Python/.rss")
assert feed is not None, f"The feed should not be None. Got: {feed}" assert feed is not None, f"The feed should not be None. Got: {feed}"
# Get the webhook.
webhook_url: str | None = os.environ.get("TEST_WEBHOOK_URL")
if not webhook_url:
reader.close()
pytest.skip("No webhook URL provided.")
assert webhook_url is not None, f"The webhook URL should not be None. Got: {webhook_url}" assert webhook_url is not None, f"The webhook URL should not be None. Got: {webhook_url}"
# Add tag to the feed and check if it is there. # Add tag to the feed and check if it is there.
@ -274,3 +280,154 @@ def test_extract_domain_special_characters() -> None:
) )
def test_extract_domain(url: str, expected: str) -> None: def test_extract_domain(url: str, expected: str) -> None:
assert extract_domain(url) == expected assert extract_domain(url) == expected
@patch("discord_rss_bot.feeds.execute_webhook")
def test_send_discord_quest_notification_text_match(mock_execute_webhook: MagicMock) -> None:
"""Send a quest link as a separate notification when plain text content contains one."""
entry = MagicMock()
entry.id = "entry-1"
entry.content = [MagicMock(type="text", value="Check this https://discord.com/quests/12345 now")]
reader = MagicMock()
send_discord_quest_notification(entry, "https://discord.com/api/webhooks/123/abc", reader)
mock_execute_webhook.assert_called_once()
webhook_sent = mock_execute_webhook.call_args[0][0]
assert webhook_sent.content == "https://discord.com/quests/12345"
@patch("discord_rss_bot.feeds.execute_webhook")
def test_send_discord_quest_notification_html_match(mock_execute_webhook: MagicMock) -> None:
"""Send a quest link when it is found inside HTML content."""
entry = MagicMock()
entry.id = "entry-2"
entry.content = [
MagicMock(
type="text/html",
value='<p>Click <a href="https://discord.com/quests/777">here</a></p>',
),
]
reader = MagicMock()
send_discord_quest_notification(entry, "https://discord.com/api/webhooks/123/abc", reader)
mock_execute_webhook.assert_called_once()
webhook_sent = mock_execute_webhook.call_args[0][0]
assert webhook_sent.content == "https://discord.com/quests/777"
@patch("discord_rss_bot.feeds.execute_webhook")
def test_send_discord_quest_notification_no_match(mock_execute_webhook: MagicMock) -> None:
"""Do nothing when no quest URL exists in entry content."""
entry = MagicMock()
entry.id = "entry-3"
entry.content = [MagicMock(type="text", value="No quest link here")]
reader = MagicMock()
send_discord_quest_notification(entry, "https://discord.com/api/webhooks/123/abc", reader)
mock_execute_webhook.assert_not_called()
def test_get_webhook_url_returns_value() -> None:
reader = MagicMock()
entry = MagicMock()
entry.feed_url = "https://example.com/feed.xml"
entry.feed.url = "https://example.com/feed.xml"
reader.get_tag.return_value = "https://discord.com/api/webhooks/123/abc"
result = get_webhook_url(reader, entry)
assert result == "https://discord.com/api/webhooks/123/abc"
def test_get_webhook_url_returns_empty_on_storage_error() -> None:
reader = MagicMock()
entry = MagicMock()
entry.feed_url = "https://example.com/feed.xml"
entry.feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = StorageError("db error")
result = get_webhook_url(reader, entry)
assert not result
def test_set_entry_as_read_handles_entry_not_found_error() -> None:
reader = MagicMock()
entry = MagicMock(id="entry-4")
reader.set_entry_read.side_effect = EntryNotFoundError("https://example.com/feed.xml", "entry-4")
set_entry_as_read(reader, entry)
reader.set_entry_read.assert_called_once_with(entry, True)
def test_set_entry_as_read_handles_storage_error() -> None:
reader = MagicMock()
entry = MagicMock(id="entry-5")
reader.set_entry_read.side_effect = StorageError("db error")
set_entry_as_read(reader, entry)
reader.set_entry_read.assert_called_once_with(entry, True)
def test_execute_webhook_skips_when_feed_paused() -> None:
webhook = MagicMock()
reader = MagicMock()
entry = MagicMock()
entry.id = "entry-6"
entry.feed.url = "https://example.com/feed.xml"
entry.feed.updates_enabled = False
execute_webhook(webhook, entry, reader)
reader.get_feed.assert_not_called()
webhook.execute.assert_not_called()
def test_execute_webhook_skips_when_feed_missing() -> None:
webhook = MagicMock()
reader = MagicMock()
reader.get_feed.side_effect = FeedNotFoundError("missing")
entry = MagicMock()
entry.id = "entry-7"
entry.feed.url = "https://example.com/feed.xml"
entry.feed.updates_enabled = True
execute_webhook(webhook, entry, reader)
webhook.execute.assert_not_called()
@patch.object(feeds, "logger")
def test_execute_webhook_logs_error_on_bad_status(mock_logger: MagicMock) -> None:
webhook = MagicMock()
webhook.json = {"content": "test"}
webhook.execute.return_value = MagicMock(status_code=500, text="fail")
reader = MagicMock()
entry = MagicMock()
entry.id = "entry-8"
entry.feed.url = "https://example.com/feed.xml"
entry.feed.updates_enabled = True
execute_webhook(webhook, entry, reader)
mock_logger.error.assert_called_once()
@patch.object(feeds, "logger")
def test_execute_webhook_logs_info_on_success(mock_logger: MagicMock) -> None:
webhook = MagicMock()
webhook.execute.return_value = MagicMock(status_code=204, text="")
reader = MagicMock()
entry = MagicMock()
entry.id = "entry-9"
entry.feed.url = "https://example.com/feed.xml"
entry.feed.updates_enabled = True
execute_webhook(webhook, entry, reader)
mock_logger.info.assert_called_once_with("Sent entry to Discord: %s", "entry-9")

View file

@ -21,6 +21,7 @@ from discord_rss_bot.git_backup import setup_backup_repo
from discord_rss_bot.main import app from discord_rss_bot.main import app
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path from pathlib import Path
@ -66,17 +67,46 @@ def test_get_backup_remote_set(monkeypatch: pytest.MonkeyPatch) -> None:
def test_setup_backup_repo_creates_git_repo(tmp_path: Path) -> None: def test_setup_backup_repo_creates_git_repo(tmp_path: Path) -> None:
"""setup_backup_repo initialises a git repo in a fresh directory.""" """setup_backup_repo initialises a git repo in a fresh directory."""
backup_path: Path = tmp_path / "backup" backup_path: Path = tmp_path / "backup"
result: bool = setup_backup_repo(backup_path)
with patch("discord_rss_bot.git_backup.subprocess.run") as mock_run:
mock_run.side_effect = [
MagicMock(returncode=0), # git init
MagicMock(returncode=1), # config user.email read
MagicMock(returncode=0), # config user.email write
MagicMock(returncode=1), # config user.name read
MagicMock(returncode=0), # config user.name write
]
result: bool = setup_backup_repo(backup_path)
assert result is True assert result is True
assert (backup_path / ".git").exists() assert backup_path.exists()
called_commands: list[list[str]] = [call.args[0] for call in mock_run.call_args_list]
assert any(cmd[:2] == [shutil.which("git") or "git", "init"] for cmd in called_commands)
@SKIP_IF_NO_GIT @SKIP_IF_NO_GIT
def test_setup_backup_repo_idempotent(tmp_path: Path) -> None: def test_setup_backup_repo_idempotent(tmp_path: Path) -> None:
"""setup_backup_repo does not fail when called on an existing repo.""" """setup_backup_repo does not re-run init when called on an existing repo."""
backup_path: Path = tmp_path / "backup" backup_path: Path = tmp_path / "backup"
assert setup_backup_repo(backup_path) is True (backup_path / ".git").mkdir(parents=True)
assert setup_backup_repo(backup_path) is True
with patch("discord_rss_bot.git_backup.subprocess.run") as mock_run:
mock_run.side_effect = [
MagicMock(returncode=1), # first call: config user.email read
MagicMock(returncode=0), # first call: config user.email write
MagicMock(returncode=1), # first call: config user.name read
MagicMock(returncode=0), # first call: config user.name write
MagicMock(returncode=0), # second call: config user.email read
MagicMock(returncode=0), # second call: config user.name read
]
assert setup_backup_repo(backup_path) is True
assert setup_backup_repo(backup_path) is True
called_commands: list[list[str]] = [call.args[0] for call in mock_run.call_args_list]
init_calls: list[list[str]] = [cmd for cmd in called_commands if "init" in cmd]
assert not init_calls
def test_setup_backup_repo_adds_origin_remote(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: def test_setup_backup_repo_adds_origin_remote(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@ -210,7 +240,7 @@ def test_commit_state_change_noop_when_not_configured(monkeypatch: pytest.Monkey
@SKIP_IF_NO_GIT @SKIP_IF_NO_GIT
def test_commit_state_change_commits(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: def test_commit_state_change_commits(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""commit_state_change creates a commit in the backup repo.""" """commit_state_change stages and commits exported state when it changes."""
backup_path: Path = tmp_path / "backup" backup_path: Path = tmp_path / "backup"
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path)) monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False) monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False)
@ -219,20 +249,25 @@ def test_commit_state_change_commits(monkeypatch: pytest.MonkeyPatch, tmp_path:
mock_reader.get_feeds.return_value = [] mock_reader.get_feeds.return_value = []
mock_reader.get_tag.return_value = [] mock_reader.get_tag.return_value = []
commit_state_change(mock_reader, "Add feed https://example.com/rss") with (
patch("discord_rss_bot.git_backup.setup_backup_repo", return_value=True) as mock_setup,
patch("discord_rss_bot.git_backup.export_state") as mock_export,
patch("discord_rss_bot.git_backup.subprocess.run") as mock_run,
):
mock_run.side_effect = [
MagicMock(returncode=0), # git add -A
MagicMock(returncode=1), # diff --cached --exit-code => staged changes present
MagicMock(returncode=0), # git commit -m <message>
]
# Verify a commit was created in the backup repo commit_state_change(mock_reader, "Add feed https://example.com/rss")
git_executable: str | None = shutil.which("git")
assert git_executable is not None, "git executable not found" mock_setup.assert_called_once_with(backup_path)
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603 mock_export.assert_called_once_with(mock_reader, backup_path)
[git_executable, "-C", str(backup_path), "log", "--oneline"], called_commands: list[list[str]] = [call.args[0] for call in mock_run.call_args_list]
capture_output=True, assert any(cmd[-2:] == ["add", "-A"] for cmd in called_commands)
text=True, assert any(cmd[-3:] == ["diff", "--cached", "--exit-code"] for cmd in called_commands)
check=False, assert any(cmd[-3:] == ["commit", "-m", "Add feed https://example.com/rss"] for cmd in called_commands)
)
assert result.returncode == 0
assert "Add feed https://example.com/rss" in result.stdout
@SKIP_IF_NO_GIT @SKIP_IF_NO_GIT
@ -246,20 +281,26 @@ def test_commit_state_change_no_double_commit(monkeypatch: pytest.MonkeyPatch, t
mock_reader.get_feeds.return_value = [] mock_reader.get_feeds.return_value = []
mock_reader.get_tag.return_value = [] mock_reader.get_tag.return_value = []
commit_state_change(mock_reader, "First commit") with (
commit_state_change(mock_reader, "Should not appear") patch("discord_rss_bot.git_backup.setup_backup_repo", return_value=True),
patch("discord_rss_bot.git_backup.export_state"),
patch("discord_rss_bot.git_backup.subprocess.run") as mock_run,
):
mock_run.side_effect = [
MagicMock(returncode=0), # first: git add -A
MagicMock(returncode=1), # first: diff => changed
MagicMock(returncode=0), # first: commit
MagicMock(returncode=0), # second: git add -A
MagicMock(returncode=0), # second: diff => no changes
]
git_executable: str | None = shutil.which("git") commit_state_change(mock_reader, "First commit")
assert git_executable is not None, "git executable not found" commit_state_change(mock_reader, "Should not appear")
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603
[git_executable, "-C", str(backup_path), "log", "--oneline"], called_commands: list[list[str]] = [call.args[0] for call in mock_run.call_args_list]
capture_output=True, commit_calls: list[list[str]] = [cmd for cmd in called_commands if "commit" in cmd]
text=True, assert len(commit_calls) == 1
check=False, assert commit_calls[0][-3:] == ["commit", "-m", "First commit"]
)
assert result.returncode == 0
assert "First commit" in result.stdout
assert "Should not appear" not in result.stdout
def test_commit_state_change_push_when_remote_set(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: def test_commit_state_change_push_when_remote_set(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@ -310,21 +351,23 @@ test_webhook_url: str = "https://discord.com/api/webhooks/999999999/testbackupwe
test_feed_url: str = "https://lovinator.space/rss_test.xml" test_feed_url: str = "https://lovinator.space/rss_test.xml"
def setup_test_feed() -> None: @pytest.fixture(scope="module", autouse=True)
"""Set up a test webhook and feed for endpoint tests.""" def feed_module_setup() -> Generator[None]:
# Clean up existing test data """Set up the test webhook and feed once for all backup endpoint tests."""
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
client.post(url="/remove", data={"feed_url": test_feed_url}) client.post(url="/remove", data={"feed_url": test_feed_url})
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
client.post(url="/delete_webhook", data={"webhook_url": test_webhook_url}) client.post(url="/delete_webhook", data={"webhook_url": test_webhook_url})
# Create webhook and feed
client.post( client.post(
url="/add_webhook", url="/add_webhook",
data={"webhook_name": test_webhook_name, "webhook_url": test_webhook_url}, data={"webhook_name": test_webhook_name, "webhook_url": test_webhook_url},
) )
client.post(url="/add", data={"feed_url": test_feed_url, "webhook_dropdown": test_webhook_name}) client.post(url="/add", data={"feed_url": test_feed_url, "webhook_dropdown": test_webhook_name})
yield
with contextlib.suppress(Exception):
client.post(url="/remove", data={"feed_url": test_feed_url})
with contextlib.suppress(Exception):
client.post(url="/delete_webhook", data={"webhook_url": test_webhook_url})
def test_post_embed_triggers_backup(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: def test_post_embed_triggers_backup(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@ -334,8 +377,6 @@ def test_post_embed_triggers_backup(monkeypatch: pytest.MonkeyPatch, tmp_path: P
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path)) monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False) monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False)
setup_test_feed()
with patch("discord_rss_bot.main.commit_state_change") as mock_commit: with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
response = client.post( response = client.post(
url="/embed", url="/embed",
@ -363,8 +404,6 @@ def test_post_use_embed_triggers_backup(monkeypatch: pytest.MonkeyPatch, tmp_pat
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path)) monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False) monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False)
setup_test_feed()
with patch("discord_rss_bot.main.commit_state_change") as mock_commit: with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
response = client.post(url="/use_embed", data={"feed_url": test_feed_url}) response = client.post(url="/use_embed", data={"feed_url": test_feed_url})
assert response.status_code == 200, f"Failed to enable embed: {response.text}" assert response.status_code == 200, f"Failed to enable embed: {response.text}"
@ -384,8 +423,6 @@ def test_post_use_text_triggers_backup(monkeypatch: pytest.MonkeyPatch, tmp_path
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path)) monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False) monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False)
setup_test_feed()
with patch("discord_rss_bot.main.commit_state_change") as mock_commit: with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
response = client.post(url="/use_text", data={"feed_url": test_feed_url}) response = client.post(url="/use_text", data={"feed_url": test_feed_url})
assert response.status_code == 200, f"Failed to disable embed: {response.text}" assert response.status_code == 200, f"Failed to disable embed: {response.text}"
@ -405,8 +442,6 @@ def test_post_custom_message_triggers_backup(monkeypatch: pytest.MonkeyPatch, tm
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path)) monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False) monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False)
setup_test_feed()
with patch("discord_rss_bot.main.commit_state_change") as mock_commit: with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
response = client.post( response = client.post(
url="/custom", url="/custom",
@ -426,6 +461,8 @@ def test_post_custom_message_triggers_backup(monkeypatch: pytest.MonkeyPatch, tm
assert test_feed_url in commit_message assert test_feed_url in commit_message
@pytest.mark.integration
@pytest.mark.slow
@SKIP_IF_NO_GIT @SKIP_IF_NO_GIT
def test_embed_backup_end_to_end(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: def test_embed_backup_end_to_end(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""End-to-end test: customizing embed creates a real commit in the backup repo.""" """End-to-end test: customizing embed creates a real commit in the backup repo."""
@ -436,8 +473,6 @@ def test_embed_backup_end_to_end(monkeypatch: pytest.MonkeyPatch, tmp_path: Path
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path)) monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False) monkeypatch.delenv("GIT_BACKUP_REMOTE", raising=False)
setup_test_feed()
# Post embed customization # Post embed customization
response = client.post( response = client.post(
url="/embed", url="/embed",

69
tests/test_healthcheck.py Normal file
View file

@ -0,0 +1,69 @@
from __future__ import annotations
from unittest.mock import MagicMock
from unittest.mock import patch
import pytest
import requests
from discord_rss_bot.healthcheck import healthcheck
def test_healthcheck_success() -> None:
"""Test that healthcheck exits with 0 when the website is up."""
mock_response = MagicMock()
mock_response.ok = True
with (
patch("discord_rss_bot.healthcheck.requests.get", return_value=mock_response),
pytest.raises(SystemExit) as exc_info,
):
healthcheck()
assert exc_info.value.code == 0
def test_healthcheck_not_ok() -> None:
"""Test that healthcheck exits with 1 when the response is not ok."""
mock_response = MagicMock()
mock_response.ok = False
with (
patch("discord_rss_bot.healthcheck.requests.get", return_value=mock_response),
pytest.raises(SystemExit) as exc_info,
):
healthcheck()
assert exc_info.value.code == 1
def test_healthcheck_request_exception(capsys: pytest.CaptureFixture) -> None:
"""Test that healthcheck exits with 1 on a request exception."""
with (
patch(
"discord_rss_bot.healthcheck.requests.get",
side_effect=requests.exceptions.ConnectionError("Connection refused"),
),
pytest.raises(SystemExit) as exc_info,
):
healthcheck()
assert exc_info.value.code == 1
captured = capsys.readouterr()
assert "Healthcheck failed" in captured.err
def test_healthcheck_timeout(capsys: pytest.CaptureFixture) -> None:
"""Test that healthcheck exits with 1 on a timeout."""
with (
patch(
"discord_rss_bot.healthcheck.requests.get",
side_effect=requests.exceptions.Timeout("Request timed out"),
),
pytest.raises(SystemExit) as exc_info,
):
healthcheck()
assert exc_info.value.code == 1
captured = capsys.readouterr()
assert "Healthcheck failed" in captured.err

View file

@ -1,6 +1,21 @@
from __future__ import annotations 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 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: class TestExtractPostIdFromHoyolabUrl:
@ -37,3 +52,197 @@ class TestExtractPostIdFromHoyolabUrl:
for url in test_cases: for url in test_cases:
assert extract_post_id_from_hoyolab_url(url) is None # type: ignore 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)
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)
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

View file

@ -45,6 +45,7 @@ def test_search() -> None:
# Delete the webhook if it already exists before we run the test. # Delete the webhook if it already exists before we run the test.
response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
assert response.status_code == 200, f"Failed to delete webhook: {response.text}"
# Add the webhook. # Add the webhook.
response: Response = client.post( response: Response = client.post(
@ -71,6 +72,7 @@ def test_add_webhook() -> None:
"""Test the /add_webhook page.""" """Test the /add_webhook page."""
# Delete the webhook if it already exists before we run the test. # Delete the webhook if it already exists before we run the test.
response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
assert response.status_code == 200, f"Failed to delete webhook: {response.text}"
# Add the webhook. # Add the webhook.
response: Response = client.post( response: Response = client.post(
@ -461,15 +463,16 @@ def test_delete_webhook() -> None:
url="/add_webhook", url="/add_webhook",
data={"webhook_name": webhook_name, "webhook_url": webhook_url}, data={"webhook_name": webhook_name, "webhook_url": webhook_url},
) )
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
# Delete the webhook. # Delete the webhook.
response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) response2: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
assert response.status_code == 200, f"Failed to delete webhook: {response.text}" assert response2.status_code == 200, f"Failed to delete webhook: {response2.text}"
# Check that the webhook was added. # Check that the webhook was added.
response = client.get(url="/webhooks") response3 = client.get(url="/webhooks")
assert response.status_code == 200, f"Failed to get /webhooks: {response.text}" assert response3.status_code == 200, f"Failed to get /webhooks: {response3.text}"
assert webhook_name not in response.text, f"Webhook found in /webhooks: {response.text}" assert webhook_name not in response3.text, f"Webhook found in /webhooks: {response3.text}"
def test_update_feed_not_found() -> None: def test_update_feed_not_found() -> None:

View file

@ -53,7 +53,6 @@ def test_get_webhook_for_entry() -> None:
# Add a feed to the database. # Add a feed to the database.
reader.add_feed("https://www.reddit.com/r/movies.rss") reader.add_feed("https://www.reddit.com/r/movies.rss")
reader.update_feed("https://www.reddit.com/r/movies.rss")
# Add a webhook to the database. # Add a webhook to the database.
reader.set_tag("https://www.reddit.com/r/movies.rss", "webhook", "https://example.com") # pyright: ignore[reportArgumentType] reader.set_tag("https://www.reddit.com/r/movies.rss", "webhook", "https://example.com") # pyright: ignore[reportArgumentType]