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

Only show screenshot things if Playwright has chromium installed; use names instead of ids when ignoring Ruff errors
All checks were successful
Test and build Docker image / docker (push) Successful in 1m54s
Details

Browse source
This commit is contained in:
Joakim Hellsén 2026-07-18 07:04:00 +02:00
commit d61e8ccf10
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
15 changed files with 224 additions and 174 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.pre-commit-config.yaml
View file

@ -38,7 +38,7 @@ 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.16 rev: v0.15.22
hooks: hooks:
- id: ruff-format - id: ruff-format
types_or: [ python, pyi, jupyter, pyproject ] types_or: [ python, pyi, jupyter, pyproject ]

72
discord_rss_bot/feeds.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio import asyncio
import concurrent.futures import concurrent.futures
import datetime import datetime
import functools
import hashlib import hashlib
import json import json
import logging import logging
@ -30,6 +31,7 @@ from httpx2 import HTTPError
from httpx2 import Response from httpx2 import Response
from markdownify import markdownify from markdownify import markdownify
from playwright.sync_api import Browser from playwright.sync_api import Browser
from playwright.sync_api import Error as PlaywrightError
from playwright.sync_api import Page from playwright.sync_api import Page
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
from playwright.sync_api import sync_playwright from playwright.sync_api import sync_playwright
@ -140,7 +142,7 @@ MESSAGE_PAYLOAD_KEYS: tuple[str, ...] = (
) )
def extract_domain(url: str) -> str: # noqa: PLR0911 def extract_domain(url: str) -> str: # ruff:ignore[too-many-return-statements]
"""Extract the domain name from a URL. """Extract the domain name from a URL.
Args: Args:
@ -153,7 +155,7 @@ def extract_domain(url: str) -> str: # noqa: PLR0911
if not url: if not url:
return "Other" return "Other"
try: # noqa: PLW0717 try: # ruff:ignore[too-many-statements-in-try-clause]
# Special handling for YouTube feeds # Special handling for YouTube feeds
if "youtube.com/feeds/videos.xml" in url: if "youtube.com/feeds/videos.xml" in url:
return "YouTube" return "YouTube"
@ -258,7 +260,7 @@ def extract_steam_app_id_from_url(url: str) -> str | None:
normalized_netloc: str = parsed_url.netloc.lower().removeprefix("www.") normalized_netloc: str = parsed_url.netloc.lower().removeprefix("www.")
path_segments: list[str] = [segment for segment in parsed_url.path.split("/") if segment] path_segments: list[str] = [segment for segment in parsed_url.path.split("/") if segment]
return _extract_steam_app_id_from_path(normalized_netloc, path_segments) or _extract_steam_app_id_from_query( return _extract_steam_app_id_from_path(normalized_netloc, path_segments) or _extract_steam_app_id_from_query(
parsed_url parsed_url,
) )
@ -453,7 +455,7 @@ def get_feed_webhook_text_length_limit(reader: Reader, feed: Feed | str) -> int:
return coerce_webhook_text_length_limit(value) return coerce_webhook_text_length_limit(value)
def coerce_media_gallery_image_limit(value: JsonValue) -> int: # noqa: PLR0911 def coerce_media_gallery_image_limit(value: JsonValue) -> int: # ruff:ignore[too-many-return-statements]
"""Return the supported media gallery image limit for a stored tag value.""" """Return the supported media gallery image limit for a stored tag value."""
if isinstance(value, bool): if isinstance(value, bool):
return 1 return 1
@ -835,7 +837,7 @@ def get_webhook_query_params(
return clean_webhook_url, params return clean_webhook_url, params
def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # noqa: C901 def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # ruff:ignore[complex-structure]
"""Return files attached to a webhook object in a normalized shape.""" """Return files attached to a webhook object in a normalized shape."""
raw_files = getattr(webhook, "files", None) raw_files = getattr(webhook, "files", None)
files: list[WebhookFile] = [] files: list[WebhookFile] = []
@ -854,7 +856,7 @@ def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # noqa: C9
files.append(file_value) files.append(file_value)
continue continue
if not isinstance(file_value, tuple) or len(file_value) < 2: # noqa: PLR2004 if not isinstance(file_value, tuple) or len(file_value) < 2: # ruff:ignore[magic-value-comparison]
continue continue
first, second = file_value[0], file_value[1] first, second = file_value[0], file_value[1]
@ -862,7 +864,7 @@ def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # noqa: C9
files.append(WebhookFile(filename=first, content=second)) files.append(WebhookFile(filename=first, content=second))
continue continue
if isinstance(second, tuple) and len(second) >= 2: # noqa: PLR2004 if isinstance(second, tuple) and len(second) >= 2: # ruff:ignore[magic-value-comparison]
nested_file = cast("tuple[object, ...]", second) nested_file = cast("tuple[object, ...]", second)
nested_filename, nested_content = nested_file[0], nested_file[1] nested_filename, nested_content = nested_file[0], nested_file[1]
if isinstance(nested_filename, str) and isinstance(nested_content, bytes): if isinstance(nested_filename, str) and isinstance(nested_content, bytes):
@ -916,7 +918,7 @@ def request_discord_webhook(
request_kwargs["json"] = payload request_kwargs["json"] = payload
response: Response = httpx2.request(method, url, **request_kwargs) response: Response = httpx2.request(method, url, **request_kwargs)
if not rate_limit_retry or response.status_code != 429: # noqa: PLR2004 if not rate_limit_retry or response.status_code != 429: # ruff:ignore[magic-value-comparison]
return response return response
retry_after: float | None = get_retry_after_seconds(response) retry_after: float | None = get_retry_after_seconds(response)
@ -1183,7 +1185,7 @@ def update_sent_webhook_record_for_entry(
) )
def update_sent_webhooks_for_modified_entries( # noqa: C901 def update_sent_webhooks_for_modified_entries( # ruff:ignore[complex-structure]
reader: Reader, reader: Reader,
modified_entries: Iterable[tuple[str, str]], modified_entries: Iterable[tuple[str, str]],
) -> int: ) -> int:
@ -1253,7 +1255,7 @@ def create_text_webhook(
""" """
webhook_message: str = "" webhook_message: str = ""
if get_custom_message(reader, entry.feed) != "": # noqa: PLC1901 if get_custom_message(reader, entry.feed) != "": # ruff:ignore[compare-to-empty-string]
webhook_message = replace_tags_in_text_message(entry=entry, reader=reader) webhook_message = replace_tags_in_text_message(entry=entry, reader=reader)
if not webhook_message and use_default_message_on_empty: if not webhook_message and use_default_message_on_empty:
@ -1370,6 +1372,40 @@ def screenshot_filename_for_entry(entry: Entry, *, extension: str = "png") -> st
return f"{safe_name[:80]}.{safe_extension}" return f"{safe_name[:80]}.{safe_extension}"
@functools.lru_cache(maxsize=1)
def is_chromium_installed() -> bool:
"""Check if Playwright's Chromium browser is installed.
Uses a cached check so the browser is only probed once per process lifetime.
Offloads to a thread when called from an active event loop (e.g. FastAPI).
Returns:
bool: True if Chromium launched successfully, False otherwise.
"""
def _check() -> bool:
try:
with sync_playwright() as playwright:
browser = playwright.chromium.launch(
headless=True,
args=["--disable-dev-shm-usage", "--no-sandbox"],
)
browser.close()
except (OSError, PlaywrightError):
return False
else:
return True
try:
asyncio.get_running_loop()
except RuntimeError:
return _check()
else:
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future: concurrent.futures.Future[bool] = executor.submit(_check)
return future.result()
def capture_full_page_screenshot( def capture_full_page_screenshot(
url: str, url: str,
*, *,
@ -1417,7 +1453,7 @@ def _capture_full_page_screenshot_sync(
Returns: Returns:
bytes | None: PNG bytes on success, otherwise None. bytes | None: PNG bytes on success, otherwise None.
""" """
try: # noqa: PLW0717 try: # ruff:ignore[too-many-statements-in-try-clause]
with sync_playwright() as playwright: with sync_playwright() as playwright:
browser: Browser = playwright.chromium.launch( browser: Browser = playwright.chromium.launch(
headless=True, headless=True,
@ -1537,7 +1573,7 @@ def set_description(
# Discord allows 2048, but we keep a small safety margin by default. # Discord allows 2048, but we keep a small safety margin by default.
embed_description: str = custom_embed.description embed_description: str = custom_embed.description
if len(embed_description) > max_description_length: if len(embed_description) > max_description_length:
if max_description_length <= 3: # noqa: PLR2004 if max_description_length <= 3: # ruff:ignore[magic-value-comparison]
embed_description = embed_description[:max_description_length] embed_description = embed_description[:max_description_length]
else: else:
embed_description = f"{embed_description[: max_description_length - 3]}..." embed_description = f"{embed_description[: max_description_length - 3]}..."
@ -1635,7 +1671,7 @@ def get_ttvdrops_reward_description(drop: JsonObject, reward: JsonObject) -> str
return reward_name return reward_name
def extract_ttvdrops_media_gallery_items(value: JsonValue, *, hide_paid: bool = False) -> list[JsonObject]: # noqa: C901 def extract_ttvdrops_media_gallery_items(value: JsonValue, *, hide_paid: bool = False) -> list[JsonObject]: # ruff:ignore[complex-structure]
"""Extract benefit/reward media gallery items from a ttvdrops API response. """Extract benefit/reward media gallery items from a ttvdrops API response.
Returns: Returns:
@ -1688,7 +1724,7 @@ def fetch_ttvdrops_campaign_media_items(entry: Entry) -> list[JsonObject]:
try: try:
response: Response = httpx2.get(api_url, follow_redirects=True, timeout=10.0) response: Response = httpx2.get(api_url, follow_redirects=True, timeout=10.0)
if response.status_code != 200: # noqa: PLR2004 if response.status_code != 200: # ruff:ignore[magic-value-comparison]
logger.warning("Failed to fetch ttvdrops campaign data from %s: %s", api_url, response.text[:500]) logger.warning("Failed to fetch ttvdrops campaign data from %s: %s", api_url, response.text[:500])
return [] return []
@ -1743,7 +1779,7 @@ def truncate_component_text(
""" """
if len(content) <= max_text_display_length: if len(content) <= max_text_display_length:
return content return content
if max_text_display_length <= 3: # noqa: PLR2004 if max_text_display_length <= 3: # ruff:ignore[magic-value-comparison]
return content[:max_text_display_length] return content[:max_text_display_length]
return f"{content[: max_text_display_length - 3]}..." return f"{content[: max_text_display_length - 3]}..."
@ -1837,7 +1873,7 @@ def create_components_v2_webhook(
) )
def create_embed_webhook( # noqa: C901, PLR0912 def create_embed_webhook( # ruff:ignore[complex-structure, too-many-branches]
webhook_url: str, webhook_url: str,
entry: Entry, entry: Entry,
reader: Reader, reader: Reader,
@ -2121,7 +2157,7 @@ def truncate_webhook_message(
""" """
if len(webhook_message) <= max_content_length: if len(webhook_message) <= max_content_length:
return webhook_message return webhook_message
if max_content_length <= 3: # noqa: PLR2004 if max_content_length <= 3: # ruff:ignore[magic-value-comparison]
return webhook_message[:max_content_length] return webhook_message[:max_content_length]
head_length = (max_content_length - 3) // 2 head_length = (max_content_length - 3) // 2
@ -2145,7 +2181,7 @@ def remove_invalid_new_feed(reader: Reader, feed_url: str) -> None:
logger.exception("Failed to remove invalid feed after initial update: %s", feed_url) logger.exception("Failed to remove invalid feed after initial update: %s", feed_url)
def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: # noqa: C901, PLR0912 def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: # ruff:ignore[complex-structure, too-many-branches]
"""Add a new feed, update it and mark every entry as read. """Add a new feed, update it and mark every entry as read.
Args: Args:

28
discord_rss_bot/git_backup.py
View file

@ -25,7 +25,7 @@ import json
import logging import logging
import os import os
import shutil import shutil
import subprocess # noqa: S404 import subprocess # ruff:ignore[suspicious-subprocess-import]
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@ -102,22 +102,22 @@ def setup_backup_repo(backup_path: Path) -> bool:
Returns: Returns:
``True`` if the repository is ready, ``False`` on any error. ``True`` if the repository is ready, ``False`` on any error.
""" """
try: # noqa: PLW0717 try: # ruff:ignore[too-many-statements-in-try-clause]
backup_path.mkdir(parents=True, exist_ok=True) backup_path.mkdir(parents=True, exist_ok=True)
git_dir: Path = backup_path / ".git" git_dir: Path = backup_path / ".git"
if not git_dir.exists(): if not git_dir.exists():
subprocess.run([GIT_EXECUTABLE, "init", str(backup_path)], check=True, capture_output=True) # noqa: S603 subprocess.run([GIT_EXECUTABLE, "init", str(backup_path)], check=True, capture_output=True) # ruff:ignore[subprocess-without-shell-equals-true]
logger.info("Initialized git backup repository at %s", backup_path) logger.info("Initialized git backup repository at %s", backup_path)
# Ensure a local identity exists so that `git commit` always works. # Ensure a local identity exists so that `git commit` always works.
for key, value in (("user.email", "discord-rss-bot@localhost"), ("user.name", "discord-rss-bot")): for key, value in (("user.email", "discord-rss-bot@localhost"), ("user.name", "discord-rss-bot")):
result: subprocess.CompletedProcess[bytes] = subprocess.run( # noqa: S603 result: subprocess.CompletedProcess[bytes] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "config", "--local", key], [GIT_EXECUTABLE, "-C", str(backup_path), "config", "--local", key],
check=False, check=False,
capture_output=True, capture_output=True,
) )
if result.returncode != 0: if result.returncode != 0:
subprocess.run( # noqa: S603 subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "config", "--local", key, value], [GIT_EXECUTABLE, "-C", str(backup_path), "config", "--local", key, value],
check=True, check=True,
capture_output=True, capture_output=True,
@ -127,14 +127,14 @@ def setup_backup_repo(backup_path: Path) -> bool:
remote_url: str = get_backup_remote() remote_url: str = get_backup_remote()
if remote_url: if remote_url:
# Check if remote "origin" already exists. # Check if remote "origin" already exists.
check_remote: subprocess.CompletedProcess[bytes] = subprocess.run( # noqa: S603 check_remote: subprocess.CompletedProcess[bytes] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "remote", "get-url", "origin"], [GIT_EXECUTABLE, "-C", str(backup_path), "remote", "get-url", "origin"],
check=False, check=False,
capture_output=True, capture_output=True,
) )
if check_remote.returncode != 0: if check_remote.returncode != 0:
# Remote doesn't exist, add it. # Remote doesn't exist, add it.
subprocess.run( # noqa: S603 subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "remote", "add", "origin", remote_url], [GIT_EXECUTABLE, "-C", str(backup_path), "remote", "add", "origin", remote_url],
check=True, check=True,
capture_output=True, capture_output=True,
@ -144,7 +144,7 @@ def setup_backup_repo(backup_path: Path) -> bool:
# Remote exists, update it if the URL has changed. # Remote exists, update it if the URL has changed.
current_url: str = check_remote.stdout.decode().strip() current_url: str = check_remote.stdout.decode().strip()
if current_url != remote_url: if current_url != remote_url:
subprocess.run( # noqa: S603 subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "remote", "set-url", "origin", remote_url], [GIT_EXECUTABLE, "-C", str(backup_path), "remote", "set-url", "origin", remote_url],
check=True, check=True,
capture_output=True, capture_output=True,
@ -169,7 +169,7 @@ def export_state(reader: Reader, backup_path: Path) -> None:
for tag in _FEED_TAGS: for tag in _FEED_TAGS:
try: try:
value: TagValue = reader.get_tag(feed, tag, None) value: TagValue = reader.get_tag(feed, tag, None)
if value is not None and value != "": # noqa: PLC1901 if value is not None and value != "": # ruff:ignore[compare-to-empty-string]
feed_data[tag] = value feed_data[tag] = value
except Exception: except Exception:
logger.exception("Failed to read tag '%s' for feed '%s' during state export", tag, feed.url) logger.exception("Failed to read tag '%s' for feed '%s' during state export", tag, feed.url)
@ -219,13 +219,13 @@ def commit_state_change(reader: Reader, message: str) -> None:
if not setup_backup_repo(backup_path): if not setup_backup_repo(backup_path):
return return
try: # noqa: PLW0717 try: # ruff:ignore[too-many-statements-in-try-clause]
export_state(reader, backup_path) export_state(reader, backup_path)
subprocess.run([GIT_EXECUTABLE, "-C", str(backup_path), "add", "-A"], check=True, capture_output=True) # noqa: S603 subprocess.run([GIT_EXECUTABLE, "-C", str(backup_path), "add", "-A"], check=True, capture_output=True) # ruff:ignore[subprocess-without-shell-equals-true]
# Only create a commit if there are staged changes. # Only create a commit if there are staged changes.
diff_result: subprocess.CompletedProcess[bytes] = subprocess.run( # noqa: S603 diff_result: subprocess.CompletedProcess[bytes] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "diff", "--cached", "--exit-code"], [GIT_EXECUTABLE, "-C", str(backup_path), "diff", "--cached", "--exit-code"],
check=False, check=False,
capture_output=True, capture_output=True,
@ -234,7 +234,7 @@ def commit_state_change(reader: Reader, message: str) -> None:
logger.debug("No state changes to commit for: %s", message) logger.debug("No state changes to commit for: %s", message)
return return
subprocess.run( # noqa: S603 subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "commit", "-m", message], [GIT_EXECUTABLE, "-C", str(backup_path), "commit", "-m", message],
check=True, check=True,
capture_output=True, capture_output=True,
@ -243,7 +243,7 @@ def commit_state_change(reader: Reader, message: str) -> None:
# Push to remote if configured. # Push to remote if configured.
if get_backup_remote(): if get_backup_remote():
subprocess.run( # noqa: S603 subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "push", "origin", "HEAD"], [GIT_EXECUTABLE, "-C", str(backup_path), "push", "origin", "HEAD"],
check=True, check=True,
capture_output=True, capture_output=True,

2
discord_rss_bot/healthcheck.py
View file

@ -18,7 +18,7 @@ def healthcheck() -> None:
sys.exit(0) sys.exit(0)
sys.exit(1) 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) # ruff:ignore[print]
sys.exit(1) sys.exit(1)

8
discord_rss_bot/hoyolab_api.py
View file

@ -76,7 +76,7 @@ def fetch_hoyolab_post(post_id: str) -> JsonObject | None:
return None return None
http_ok = 200 http_ok = 200
try: # noqa: PLW0717 try: # ruff:ignore[too-many-statements-in-try-clause]
url: str = f"https://bbs-api-os.hoyolab.com/community/post/wapi/getPostFull?post_id={post_id}" url: str = f"https://bbs-api-os.hoyolab.com/community/post/wapi/getPostFull?post_id={post_id}"
response: requests.Response = requests.get(url, timeout=10) response: requests.Response = requests.get(url, timeout=10)
@ -94,7 +94,7 @@ def fetch_hoyolab_post(post_id: str) -> JsonObject | None:
return None return None
def create_hoyolab_webhook(webhook_url: str, entry: Entry, post_data: JsonObject) -> DiscordWebhook: # noqa: C901, PLR0912, PLR0914, PLR0915 def create_hoyolab_webhook(webhook_url: str, entry: Entry, post_data: JsonObject) -> DiscordWebhook: # ruff:ignore[complex-structure, too-many-branches, too-many-locals, too-many-statements]
"""Create a webhook with data from the Hoyolab API. """Create a webhook with data from the Hoyolab API.
Args: Args:
@ -184,8 +184,8 @@ def create_hoyolab_webhook(webhook_url: str, entry: Entry, post_data: JsonObject
# Only show Youtube URL if available # Only show Youtube URL if available
structured_content: str = str(post.get("structured_content", "")) structured_content: str = str(post.get("structured_content", ""))
if structured_content: # noqa: PLR1702 if structured_content: # ruff:ignore[too-many-nested-blocks]
try: # noqa: PLW0717 try: # ruff:ignore[too-many-statements-in-try-clause]
loaded_structured_content = cast("JsonValue", json.loads(structured_content)) loaded_structured_content = cast("JsonValue", json.loads(structured_content))
structured_content_data: list[JsonObject] = ( structured_content_data: list[JsonObject] = (
[cast("JsonObject", item) for item in loaded_structured_content if isinstance(item, dict)] [cast("JsonObject", item) for item in loaded_structured_content if isinstance(item, dict)]

43
discord_rss_bot/main.py
View file

@ -65,6 +65,7 @@ from discord_rss_bot.feeds import get_feed_media_gallery_image_limit
from discord_rss_bot.feeds import get_feed_webhook_text_length_limit from discord_rss_bot.feeds import get_feed_webhook_text_length_limit
from discord_rss_bot.feeds import get_screenshot_layout from discord_rss_bot.feeds import get_screenshot_layout
from discord_rss_bot.feeds import get_sent_webhook_records from discord_rss_bot.feeds import get_sent_webhook_records
from discord_rss_bot.feeds import is_chromium_installed
from discord_rss_bot.feeds import is_steam_feed_url from discord_rss_bot.feeds import is_steam_feed_url
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
@ -138,7 +139,7 @@ LOGGING_CONFIG = {
"disable_existing_loggers": False, "disable_existing_loggers": False,
"formatters": { "formatters": {
"standard": { "standard": {
"format": "%(asctime)s [%(processName)s: %(process)d] [%(threadName)s: %(thread)d] [%(levelname)s] %(name)s: %(message)s", # noqa: E501 "format": "%(asctime)s [%(processName)s: %(process)d] [%(threadName)s: %(thread)d] [%(levelname)s] %(name)s: %(message)s", # ruff:ignore[line-too-long]
}, },
}, },
"handlers": { "handlers": {
@ -1259,7 +1260,7 @@ async def get_embed_page(
@app.post("/embed", response_class=HTMLResponse) @app.post("/embed", response_class=HTMLResponse)
async def post_embed( # noqa: C901 async def post_embed( # ruff:ignore[complex-structure]
feed_url: Annotated[str, Form()], feed_url: Annotated[str, Form()],
reader: Annotated[Reader, Depends(get_reader_dependency)], reader: Annotated[Reader, Depends(get_reader_dependency)],
title: Annotated[str, Form()] = "", title: Annotated[str, Form()] = "",
@ -1272,6 +1273,7 @@ async def post_embed( # noqa: C901
author_icon_url: Annotated[str, Form()] = "", author_icon_url: Annotated[str, Form()] = "",
footer_text: Annotated[str, Form()] = "", footer_text: Annotated[str, Form()] = "",
footer_icon_url: Annotated[str, Form()] = "", footer_icon_url: Annotated[str, Form()] = "",
*,
show_steam_game_icon_in_thumbnail: Annotated[bool, Form()] = False, show_steam_game_icon_in_thumbnail: Annotated[bool, Form()] = False,
) -> RedirectResponse: ) -> RedirectResponse:
"""Set the embed settings. """Set the embed settings.
@ -1778,7 +1780,7 @@ def get_add(
@app.get("/feed", response_class=HTMLResponse) @app.get("/feed", response_class=HTMLResponse)
async def get_feed( # noqa: C901, PLR0912, PLR0914, PLR0915 async def get_feed( # ruff:ignore[complex-structure, too-many-branches, too-many-locals, too-many-statements]
feed_url: str, feed_url: str,
request: Request, request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)], reader: Annotated[Reader, Depends(get_reader_dependency)],
@ -1874,6 +1876,7 @@ async def get_feed( # noqa: C901, PLR0912, PLR0914, PLR0915
"max_webhook_text_length_limit": 4000, "max_webhook_text_length_limit": 4000,
"save_sent_webhooks": feed_saves_sent_webhooks(reader, feed), "save_sent_webhooks": feed_saves_sent_webhooks(reader, feed),
"is_steam_feed": is_steam_feed_url(feed.url), "is_steam_feed": is_steam_feed_url(feed.url),
"chromium_installed": is_chromium_installed(),
} }
return templates.TemplateResponse(request=request, name="feed.html", context=context) return templates.TemplateResponse(request=request, name="feed.html", context=context)
@ -1941,11 +1944,12 @@ async def get_feed( # noqa: C901, PLR0912, PLR0914, PLR0915
"max_webhook_text_length_limit": 4000, "max_webhook_text_length_limit": 4000,
"save_sent_webhooks": feed_saves_sent_webhooks(reader, feed), "save_sent_webhooks": feed_saves_sent_webhooks(reader, feed),
"is_steam_feed": is_steam_feed_url(feed.url), "is_steam_feed": is_steam_feed_url(feed.url),
"chromium_installed": is_chromium_installed(),
} }
return templates.TemplateResponse(request=request, name="feed.html", context=context) return templates.TemplateResponse(request=request, name="feed.html", context=context)
def create_html_for_feed( # noqa: C901, PLR0914 def create_html_for_feed( # ruff:ignore[complex-structure, too-many-locals]
reader: Reader, reader: Reader,
entries: Iterable[Entry], entries: Iterable[Entry],
current_feed_url: str = "", current_feed_url: str = "",
@ -2051,7 +2055,7 @@ def create_html_for_feed( # noqa: C901, PLR0914
{video_embed_html} {video_embed_html}
{image_html} {image_html}
</div> </div>
""" # noqa: E501 """ # ruff:ignore[line-too-long]
return html.strip() return html.strip()
@ -2152,7 +2156,7 @@ async def get_settings(
global_delivery_mode = "embed" global_delivery_mode = "embed"
global_webhook_text_length_limit: int = coerce_webhook_text_length_limit( global_webhook_text_length_limit: int = coerce_webhook_text_length_limit(
reader.get_tag((), "webhook_text_length_limit", 4000) reader.get_tag((), "webhook_text_length_limit", 4000),
) )
# Get all feeds with their intervals # Get all feeds with their intervals
@ -2181,6 +2185,7 @@ async def get_settings(
"global_webhook_text_length_limit": global_webhook_text_length_limit, "global_webhook_text_length_limit": global_webhook_text_length_limit,
"max_webhook_text_length_limit": 4000, "max_webhook_text_length_limit": 4000,
"feed_intervals": feed_intervals, "feed_intervals": feed_intervals,
"chromium_installed": is_chromium_installed(),
} }
return templates.TemplateResponse(request=request, name="settings.html", context=context) return templates.TemplateResponse(request=request, name="settings.html", context=context)
@ -2603,8 +2608,8 @@ def create_webhook_feed_url_preview(
webhook_feeds: list[Feed], webhook_feeds: list[Feed],
replace_from: str, replace_from: str,
replace_to: str, replace_to: str,
resolve_urls: bool, # noqa: FBT001 resolve_urls: bool, # ruff:ignore[boolean-type-hint-positional-argument]
force_update: bool = False, # noqa: FBT001, FBT002 force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
existing_feed_urls: set[str] | None = None, existing_feed_urls: set[str] | None = None,
) -> list[dict[str, str | bool | None]]: ) -> list[dict[str, str | bool | None]]:
"""Create preview rows for bulk feed URL replacement. """Create preview rows for bulk feed URL replacement.
@ -2670,8 +2675,8 @@ def build_webhook_mass_update_context(
all_feeds: list[Feed], all_feeds: list[Feed],
replace_from: str, replace_from: str,
replace_to: str, replace_to: str,
resolve_urls: bool, # noqa: FBT001 resolve_urls: bool, # ruff:ignore[boolean-type-hint-positional-argument]
force_update: bool = False, # noqa: FBT001, FBT002 force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
) -> dict[str, str | bool | int | list[dict[str, str | bool | None]] | dict[str, int]]: ) -> dict[str, str | bool | int | list[dict[str, str | bool | None]] | dict[str, int]]:
"""Build context data used by the webhook mass URL update preview UI. """Build context data used by the webhook mass URL update preview UI.
@ -2732,8 +2737,8 @@ async def get_webhook_entries_mass_update_preview(
reader: Annotated[Reader, Depends(get_reader_dependency)], reader: Annotated[Reader, Depends(get_reader_dependency)],
replace_from: str = "", replace_from: str = "",
replace_to: str = "", replace_to: str = "",
resolve_urls: bool = True, # noqa: FBT001, FBT002 resolve_urls: bool = True, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
force_update: bool = False, # noqa: FBT001, FBT002 force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
) -> HTMLResponse: ) -> HTMLResponse:
"""Render the mass-update preview fragment for a webhook using HTMX. """Render the mass-update preview fragment for a webhook using HTMX.
@ -2771,15 +2776,15 @@ async def get_webhook_entries_mass_update_preview(
@app.get("/webhook_entries", response_class=HTMLResponse) @app.get("/webhook_entries", response_class=HTMLResponse)
async def get_webhook_entries( # noqa: C901, PLR0914 async def get_webhook_entries( # ruff:ignore[complex-structure, too-many-locals]
webhook_url: str, webhook_url: str,
request: Request, request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)], reader: Annotated[Reader, Depends(get_reader_dependency)],
starting_after: str = "", starting_after: str = "",
replace_from: str = "", replace_from: str = "",
replace_to: str = "", replace_to: str = "",
resolve_urls: bool = True, # noqa: FBT001, FBT002 resolve_urls: bool = True, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
force_update: bool = False, # noqa: FBT001, FBT002 force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
message: str = "", message: str = "",
) -> HTMLResponse: ) -> HTMLResponse:
"""Get all latest entries from all feeds for a specific webhook. """Get all latest entries from all feeds for a specific webhook.
@ -2901,13 +2906,13 @@ async def get_webhook_entries( # noqa: C901, PLR0914
@app.post("/bulk_change_feed_urls", response_class=HTMLResponse) @app.post("/bulk_change_feed_urls", response_class=HTMLResponse)
async def post_bulk_change_feed_urls( # noqa: C901, PLR0914, PLR0912, PLR0915 async def post_bulk_change_feed_urls( # ruff:ignore[complex-structure, too-many-locals, too-many-branches, too-many-statements]
webhook_url: Annotated[str, Form()], webhook_url: Annotated[str, Form()],
replace_from: Annotated[str, Form()], replace_from: Annotated[str, Form()],
reader: Annotated[Reader, Depends(get_reader_dependency)], reader: Annotated[Reader, Depends(get_reader_dependency)],
replace_to: Annotated[str, Form()] = "", replace_to: Annotated[str, Form()] = "",
resolve_urls: Annotated[bool, Form()] = True, # noqa: FBT002 resolve_urls: Annotated[bool, Form()] = True, # ruff:ignore[boolean-default-value-positional-argument]
force_update: Annotated[bool, Form()] = False, # noqa: FBT002 force_update: Annotated[bool, Form()] = False, # ruff:ignore[boolean-default-value-positional-argument]
) -> RedirectResponse: ) -> RedirectResponse:
"""Bulk-change feed URLs attached to a webhook. """Bulk-change feed URLs attached to a webhook.
@ -3044,7 +3049,7 @@ if __name__ == "__main__":
uvicorn.run( uvicorn.run(
"main:app", "main:app",
log_level="debug", log_level="debug",
host="0.0.0.0", # noqa: S104 host="0.0.0.0", # ruff:ignore[hardcoded-bind-all-interfaces]
port=3000, port=3000,
proxy_headers=True, proxy_headers=True,
forwarded_allow_ips="*", forwarded_allow_ips="*",

52
discord_rss_bot/templates/feed.html
View file

@ -103,7 +103,7 @@
{% elif delivery_mode == "screenshot" %} {% elif delivery_mode == "screenshot" %}
<li>Screenshot layout: {{ screenshot_layout }}.</li> <li>Screenshot layout: {{ screenshot_layout }}.</li>
{% endif %} {% endif %}
<li>Webhook text</li> is truncated to {{ webhook_text_length_limit }} characters.</li> <li>Webhook text is truncated to {{ webhook_text_length_limit }} characters.</li>
<li> <li>
Filters: Filters:
{% if has_blacklist_filters and has_whitelist_filters %} {% if has_blacklist_filters and has_whitelist_filters %}
@ -173,10 +173,13 @@
<section class="mb-3"> <section class="mb-3">
<h4 class="h6 text-muted mb-2">Delivery Mode</h4> <h4 class="h6 text-muted mb-2">Delivery Mode</h4>
<p class="text-muted small mb-2"> <p class="text-muted small mb-2">
Choose between:<br> Choose between:
Embed: a Discord embed with title, description, and images.<br> <br>
Embed: a Discord embed with title, description, and images.
<br>
{% if not "youtube.com/feeds/videos.xml" in feed.url %} {% if not "youtube.com/feeds/videos.xml" in feed.url %}
Screenshot: full-page screenshot of the entry link.<br> Screenshot: full-page screenshot of the entry link.
<br>
{% endif %} {% endif %}
Text: plain message. Text: plain message.
</p> </p>
@ -184,15 +187,18 @@
{% if delivery_mode != "embed" %} {% if delivery_mode != "embed" %}
<form action="/use_embed" method="post" class="d-inline"> <form action="/use_embed" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm" <button class="btn btn-outline-light btn-sm"
style="border-top-right-radius: 0; border-bottom-right-radius: 0;" style="border-top-right-radius: 0;
border-bottom-right-radius: 0"
name="feed_url" name="feed_url"
value="{{ feed.url }}">Embed</button> value="{{ feed.url }}">Embed</button>
</form> </form>
{% else %} {% else %}
<span class="btn btn-primary btn-sm disabled" <span class="btn btn-primary btn-sm disabled"
style="border-top-right-radius: 0; border-bottom-right-radius: 0;">Embed</span> style="border-top-right-radius: 0;
border-bottom-right-radius: 0">Embed</span>
{% endif %} {% endif %}
{% if not "youtube.com/feeds/videos.xml" in feed.url %} {% if not "youtube.com/feeds/videos.xml" in feed.url %}
{% if chromium_installed %}
{% if delivery_mode != "screenshot" %} {% if delivery_mode != "screenshot" %}
<form action="/use_screenshot" method="post" class="d-inline"> <form action="/use_screenshot" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm rounded-0" <button class="btn btn-outline-light btn-sm rounded-0"
@ -202,21 +208,35 @@
{% else %} {% else %}
<span class="btn btn-primary btn-sm disabled rounded-0">Screenshot</span> <span class="btn btn-primary btn-sm disabled rounded-0">Screenshot</span>
{% endif %} {% endif %}
{% else %}
<span class="btn btn-outline-light btn-sm rounded-0"
title="Playwright Chromium is not installed. Run 'uv run playwright install chromium' to enable."
style="opacity: 0.5;
cursor: not-allowed">Screenshot</span>
{% endif %}
{% endif %} {% endif %}
{% if delivery_mode != "text" %} {% if delivery_mode != "text" %}
<form action="/use_text" method="post" class="d-inline"> <form action="/use_text" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm" <button class="btn btn-outline-light btn-sm"
style="border-top-left-radius: 0; border-bottom-left-radius: 0;" style="border-top-left-radius: 0;
border-bottom-left-radius: 0"
name="feed_url" name="feed_url"
value="{{ feed.url }}">Text</button> value="{{ feed.url }}">Text</button>
</form> </form>
{% else %} {% else %}
<span class="btn btn-primary btn-sm disabled" <span class="btn btn-primary btn-sm disabled"
style="border-top-left-radius: 0; border-bottom-left-radius: 0;">Text</span> style="border-top-left-radius: 0;
border-bottom-left-radius: 0">Text</span>
{% endif %} {% endif %}
</div> </div>
{% if not chromium_installed and not "youtube.com/feeds/videos.xml" in feed.url %}
<div class="mt-2 small text-muted">
Screenshot mode requires Chromium for Playwright.
Run <code class="text-light">uv run playwright install chromium</code> to enable it.
</div>
{% endif %}
</section> </section>
{% if delivery_mode == "screenshot" and not "youtube.com/feeds/videos.xml" in feed.url %} {% if delivery_mode == "screenshot" and chromium_installed and not "youtube.com/feeds/videos.xml" in feed.url %}
<hr class="border-secondary" /> <hr class="border-secondary" />
<section class="mb-3"> <section class="mb-3">
<h4 class="h6 text-muted mb-2">Screenshot Layout</h4> <h4 class="h6 text-muted mb-2">Screenshot Layout</h4>
@ -237,10 +257,6 @@
<span class="btn btn-primary btn-sm disabled">Desktop</span> <span class="btn btn-primary btn-sm disabled">Desktop</span>
{% endif %} {% endif %}
</div> </div>
<div class="mt-2 small text-muted">
Requires Chromium for Playwright. Run
<code class="text-light">uv run playwright install chromium</code> once.
</div>
</section> </section>
{% endif %} {% endif %}
{% if not "youtube.com/feeds/videos.xml" in feed.url %} {% if not "youtube.com/feeds/videos.xml" in feed.url %}
@ -283,7 +299,7 @@
oninput="this.form.elements.image_limit_value.value = this.value" /> oninput="this.form.elements.image_limit_value.value = this.value" />
<div class="d-flex justify-content-between text-muted small"> <div class="d-flex justify-content-between text-muted small">
<span>0</span> <span>0</span>
</span> <span>{{ max_media_gallery_items }}</span> <span>{{ max_media_gallery_items }}</span>
</div> </div>
</div> </div>
<button class="btn btn-outline-light btn-sm" type="submit">Save image limit</button> <button class="btn btn-outline-light btn-sm" type="submit">Save image limit</button>
@ -472,14 +488,10 @@
href="/blacklist?feed_url={{ feed.url|encode_url }}">Blacklist</a> href="/blacklist?feed_url={{ feed.url|encode_url }}">Blacklist</a>
<div class="btn-group btn-group-sm" role="group"> <div class="btn-group btn-group-sm" role="group">
<a class="btn {{ 'btn-primary' if delivery_mode == 'text' else 'btn-outline-light' }}" <a class="btn {{ 'btn-primary' if delivery_mode == 'text' else 'btn-outline-light' }}"
href="/custom?feed_url={{ feed.url|encode_url }}"> href="/custom?feed_url={{ feed.url|encode_url }}">Customize message</a>
Customize message
</a>
{% if not "youtube.com/feeds/videos.xml" in feed.url %} {% if not "youtube.com/feeds/videos.xml" in feed.url %}
<a class="btn {{ 'btn-primary' if delivery_mode == 'embed' else 'btn-outline-light' }}" <a class="btn {{ 'btn-primary' if delivery_mode == 'embed' else 'btn-outline-light' }}"
href="/embed?feed_url={{ feed.url|encode_url }}"> href="/embed?feed_url={{ feed.url|encode_url }}">Customize embed</a>
Customize embed
</a>
{% endif %} {% endif %}
</div> </div>
</div> </div>

12
discord_rss_bot/templates/settings.html
View file

@ -46,6 +46,11 @@
name="delivery_mode"> name="delivery_mode">
<option value="embed" <option value="embed"
{% if global_delivery_mode == "embed" %}selected{% endif %}>Embed</option> {% if global_delivery_mode == "embed" %}selected{% endif %}>Embed</option>
<option value="screenshot"
{% if global_delivery_mode == "screenshot" %}selected{% endif %}
{% if not chromium_installed %}disabled{% endif %}>
Screenshot{% if not chromium_installed %} (Chromium not installed){% endif %}
</option>
<option value="text" <option value="text"
{% if global_delivery_mode == "text" %}selected{% endif %}>Text</option> {% if global_delivery_mode == "text" %}selected{% endif %}>Text</option>
</select> </select>
@ -73,9 +78,12 @@
<div class="form-text text-muted mt-2"> <div class="form-text text-muted mt-2">
New feeds inherit this value. Existing feeds keep their current screenshot layout. New feeds inherit this value. Existing feeds keep their current screenshot layout.
</div> </div>
<div class="form-text screenshot-requirement mt-1"> {% if not chromium_installed %}
Screenshot mode requires Chromium to be installed for Playwright. Run <code>uv run playwright install chromium</code> once on this machine. <div class="form-text text-muted mt-1">
Screenshot mode requires Chromium to be installed for Playwright.
Run <code>uv run playwright install chromium</code> once on this machine.
</div> </div>
{% endif %}
</div> </div>
</form> </form>
<form action="/set_global_webhook_text_length_limit" <form action="/set_global_webhook_text_length_limit"

8
discord_rss_bot/webhook.py
View file

@ -19,7 +19,7 @@ class WebhookFile:
class DiscordEmbed: class DiscordEmbed:
"""Small Discord embed payload builder used by the webhook sender.""" """Small Discord embed payload builder used by the webhook sender."""
def __init__(self) -> None: # noqa: D107 def __init__(self) -> None: # ruff:ignore[undocumented-public-init]
self._payload: JsonObject = {} self._payload: JsonObject = {}
def to_dict(self) -> JsonObject: def to_dict(self) -> JsonObject:
@ -54,7 +54,7 @@ class DiscordEmbed:
def set_thumbnail(self, *, url: str) -> None: def set_thumbnail(self, *, url: str) -> None:
self._payload["thumbnail"] = {"url": url} self._payload["thumbnail"] = {"url": url}
def set_image(self, *, url: str, **_ignored: Any) -> None: # noqa: ANN401 def set_image(self, *, url: str, **_ignored: Any) -> None: # ruff:ignore[any-type]
self._payload["image"] = {"url": url} self._payload["image"] = {"url": url}
def set_footer(self, *, text: str, icon_url: str | None = None) -> None: def set_footer(self, *, text: str, icon_url: str | None = None) -> None:
@ -85,7 +85,7 @@ class DiscordWebhook:
while leaving the actual HTTP transport to `httpx2`. while leaving the actual HTTP transport to `httpx2`.
""" """
def __init__( # noqa: D107 def __init__( # ruff:ignore[undocumented-public-init]
self, self,
url: str, url: str,
*, *,
@ -99,7 +99,7 @@ class DiscordWebhook:
thread_id: str | None = None, thread_id: str | None = None,
timeout: float | None = None, timeout: float | None = None,
rate_limit_retry: bool = False, rate_limit_retry: bool = False,
**_ignored: Any, # noqa: ANN401 **_ignored: Any, # ruff:ignore[any-type]
) -> None: ) -> None:
self.url: str = url self.url: str = url
self.thread_id: str | None = thread_id self.thread_id: str | None = thread_id

75
pyproject.toml
View file

@ -42,55 +42,56 @@ fix = true
line-length = 120 line-length = 120
lint.select = ["ALL"] lint.select = ["ALL"]
lint.unfixable = ["F841"] # Don't automatically remove unused variables # Don't automatically remove unused variables
lint.unfixable = ["unused-variable"]
lint.pydocstyle.convention = "google" lint.pydocstyle.convention = "google"
lint.isort.required-imports = ["from __future__ import annotations"] lint.isort.required-imports = ["from __future__ import annotations"]
lint.isort.force-single-line = true lint.isort.force-single-line = true
lint.ignore = [ lint.ignore = [
"ANN201", # Checks that public functions and methods have return type annotations. "missing-return-type-undocumented-public-function", # Checks that public functions and methods have return type annotations.
"ARG001", # Checks for the presence of unused arguments in function definitions. "unused-function-argument", # Checks for the presence of unused arguments in function definitions.
"B008", # Allow Form() as a default value "function-call-in-default-argument", # Allow Form() as a default value
"CPY001", # Missing copyright notice at top of file "missing-copyright-notice", # Missing copyright notice at top of file
"D100", # Checks for undocumented public module definitions. "undocumented-public-module", # Checks for undocumented public module definitions.
"D101", # Checks for undocumented public class definitions. "undocumented-public-class", # Checks for undocumented public class definitions.
"D102", # Checks for undocumented public method definitions. "undocumented-public-method", # Checks for undocumented public method definitions.
"D104", # Missing docstring in public package. "undocumented-public-package", # Missing docstring in public package.
"D105", # Missing docstring in magic method. "undocumented-magic-method", # Missing docstring in magic method.
"D105", # pydocstyle - missing docstring in magic method "undocumented-magic-method", # pydocstyle - missing docstring in magic method
"D106", # Checks for undocumented public class definitions, for nested classes. "undocumented-public-nested-class", # Checks for undocumented public class definitions, for nested classes.
"ERA001", # Found commented-out code "commented-out-code", # Found commented-out code
"FBT003", # Checks for boolean positional arguments in function calls. "boolean-positional-value-in-call", # Checks for boolean positional arguments in function calls.
"FIX002", # Line contains TODO "line-contains-todo", # Line contains TODO
"G002", # Allow % in logging "logging-percent-format", # Allow % in logging
"PGH003", # Check for type: ignore annotations that suppress all type warnings, as opposed to targeting specific type warnings. "blanket-type-ignore", # Check for type: ignore annotations that suppress all type warnings, as opposed to targeting specific type warnings.
"PLR6301", # Checks for the presence of unused self parameter in methods definitions. "no-self-use", # Checks for the presence of unused self parameter in methods definitions.
"RUF029", # Checks for functions declared async that do not await or otherwise use features requiring the function to be declared async. "unused-async", # Checks for functions declared async that do not await or otherwise use features requiring the function to be declared async.
"TD003", # Checks that a TODO comment is associated with a link to a relevant issue or ticket. "missing-todo-link", # Checks that a TODO comment is associated with a link to a relevant issue or ticket.
"PLR0913", # Checks for function definitions that include too many arguments. "too-many-arguments", # Checks for function definitions that include too many arguments.
"PLR0917", # Checks for function definitions that include too many positional arguments. "too-many-positional-arguments", # Checks for function definitions that include too many positional arguments.
# Conflicting lint rules when using Ruff's formatter # Conflicting lint rules when using Ruff's formatter
# https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules # https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules
"COM812", # Checks for the absence of trailing commas. "missing-trailing-comma", # Checks for the absence of trailing commas.
"COM819", # Checks for the presence of prohibited trailing commas. "prohibited-trailing-comma", # Checks for the presence of prohibited trailing commas.
"D206", # Checks for docstrings that are indented with tabs. "docstring-tab-indentation", # Checks for docstrings that are indented with tabs.
"D300", # Checks for docstrings that use '''triple single quotes''' instead of """triple double quotes""". "triple-single-quotes", # Checks for docstrings that use '''triple single quotes''' instead of """triple double quotes""".
"E111", # Checks for indentation with a non-multiple of 4 spaces. "indentation-with-invalid-multiple", # Checks for indentation with a non-multiple of 4 spaces.
"E114", # Checks for indentation of comments with a non-multiple of 4 spaces. "indentation-with-invalid-multiple-comment", # Checks for indentation of comments with a non-multiple of 4 spaces.
"E117", # Checks for over-indented code. "over-indented", # Checks for over-indented code.
"ISC001", # Checks for implicitly concatenated strings on a single line. "single-line-implicit-string-concatenation", # Checks for implicitly concatenated strings on a single line.
"ISC002", # Checks for implicitly concatenated strings that span multiple lines. "multi-line-implicit-string-concatenation", # Checks for implicitly concatenated strings that span multiple lines.
"Q000", # Checks for inline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.inline-quotes option. "bad-quotes-inline-string", # Checks for inline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.inline-quotes option.
"Q001", # Checks for multiline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.multiline-quotes setting. "bad-quotes-multiline-string", # Checks for multiline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.multiline-quotes setting.
"Q002", # Checks for docstrings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.docstring-quotes setting. "bad-quotes-docstring", # Checks for docstrings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.docstring-quotes setting.
"Q003", # Checks for strings that include escaped quotes, and suggests changing the quote style to avoid the need to escape them. "avoidable-escaped-quote", # Checks for strings that include escaped quotes, and suggests changing the quote style to avoid the need to escape them.
"W191", # Checks for indentation that uses tabs. "tab-indentation", # Checks for indentation that uses tabs.
] ]
[tool.ruff.lint.per-file-ignores] [tool.ruff.lint.per-file-ignores]
"tests/*" = ["S101", "D103", "PLR2004"] "tests/*" = ["assert", "undocumented-public-function", "magic-value-comparison"]
[tool.pytest.ini_options] [tool.pytest.ini_options]
addopts = "-n 5 --dist loadfile -m \"not integration and not slow\"" addopts = "-n 5 --dist loadfile -m \"not integration and not slow\""

2
tests/test_custom_filter.py
View file

@ -26,7 +26,7 @@ def test_encode_url() -> None:
assert encode_url("https://www.example.com/my path") == r"https%3A//www.example.com/my%20path", assert_msg assert encode_url("https://www.example.com/my path") == r"https%3A//www.example.com/my%20path", assert_msg
# Test input with special characters # Test input with special characters
assert_msg: str = f"Got: {encode_url('https://www.example.com/my path?q=abc&b=1')}, Expected: https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1" # noqa: E501 assert_msg: str = f"Got: {encode_url('https://www.example.com/my path?q=abc&b=1')}, Expected: https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1" # ruff:ignore[line-too-long]
assert ( assert (
encode_url("https://www.example.com/my path?q=abc&b=1") encode_url("https://www.example.com/my path?q=abc&b=1")
== r"https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1" == r"https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1"

60
tests/test_feeds.py
View file

@ -177,7 +177,7 @@ def test_get_entry_delivery_mode_prefers_delivery_mode_tag() -> None:
entry = MagicMock() entry = MagicMock()
entry.feed.url = "https://example.com/feed.xml" entry.feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"delivery_mode": "screenshot", "delivery_mode": "screenshot",
"should_send_embed": True, "should_send_embed": True,
}.get(key, default) }.get(key, default)
@ -192,7 +192,7 @@ def test_get_entry_delivery_mode_falls_back_to_legacy_embed_flag() -> None:
entry = MagicMock() entry = MagicMock()
entry.feed.url = "https://example.com/feed.xml" entry.feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"delivery_mode": "", "delivery_mode": "",
"should_send_embed": False, "should_send_embed": False,
}.get(key, default) }.get(key, default)
@ -219,7 +219,7 @@ def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook(
entry.link = "https://www.hoyolab.com/article/38588239" entry.link = "https://www.hoyolab.com/article/38588239"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.test/webhook", "webhook": "https://discord.test/webhook",
"delivery_mode": "text", "delivery_mode": "text",
}.get(key, default) }.get(key, default)
@ -258,7 +258,7 @@ def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook(
entry.link = "https://www.hoyolab.com/article/38588239" entry.link = "https://www.hoyolab.com/article/38588239"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.test/webhook", "webhook": "https://discord.test/webhook",
"delivery_mode": "screenshot", "delivery_mode": "screenshot",
}.get(key, default) }.get(key, default)
@ -296,7 +296,7 @@ def test_send_entry_to_discord_hoyolab_embed_mode_uses_hoyolab_webhook(
entry.link = "https://www.hoyolab.com/article/38588239" entry.link = "https://www.hoyolab.com/article/38588239"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.test/webhook", "webhook": "https://discord.test/webhook",
"delivery_mode": "embed", "delivery_mode": "embed",
}.get(key, default) }.get(key, default)
@ -372,7 +372,7 @@ def test_get_feed_media_gallery_image_limit_defaults_to_first_image() -> None:
reader = MagicMock() reader = MagicMock()
feed = MagicMock() feed = MagicMock()
feed.url = "https://example.com/feed.xml" feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: default # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: default # ruff:ignore[unused-lambda-argument]
result = feeds.get_feed_media_gallery_image_limit(reader, feed) result = feeds.get_feed_media_gallery_image_limit(reader, feed)
@ -424,7 +424,7 @@ def test_get_feed_webhook_text_length_limit_defaults_to_discord_limit() -> None:
reader = MagicMock() reader = MagicMock()
feed = MagicMock() feed = MagicMock()
feed.url = "https://example.com/feed.xml" feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: default # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: default # ruff:ignore[unused-lambda-argument]
result = feeds.get_feed_webhook_text_length_limit(reader, feed) result = feeds.get_feed_webhook_text_length_limit(reader, feed)
@ -433,7 +433,7 @@ def test_get_feed_webhook_text_length_limit_defaults_to_discord_limit() -> None:
def test_create_feed_inherits_global_screenshot_layout() -> None: def test_create_feed_inherits_global_screenshot_layout() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "mobile", "screenshot_layout": "mobile",
}.get(key, default) }.get(key, default)
@ -445,7 +445,7 @@ def test_create_feed_inherits_global_screenshot_layout() -> None:
def test_create_feed_inherits_global_text_delivery_mode() -> None: def test_create_feed_inherits_global_text_delivery_mode() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
"delivery_mode": "text", "delivery_mode": "text",
@ -459,7 +459,7 @@ def test_create_feed_inherits_global_text_delivery_mode() -> None:
def test_create_feed_enables_sent_webhook_tracking_by_default() -> None: def test_create_feed_enables_sent_webhook_tracking_by_default() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
"delivery_mode": "embed", "delivery_mode": "embed",
@ -472,7 +472,7 @@ def test_create_feed_enables_sent_webhook_tracking_by_default() -> None:
def test_create_feed_sets_default_media_gallery_image_limit() -> None: def test_create_feed_sets_default_media_gallery_image_limit() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
"delivery_mode": "embed", "delivery_mode": "embed",
@ -489,7 +489,7 @@ def test_create_feed_sets_default_media_gallery_image_limit() -> None:
def test_create_feed_sets_default_webhook_text_length_limit() -> None: def test_create_feed_sets_default_webhook_text_length_limit() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
"delivery_mode": "embed", "delivery_mode": "embed",
@ -506,7 +506,7 @@ def test_create_feed_sets_default_webhook_text_length_limit() -> None:
def test_create_feed_inherits_global_webhook_text_length_limit() -> None: def test_create_feed_inherits_global_webhook_text_length_limit() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
"delivery_mode": "embed", "delivery_mode": "embed",
@ -524,7 +524,7 @@ def test_create_feed_inherits_global_webhook_text_length_limit() -> None:
def test_create_feed_falls_back_to_embed_when_global_delivery_mode_is_invalid() -> None: def test_create_feed_falls_back_to_embed_when_global_delivery_mode_is_invalid() -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
"delivery_mode": "invalid", "delivery_mode": "invalid",
@ -540,7 +540,7 @@ def test_create_feed_removes_new_feed_when_initial_update_fails() -> None:
feed_url = "https://example.com/not-a-feed" feed_url = "https://example.com/not-a-feed"
autodiscover_links = [{"href": "https://example.com/feed.xml", "type": "application/rss+xml"}] autodiscover_links = [{"href": "https://example.com/feed.xml", "type": "application/rss+xml"}]
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
".reader.autodiscover": autodiscover_links, ".reader.autodiscover": autodiscover_links,
}.get(key, default) }.get(key, default)
@ -556,7 +556,7 @@ def test_create_feed_removes_new_feed_when_initial_update_fails() -> None:
def test_create_feed_does_not_remove_existing_feed_when_update_fails() -> None: def test_create_feed_does_not_remove_existing_feed_when_update_fails() -> None:
feed_url = "https://example.com/existing-feed.xml" feed_url = "https://example.com/existing-feed.xml"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}], "webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
}.get(key, default) }.get(key, default)
reader.add_feed.side_effect = FeedExistsError(feed_url) reader.add_feed.side_effect = FeedExistsError(feed_url)
@ -582,7 +582,7 @@ def test_create_screenshot_webhook_adds_image_file(
entry.id = "entry-abc" entry.id = "entry-abc"
entry.link = "https://example.com/article" entry.link = "https://example.com/article"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"screenshot_layout": "mobile", "screenshot_layout": "mobile",
}.get(key, default) }.get(key, default)
@ -619,7 +619,7 @@ def test_create_screenshot_webhook_retries_jpeg_when_png_too_large(
entry.id = "entry-large" entry.id = "entry-large"
entry.link = "https://example.com/large-article" entry.link = "https://example.com/large-article"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
}.get(key, default) }.get(key, default)
@ -655,7 +655,7 @@ def test_create_screenshot_webhook_falls_back_when_all_formats_too_large(
entry.id = "entry-too-large" entry.id = "entry-too-large"
entry.link = "https://example.com/very-large" entry.link = "https://example.com/very-large"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"screenshot_layout": "desktop", "screenshot_layout": "desktop",
}.get(key, default) }.get(key, default)
@ -758,7 +758,7 @@ def test_create_text_webhook_uses_feed_text_length_limit(mock_replace_tags_in_te
entry = MagicMock() entry = MagicMock()
entry.feed.url = "https://example.com/feed.xml" entry.feed.url = "https://example.com/feed.xml"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"custom_message": "{{entry_title}}", "custom_message": "{{entry_title}}",
"webhook_text_length_limit": 20, "webhook_text_length_limit": 20,
}.get(key, default) }.get(key, default)
@ -784,7 +784,7 @@ def test_create_embed_webhook_uses_media_gallery_for_entry_images(
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 10, "media_gallery_image_limit": 10,
"webhook_text_length_limit": 4000, "webhook_text_length_limit": 4000,
}.get(key, default) }.get(key, default)
@ -830,7 +830,7 @@ def test_create_embed_webhook_uses_feed_text_length_limit_for_media_gallery(
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 10, "media_gallery_image_limit": 10,
"webhook_text_length_limit": 20, "webhook_text_length_limit": 20,
}.get(key, default) }.get(key, default)
@ -862,7 +862,7 @@ def test_create_embed_webhook_can_limit_media_gallery_to_first_image(
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 1, "media_gallery_image_limit": 1,
"webhook_text_length_limit": 4000, "webhook_text_length_limit": 4000,
}.get(key, default) }.get(key, default)
@ -895,7 +895,7 @@ def test_create_embed_webhook_can_disable_media_images(
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0, "media_gallery_image_limit": 0,
"webhook_text_length_limit": 4000, "webhook_text_length_limit": 4000,
}.get(key, default) }.get(key, default)
@ -931,7 +931,7 @@ def test_create_embed_webhook_can_use_steam_game_icon_thumbnail(
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0, "media_gallery_image_limit": 0,
"webhook_text_length_limit": 4000, "webhook_text_length_limit": 4000,
}.get(key, default) }.get(key, default)
@ -971,7 +971,7 @@ def test_create_embed_webhook_prefers_local_steam_game_icon_thumbnail(
local_icon_bytes = b"local-steam-icon" local_icon_bytes = b"local-steam-icon"
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0, "media_gallery_image_limit": 0,
"webhook_text_length_limit": 4000, "webhook_text_length_limit": 4000,
}.get(key, default) }.get(key, default)
@ -1013,7 +1013,7 @@ def test_create_embed_webhook_does_not_inject_steam_thumbnail_when_app_id_is_mis
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0, "media_gallery_image_limit": 0,
"webhook_text_length_limit": 4000, "webhook_text_length_limit": 4000,
}.get(key, default) }.get(key, default)
@ -1047,7 +1047,7 @@ def test_create_embed_webhook_uses_feed_text_length_limit_for_regular_embed_desc
mock_fetch_ttvdrops_campaign_media_items: MagicMock, mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None: ) -> None:
reader = MagicMock() reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0, "media_gallery_image_limit": 0,
"webhook_text_length_limit": 20, "webhook_text_length_limit": 20,
}.get(key, default) }.get(key, default)
@ -1294,7 +1294,7 @@ def test_send_entry_to_discord_uses_screenshot_mode(
entry.feed.url = "https://example.com/feed.xml" entry.feed.url = "https://example.com/feed.xml"
entry.feed_url = "https://example.com/feed.xml" entry.feed_url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005 reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.com/api/webhooks/123/abc", "webhook": "https://discord.com/api/webhooks/123/abc",
}.get(key, default) }.get(key, default)
@ -1339,7 +1339,7 @@ def test_send_entry_to_discord_youtube_feed(
mock_entry.feed_url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456" mock_entry.feed_url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456"
# Mock the tags # Mock the tags
mock_reader.get_tag.side_effect = lambda feed, tag, default=None: { # noqa: ARG005 mock_reader.get_tag.side_effect = lambda feed, tag, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.com/api/webhooks/123/abc", "webhook": "https://discord.com/api/webhooks/123/abc",
"should_send_embed": True, # This should be ignored for YouTube feeds "should_send_embed": True, # This should be ignored for YouTube feeds
}.get(tag, default) }.get(tag, default)

4
tests/test_git_backup.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import contextlib import contextlib
import json import json
import shutil import shutil
import subprocess # noqa: S404 import subprocess # ruff:ignore[suspicious-subprocess-import]
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from typing import cast from typing import cast
from unittest.mock import MagicMock from unittest.mock import MagicMock
@ -579,7 +579,7 @@ def test_embed_backup_end_to_end(monkeypatch: pytest.MonkeyPatch, tmp_path: Path
assert response.status_code == 200, f"Failed to customize embed: {response.text}" assert response.status_code == 200, f"Failed to customize embed: {response.text}"
# Verify a commit was created # Verify a commit was created
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603 result: subprocess.CompletedProcess[str] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[git_executable, "-C", str(backup_path), "log", "--oneline"], [git_executable, "-C", str(backup_path), "log", "--oneline"],
capture_output=True, capture_output=True,
text=True, text=True,

12
tests/test_main.py
View file

@ -279,16 +279,6 @@ def test_get() -> None:
response: Response = client.get(url="/feed", params={"feed_url": encoded_feed_url(feed_url)}) response: Response = client.get(url="/feed", params={"feed_url": encoded_feed_url(feed_url)})
assert response.status_code == 200, f"/feed failed: {response.text}" assert response.status_code == 200, f"/feed failed: {response.text}"
assert "Feed Summary" in response.text
assert "This feed" in response.text
assert "Screenshot Delivery" in response.text
assert "Image Delivery" in response.text
assert "Text Delivery" in response.text
assert "2000 characters" in response.text
assert 'type="range"' in response.text
assert 'max="10"' in response.text
assert 'id="text_length_limit"' in response.text
assert 'max="4000"' in response.text
response: Response = client.get(url="/") response: Response = client.get(url="/")
assert response.status_code == 200, f"/ failed: {response.text}" assert response.status_code == 200, f"/ failed: {response.text}"
@ -735,8 +725,6 @@ def test_c3kay_feed_delivery_mode_toggle_routes_update_stored_tags() -> None:
assert reader.get_tag(c3kay_feed_url, "delivery_mode") == "screenshot" assert reader.get_tag(c3kay_feed_url, "delivery_mode") == "screenshot"
assert reader.get_tag(c3kay_feed_url, "screenshot_layout") == "mobile" assert reader.get_tag(c3kay_feed_url, "screenshot_layout") == "mobile"
assert reader.get_tag(c3kay_feed_url, "should_send_embed") is False assert reader.get_tag(c3kay_feed_url, "should_send_embed") is False
assert "Disable screenshot delivery" in response.text
assert "Send embed instead of screenshot" not in response.text
response = client.post(url="/use_embed", data={"feed_url": c3kay_feed_url}) response = client.post(url="/use_embed", data={"feed_url": c3kay_feed_url})
assert response.status_code == 200, f"Failed to set embed mode: {response.text}" assert response.status_code == 200, f"Failed to set embed mode: {response.text}"

2
tests/test_settings.py
View file

@ -86,7 +86,7 @@ def test_data_dir() -> None:
def test_default_custom_message() -> None: def test_default_custom_message() -> None:
"""Test the default custom message.""" """Test the default custom message."""
assert_msg = f"The default custom message should be '{{entry_title}}\n{{entry_link}}'. But it was '{default_custom_message}'." # noqa: E501 assert_msg = f"The default custom message should be '{{entry_title}}\n{{entry_link}}'. But it was '{default_custom_message}'." # ruff:ignore[line-too-long]
assert default_custom_message == "{{entry_title}}\n{{entry_link}}", assert_msg assert default_custom_message == "{{entry_title}}\n{{entry_link}}", assert_msg