Watch 1 content
Watch 2 content
""" with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f: f.write(test_return_data) return None def set_modified_response(datastore_path): test_return_data = """ Some initial textWatch 1 content MODIFIED
Watch 2 content CHANGED
""" with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f: f.write(test_return_data) return None def test_rss_group(client, live_server, measure_memory_usage, datastore_path): """ Test that RSS feed for a specific tag/group shows only watches in that group and displays changes correctly. """ set_original_response(datastore_path=datastore_path) # Create a tag/group res = client.post( url_for("tags.form_tag_add"), data={"name": "test-rss-group"}, follow_redirects=True ) assert b"Tag added" in res.data assert b"test-rss-group" in res.data # Get the tag UUID tag_uuid = get_UUID_for_tag_name(client, name="test-rss-group") assert tag_uuid is not None # Add first watch with the tag test_url_1 = url_for('test_endpoint', _external=True) + "?watch=1" res = client.post( url_for("ui.ui_views.form_quick_watch_add"), data={"url": test_url_1, "tags": 'test-rss-group'}, follow_redirects=True ) assert b"Watch added" in res.data # Add second watch with the tag test_url_2 = url_for('test_endpoint', _external=True) + "?watch=2" res = client.post( url_for("ui.ui_views.form_quick_watch_add"), data={"url": test_url_2, "tags": 'test-rss-group'}, follow_redirects=True ) assert b"Watch added" in res.data # Add a third watch WITHOUT the tag (should not appear in RSS) test_url_3 = url_for('test_endpoint', _external=True) + "?watch=3" res = client.post( url_for("ui.ui_views.form_quick_watch_add"), data={"url": test_url_3, "tags": 'other-tag'}, follow_redirects=True ) assert b"Watch added" in res.data # Wait for initial checks to complete wait_for_all_checks(client) # Trigger a change set_modified_response(datastore_path=datastore_path) # Recheck all watches 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) assert rss_token is not None # Request RSS feed for the specific tag/group using the new endpoint res = client.get( url_for("rss.rss_tag_feed", tag_uuid=tag_uuid, token=rss_token, _external=True), follow_redirects=True ) # Verify response is successful assert res.status_code == 200 assert b"