From 5b53b76afe48e51c48b1841c5a83681663f27b36 Mon Sep 17 00:00:00 2001 From: Mohammad Iskandarany Date: Wed, 8 Oct 2025 10:10:14 +0300 Subject: [PATCH 1/4] feat: updates rss feed --- requirements.txt | 1 + tests/tests_feeds.py | 80 ++++++++++++++++++++++++++ webapp/app.py | 2 + webapp/feeds/__init__.py | 0 webapp/feeds/feeds.py | 119 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 202 insertions(+) create mode 100644 tests/tests_feeds.py create mode 100644 webapp/feeds/__init__.py create mode 100644 webapp/feeds/feeds.py diff --git a/requirements.txt b/requirements.txt index ab026f523b..95bb33480f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/tests/tests_feeds.py b/tests/tests_feeds.py new file mode 100644 index 0000000000..c4016b73bf --- /dev/null +++ b/tests/tests_feeds.py @@ -0,0 +1,80 @@ +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", + "description": "This is a test snap for testing", + "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('', content) + self.assertIn('', content) + self.assertIn('Snapcraft – recently updated applications', content) + self.assertIn('Test Snap', content) + self.assertIn('', content) + self.assertIn('Test Publisher', content) + self.assertIn('https://snapcraft.io/test-snap', content) + + @patch('webapp.feeds.feeds.session.get') + def test_feeds_updates_api_error(self, mock_get): + """Test RSS feed generation when API fails""" + # Mock API error + mock_get.side_effect = Exception("API Error") + + # Make request to RSS endpoint + response = self.client.get('/feeds/updates') + + # Should still return valid RSS, just empty + self.assertEqual(response.status_code, 200) + self.assertEqual(response.content_type, 'application/rss+xml; charset=utf-8') + + content = response.get_data(as_text=True) + self.assertIn('', content) + self.assertIn('', content) + self.assertIn('Snapcraft – recently updated applications', content) + + +if __name__ == '__main__': + unittest.main() diff --git a/webapp/app.py b/webapp/app.py index b84a06cb39..c6880157bb 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -37,6 +37,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 TALISKER_WSGI_LOGGER = logging.getLogger("talisker.wsgi") @@ -87,6 +88,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") diff --git a/webapp/feeds/__init__.py b/webapp/feeds/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/webapp/feeds/feeds.py b/webapp/feeds/feeds.py new file mode 100644 index 0000000000..7f12f30810 --- /dev/null +++ b/webapp/feeds/feeds.py @@ -0,0 +1,119 @@ +import flask +import requests +import talisker.requests +from datetime import datetime, timezone +from flask import Response +from feedgen.feed import FeedGenerator +from requests import Session + +feeds = flask.Blueprint( + "feeds", + __name__, +) + +session = Session() + + +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"): + description_parts.append(f'') + + if snap.get("summary"): + description_parts.append(f"

{snap['summary']}

") + + if snap.get("description"): + desc_html = snap["description"].replace("\n", "
") + description_parts.append(f"

{desc_html}

") + + additional_info = [] + if snap.get("publisher"): + additional_info.append(f"
  • Developer: {snap['publisher']}
    • ") + if snap.get("version"): + additional_info.append(f"
  • Version: {snap['version']}
    • ") + + if additional_info: + description_parts.append("
      " + "".join(additional_info) + "
    ") + + if snap.get("media"): + for media in snap["media"]: + if media.get("type") == "screenshot" and media.get("url"): + description_parts.append(f'') + + 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') + + # can take in size and page parameters + size = flask.request.args.get('size', '50') + page = 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: + # Log the error and return an empty feed + flask.current_app.logger.error(f"Failed to fetch recommendations: {e}") + snaps = [] + + # Add feed entries + for snap in snaps: + try: + fe = fg.add_entry() + + title = snap.get("title") + fe.title(title) + + snap_name = snap.get("name", "") + snap_url = f"https://snapcraft.io/{snap_name}" + fe.link(href=snap_url) + + # Set description + description = create_snap_description(snap) + fe.description(description) + + if snap.get("last_updated"): + pub_date = parse_snap_date(snap["last_updated"]) + fe.pubDate(pub_date) + else: + fe.pubDate(datetime.now(timezone.utc)) + + except Exception as e: + flask.current_app.logger.error(f"Failed to add snap to RSS feed: {e}") + raise + # 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 \ No newline at end of file From 1f99138bbc5ebce7cb4422865805e9e66054831a Mon Sep 17 00:00:00 2001 From: Mohammad Iskandarany Date: Wed, 8 Oct 2025 13:09:39 +0300 Subject: [PATCH 2/4] rss feed --- tests/tests_feeds.py | 52 ++++++---------- webapp/feeds/feeds.py | 138 ++++++++++++++++++++++++------------------ 2 files changed, 96 insertions(+), 94 deletions(-) diff --git a/tests/tests_feeds.py b/tests/tests_feeds.py index c4016b73bf..5a09f760ab 100644 --- a/tests/tests_feeds.py +++ b/tests/tests_feeds.py @@ -8,7 +8,7 @@ def setUp(self): self.app = create_app(testing=True) self.client = self.app.test_client() - @patch('webapp.feeds.feeds.session.get') + @patch("webapp.feeds.feeds.session.get") def test_feeds_updates_success(self, mock_get): """Test successful RSS feed generation using feedgen""" # Mock API response @@ -21,7 +21,6 @@ def test_feeds_updates_success(self, mock_get): "name": "test-snap", "title": "Test Snap", "summary": "A test snap", - "description": "This is a test snap for testing", "publisher": "Test Publisher", "license": "MIT", "version": "1.0.0", @@ -31,50 +30,33 @@ def test_feeds_updates_success(self, mock_get): "media": [ { "type": "screenshot", - "url": "https://example.com/screenshot.png" + "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') - + 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') - + self.assertEqual( + response.content_type, "application/rss+xml; charset=utf-8" + ) # Check RSS content content = response.get_data(as_text=True) - self.assertIn('', content) - self.assertIn('', content) - self.assertIn('Snapcraft – recently updated applications', content) - self.assertIn('Test Snap', content) - self.assertIn('', content) - self.assertIn('Test Publisher', content) - self.assertIn('https://snapcraft.io/test-snap', content) + self.assertIn("", content) + self.assertIn( + "Snapcraft - recently updated snaps", content + ) + self.assertIn("Test Snap", content) + self.assertIn("", content) + self.assertIn("Test Publisher", content) + self.assertIn("https://snapcraft.io/test-snap", content) - @patch('webapp.feeds.feeds.session.get') - def test_feeds_updates_api_error(self, mock_get): - """Test RSS feed generation when API fails""" - # Mock API error - mock_get.side_effect = Exception("API Error") - # Make request to RSS endpoint - response = self.client.get('/feeds/updates') - - # Should still return valid RSS, just empty - self.assertEqual(response.status_code, 200) - self.assertEqual(response.content_type, 'application/rss+xml; charset=utf-8') - - content = response.get_data(as_text=True) - self.assertIn('', content) - self.assertIn('', content) - self.assertIn('Snapcraft – recently updated applications', content) - - -if __name__ == '__main__': +if __name__ == "__main__": unittest.main() diff --git a/webapp/feeds/feeds.py b/webapp/feeds/feeds.py index 7f12f30810..1caa30885a 100644 --- a/webapp/feeds/feeds.py +++ b/webapp/feeds/feeds.py @@ -1,10 +1,11 @@ import flask import requests -import talisker.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", @@ -14,6 +15,17 @@ 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: @@ -26,94 +38,102 @@ def parse_snap_date(date_string): def create_snap_description(snap): """Create HTML description for RSS item.""" description_parts = [] - - if snap.get("icon"): - description_parts.append(f'') - + + if snap.get("icon") and is_safe_url(snap["icon"]): + icon_url = escape(snap["icon"]) + description_parts.append(f'Snap icon') + if snap.get("summary"): - description_parts.append(f"

    {snap['summary']}

    ") - - if snap.get("description"): - desc_html = snap["description"].replace("\n", "
    ") - description_parts.append(f"

    {desc_html}

    ") - + summary = escape(snap["summary"]) + description_parts.append(f"

    {summary}

    ") + additional_info = [] + if snap.get("publisher"): - additional_info.append(f"
  • Developer: {snap['publisher']}
    • ") + publisher = escape(snap["publisher"]) + additional_info.append(f"
  • Developer: {publisher}
    • ") + if snap.get("version"): - additional_info.append(f"
  • Version: {snap['version']}
    • ") - + version = escape(snap["version"]) + additional_info.append(f"
  • Version: {version}
    • ") + if additional_info: description_parts.append("
      " + "".join(additional_info) + "
    ") - + if snap.get("media"): for media in snap["media"]: - if media.get("type") == "screenshot" and media.get("url"): - description_parts.append(f'') - + if ( + media.get("type") == "screenshot" + and media.get("url") + and is_safe_url(media["url"]) + ): + media_url = escape(media["url"]) + description_parts.append( + f'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') - - # can take in size and page parameters - size = flask.request.args.get('size', '50') - page = flask.request.args.get('page', '1') - + 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} + 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: - # Log the error and return an empty feed flask.current_app.logger.error(f"Failed to fetch recommendations: {e}") snaps = [] - - # Add feed entries + for snap in snaps: try: fe = fg.add_entry() - - title = snap.get("title") + + title = escape(snap.get("title")) + fe.title(title) - - snap_name = snap.get("name", "") + + snap_name = snap.get("name") snap_url = f"https://snapcraft.io/{snap_name}" fe.link(href=snap_url) - - # Set description + description = create_snap_description(snap) fe.description(description) - - if snap.get("last_updated"): - pub_date = parse_snap_date(snap["last_updated"]) - fe.pubDate(pub_date) - else: - fe.pubDate(datetime.now(timezone.utc)) - + + 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}") - raise - # continue - + 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 \ No newline at end of file + + response = Response(rss_str, mimetype="application/rss+xml") + response.headers["Cache-Control"] = "public, max-age=86400" + + return response From 24ad688ef40457289bfa5a8a52a5234f19868e2f Mon Sep 17 00:00:00 2001 From: Mohammad Iskandarany Date: Thu, 9 Oct 2025 14:25:33 +0300 Subject: [PATCH 3/4] fix image alt and size --- webapp/feeds/feeds.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/webapp/feeds/feeds.py b/webapp/feeds/feeds.py index 1caa30885a..cbe6c7bcc7 100644 --- a/webapp/feeds/feeds.py +++ b/webapp/feeds/feeds.py @@ -41,7 +41,9 @@ def create_snap_description(snap): if snap.get("icon") and is_safe_url(snap["icon"]): icon_url = escape(snap["icon"]) - description_parts.append(f'Snap icon') + description_parts.append( + f'Snap icon' + ) if snap.get("summary"): summary = escape(snap["summary"]) @@ -68,9 +70,7 @@ def create_snap_description(snap): and is_safe_url(media["url"]) ): media_url = escape(media["url"]) - description_parts.append( - f'Screenshot' - ) + description_parts.append(f'') return "".join(description_parts) From f58627dc398719d0b793ca6c3be6af5a173c3b40 Mon Sep 17 00:00:00 2001 From: Mohammad Iskandarany Date: Mon, 13 Oct 2025 10:42:25 +0300 Subject: [PATCH 4/4] better tests --- tests/tests_feeds.py | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/tests/tests_feeds.py b/tests/tests_feeds.py index 5a09f760ab..628bf06422 100644 --- a/tests/tests_feeds.py +++ b/tests/tests_feeds.py @@ -1,4 +1,5 @@ import unittest +import xml.etree.ElementTree as ET from unittest.mock import patch, Mock from webapp.app import create_app @@ -11,7 +12,6 @@ def setUp(self): @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, @@ -39,20 +39,30 @@ def test_feeds_updates_success(self, mock_get): 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("", content) self.assertIn( "Snapcraft - recently updated snaps", content ) - self.assertIn("Test Snap", content) + + try: + root = ET.fromstring(content) + except ET.ParseError as e: + self.fail(f"Returned content is not valid XML: {e}") + + self.assertEqual(root.tag, "rss") + channel = root.find("channel") + self.assertIsNotNone(channel, "No element found in RSS feed") + + items = channel.findall("item") + self.assertTrue( + any(item.findtext("title") == "Test Snap" for item in items) + ) self.assertIn("", content) self.assertIn("Test Publisher", content) self.assertIn("https://snapcraft.io/test-snap", content)