Add domain-wide blacklist and whitelist functionality
This commit is contained in:
parent
aa8a74ba67
commit
bdbd46ebd4
14 changed files with 930 additions and 305 deletions
|
|
@ -203,3 +203,33 @@ def test_regex_should_be_skipped() -> None:
|
|||
)
|
||||
reader.delete_tag(feed, "regex_blacklist_author")
|
||||
assert entry_should_be_skipped(reader, first_entry[0]) is False, f"Entry should not be skipped: {first_entry[0]}"
|
||||
|
||||
|
||||
def test_domain_blacklist_should_be_skipped() -> None:
|
||||
"""Domain-wide blacklist should apply to feeds on the same domain."""
|
||||
reader: Reader = get_reader()
|
||||
|
||||
reader.add_feed(feed_url)
|
||||
feed: Feed = reader.get_feed(feed_url)
|
||||
reader.update_feeds()
|
||||
|
||||
entries: Iterable[Entry] = reader.get_entries(feed=feed)
|
||||
first_entry: Entry | None = next(iter(entries), None)
|
||||
assert first_entry is not None, "Expected at least one entry"
|
||||
|
||||
assert feed_has_blacklist_tags(reader, feed) is False, "Feed should not have blacklist tags"
|
||||
assert entry_should_be_skipped(reader, first_entry) is False, "Entry should not be skipped"
|
||||
|
||||
reader.set_tag(
|
||||
(),
|
||||
"domain_blacklist",
|
||||
{
|
||||
"lovinator.space": {
|
||||
"blacklist_author": "TheLovinator",
|
||||
"regex_blacklist_title": r"fvnnn\\w+",
|
||||
},
|
||||
},
|
||||
) # pyright: ignore[reportArgumentType]
|
||||
|
||||
assert feed_has_blacklist_tags(reader, feed) is True, "Domain blacklist should count as blacklist tags"
|
||||
assert entry_should_be_skipped(reader, first_entry) is True, "Entry should be skipped by domain blacklist"
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ from __future__ import annotations
|
|||
import asyncio
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import UTC
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import LiteralString
|
||||
from unittest.mock import MagicMock
|
||||
|
|
@ -190,6 +192,67 @@ def test_get_entry_delivery_mode_falls_back_to_legacy_embed_flag() -> None:
|
|||
assert result == "text"
|
||||
|
||||
|
||||
@patch("discord_rss_bot.feeds.execute_webhook")
|
||||
@patch("discord_rss_bot.feeds.create_text_webhook")
|
||||
@patch("discord_rss_bot.feeds.should_be_sent", return_value=True)
|
||||
@patch("discord_rss_bot.feeds.has_white_tags", return_value=True)
|
||||
@patch("discord_rss_bot.feeds.entry_should_be_skipped", return_value=True)
|
||||
def test_send_to_discord_whitelist_precedence_over_blacklist(
|
||||
mock_entry_should_be_skipped: MagicMock,
|
||||
mock_has_white_tags: MagicMock,
|
||||
mock_should_be_sent: MagicMock,
|
||||
mock_create_text_webhook: MagicMock,
|
||||
mock_execute_webhook: MagicMock,
|
||||
) -> None:
|
||||
"""When whitelist is configured and matches, entry should still be sent even if blacklist matches."""
|
||||
reader = MagicMock()
|
||||
feed = MagicMock()
|
||||
feed.url = "https://example.com/feed.xml"
|
||||
|
||||
entry = MagicMock()
|
||||
entry.id = "entry-1"
|
||||
entry.feed = feed
|
||||
entry.feed_url = feed.url
|
||||
entry.added = datetime.now(tz=UTC)
|
||||
|
||||
reader.get_entries.return_value = [entry]
|
||||
|
||||
def get_tag_side_effect(
|
||||
resource: str | Feed,
|
||||
key: str,
|
||||
default: str | None = None,
|
||||
) -> str | None:
|
||||
"""Side effect function for reader.get_tag to return specific values based on the key.
|
||||
|
||||
Args:
|
||||
resource: The resource for which the tag is being requested (ignored in this case).
|
||||
key: The tag key being requested.
|
||||
default: The default value to return if the key is not found.
|
||||
|
||||
Returns:
|
||||
- "https://discord.test/webhook" for "webhook" key
|
||||
- "text" for "delivery_mode" key
|
||||
- default value for any other key
|
||||
"""
|
||||
if key == "webhook":
|
||||
return "https://discord.test/webhook"
|
||||
if key == "delivery_mode":
|
||||
return "text"
|
||||
return default
|
||||
|
||||
reader.get_tag.side_effect = get_tag_side_effect
|
||||
|
||||
webhook = MagicMock()
|
||||
mock_create_text_webhook.return_value = webhook
|
||||
|
||||
send_to_discord(reader=reader, feed=feed, do_once=True)
|
||||
|
||||
mock_has_white_tags.assert_called_once_with(reader, feed)
|
||||
mock_should_be_sent.assert_called_once_with(reader, entry)
|
||||
mock_entry_should_be_skipped.assert_not_called()
|
||||
mock_execute_webhook.assert_called_once_with(webhook, entry, reader=reader)
|
||||
|
||||
|
||||
@patch("discord_rss_bot.feeds.execute_webhook")
|
||||
@patch("discord_rss_bot.feeds.create_text_webhook")
|
||||
@patch("discord_rss_bot.feeds.create_hoyolab_webhook")
|
||||
|
|
|
|||
|
|
@ -173,9 +173,11 @@ def test_export_state_creates_state_json(tmp_path: Path) -> None:
|
|||
tag: str | None = None,
|
||||
default: str | None = None,
|
||||
) -> list[Any] | str | None:
|
||||
if feed_or_key == () and tag is None:
|
||||
# Called for global webhooks list
|
||||
return []
|
||||
if feed_or_key == () and tag == "domain_blacklist":
|
||||
return {"example.com": {"blacklist_title": "spoiler"}}
|
||||
|
||||
if feed_or_key == () and tag == "domain_whitelist":
|
||||
return {"example.com": {"whitelist_title": "release"}}
|
||||
|
||||
if tag == "webhook":
|
||||
return "https://discord.com/api/webhooks/123/abc"
|
||||
|
|
@ -194,6 +196,8 @@ def test_export_state_creates_state_json(tmp_path: Path) -> None:
|
|||
data: dict[str, Any] = json.loads(state_file.read_text(encoding="utf-8"))
|
||||
assert "feeds" in data
|
||||
assert "webhooks" in data
|
||||
assert data["domain_blacklist"]["example.com"]["blacklist_title"] == "spoiler"
|
||||
assert data["domain_whitelist"]["example.com"]["whitelist_title"] == "release"
|
||||
assert data["feeds"][0]["url"] == "https://example.com/feed.rss"
|
||||
assert data["feeds"][0]["webhook"] == "https://discord.com/api/webhooks/123/abc"
|
||||
|
||||
|
|
|
|||
|
|
@ -221,6 +221,264 @@ def test_get() -> None:
|
|||
assert response.status_code == 200, f"/whitelist failed: {response.text}"
|
||||
|
||||
|
||||
def test_post_blacklist_apply_to_domain_updates_global_domain_blacklist() -> None:
|
||||
"""Posting blacklist with apply_to_domain should save domain-wide blacklist values."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
|
||||
# Ensure webhook exists and feed can be created.
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
response = client.post(
|
||||
url="/blacklist",
|
||||
data={
|
||||
"feed_url": feed_url,
|
||||
"blacklist_author": "TheLovinator",
|
||||
"apply_to_domain": "true",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post blacklist: {response.text}"
|
||||
|
||||
domain_blacklist = reader.get_tag((), "domain_blacklist", {})
|
||||
assert isinstance(domain_blacklist, dict), "domain_blacklist should be a dict"
|
||||
assert "lovinator.space" in domain_blacklist, "Expected domain key in domain_blacklist"
|
||||
assert domain_blacklist["lovinator.space"]["blacklist_author"] == "TheLovinator"
|
||||
|
||||
|
||||
def test_post_whitelist_apply_to_domain_updates_global_domain_whitelist() -> None:
|
||||
"""Posting whitelist with apply_to_domain should save domain-wide whitelist values."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
response = client.post(
|
||||
url="/whitelist",
|
||||
data={
|
||||
"feed_url": feed_url,
|
||||
"whitelist_author": "TheLovinator",
|
||||
"apply_to_domain": "true",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post whitelist: {response.text}"
|
||||
|
||||
domain_whitelist = reader.get_tag((), "domain_whitelist", {})
|
||||
assert isinstance(domain_whitelist, dict), "domain_whitelist should be a dict"
|
||||
assert "lovinator.space" in domain_whitelist, "Expected domain key in domain_whitelist"
|
||||
assert domain_whitelist["lovinator.space"]["whitelist_author"] == "TheLovinator"
|
||||
|
||||
|
||||
def test_domain_filter_pages_show_domain_enabled_notice() -> None:
|
||||
"""Blacklist and whitelist pages should show domain-wide enabled notices when configured."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
reader.set_tag(
|
||||
(),
|
||||
"domain_blacklist",
|
||||
{"lovinator.space": {"blacklist_title": "spoiler"}},
|
||||
) # pyright: ignore[reportArgumentType]
|
||||
reader.set_tag(
|
||||
(),
|
||||
"domain_whitelist",
|
||||
{"lovinator.space": {"whitelist_title": "release"}},
|
||||
) # pyright: ignore[reportArgumentType]
|
||||
|
||||
response = client.get(url="/blacklist", params={"feed_url": encoded_feed_url(feed_url)})
|
||||
assert response.status_code == 200, f"/blacklist failed: {response.text}"
|
||||
assert "Domain-wide blacklist is enabled for lovinator.space." in response.text
|
||||
|
||||
response = client.get(url="/whitelist", params={"feed_url": encoded_feed_url(feed_url)})
|
||||
assert response.status_code == 200, f"/whitelist failed: {response.text}"
|
||||
assert "Domain-wide whitelist is enabled for lovinator.space." in response.text
|
||||
|
||||
|
||||
def test_domain_blacklist_isolation_between_domains() -> None:
|
||||
"""Applying domain blacklist should not overwrite other domains."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
reader.set_tag((), "domain_blacklist", {"example.com": {"blacklist_title": "existing"}}) # pyright: ignore[reportArgumentType]
|
||||
|
||||
response = client.post(
|
||||
url="/blacklist",
|
||||
data={
|
||||
"feed_url": feed_url,
|
||||
"blacklist_author": "TheLovinator",
|
||||
"apply_to_domain": "true",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post blacklist: {response.text}"
|
||||
|
||||
domain_blacklist = reader.get_tag((), "domain_blacklist", {})
|
||||
assert isinstance(domain_blacklist, dict)
|
||||
assert domain_blacklist["example.com"]["blacklist_title"] == "existing"
|
||||
assert domain_blacklist["lovinator.space"]["blacklist_author"] == "TheLovinator"
|
||||
|
||||
|
||||
def test_domain_whitelist_isolation_between_domains() -> None:
|
||||
"""Applying domain whitelist should not overwrite other domains."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
reader.set_tag((), "domain_whitelist", {"example.com": {"whitelist_title": "existing"}}) # pyright: ignore[reportArgumentType]
|
||||
|
||||
response = client.post(
|
||||
url="/whitelist",
|
||||
data={
|
||||
"feed_url": feed_url,
|
||||
"whitelist_author": "TheLovinator",
|
||||
"apply_to_domain": "true",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post whitelist: {response.text}"
|
||||
|
||||
domain_whitelist = reader.get_tag((), "domain_whitelist", {})
|
||||
assert isinstance(domain_whitelist, dict)
|
||||
assert domain_whitelist["example.com"]["whitelist_title"] == "existing"
|
||||
assert domain_whitelist["lovinator.space"]["whitelist_author"] == "TheLovinator"
|
||||
|
||||
|
||||
def test_domain_blacklist_removed_when_apply_to_domain_and_empty_values() -> None:
|
||||
"""Submitting empty domain blacklist values should remove existing domain entry."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
reader.set_tag((), "domain_blacklist", {"lovinator.space": {"blacklist_title": "existing"}}) # pyright: ignore[reportArgumentType]
|
||||
|
||||
response = client.post(
|
||||
url="/blacklist",
|
||||
data={"feed_url": feed_url, "apply_to_domain": "true"},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post blacklist: {response.text}"
|
||||
|
||||
domain_blacklist = reader.get_tag((), "domain_blacklist", {})
|
||||
assert isinstance(domain_blacklist, dict)
|
||||
assert "lovinator.space" not in domain_blacklist
|
||||
|
||||
|
||||
def test_domain_whitelist_removed_when_apply_to_domain_and_empty_values() -> None:
|
||||
"""Submitting empty domain whitelist values should remove existing domain entry."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
reader.set_tag((), "domain_whitelist", {"lovinator.space": {"whitelist_title": "existing"}}) # pyright: ignore[reportArgumentType]
|
||||
|
||||
response = client.post(
|
||||
url="/whitelist",
|
||||
data={"feed_url": feed_url, "apply_to_domain": "true"},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post whitelist: {response.text}"
|
||||
|
||||
domain_whitelist = reader.get_tag((), "domain_whitelist", {})
|
||||
assert isinstance(domain_whitelist, dict)
|
||||
assert "lovinator.space" not in domain_whitelist
|
||||
|
||||
|
||||
def test_apply_to_domain_missing_does_not_update_domain_tags() -> None:
|
||||
"""When apply_to_domain is omitted, domain tags should not change."""
|
||||
reader: Reader = get_reader_dependency()
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
reader.set_tag((), "domain_blacklist", {}) # pyright: ignore[reportArgumentType]
|
||||
reader.set_tag((), "domain_whitelist", {}) # pyright: ignore[reportArgumentType]
|
||||
|
||||
response = client.post(
|
||||
url="/blacklist",
|
||||
data={"feed_url": feed_url, "blacklist_author": "TheLovinator"},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post blacklist: {response.text}"
|
||||
|
||||
response = client.post(
|
||||
url="/whitelist",
|
||||
data={"feed_url": feed_url, "whitelist_author": "TheLovinator"},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to post whitelist: {response.text}"
|
||||
|
||||
assert reader.get_tag((), "domain_blacklist", {}) == {}
|
||||
assert reader.get_tag((), "domain_whitelist", {}) == {}
|
||||
|
||||
|
||||
def test_apply_to_domain_invalid_value_rejected() -> None:
|
||||
"""Invalid boolean value for apply_to_domain should return validation error."""
|
||||
response = client.post(
|
||||
url="/blacklist",
|
||||
data={
|
||||
"feed_url": feed_url,
|
||||
"blacklist_author": "TheLovinator",
|
||||
"apply_to_domain": "invalid-bool",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 422, f"Expected 422 for invalid boolean: {response.text}"
|
||||
|
||||
|
||||
def test_index_shows_domain_filter_shortcuts() -> None:
|
||||
"""Index should show domain whitelist/blacklist shortcut buttons."""
|
||||
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
|
||||
response: Response = client.post(
|
||||
url="/add_webhook",
|
||||
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
|
||||
|
||||
client.post(url="/remove", data={"feed_url": feed_url})
|
||||
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
|
||||
assert response.status_code == 200, f"Failed to add feed: {response.text}"
|
||||
|
||||
response = client.get(url="/")
|
||||
assert response.status_code == 200, f"Failed to get /: {response.text}"
|
||||
assert "Domain whitelist" in response.text
|
||||
assert "Domain blacklist" in response.text
|
||||
assert f"/whitelist?feed_url={encoded_feed_url(feed_url)}" in response.text
|
||||
assert f"/blacklist?feed_url={encoded_feed_url(feed_url)}" in response.text
|
||||
|
||||
|
||||
def test_settings_page_shows_screenshot_layout_setting() -> None:
|
||||
response: Response = client.get(url="/settings")
|
||||
assert response.status_code == 200, f"/settings failed: {response.text}"
|
||||
|
|
|
|||
|
|
@ -184,3 +184,33 @@ def test_regex_should_be_sent() -> None:
|
|||
assert should_be_sent(reader, first_entry[0]) is True, "Entry should be sent with newline-separated patterns"
|
||||
reader.delete_tag(feed, "regex_whitelist_author")
|
||||
assert should_be_sent(reader, first_entry[0]) is False, "Entry should not be sent"
|
||||
|
||||
|
||||
def test_domain_whitelist_should_be_sent() -> None:
|
||||
"""Domain-wide whitelist should apply to feeds on the same domain."""
|
||||
reader: Reader = get_reader()
|
||||
|
||||
reader.add_feed(feed_url)
|
||||
feed: Feed = reader.get_feed(feed_url)
|
||||
reader.update_feeds()
|
||||
|
||||
entries: Iterable[Entry] = reader.get_entries(feed=feed)
|
||||
first_entry: Entry | None = next(iter(entries), None)
|
||||
assert first_entry is not None, "Expected at least one entry"
|
||||
|
||||
assert has_white_tags(reader, feed) is False, "Feed should not have whitelist tags"
|
||||
assert should_be_sent(reader, first_entry) is False, "Entry should not be sent"
|
||||
|
||||
reader.set_tag(
|
||||
(),
|
||||
"domain_whitelist",
|
||||
{
|
||||
"lovinator.space": {
|
||||
"whitelist_author": "TheLovinator",
|
||||
"regex_whitelist_title": r"fvnnn\\w+",
|
||||
},
|
||||
},
|
||||
) # pyright: ignore[reportArgumentType]
|
||||
|
||||
assert has_white_tags(reader, feed) is True, "Domain whitelist should count as whitelist tags"
|
||||
assert should_be_sent(reader, first_entry) is True, "Entry should be sent by domain whitelist"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue