From ff9b4fc32c46f53ba84043e6e94454b954fb3d0f Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Thu, 13 Nov 2025 20:07:03 +0100 Subject: [PATCH] Adding tests, adding group/tag name to RSS --- changedetectionio/blueprint/rss/main_feed.py | 8 + .../blueprint/rss/single_watch.py | 8 + changedetectionio/blueprint/rss/tag.py | 8 + .../tests/test_rss_single_watch.py | 245 ++++++++++++++++++ 4 files changed, 269 insertions(+) create mode 100644 changedetectionio/tests/test_rss_single_watch.py diff --git a/changedetectionio/blueprint/rss/main_feed.py b/changedetectionio/blueprint/rss/main_feed.py index bb35240a6..fec9d9f12 100644 --- a/changedetectionio/blueprint/rss/main_feed.py +++ b/changedetectionio/blueprint/rss/main_feed.py @@ -95,6 +95,14 @@ def construct_main_feed_routes(rss_blueprint, datastore): dt = dt.replace(tzinfo=pytz.UTC) fe.pubDate(dt) + # Add categories based on watch tags + for tag_uuid in watch.get('tags', []): + tag = datastore.data['settings']['application'].get('tags', {}).get(tag_uuid) + if tag: + tag_title = tag.get('title', '') + if tag_title: + fe.category(term=tag_title) + response = make_response(fg.rss_str()) response.headers.set('Content-Type', 'application/rss+xml;charset=utf-8') logger.trace(f"RSS generated in {time.time() - now:.3f}s") diff --git a/changedetectionio/blueprint/rss/single_watch.py b/changedetectionio/blueprint/rss/single_watch.py index 3e2c844e0..d47a2abbe 100644 --- a/changedetectionio/blueprint/rss/single_watch.py +++ b/changedetectionio/blueprint/rss/single_watch.py @@ -142,6 +142,14 @@ def construct_single_watch_routes(rss_blueprint, datastore): # Use the timestamp of the "to" snapshot for pubDate fe.pubDate(dt) + # Add categories based on watch tags + for tag_uuid in watch.get('tags', []): + tag = datastore.data['settings']['application'].get('tags', {}).get(tag_uuid) + if tag: + tag_title = tag.get('title', '') + if tag_title: + fe.category(term=tag_title) + except (IndexError, FileNotFoundError) as e: # Skip this diff if we can't generate it continue diff --git a/changedetectionio/blueprint/rss/tag.py b/changedetectionio/blueprint/rss/tag.py index 0d6fa311d..86aeb290a 100644 --- a/changedetectionio/blueprint/rss/tag.py +++ b/changedetectionio/blueprint/rss/tag.py @@ -81,6 +81,14 @@ def construct_tag_routes(rss_blueprint, datastore): dt = dt.replace(tzinfo=pytz.UTC) fe.pubDate(dt) + # Add categories based on watch tags + for tag_uuid in watch.get('tags', []): + tag = datastore.data['settings']['application'].get('tags', {}).get(tag_uuid) + if tag: + tag_title = tag.get('title', '') + if tag_title: + fe.category(term=tag_title) + response = make_response(fg.rss_str()) response.headers.set('Content-Type', 'application/rss+xml;charset=utf-8') return response diff --git a/changedetectionio/tests/test_rss_single_watch.py b/changedetectionio/tests/test_rss_single_watch.py new file mode 100644 index 000000000..be8f553ee --- /dev/null +++ b/changedetectionio/tests/test_rss_single_watch.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 + +import time +import os +import xml.etree.ElementTree as ET +from flask import url_for +from .util import live_server_setup, wait_for_all_checks, extract_rss_token_from_UI, extract_UUID_from_client, delete_all_watches + + +def test_rss_single_watch_order(client, live_server, measure_memory_usage, datastore_path): + """ + Test that single watch RSS feed shows changes in correct order (newest first). + """ + + # Create initial content + def set_response(datastore_path, version): + test_return_data = f""" + +

Version {version} content

+ + + """ + with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f: + f.write(test_return_data) + + # Start with version 1 + set_response(datastore_path, 1) + + # Add a watch + test_url = url_for('test_endpoint', _external=True) + "?order_test=1" + res = client.post( + url_for("ui.ui_views.form_quick_watch_add"), + data={"url": test_url, "tags": 'test-tag'}, + follow_redirects=True + ) + assert b"Watch added" in res.data + + # Get the watch UUID + watch_uuid = extract_UUID_from_client(client) + + # Wait for initial check + wait_for_all_checks(client) + + # Create multiple versions by triggering changes + for version in range(2, 6): # Create versions 2, 3, 4, 5 + set_response(datastore_path, version) + res = client.get(url_for("ui.form_watch_checknow"), follow_redirects=True) + wait_for_all_checks(client) + time.sleep(0.5) # Small delay to ensure different timestamps + + # Get RSS token + rss_token = extract_rss_token_from_UI(client) + + # Request RSS feed for the single watch + res = client.get( + url_for("rss.rss_single_watch", uuid=watch_uuid, token=rss_token, _external=True), + follow_redirects=True + ) + + # Should return valid RSS + assert res.status_code == 200 + assert b"= 3, f"Expected at least 3 items, got {len(items)}" + + # Get the descriptions/content from first 3 items + descriptions = [] + for item in items[:3]: + # Try RSS format first + desc = item.findtext('description') + if not desc: + # Try Atom format + content_elem = item.find('{http://www.w3.org/2005/Atom}content') + if content_elem is not None: + desc = content_elem.text + descriptions.append(desc if desc else "") + + print(f"First item content: {descriptions[0][:100] if descriptions[0] else 'None'}") + print(f"Second item content: {descriptions[1][:100] if descriptions[1] else 'None'}") + print(f"Third item content: {descriptions[2][:100] if descriptions[2] else 'None'}") + + # The FIRST item should contain the NEWEST change (Version 5) + # The SECOND item should contain Version 4 + # The THIRD item should contain Version 3 + assert b"Version 5" in descriptions[0].encode() or "Version 5" in descriptions[0], \ + f"First item should show newest change (Version 5), but got: {descriptions[0][:200]}" + + # Verify the order is correct + assert b"Version 4" in descriptions[1].encode() or "Version 4" in descriptions[1], \ + f"Second item should show Version 4, but got: {descriptions[1][:200]}" + + assert b"Version 3" in descriptions[2].encode() or "Version 3" in descriptions[2], \ + f"Third item should show Version 3, but got: {descriptions[2][:200]}" + + # Clean up + delete_all_watches(client) + + +def test_rss_categories_from_tags(client, live_server, measure_memory_usage, datastore_path): + """ + Test that RSS feeds include category tags from watch tags. + """ + + # Create initial content + test_return_data = """ + +

Test content for RSS categories

+ + + """ + with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f: + f.write(test_return_data) + + # Create some tags first + res = client.post( + url_for("tags.form_tag_add"), + data={"name": "Security"}, + follow_redirects=True + ) + + res = client.post( + url_for("tags.form_tag_add"), + data={"name": "Python"}, + follow_redirects=True + ) + + res = client.post( + url_for("tags.form_tag_add"), + data={"name": "Tech News"}, + follow_redirects=True + ) + + # Add a watch with tags + test_url = url_for('test_endpoint', _external=True) + "?category_test=1" + res = client.post( + url_for("ui.ui_views.form_quick_watch_add"), + data={"url": test_url, "tags": "Security, Python, Tech News"}, + follow_redirects=True + ) + assert b"Watch added" in res.data + + # Get the watch UUID + watch_uuid = extract_UUID_from_client(client) + + # Wait for initial check + wait_for_all_checks(client) + + # Trigger one change + test_return_data_v2 = """ + +

Updated content for RSS categories

+ + + """ + with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f: + f.write(test_return_data_v2) + + res = client.get(url_for("ui.form_watch_checknow"), follow_redirects=True) + wait_for_all_checks(client) + + # Get RSS token + rss_token = extract_rss_token_from_UI(client) + + # Test 1: Check single watch RSS feed + res = client.get( + url_for("rss.rss_single_watch", uuid=watch_uuid, token=rss_token, _external=True), + follow_redirects=True + ) + assert res.status_code == 200 + assert b"= 1, "Expected at least 1 item in RSS feed" + + # Get categories from first item + categories = [cat.text for cat in items[0].findall('category')] + + print(f"Found categories in single watch RSS: {categories}") + + # Should have all three categories + assert "Security" in categories, f"Expected 'Security' category, got: {categories}" + assert "Python" in categories, f"Expected 'Python' category, got: {categories}" + assert "Tech News" in categories, f"Expected 'Tech News' category, got: {categories}" + assert len(categories) == 3, f"Expected 3 categories, got {len(categories)}: {categories}" + + # Test 2: Check main RSS feed + res = client.get( + url_for("rss.feed", token=rss_token, _external=True), + follow_redirects=True + ) + assert res.status_code == 200 + + root = ET.fromstring(res.data) + items = root.findall('.//item') + assert len(items) >= 1, "Expected at least 1 item in main RSS feed" + + # Get categories from first item in main feed + categories = [cat.text for cat in items[0].findall('category')] + + print(f"Found categories in main RSS feed: {categories}") + + # Should have all three categories + assert "Security" in categories, f"Expected 'Security' category in main feed, got: {categories}" + assert "Python" in categories, f"Expected 'Python' category in main feed, got: {categories}" + assert "Tech News" in categories, f"Expected 'Tech News' category in main feed, got: {categories}" + + # Test 3: Check tag-specific RSS feed (should also have categories) + # Get the tag UUID for "Security" and verify the tag feed also has categories + from .util import get_UUID_for_tag_name + security_tag_uuid = get_UUID_for_tag_name(client, name="Security") + + if security_tag_uuid: + res = client.get( + url_for("rss.rss_tag_feed", tag_uuid=security_tag_uuid, token=rss_token, _external=True), + follow_redirects=True + ) + assert res.status_code == 200 + + root = ET.fromstring(res.data) + items = root.findall('.//item') + + if len(items) >= 1: + categories = [cat.text for cat in items[0].findall('category')] + print(f"Found categories in tag RSS feed: {categories}") + + # Should still have all three categories + assert "Security" in categories, f"Expected 'Security' category in tag feed, got: {categories}" + assert "Python" in categories, f"Expected 'Python' category in tag feed, got: {categories}" + assert "Tech News" in categories, f"Expected 'Tech News' category in tag feed, got: {categories}" + + # Clean up + delete_all_watches(client)