Compare commits
3 commits
ece6e42b68
...
c55610affa
| Author | SHA1 | Date | |
|---|---|---|---|
|
c55610affa |
|||
|
cba35edb19 |
|||
|
e9d219676f |
13 changed files with 881 additions and 75 deletions
|
|
@ -61,7 +61,7 @@ jobs:
|
|||
run: uv sync --all-extras --all-groups
|
||||
|
||||
- name: Run tests
|
||||
run: uv run pytest
|
||||
run: uv run pytest -o addopts='-n 5 --dist loadfile -m ""'
|
||||
|
||||
- id: tags
|
||||
name: Compute image tags
|
||||
|
|
|
|||
|
|
@ -38,14 +38,16 @@ repos:
|
|||
|
||||
# An extremely fast Python linter and formatter.
|
||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||
rev: v0.15.5
|
||||
rev: v0.15.10
|
||||
hooks:
|
||||
- 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"]
|
||||
|
||||
# Static checker for GitHub Actions workflow files.
|
||||
- repo: https://github.com/rhysd/actionlint
|
||||
rev: v1.7.11
|
||||
rev: v1.7.12
|
||||
hooks:
|
||||
- id: actionlint
|
||||
|
|
|
|||
|
|
@ -256,6 +256,7 @@ def replace_tags_in_embed(feed: Feed, entry: Entry, reader: Reader) -> CustomEmb
|
|||
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_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_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.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]] = [
|
||||
{"{{feed_author}}": feed.author or ""},
|
||||
{"{{feed_added}}": feed_added or ""},
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ def healthcheck() -> None:
|
|||
r: requests.Response = requests.get(url="http://localhost:5000", timeout=5)
|
||||
if r.ok:
|
||||
sys.exit(0)
|
||||
sys.exit(1)
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Healthcheck failed: {e}", file=sys.stderr) # noqa: T201
|
||||
sys.exit(1)
|
||||
|
|
|
|||
|
|
@ -22,14 +22,20 @@ dependencies = [
|
|||
]
|
||||
|
||||
[dependency-groups]
|
||||
dev = ["djlint", "pytest", "pytest-randomly", "pytest-xdist"]
|
||||
dev = [
|
||||
"djlint",
|
||||
"identify",
|
||||
"pytest",
|
||||
"pytest-cov",
|
||||
"pytest-randomly",
|
||||
"pytest-xdist",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core>=1.0.0"]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
|
||||
[tool.ruff]
|
||||
|
||||
preview = true
|
||||
unsafe-fixes = true
|
||||
fix = true
|
||||
|
|
@ -87,7 +93,11 @@ lint.ignore = [
|
|||
"tests/*" = ["S101", "D103", "PLR2004"]
|
||||
|
||||
[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 = [
|
||||
"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",
|
||||
|
|
|
|||
107
tests/test_conftest_hooks.py
Normal file
107
tests/test_conftest_hooks.py
Normal 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")
|
||||
|
|
@ -9,11 +9,18 @@ import pytest
|
|||
|
||||
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 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_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:
|
||||
from reader import Entry
|
||||
from reader.types import Feed
|
||||
|
||||
# https://docs.discord.com/developers/reference#message-formatting
|
||||
TIMESTAMP_FORMATS: tuple[str, ...] = (
|
||||
|
|
@ -138,3 +145,217 @@ def test_replace_tags_in_embed_preserves_timestamp_tags(
|
|||
|
||||
for timestamp_tag in TIMESTAMP_FORMATS:
|
||||
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"
|
||||
|
|
|
|||
|
|
@ -8,20 +8,33 @@ from unittest.mock import MagicMock
|
|||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from reader import EntryNotFoundError
|
||||
from reader import Feed
|
||||
from reader import FeedNotFoundError
|
||||
from reader import Reader
|
||||
from reader import StorageError
|
||||
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 get_webhook_url
|
||||
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_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
|
||||
|
||||
|
||||
def test_send_to_discord() -> None:
|
||||
"""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:
|
||||
# Create the temp directory.
|
||||
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")
|
||||
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}"
|
||||
|
||||
# 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:
|
||||
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")
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from discord_rss_bot.git_backup import setup_backup_repo
|
|||
from discord_rss_bot.main import app
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Generator
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
|
|
@ -66,18 +67,47 @@ def test_get_backup_remote_set(monkeypatch: pytest.MonkeyPatch) -> None:
|
|||
def test_setup_backup_repo_creates_git_repo(tmp_path: Path) -> None:
|
||||
"""setup_backup_repo initialises a git repo in a fresh directory."""
|
||||
backup_path: Path = tmp_path / "backup"
|
||||
|
||||
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 (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
|
||||
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 / ".git").mkdir(parents=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:
|
||||
"""setup_backup_repo adds remote 'origin' when GIT_BACKUP_REMOTE is set."""
|
||||
|
|
@ -210,7 +240,7 @@ def test_commit_state_change_noop_when_not_configured(monkeypatch: pytest.Monkey
|
|||
|
||||
@SKIP_IF_NO_GIT
|
||||
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"
|
||||
monkeypatch.setenv("GIT_BACKUP_PATH", str(backup_path))
|
||||
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_tag.return_value = []
|
||||
|
||||
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>
|
||||
]
|
||||
|
||||
commit_state_change(mock_reader, "Add feed https://example.com/rss")
|
||||
|
||||
# Verify a commit was created in the backup repo
|
||||
git_executable: str | None = shutil.which("git")
|
||||
|
||||
assert git_executable is not None, "git executable not found"
|
||||
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603
|
||||
[git_executable, "-C", str(backup_path), "log", "--oneline"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "Add feed https://example.com/rss" in result.stdout
|
||||
mock_setup.assert_called_once_with(backup_path)
|
||||
mock_export.assert_called_once_with(mock_reader, backup_path)
|
||||
called_commands: list[list[str]] = [call.args[0] for call in mock_run.call_args_list]
|
||||
assert any(cmd[-2:] == ["add", "-A"] for cmd in called_commands)
|
||||
assert any(cmd[-3:] == ["diff", "--cached", "--exit-code"] for cmd in called_commands)
|
||||
assert any(cmd[-3:] == ["commit", "-m", "Add feed https://example.com/rss"] for cmd in called_commands)
|
||||
|
||||
|
||||
@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_tag.return_value = []
|
||||
|
||||
with (
|
||||
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
|
||||
]
|
||||
|
||||
commit_state_change(mock_reader, "First commit")
|
||||
commit_state_change(mock_reader, "Should not appear")
|
||||
|
||||
git_executable: str | None = shutil.which("git")
|
||||
assert git_executable is not None, "git executable not found"
|
||||
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603
|
||||
[git_executable, "-C", str(backup_path), "log", "--oneline"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "First commit" in result.stdout
|
||||
assert "Should not appear" not in result.stdout
|
||||
called_commands: list[list[str]] = [call.args[0] for call in mock_run.call_args_list]
|
||||
commit_calls: list[list[str]] = [cmd for cmd in called_commands if "commit" in cmd]
|
||||
assert len(commit_calls) == 1
|
||||
assert commit_calls[0][-3:] == ["commit", "-m", "First commit"]
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
def setup_test_feed() -> None:
|
||||
"""Set up a test webhook and feed for endpoint tests."""
|
||||
# Clean up existing test data
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def feed_module_setup() -> Generator[None]:
|
||||
"""Set up the test webhook and feed once for all backup endpoint tests."""
|
||||
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})
|
||||
|
||||
# Create webhook and feed
|
||||
client.post(
|
||||
url="/add_webhook",
|
||||
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})
|
||||
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:
|
||||
|
|
@ -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.delenv("GIT_BACKUP_REMOTE", raising=False)
|
||||
|
||||
setup_test_feed()
|
||||
|
||||
with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
|
||||
response = client.post(
|
||||
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.delenv("GIT_BACKUP_REMOTE", raising=False)
|
||||
|
||||
setup_test_feed()
|
||||
|
||||
with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
|
||||
response = client.post(url="/use_embed", data={"feed_url": test_feed_url})
|
||||
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.delenv("GIT_BACKUP_REMOTE", raising=False)
|
||||
|
||||
setup_test_feed()
|
||||
|
||||
with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
|
||||
response = client.post(url="/use_text", data={"feed_url": test_feed_url})
|
||||
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.delenv("GIT_BACKUP_REMOTE", raising=False)
|
||||
|
||||
setup_test_feed()
|
||||
|
||||
with patch("discord_rss_bot.main.commit_state_change") as mock_commit:
|
||||
response = client.post(
|
||||
url="/custom",
|
||||
|
|
@ -426,6 +461,8 @@ def test_post_custom_message_triggers_backup(monkeypatch: pytest.MonkeyPatch, tm
|
|||
assert test_feed_url in commit_message
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.slow
|
||||
@SKIP_IF_NO_GIT
|
||||
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."""
|
||||
|
|
@ -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.delenv("GIT_BACKUP_REMOTE", raising=False)
|
||||
|
||||
setup_test_feed()
|
||||
|
||||
# Post embed customization
|
||||
response = client.post(
|
||||
url="/embed",
|
||||
|
|
|
|||
69
tests/test_healthcheck.py
Normal file
69
tests/test_healthcheck.py
Normal 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
|
||||
|
|
@ -1,6 +1,21 @@
|
|||
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:
|
||||
|
|
@ -37,3 +52,197 @@ class TestExtractPostIdFromHoyolabUrl:
|
|||
|
||||
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)
|
||||
|
||||
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
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ def test_search() -> None:
|
|||
|
||||
# Delete the webhook if it already exists before we run the test.
|
||||
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.
|
||||
response: Response = client.post(
|
||||
|
|
@ -71,6 +72,7 @@ def test_add_webhook() -> None:
|
|||
"""Test the /add_webhook page."""
|
||||
# Delete the webhook if it already exists before we run the test.
|
||||
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.
|
||||
response: Response = client.post(
|
||||
|
|
@ -461,15 +463,16 @@ def test_delete_webhook() -> None:
|
|||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
|
||||
# Delete the webhook.
|
||||
response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
assert response.status_code == 200, f"Failed to delete webhook: {response.text}"
|
||||
response2: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
assert response2.status_code == 200, f"Failed to delete webhook: {response2.text}"
|
||||
|
||||
# Check that the webhook was added.
|
||||
response = client.get(url="/webhooks")
|
||||
assert response.status_code == 200, f"Failed to get /webhooks: {response.text}"
|
||||
assert webhook_name not in response.text, f"Webhook found in /webhooks: {response.text}"
|
||||
response3 = client.get(url="/webhooks")
|
||||
assert response3.status_code == 200, f"Failed to get /webhooks: {response3.text}"
|
||||
assert webhook_name not in response3.text, f"Webhook found in /webhooks: {response3.text}"
|
||||
|
||||
|
||||
def test_update_feed_not_found() -> None:
|
||||
|
|
|
|||
|
|
@ -53,7 +53,6 @@ def test_get_webhook_for_entry() -> None:
|
|||
|
||||
# Add a feed to the database.
|
||||
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.
|
||||
reader.set_tag("https://www.reddit.com/r/movies.rss", "webhook", "https://example.com") # pyright: ignore[reportArgumentType]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue