Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ Flask-OpenID==1.3.0
Flask-WTF==1.2.1
bleach==6.1.0
emoji==2.9.0
feedgen==1.0.0
humanize==4.9.0
mistune==2.0.5
pybadges==3.0.1
Expand Down
62 changes: 62 additions & 0 deletions tests/tests_feeds.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
import unittest
from unittest.mock import patch, Mock
from webapp.app import create_app


class TestFeeds(unittest.TestCase):
def setUp(self):
self.app = create_app(testing=True)
self.client = self.app.test_client()

@patch("webapp.feeds.feeds.session.get")
def test_feeds_updates_success(self, mock_get):
"""Test successful RSS feed generation using feedgen"""
# Mock API response
mock_response = Mock()
mock_response.json.return_value = {
"page": 1,
"size": 2,
"snaps": [
{
"name": "test-snap",
"title": "Test Snap",
"summary": "A test snap",
"publisher": "Test Publisher",
"license": "MIT",
"version": "1.0.0",
"last_updated": "Thu, 02 Oct 2025 22:07:58 GMT",
"icon": "https://example.com/icon.png",
"snap_id": "test-snap-id-123",
"media": [
{
"type": "screenshot",
"url": "https://example.com/screenshot.png",
}
],
}
],
}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response

# Make request to RSS endpoint
response = self.client.get("/feeds/updates")
# Check response
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.content_type, "application/rss+xml; charset=utf-8"
)
# Check RSS content
content = response.get_data(as_text=True)
self.assertIn("<?xml version='1.0' encoding='UTF-8'?>", content)
self.assertIn(
"<title>Snapcraft - recently updated snaps</title>", content
)
self.assertIn("<title>Test Snap</title>", content)
self.assertIn("<description>", content)
self.assertIn("Test Publisher", content)
self.assertIn("https://snapcraft.io/test-snap", content)


if __name__ == "__main__":
unittest.main()
2 changes: 2 additions & 0 deletions webapp/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
from webapp.endpoints.validation_sets import validation_sets
from webapp.endpoints.invites import invites
from webapp.endpoints.settings import settings
from webapp.feeds.feeds import feeds
from webapp.config import SENTRY_DSN


Expand Down Expand Up @@ -81,6 +82,7 @@ def inject_csrf_token():
app.register_blueprint(validation_sets)
app.register_blueprint(invites)
app.register_blueprint(settings)
app.register_blueprint(feeds)
init_docs(app, "/docs")
init_blog(app, "/blog")
init_tutorials(app, "/tutorials")
Expand Down
Empty file added webapp/feeds/__init__.py
Empty file.
139 changes: 139 additions & 0 deletions webapp/feeds/feeds.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
import flask
import requests
from datetime import datetime, timezone
from flask import Response
from feedgen.feed import FeedGenerator
from requests import Session
from html import escape
from urllib.parse import urlparse

feeds = flask.Blueprint(
"feeds",
__name__,
)

session = Session()


def is_safe_url(url):
"""Check if URL is safe (http/https only)."""
if not url:
return False
try:
parsed = urlparse(url)
return parsed.scheme in ("http", "https") and parsed.netloc
except Exception:
return False


def parse_snap_date(date_string):
"""Parse date string from API to datetime object."""
try:
dt = datetime.strptime(date_string, "%a, %d %b %Y %H:%M:%S %Z")
return dt.replace(tzinfo=timezone.utc)
except ValueError:
return datetime.now(timezone.utc)


def create_snap_description(snap):
"""Create HTML description for RSS item."""
description_parts = []

if snap.get("icon") and is_safe_url(snap["icon"]):
icon_url = escape(snap["icon"])
description_parts.append(f'<img src="{icon_url}" alt="Snap icon">')

if snap.get("summary"):
summary = escape(snap["summary"])
description_parts.append(f"<p>{summary}</p>")

additional_info = []

if snap.get("publisher"):
publisher = escape(snap["publisher"])
additional_info.append(f"<li>Developer: {publisher}</li>")

if snap.get("version"):
version = escape(snap["version"])
additional_info.append(f"<li>Version: {version}</li>")

if additional_info:
description_parts.append("<ul>" + "".join(additional_info) + "</ul>")

if snap.get("media"):
for media in snap["media"]:
if (
media.get("type") == "screenshot"
and media.get("url")
and is_safe_url(media["url"])
):
media_url = escape(media["url"])
description_parts.append(
f'<img src="{media_url}" alt="Screenshot">'
)

return "".join(description_parts)


@feeds.route("/feeds/updates")
def recently_updated_feed():
"""Generate RSS feed for recently updated snaps."""

fg = FeedGenerator()
fg.title("Snapcraft - recently updated snaps")
fg.link(href="https://snapcraft.io/store", rel="alternate")
fg.description("Recently updated snaps published on Snapcraft")
fg.language("en")
fg.docs("http://www.rssboard.org/rss-specification")
fg.generator("python-feedgen")

size = int(flask.request.args.get("size", "50"))

page = int(flask.request.args.get("page", "1"))

try:
api_url = flask.current_app.config.get(
"RECOMMENDATION_API_URL",
"https://recommendations.snapcraft.io/api/recently-updated",
)
params = {"size": size, "page": page}
response = session.get(api_url, params=params, timeout=10)
response.raise_for_status()

data = response.json()
snaps = data.get("snaps", [])

except (requests.RequestException, ValueError) as e:
flask.current_app.logger.error(f"Failed to fetch recommendations: {e}")
snaps = []

for snap in snaps:
try:
fe = fg.add_entry()

title = escape(snap.get("title"))

fe.title(title)

snap_name = snap.get("name")
snap_url = f"https://snapcraft.io/{snap_name}"
fe.link(href=snap_url)

description = create_snap_description(snap)
fe.description(description)

pub_date = parse_snap_date(snap["last_updated"])
fe.pubDate(pub_date)

except Exception as e:
flask.current_app.logger.error(
f"Failed to add snap to RSS feed: {e}"
)
continue

rss_str = fg.rss_str(pretty=True)

response = Response(rss_str, mimetype="application/rss+xml")
response.headers["Cache-Control"] = "public, max-age=86400"

return response