Compare commits

...
Author SHA1 Message Date
dgtlmoon 041ddb0e95 API - DELETE for Watch history #4397 2026-09-09 15:07:32 +02:00
dgtlmoon 0f4b556af0 0.60.3
Build and push containers / metadata (push) Canceled after 0s
Build and push containers / build-push-containers (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Build distribution 📦 (push) Canceled after 0s
ChangeDetection.io App Test / lint-code (push) Canceled after 0s
ChangeDetection.io App Test / lint-translations (push) Canceled after 0s
ChangeDetection.io App Test / lint-template-i18n (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Test the built package works basically. (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Publish Python 🐍 distribution 📦 to PyPI (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-11 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-12 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-13 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-14 (push) Canceled after 0s
2026-09-04 14:12:00 +02:00
dgtlmoon 7a3bc2ab9e GHSA-56fq-63vj-9992 Add-watch-UI should be POST/CSRF protected @AmerMrkaljevic (#4375) 2026-09-04 12:39:36 +02:00
dgtlmoon 3a71777499 UI - Fixing lots of actions that should be a POST style action which led to a 404
Also solves some of GHSA-56fq-63vj-9992 Add-watch-UI should be POST/CSRF protected
2026-09-04 12:38:20 +02:00
dgtlmoon e0fb224d41 UI - Fixing lots of actions that should be a POST style action which led to a 404 (#4370) 2026-09-04 10:56:23 +02:00
dgtlmoon 050172129e Translations - support for translation overlays/local translations of existing strings (#4365)
Build and push containers / metadata (push) Canceled after 0s
Build and push containers / build-push-containers (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Build distribution 📦 (push) Canceled after 0s
ChangeDetection.io Container Build Test / Build linux/amd64 (alpine) (push) Canceled after 0s
ChangeDetection.io Container Build Test / Build linux/arm64 (alpine) (push) Canceled after 0s
ChangeDetection.io Container Build Test / Build linux/amd64 (main) (push) Canceled after 0s
ChangeDetection.io Container Build Test / Build linux/arm/v7 (main) (push) Canceled after 0s
ChangeDetection.io Container Build Test / Build linux/arm/v8 (main) (push) Canceled after 0s
ChangeDetection.io Container Build Test / Build linux/arm64 (main) (push) Canceled after 0s
ChangeDetection.io App Test / lint-code (push) Canceled after 0s
ChangeDetection.io App Test / lint-translations (push) Canceled after 0s
ChangeDetection.io App Test / lint-template-i18n (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Test the built package works basically. (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Publish Python 🐍 distribution 📦 to PyPI (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-11 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-12 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-13 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-14 (push) Canceled after 0s
* Translations - support for translation overlays/local translations of existing strings

* Dont forget translations_overlay for pip
2026-09-03 11:15:46 +02:00
dgtlmoon e573837789 0.60.2 2026-09-03 09:15:35 +02:00
Andrew PeabodyandAndrew Peabody 3d6de42924 fix(conditions): support zero values in condition filtering and json logic conversion (#4362)
* fix(conditions): support zero values in condition filtering and json logic conversion

- Fix filter_complete_rules dropping rules where value is 0/0.0 due to 0 == False in Python
- Fix convert_to_jsonlogic raising EmptyConditionRuleRowNotUsable on value=0 due to truthiness check
- Fix str != 'None' type comparison typo in convert_to_jsonlogic
- Add comprehensive unit tests covering zero value condition filtering, conversion, and execution

* chore: re-trigger CI checks

---------

Co-authored-by: Andrew Peabody <apeabody@users.noreply.github.com>
2026-09-03 08:57:34 +02:00
dgtlmoonandClaude Opus 5 fbd9472218 fix(api): accept an existing tag UUID in the watch tag field, and deprecate it (#4361)
Build and push containers / metadata (push) Canceled after 0s
Build and push containers / build-push-containers (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Build distribution 📦 (push) Canceled after 0s
ChangeDetection.io App Test / lint-code (push) Canceled after 0s
ChangeDetection.io App Test / lint-translations (push) Canceled after 0s
ChangeDetection.io App Test / lint-template-i18n (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Test the built package works basically. (push) Canceled after 0s
Publish Python 🐍distribution 📦 to PyPI and TestPyPI / Publish Python 🐍 distribution 📦 to PyPI (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-11 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-12 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-13 (push) Canceled after 0s
ChangeDetection.io App Test / test-application-3-14 (push) Canceled after 0s
`tag` on POST /watch was documented as taking a tag UUID, but the value went to
add_tag(title): a UUID silently created a junk tag *titled* with that UUID and never
applied the tag the caller asked for. `tags` (UUIDs) was the only thing that worked.

- `tag=` now resolves an existing tag UUID to that tag, still falling back to title
  matching/creation for names. A UUID-shaped value matching nothing is skipped with a
  warning rather than becoming a group named after a UUID.
- Blank tokens ("One,,Two,") no longer store False in watch['tags'] - add_tag() returns
  False for an empty title and it was appended unguarded. Consumers tolerate it
  (get_all_tags_for_watch() dictfilt()s over known tags) but it is not valid data.
- add_tag()'s title search is extracted to tag_uuid_for_title(), so existence can be
  tested without creating as a side effect. add_tag()'s contract is unchanged.
- api-spec: `tag` is marked `deprecated: true` (so Redoc renders the badge) and states
  plainly that it takes names, not UUIDs. `tags` now says what it really does - applied
  verbatim, never creates, unknown UUIDs stored as dangling refs. Rendered docs rebuilt.

Every claim in the new field docs is asserted in test_api_tags.py against the real API.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-02 16:46:27 +02:00
dgtlmoonandClaude Opus 5 2669a5cee6 Feature/Watch limit - Adding test (#4360)
* Env var - PAGE_WATCH_LIMIT enhancements

* Rebuild API docs

* Bump APi doc version

* test: cover PAGE_WATCH_LIMIT across every add path

- API create returns 429; API import returns 429 and refuses the batch whole
- quick-add and the UI importer flash the limit (importer once per file, not per row)
  and hand unimported URLs back
- clone at the limit no longer KeyErrors
- an instance already over the limit still loads from disk and stays editable, only
  new watches are refused
- the Info tab shows the limit only when one is set
- add_watch() with no request context returns None instead of raising from flash()
- an absent, empty or unparseable env var all mean unlimited

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-02 15:43:59 +02:00
74 changed files with 2050 additions and 359 deletions
+13
View File
@@ -52,6 +52,19 @@ jobs:
git diff --stat changedetectionio/translations
exit 1
fi
- name: Check translation overlay
# Deliberately after extract_messages above, so overrides are validated against a freshly
# extracted messages.pot. An overlay entry is keyed on the exact upstream msgid: when a
# string is reworded upstream the override stops matching and silently reverts to upstream
# wording. This is the only thing that makes that visible.
# See changedetectionio/translations_overlay/README.md
if: hashFiles('changedetectionio/translations_overlay/**/*.po') != ''
run: |
find changedetectionio/translations_overlay -name "*.po" | while read f; do
echo "Checking $f"
msgfmt --check-format -o /dev/null "$f"
done
python changedetectionio/translations_overlay/manage.py check
lint-template-i18n:
runs-on: ubuntu-latest
+1
View File
@@ -13,6 +13,7 @@ recursive-include changedetectionio/store *
recursive-include changedetectionio/templates *
recursive-include changedetectionio/tests *
recursive-include changedetectionio/translations *
recursive-include changedetectionio/translations_overlay *
recursive-include changedetectionio/widgets *
prune changedetectionio/static/package-lock.json
prune changedetectionio/static/styles/node_modules
+1 -1
View File
@@ -2,7 +2,7 @@
# Read more https://github.com/dgtlmoon/changedetection.io/wiki
# Semver means never use .01, or 00. Should be .1.
__version__ = '0.55.8'
__version__ = '0.60.3'
from changedetectionio.strtobool import strtobool
from json.decoder import JSONDecodeError
+14
View File
@@ -293,6 +293,20 @@ class WatchHistory(Resource):
abort(404, message='No watch exists with the UUID of {}'.format(uuid))
return watch.history, 200
# Delete all history/snapshots for a watch, but keep the watch itself
# curl -X DELETE http://localhost:5000/api/v1/watch/<uuid_str:uuid>/history
@auth.check_token
@validate_openapi_request('deleteWatchHistory')
def delete(self, uuid):
"""Clear all snapshot history for a watch (the watch itself is kept)."""
if not self.datastore.data['watching'].get(uuid):
abort(404, message='No watch exists with the UUID of {}'.format(uuid))
# Same call as the UI "Clear history" button - wipes snapshots/screenshots and
# resets last_checked etc, while preserving the watch and its processor config
self.datastore.clear_watch_history(uuid)
return 'OK', 204
class WatchSingleHistory(Resource):
def __init__(self, **kwargs):
@@ -42,7 +42,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
system_default_browser=browser_config.system_default_description(datastore),
)
@add_watch_ui_blueprint.route("/snapshot", methods=['GET'])
@add_watch_ui_blueprint.route("/snapshot", methods=['POST'])
@login_optionally_required
def add_watch_ui_snapshot():
"""One-shot live fetch of an arbitrary URL for the Add Watch visual selector.
@@ -52,6 +52,10 @@ def construct_blueprint(datastore: ChangeDetectionStore):
connect, "Goto site", grab the screenshot + xpath element data, then tear
the browser down again. Element selection then happens client-side on the
returned data, exactly like the watch Edit page's visual selector.
POST-only and CSRF protected on purpose: this drives a real browser fetch and
writes a temporary watch dir, so as a GET it could be triggered cross-origin
(or by any tag/link that issues a GET) without the operator's consent.
"""
import base64
from changedetectionio.blueprint.browser_steps import (
@@ -71,7 +75,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# backslash/parser-differential rejection of GHSA-rph4-96w6-q594 (GHSA-56fq-63vj-9992).
# Note this fetch never reaches difference_detection_processor.call_browser(), so it gets
# no gating from there - it has to validate for itself.
url = (request.args.get('url') or '').strip()
url = (request.form.get('url') or '').strip()
ok, reason = is_fetch_url_allowed(url)
if not ok:
logger.warning(f"Add-watch snapshot: refused '{url}' - {reason}")
@@ -82,7 +86,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
# Either way it has to be able to render a preview - the plain HTTP client
# produces no screenshot and no element data, so previewing with it is pointless
# (and it used to be the silent default here, see the system-default bug).
fetcher_name = (request.args.get('fetch_backend') or '').strip() or browser_config.default_visual_browser(datastore)
fetcher_name = (request.form.get('fetch_backend') or '').strip() or browser_config.default_visual_browser(datastore)
if not fetcher_name or not browser_config.is_visual_capable(fetcher_name, datastore):
logger.warning(f"Add-watch snapshot: refused browser '{fetcher_name}' for '{url}'")
return make_response('No interactive browser available that can render a live preview '
@@ -59,9 +59,18 @@ $(document).ready(() => {
$.ajax({
url: add_watch_snapshot_url,
// POST, never GET - this makes the server-side browser fetch a URL of our
// choosing, so it must not be triggerable cross-origin. csrf.js adds the
// X-CSRFToken header to every non-GET ajax call; the CSRF field on the form
// is sent too so it works even if that handler hasn't run yet.
method: 'POST',
// Preview with the browser picked in the list - that same browser is what
// gets saved on the watch, so what you see here is what it will check with.
data: {url: url, fetch_backend: $('input[name="fetch_backend"]:checked').val() || ''},
data: {
url: url,
fetch_backend: $('input[name="fetch_backend"]:checked').val() || '',
csrf_token: $('#new-watch-form input[name="csrf_token"]').val() || '',
},
dataType: 'json',
}).done((data) => {
showState('ready');
@@ -98,7 +98,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
backups_blueprint.register_blueprint(construct_restore_blueprint(datastore))
backup_threads = []
@backups_blueprint.route("/request-backup", methods=['GET'])
@backups_blueprint.route("/request-backup", methods=['POST'])
@login_optionally_required
def request_backup():
if any(thread.is_alive() for thread in backup_threads):
@@ -35,8 +35,10 @@
</p>
{% endif %}
<a class="pure-button pure-button-primary"
href="{{ url_for('backups.request_backup') }}">{{ _('Create backup') }}</a>
<form method="POST" action="{{ url_for('backups.request_backup') }}" style="display: inline;">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="pure-button pure-button-primary">{{ _('Create backup') }}</button>
</form>
{% if available_backups %}
{# POST + CSRF token: this permanently deletes every backup archive, so it must
not be reachable from a bare GET (an <img src=...> on any page the operator
@@ -296,7 +296,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
return browsersteps_start_session
@browser_steps_blueprint.route("/browsersteps_start_session", methods=['GET'])
@browser_steps_blueprint.route("/browsersteps_start_session", methods=['POST'])
@login_optionally_required
def browsersteps_start_session():
# A new session was requested, return sessionID
@@ -100,7 +100,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
results = _recalc_check_status(uuid=uuid)
return results
@check_proxies_blueprint.route("/<uuid_str:uuid>/start", methods=['GET'])
@check_proxies_blueprint.route("/<uuid_str:uuid>/start", methods=['POST'])
@login_optionally_required
def start_check(uuid):
@@ -14,7 +14,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q: PriorityQueue
price_data_follower_blueprint = Blueprint('price_data_follower', __name__)
@price_data_follower_blueprint.route("/<uuid_str:uuid>/accept", methods=['GET'])
@price_data_follower_blueprint.route("/<uuid_str:uuid>/accept", methods=['POST'])
@login_optionally_required
def accept(uuid):
datastore.data['watching'][uuid]['track_ldjson_price_data'] = PRICE_DATA_TRACK_ACCEPT
@@ -24,7 +24,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q: PriorityQueue
worker_pool.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid}))
return redirect(url_for("watchlist.index"))
@price_data_follower_blueprint.route("/<uuid_str:uuid>/reject", methods=['GET'])
@price_data_follower_blueprint.route("/<uuid_str:uuid>/reject", methods=['POST'])
@login_optionally_required
def reject(uuid):
datastore.data['watching'][uuid]['track_ldjson_price_data'] = PRICE_DATA_TRACK_REJECT
@@ -278,7 +278,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
return output
@settings_blueprint.route("/reset-api-key", methods=['GET'])
@settings_blueprint.route("/reset-api-key", methods=['POST'])
@login_optionally_required
def settings_reset_api_key():
secret = secrets.token_hex(16)
@@ -295,7 +295,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
logs=notification_debug_log if len(notification_debug_log) else ["Notification logs are empty - no notifications sent yet."])
return output
@settings_blueprint.route("/toggle-all-paused", methods=['GET'])
@settings_blueprint.route("/toggle-all-paused", methods=['POST'])
@login_optionally_required
def toggle_all_paused():
current_state = datastore.data['settings']['application'].get('all_paused', False)
@@ -309,7 +309,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
return redirect(url_for('watchlist.index'))
@settings_blueprint.route("/toggle-all-muted", methods=['GET'])
@settings_blueprint.route("/toggle-all-muted", methods=['POST'])
@login_optionally_required
def toggle_all_muted():
current_state = datastore.data['settings']['application'].get('all_muted', False)
+1 -1
View File
@@ -130,7 +130,7 @@ def construct_llm_blueprint(datastore: ChangeDetectionStore):
logger.exception("LLM model list full traceback:")
return jsonify({'models': [], 'error': str(e)}), 400
@llm_blueprint.route("/test", methods=['GET'])
@llm_blueprint.route("/test", methods=['POST'])
@login_optionally_required
def llm_test():
from flask import request
@@ -208,7 +208,7 @@ nav
</div>
</div>
<div class="pure-control-group">
<a href="{{url_for('settings.settings_reset_api_key')}}" class="pure-button button-small button-cancel">{{ _('Regenerate API key') }}</a>
<button type="submit" formmethod="post" formnovalidate formaction="{{url_for('settings.settings_reset_api_key')}}" class="pure-button button-small button-cancel">{{ _('Regenerate API key') }}</button>
</div>
<div class="pure-control-group">
<h4>{{ _('Chrome Extension') }}</h4>
@@ -577,7 +577,10 @@
if (mult.trim()) params.set('local_token_multiplier', mult.trim());
try {
const resp = await fetch('{{ url_for("settings.llm.llm_test") }}?' + params);
const resp = await fetch('{{ url_for("settings.llm.llm_test") }}?' + params, {
method: 'POST',
headers: {'X-CSRFToken': csrftoken}
});
const data = await resp.json();
if (data.ok) {
result.style.cssText = 'display:block; background:rgba(39,174,96,0.08); border:1px solid rgba(39,174,96,0.3); border-radius:5px; padding:0.6em 0.85em; font-size:0.88em; line-height:1.45;';
+1 -1
View File
@@ -62,7 +62,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
return redirect(url_for('tags.tags_overview_page'))
@tags_blueprint.route("/mute/<uuid_str:uuid>", methods=['GET'])
@tags_blueprint.route("/mute/<uuid_str:uuid>", methods=['POST'])
@login_optionally_required
def mute(uuid):
tag = datastore.data['settings']['application']['tags'].get(uuid)
@@ -68,7 +68,10 @@ html[data-darkmode="true"] .watch-tag-list.tag-{{ class_name }} {
{#-{{ loop.cycle('pure-table-odd', 'pure-table-even') }}-#}
<tr id="{{ uuid }}" class="">
<td class="watch-controls">
<a class="link-mute state-{{'on' if tag.notification_muted else 'off'}}" href="{{url_for('tags.mute', uuid=tag.uuid)}}" aria-label="{{ _('Mute notifications') }}" title="{{ _('Mute notifications') }}"><i data-feather="{{ 'bell-off' if tag.notification_muted else 'bell' }}" class="icon icon-mute"></i></a>
<form method="POST" action="{{url_for('tags.mute', uuid=tag.uuid)}}" style="display: inline;">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bare-btn link-mute state-{{'on' if tag.notification_muted else 'off'}}" aria-label="{{ _('Mute notifications') }}" title="{{ _('Mute notifications') }}"><i data-feather="{{ 'bell-off' if tag.notification_muted else 'bell' }}" class="icon icon-mute"></i></button>
</form>
</td>
<td class="watch-count">{{ "{:,}".format(tag_count[uuid]) if uuid in tag_count else 0 }}</td>
<td class="title-col inline"> <a href="{{url_for('watchlist.index', tag=uuid) }}" class="watch-tag-list tag-{{ tag.title|sanitize_tag_class }}">{{ tag.title }}</a></td>
+2 -2
View File
@@ -407,7 +407,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_pool,
return redirect(url_for('watchlist.index'))
@ui_blueprint.route("/share-url/<uuid_str:uuid>", methods=['GET'])
@ui_blueprint.route("/share-url/<uuid_str:uuid>", methods=['POST'])
@login_optionally_required
def form_share_put_watch(uuid):
"""Given a watch UUID, upload the info and return a share-link
@@ -455,7 +455,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_pool,
return redirect(url_for('watchlist.index'))
@ui_blueprint.route("/language/auto-detect", methods=['GET'])
@ui_blueprint.route("/language/auto-detect", methods=['POST'])
def delete_locale_language_session_var_if_it_exists():
"""Clear the session locale preference to auto-detect from browser Accept-Language header"""
if 'locale' in session:
@@ -18,6 +18,24 @@ from changedetectionio.blueprint.watchlist.row_context import watch_row_context
def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMetaData):
watchlist_blueprint = Blueprint('watchlist', __name__, template_folder="templates")
@watchlist_blueprint.route("/toggle", methods=['POST'])
@login_optionally_required
def toggle():
op = request.args.get('op')
uuid = request.args.get('uuid')
watch = datastore.data['watching'].get(uuid)
if not watch:
flash(_('Watch not found'), 'error')
else:
if op == 'pause':
watch.toggle_pause()
elif op == 'mute':
watch.toggle_mute()
watch.commit()
return redirect(url_for('watchlist.index', tag=request.args.get('tag')))
@watchlist_blueprint.route("/", methods=['GET'])
@login_optionally_required
def index():
@@ -36,17 +54,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
if request.args.get('rss'):
return redirect(url_for('rss.feed', tag=active_tag_uuid))
op = request.args.get('op')
if op:
uuid = request.args.get('uuid')
if op == 'pause':
datastore.data['watching'][uuid].toggle_pause()
elif op == 'mute':
datastore.data['watching'][uuid].toggle_mute()
datastore.data['watching'][uuid].commit()
return redirect(url_for('watchlist.index', tag = active_tag_uuid))
# Sort by last_changed and add the uuid which is usually the key..
sorted_watches = []
active_processor = request.args.get('processor', '').strip()
@@ -35,10 +35,12 @@
<td class="inline checkbox-uuid" ><div><input name="uuids" type="checkbox" value="{{ watch.uuid}} " >{# <span class="counter-i">{{ loop.index+pagination.skip }}</span>#}</div></td>
<td class="inline watch-controls">
<div>
<a class="ajax-op state-off pause-toggle" data-op="pause" aria-label="{{ _('Pause checks') }}" title="{{ _('Pause checks') }}" href="{{url_for('watchlist.index', op='pause', uuid=watch.uuid, tag=active_tag_uuid)}}"><i data-feather="pause" class="icon icon-pause"></i></a>
<a class="ajax-op state-on pause-toggle" data-op="pause" style="display: none" aria-label="{{ _('UnPause checks') }}" title="{{ _('UnPause checks') }}" href="{{url_for('watchlist.index', op='pause', uuid=watch.uuid, tag=active_tag_uuid)}}"><i data-feather="play" class="icon icon-unpause"></i></a>
<a class="ajax-op state-off mute-toggle" data-op="mute" aria-label="{{ _('Mute notification') }}" title="{{ _('Mute notification') }}" href="{{url_for('watchlist.index', op='mute', uuid=watch.uuid, tag=active_tag_uuid)}}"><i data-feather="bell" class="icon icon-mute"></i></a>
<a class="ajax-op state-on mute-toggle" data-op="mute" style="display: none" aria-label="{{ _('UnMute notification') }}" title="{{ _('UnMute notification') }}" href="{{url_for('watchlist.index', op='mute', uuid=watch.uuid, tag=active_tag_uuid)}}"><i data-feather="bell-off" class="icon icon-mute"></i></a>
{%- set pause_action = url_for('watchlist.toggle', op='pause', uuid=watch.uuid, tag=active_tag_uuid) -%}
{%- set mute_action = url_for('watchlist.toggle', op='mute', uuid=watch.uuid, tag=active_tag_uuid) -%}
<button type="submit" formmethod="post" formaction="{{ pause_action }}" class="bare-btn ajax-op state-off pause-toggle" data-op="pause" aria-label="{{ _('Pause checks') }}" title="{{ _('Pause checks') }}"><i data-feather="pause" class="icon icon-pause"></i></button>
<button type="submit" formmethod="post" formaction="{{ pause_action }}" class="bare-btn ajax-op state-on pause-toggle" data-op="pause" style="display: none" aria-label="{{ _('UnPause checks') }}" title="{{ _('UnPause checks') }}"><i data-feather="play" class="icon icon-unpause"></i></button>
<button type="submit" formmethod="post" formaction="{{ mute_action }}" class="bare-btn ajax-op state-off mute-toggle" data-op="mute" aria-label="{{ _('Mute notification') }}" title="{{ _('Mute notification') }}"><i data-feather="bell" class="icon icon-mute"></i></button>
<button type="submit" formmethod="post" formaction="{{ mute_action }}" class="bare-btn ajax-op state-on mute-toggle" data-op="mute" style="display: none" aria-label="{{ _('UnMute notification') }}" title="{{ _('UnMute notification') }}"><i data-feather="bell-off" class="icon icon-mute"></i></button>
</div>
</td>
@@ -81,13 +83,13 @@
<div class="error-text" style="display:none;">{{ error_texts|safe }}</div>
{%- if watch['processor'] == 'text_json_diff' -%}
{%- if watch['has_ldjson_price_data'] and not watch['track_ldjson_price_data'] -%}
<div class="ldjson-price-track-offer">Switch to Restock & Price watch mode? <a href="{{url_for('price_data_follower.accept', uuid=watch.uuid)}}" class="pure-button button-xsmall">Yes</a> <a href="{{url_for('price_data_follower.reject', uuid=watch.uuid)}}" class="">No</a></div>
<div class="ldjson-price-track-offer">Switch to Restock & Price watch mode? <button type="submit" formmethod="post" formaction="{{url_for('price_data_follower.accept', uuid=watch.uuid)}}" class="pure-button button-xsmall">Yes</button> <button type="submit" formmethod="post" formaction="{{url_for('price_data_follower.reject', uuid=watch.uuid)}}" class="bare-btn bare-btn--link">No</button></div>
{%- endif -%}
{%- endif -%}
</div>
<div class="status-icons">
<a class="link-spread" href="{{url_for('ui.form_share_put_watch', uuid=watch.uuid)}}"><img src="{{url_for('static_content', group='images', filename='spread.svg')}}" class="status-icon icon icon-spread" title="{{ _('Create a link to share watch config with others') }}" ></a>
<button type="submit" formmethod="post" formaction="{{url_for('ui.form_share_put_watch', uuid=watch.uuid)}}" class="bare-btn link-spread"><img src="{{url_for('static_content', group='images', filename='spread.svg')}}" class="status-icon icon icon-spread" title="{{ _('Create a link to share watch config with others') }}" ></button>
{%- set effective_fetcher = watch.get_fetch_backend if watch.get_fetch_backend != "system" else system_default_fetcher -%}
{%- if effective_fetcher and ("html_webdriver" in effective_fetcher or "html_" in effective_fetcher or "extra_browser_" in effective_fetcher) -%}
{{ effective_fetcher|fetcher_status_icons }}
@@ -163,7 +165,9 @@
<div>
{%- set target_attr = ' target="' ~ watch.uuid ~ '"' if datastore.data['settings']['application']['ui'].get('open_diff_in_new_tab') else '' -%}
<a href="" class="already-in-queue-button recheck cdio-btn cdio-btn--primary cdio-btn--sm" style="display: none;" disabled="disabled"><i data-feather="clock"></i>{{ _('Queued') }}</a>
<a href="{{ url_for('ui.form_watch_checknow', uuid=watch.uuid, tag=request.args.get('tag')) }}" data-op='recheck' class="ajax-op recheck cdio-btn cdio-btn--primary cdio-btn--sm"><i data-feather="refresh-cw"></i>{{ _('Recheck') }}</a>
<button type="submit" formmethod="post"
formaction="{{ url_for('ui.form_watch_checknow', uuid=watch.uuid, tag=request.args.get('tag')) }}"
data-op='recheck' class="ajax-op recheck cdio-btn cdio-btn--primary cdio-btn--sm"><i data-feather="refresh-cw"></i>{{ _('Recheck') }}</button>
<a href="{{ url_for('ui.ui_edit.edit_page', uuid=watch.uuid, tag=active_tag_uuid)}}#general" class="cdio-btn cdio-btn--primary cdio-btn--sm">{{ _('Edit') }}</a>
<a href="{{ url_for('ui.ui_diff.diff_history_page', uuid=watch.uuid)}}" {{target_attr}} class="cdio-btn cdio-btn--primary cdio-btn--sm history-link ai-history-btn" style="display: none;" data-uuid="{{ watch.uuid }}" data-summary-url="{{ url_for('ui.ui_diff.diff_llm_summary', uuid=watch.uuid) }}" data-processor-data-url="{{ url_for('ui.ui_diff.diff_history_page_processor_data', uuid=watch.uuid) }}"><span class="btn-label-history">{{ _('History') }}</span><span class="btn-label-summary">&#x2728; {{ _('Summary') }}</span></a>
<a href="{{ url_for('ui.ui_preview.preview_page', uuid=watch.uuid)}}" {{target_attr}} class="cdio-btn cdio-btn--primary cdio-btn--sm preview-link" style="display: none;">{{ _('Preview') }}</a>
+17 -3
View File
@@ -34,7 +34,13 @@ CUSTOM_OPERATIONS = {
def filter_complete_rules(ruleset):
rules = [
rule for rule in ruleset
if all(value not in ("", False, "None", None) for value in [rule["operator"], rule["field"], rule["value"]])
if all(
rule.get(k) is not None
and rule.get(k) is not False
and rule.get(k) != ""
and rule.get(k) != "None"
for k in ("operator", "field", "value")
)
]
return rules
@@ -54,12 +60,20 @@ def convert_to_jsonlogic(logic_operator: str, rule_dict: list):
field = condition["field"]
value = condition["value"]
if not operator or operator == 'None' or not value or not field:
if (
not operator
or operator == 'None'
or not field
or value is None
or value is False
or value == ""
or value == "None"
):
raise EmptyConditionRuleRowNotUsable()
# Convert value to int/float if possible
try:
if isinstance(value, str) and "." in value and str != "None":
if isinstance(value, str) and "." in value and value != "None":
value = float(value)
else:
value = int(value)
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -284,7 +284,7 @@ $(document).ready(function () {
$('#browser-steps-ui .loader .spinner').show();
// Request a new session
$.ajax({
type: "GET",
type: "POST",
url: browser_steps_start_url,
statusCode: {
400: function () {
+1 -1
View File
@@ -73,7 +73,7 @@ $(function () {
// Request start, needs CSRF?
$.ajax({
type: "GET",
type: "POST",
url: recheck_proxy_start_url,
}).done(function (data) {
$.each(data, function (proxy_key, state) {
@@ -91,3 +91,26 @@
&:hover { color: #d68a00; border-color: #d68a00; }
}
}
// State-mutating controls have to POST, and only a <button> can submit - these strip
// the browser's button chrome so they render exactly like the <a> they replaced.
.bare-btn {
background: none;
border: 0;
padding: 0;
margin: 0;
font: inherit;
// Matches `a { color: var(--color-link) }` - these replace anchors, and the row
// icons stroke with currentColor, so `inherit` would pick up .watch-controls red.
color: var(--color-link);
cursor: pointer;
&:focus-visible {
outline: 2px solid var(--color-link);
outline-offset: 2px;
}
&--link {
text-decoration: underline;
}
}
@@ -122,7 +122,7 @@
li {
border-bottom: 1px solid var(--color-border-table-cell);
>* {
>*, >form>button {
display: block;
padding: 1rem 1.5rem;
color: var(--color-text);
@@ -134,6 +134,14 @@
background: var(--color-background-menu-link-hover);
}
}
// Buttons shrink-wrap and centre their label; anchors don't. No global
// border-box reset here, so this must not be folded into the rule above.
>form>button {
box-sizing: border-box;
width: 100%;
text-align: left;
}
&#menu-pause, &#menu-mute {
display: none;
}
@@ -18,10 +18,20 @@
flex-direction: column;
gap: 0.5rem;
padding: 0.5rem 0;
// The csrf mini-form around each option is layout-transparent, so the buttons
// stay the flex items.
> form {
display: contents;
}
}
.language-option {
display: flex;
background: none;
font: inherit;
text-align: left;
cursor: pointer;
align-items: center;
gap: 1rem;
padding: 0.25rem;
@@ -15,6 +15,13 @@
.pure-menu-item {
height: initial;
// Mini POST forms (pause/mute/log out need a csrf_token) are layout-transparent,
// so the button inside sits where the plain <a> used to.
> form {
display: contents;
}
svg {
height: 1.2rem;
}
@@ -128,6 +128,8 @@ ul#top-right-menu {
font-weight: 600;
white-space: nowrap;
text-decoration: none;
font-family: inherit;
cursor: pointer;
&:hover {
background: rgba(255, 255, 255, 0.18);
@@ -205,7 +205,7 @@ body.watch-selection-active #checkbox-operations {
}
&.queued {
a.recheck {
.recheck {
display: none !important;
}
@@ -216,7 +216,7 @@ body.watch-selection-active #checkbox-operations {
}
&.paused {
a.pause-toggle {
.pause-toggle {
&.state-on {
display: inline !important;
}
@@ -228,7 +228,7 @@ body.watch-selection-active #checkbox-operations {
}
&.notification_muted {
a.mute-toggle {
.mute-toggle {
&.state-on {
display: inline !important;
}
File diff suppressed because one or more lines are too long
+44 -9
View File
@@ -49,6 +49,9 @@ dictfilt = lambda x, y: dict([(i, x[i]) for i in x if i in set(y)])
# Is there an existing library to ensure some data store (JSON etc) is in sync with CRUD methods?
# Open a github issue if you know something :)
# https://stackoverflow.com/questions/6190468/how-to-trigger-function-on-value-change
_TAG_UUID_RE = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE)
class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore):
__version_check = True
@@ -791,11 +794,31 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore):
return None
if tag and type(tag) == str:
# Then it's probably a string of the actual tag by name, split and add it
for t in tag.split(','):
# for each stripped tag, add tag as UUID
for a_t in t.split(','):
tag_uuid = self.add_tag(a_t)
# A comma separated string of tag *titles*, created when they don't exist yet.
# An existing tag's UUID is accepted here too: the API documented this field as taking
# a UUID for years, and honouring that beats creating a tag *titled* with the UUID.
existing_tag_uuids = self.__data['settings']['application'].get('tags', {})
for tag_name in tag.split(','):
tag_name = tag_name.strip()
if not tag_name:
continue
if _TAG_UUID_RE.match(tag_name):
if tag_name in existing_tag_uuids:
apply_extras['tags'].append(tag_name)
continue
# UUID-shaped but no such tag, and no tag literally titled that either -
# a stale or foreign ID. Skip it rather than leave behind a group named
# after a UUID, which is never what the caller wanted.
if not self.tag_uuid_for_title(tag_name):
logger.warning(f"Tag '{tag_name}' looks like a UUID but no such tag exists, skipping")
continue
tag_uuid = self.add_tag(tag_name)
# add_tag() returns False for a title it won't create - never let that into the list,
# a falsy entry blows up every lookup of watch['tags']
if tag_uuid:
apply_extras['tags'].append(tag_uuid)
# Or if UUIDs given directly
@@ -1069,6 +1092,18 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore):
return ret
def tag_uuid_for_title(self, title):
"""UUID of the tag with this title (case/space insensitive), or None. Creates nothing."""
n = title.strip().lower()
if not n:
return None
for uuid, tag in self.__data['settings']['application'].get('tags', {}).items():
if n == tag.get('title', '').lower().strip():
return uuid
return None
def add_tag(self, title):
# If name exists, return that
n = title.strip().lower()
@@ -1076,10 +1111,10 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore):
if not n:
return False
for uuid, tag in self.__data['settings']['application'].get('tags', {}).items():
if n == tag.get('title', '').lower().strip():
logger.warning(f"Tag '{title}' already exists, skipping creation.")
return uuid
existing_uuid = self.tag_uuid_for_title(title)
if existing_uuid:
logger.warning(f"Tag '{title}' already exists, skipping creation.")
return existing_uuid
# Eventually almost everything todo with a watch will apply as a Tag
# So we use the same model as a Watch
+10 -4
View File
@@ -271,13 +271,19 @@
<div class="modal-body">
<div class="language-list">
{% for locale, lang_data in available_languages.items()|sort %}
<a href="{{ url_for('set_language', locale=locale, redirect=request.path) }}" class="language-option" data-locale="{{ locale }}">
<span class="lang-option {{ lang_data.flag }}"></span> <span class="language-name">{{ lang_data.name }}</span>
</a>
<form method="POST" action="{{ url_for('set_language', locale=locale, redirect=request.path) }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="language-option" data-locale="{{ locale }}">
<span class="lang-option {{ lang_data.flag }}"></span> <span class="language-name">{{ lang_data.name }}</span>
</button>
</form>
{% endfor %}
</div>
<div>
<a href="{{ url_for('ui.delete_locale_language_session_var_if_it_exists', redirect=request.path) }}" >{{ _('Auto-detect from browser') }}</a>
<form method="POST" action="{{ url_for('ui.delete_locale_language_session_var_if_it_exists', redirect=request.path) }}" style="display: inline;">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bare-btn">{{ _('Auto-detect from browser') }}</button>
</form>
</div>
<div>
{{ _('Language support is in beta, please help us improve by opening a PR on GitHub with any updates.') }}
+12 -3
View File
@@ -5,14 +5,23 @@
{% if current_user.is_authenticated or not has_password %}
{% if not current_diff_url %}
<li class="pure-menu-item" id="menu-pause">
<a class="status-pill {{ 'paused' if all_paused }}" href="{{ url_for('settings.toggle_all_paused') }}" aria-label="{% if all_paused %}{{ _('Resume automatic scheduling') }}{% else %}{{ _('Pause auto-queue scheduling of watches') }}{% endif %}" title="{% if all_paused %}{{ _('Scheduling paused — click to resume') }}{% else %}{{ _('Scheduling active — click to pause all') }}{% endif %}"><span class="live-dot"></span>{% if all_paused %}{{ _('Paused') }}{% else %}{{ _('Running') }}{% endif %}</a>
<form method="POST" action="{{ url_for('settings.toggle_all_paused') }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="status-pill {{ 'paused' if all_paused }}" aria-label="{% if all_paused %}{{ _('Resume automatic scheduling') }}{% else %}{{ _('Pause auto-queue scheduling of watches') }}{% endif %}" title="{% if all_paused %}{{ _('Scheduling paused — click to resume') }}{% else %}{{ _('Scheduling active — click to pause all') }}{% endif %}"><span class="live-dot"></span>{% if all_paused %}{{ _('Paused') }}{% else %}{{ _('Running') }}{% endif %}</button>
</form>
</li>
<li class="pure-menu-item " id="menu-mute">
<a class="status-pill {{ 'muted' if all_muted }}" href="{{ url_for('settings.toggle_all_muted') }}" aria-label="{% if all_muted %}{{ _('Unmute notifications') }}{% else %}{{ _('Mute notifications') }}{% endif %}" title="{% if all_muted %}{{ _('Notifications are muted - click to unmute') }}{% else %}{{ _('Mute notifications') }}{% endif %}"><i data-feather="{{ 'bell-off' if all_muted else 'bell' }}" class="action-icon"></i>{% if all_muted %}{{ _('Muted') }}{% else %}{{ _('Alerts on') }}{% endif %}</a>
<form method="POST" action="{{ url_for('settings.toggle_all_muted') }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="status-pill {{ 'muted' if all_muted }}" aria-label="{% if all_muted %}{{ _('Unmute notifications') }}{% else %}{{ _('Mute notifications') }}{% endif %}" title="{% if all_muted %}{{ _('Notifications are muted - click to unmute') }}{% else %}{{ _('Mute notifications') }}{% endif %}"><i data-feather="{{ 'bell-off' if all_muted else 'bell' }}" class="action-icon"></i>{% if all_muted %}{{ _('Muted') }}{% else %}{{ _('Alerts on') }}{% endif %}</button>
</form>
</li>
{%- if current_user.is_authenticated -%}
<li class="pure-menu-item menu-collapsible">
<a href="{{ url_for('logout', redirect=request.path) }}" ><i data-feather="log-out" class="action-icon"></i>&nbsp;{{ _('Log out') }}</a>
<form method="POST" action="{{ url_for('logout', redirect=request.path) }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bare-btn"><i data-feather="log-out" class="action-icon"></i>&nbsp;{{ _('Log out') }}</button>
</form>
</li>
{%- endif -%}
@@ -84,10 +84,12 @@ def test_socks5(client, live_server, measure_memory_usage, datastore_path):
# PROXY CHECKER WIDGET CHECK - this needs more checking
uuid = next(iter(live_server.app.config['DATASTORE'].data['watching']))
res = client.get(
# POST only - it kicks off real fetches through every configured proxy
res = client.post(
url_for("check_proxies.start_check", uuid=uuid),
follow_redirects=True
)
assert res.status_code == 200
# It's probably already finished super fast :(
#assert b"RUNNING" in res.data
@@ -125,7 +125,7 @@ def test_check_access_control(app, client, live_server, measure_memory_usage, da
follow_redirects=True
)
res = c.get(url_for("logout"),
res = c.post(url_for("logout"),
follow_redirects=True)
assert b"Login" in res.data
@@ -79,26 +79,41 @@ def test_snapshot_refuses_browser_that_cannot_preview(client, live_server, measu
from changedetectionio.blueprint.add_watch_ui import browser_config
monkeypatch.setattr(browser_config, 'is_visual_capable', lambda name, datastore: False)
snapshot_url = url_for('add_watch_ui.add_watch_ui_snapshot')
# Nothing capable, and no explicit browser asked for -> nothing to preview with
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com'))
res = client.post(snapshot_url, data={'url': 'https://example.com'})
assert res.status_code == 400
assert b'No interactive browser' in res.data
# Explicitly asking for a browser that can't preview is refused just the same
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com',
fetch_backend='html_requests'))
res = client.post(snapshot_url, data={'url': 'https://example.com',
'fetch_backend': 'html_requests'})
assert res.status_code == 400
# A made-up name never resolves to a capable fetcher either (real capability lookup here)
monkeypatch.undo()
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com',
fetch_backend='../../etc/passwd'))
res = client.post(snapshot_url, data={'url': 'https://example.com',
'fetch_backend': '../../etc/passwd'})
assert res.status_code == 400
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot', url='https://example.com',
fetch_backend='os'))
res = client.post(snapshot_url, data={'url': 'https://example.com',
'fetch_backend': 'os'})
assert res.status_code == 400
def test_snapshot_is_post_only(client, live_server, measure_memory_usage, datastore_path):
"""A GET must not reach the endpoint at all.
/snapshot drives a real server-side browser fetch and hands the rendered result back in
the response (GHSA-56fq-63vj-9992). As a GET that is reachable by anything that can make
the operator's browser issue a request - an <img>/<iframe>/link from another site - with
no CSRF token in play. POST-only + CSRFProtect means only our own page can trigger it.
"""
# Method mismatch surfaces as 404 here rather than 405
res = client.get(url_for('add_watch_ui.add_watch_ui_snapshot') + '?url=https://example.com')
assert res.status_code in (404, 405)
def test_submit_rejects_unknown_fetcher(client, live_server, measure_memory_usage, datastore_path):
"""A posted browser is checked server side, so a doctored form can't pin a junk fetcher."""
datastore = _datastore(client)
+78
View File
@@ -338,6 +338,84 @@ def test_api_simple(client, live_server, measure_memory_usage, datastore_path):
)
assert len(res.json) == 0, "Watch list should be empty"
def test_api_delete_watch_history(client, live_server, measure_memory_usage, datastore_path):
"""DELETE /api/v1/watch/<uuid>/history should wipe the snapshots but keep the watch (#4397)"""
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
set_original_response(datastore_path=datastore_path)
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": test_url}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
follow_redirects=True
)
assert res.status_code == 201
watch_uuid = res.json.get('uuid')
wait_for_all_checks(client)
# A second snapshot so we know we're clearing more than one
set_modified_response(datastore_path=datastore_path)
client.get(url_for("watch", uuid=watch_uuid, recheck='1'), headers={'x-api-key': api_key})
wait_for_all_checks(client)
res = client.get(
url_for("watchhistory", uuid=watch_uuid),
headers={'x-api-key': api_key},
)
assert len(res.json) == 2, "Should have two history entries before clearing"
# Unknown watch UUID should 404 and not blow up
res = client.delete(
url_for("watchhistory", uuid='4d8b5b4a-8e0b-4d4a-9f57-3f2b1c0d9e11'),
headers={'x-api-key': api_key},
)
assert res.status_code == 404
# Requires the API key
res = client.delete(url_for("watchhistory", uuid=watch_uuid))
assert res.status_code == 403
# Pause it first - clearing resets last_checked to 0 which otherwise makes the ticker
# queue an instant recheck, and that would race with the assertions below
client.get(url_for("watch", uuid=watch_uuid, paused='paused'), headers={'x-api-key': api_key})
# Now really clear it
res = client.delete(
url_for("watchhistory", uuid=watch_uuid),
headers={'x-api-key': api_key},
)
assert res.status_code == 204
res = client.get(
url_for("watchhistory", uuid=watch_uuid),
headers={'x-api-key': api_key},
)
assert res.json == {}, "History should be empty after DELETE"
# The watch itself must survive, with its state reset
res = client.get(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key}
)
assert res.status_code == 200
assert res.json.get('url') == test_url
assert res.json.get('history_n') == 0
assert res.json.get('last_checked') == 0
assert res.json.get('previous_md5') == False
# And a snapshot fetch now has nothing to give
res = client.get(
url_for("watchsinglehistory", uuid=watch_uuid, timestamp='latest'),
headers={'x-api-key': api_key},
)
assert res.status_code == 404
delete_all_watches(client)
def test_roundtrip_API(client, live_server, measure_memory_usage, datastore_path):
"""
Test the full round trip, this way we test the default Model fits back into OpenAPI spec
+77
View File
@@ -321,3 +321,80 @@ def test_roundtrip_API(client, live_server, measure_memory_usage, datastore_path
date_created = res.json.get('date_created')
assert date_created != 454444444444, "ReadOnly date_created should not be updateable"
assert date_created != "454444444444", "ReadOnly date_created should not be updateable"
def test_api_watch_tag_field_accepts_names_and_uuids(client, live_server, measure_memory_usage, datastore_path):
"""The `tag` field on a watch takes tag *names*, `tags` takes UUIDs.
`tag` was documented as taking a UUID for years while the code fed it to add_tag(title),
so a UUID silently created a junk group *titled* with that UUID and never applied the tag
the caller asked for. Both spellings now resolve to the same tag.
"""
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
hdr = {'x-api-key': api_key, 'content-type': 'application/json'}
def titles_of(watch_uuid):
tags = datastore.data['settings']['application']['tags']
return sorted(tags[t].get('title') for t in datastore.data['watching'][watch_uuid].get('tags'))
# A name creates the group
res = client.post(url_for("createwatch"), data=json.dumps({"url": test_url, "tag": "helloworld"}), headers=hdr)
assert res.status_code == 201
assert titles_of(res.json['uuid']) == ['helloworld']
# An existing tag's UUID links to that tag rather than making a group named after the UUID
res = client.post(url_for("tag"), data=json.dumps({"title": "RealTag"}), headers=hdr)
assert res.status_code == 201
real_tag_uuid = res.json['uuid']
tag_count_before = len(datastore.data['settings']['application']['tags'])
res = client.post(url_for("createwatch"), data=json.dumps({"url": f"{test_url}?p=2", "tag": real_tag_uuid}), headers=hdr)
assert res.status_code == 201
assert real_tag_uuid in datastore.data['watching'][res.json['uuid']].get('tags')
assert titles_of(res.json['uuid']) == ['RealTag']
assert len(datastore.data['settings']['application']['tags']) == tag_count_before, "No junk tag titled with a UUID"
# `tags` with UUIDs keeps working
res = client.post(url_for("createwatch"), data=json.dumps({"url": f"{test_url}?p=3", "tags": [real_tag_uuid]}), headers=hdr)
assert res.status_code == 201
assert titles_of(res.json['uuid']) == ['RealTag']
# Names and UUIDs can be mixed, and blank entries from a trailing comma are dropped -
# add_tag() returns False for those and a falsy entry breaks every watch['tags'] lookup
res = client.post(url_for("createwatch"),
data=json.dumps({"url": f"{test_url}?p=4", "tag": f"Mixed,,{real_tag_uuid},"}), headers=hdr)
assert res.status_code == 201
assert titles_of(res.json['uuid']) == ['Mixed', 'RealTag']
assert all(datastore.data['watching'][res.json['uuid']].get('tags')), "No falsy entries in tags"
# A UUID that matches no tag is skipped rather than becoming a group named after it
unknown_uuid = '0be0272a-19dc-4c97-8aae-5a68df319489'
tag_count_before = len(datastore.data['settings']['application']['tags'])
res = client.post(url_for("createwatch"),
data=json.dumps({"url": f"{test_url}?p=5", "tag": unknown_uuid}), headers=hdr)
assert res.status_code == 201
assert datastore.data['watching'][res.json['uuid']].get('tags') == []
assert len(datastore.data['settings']['application']['tags']) == tag_count_before
# Names are matched case-insensitively against existing tags, as the spec claims
tag_count_before = len(datastore.data['settings']['application']['tags'])
res = client.post(url_for("createwatch"),
data=json.dumps({"url": f"{test_url}?p=6", "tag": "rEaLtAg"}), headers=hdr)
assert res.status_code == 201
assert datastore.data['watching'][res.json['uuid']].get('tags') == [real_tag_uuid]
assert len(datastore.data['settings']['application']['tags']) == tag_count_before, "Casing must not fork a second tag"
# `tags` is applied verbatim and never creates: an unknown UUID is stored as a dangling
# reference that simply resolves to no group. Documented, and harmless because the lookup
# is a dictfilt() over known tags - pinned here so changing it has to be deliberate.
bogus = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
tag_count_before = len(datastore.data['settings']['application']['tags'])
res = client.post(url_for("createwatch"),
data=json.dumps({"url": f"{test_url}?p=7", "tags": [bogus]}), headers=hdr)
assert res.status_code == 201
assert datastore.data['watching'][res.json['uuid']].get('tags') == [bogus]
assert len(datastore.data['settings']['application']['tags']) == tag_count_before
assert datastore.get_all_tags_for_watch(res.json['uuid']) == {}
assert client.get(url_for("watchlist.index")).status_code == 200, "A dangling tag ref must not break the list"
@@ -96,7 +96,7 @@ def test_check_ldjson_price_autodetect(client, live_server, measure_memory_usage
assert b'ldjson-price-track-offer' in res.data
# Accept it
client.get(url_for('price_data_follower.accept', uuid=uuid, follow_redirects=True))
client.post(url_for('price_data_follower.accept', uuid=uuid), follow_redirects=True)
client.post(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
# Offer should be gone
+2 -2
View File
@@ -24,7 +24,7 @@ def test_backup(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
# Launch the thread in the background to create the backup
res = client.get(
res = client.post(
url_for("backups.request_backup"),
follow_redirects=True
)
@@ -136,7 +136,7 @@ def test_backup_restore(client, live_server, measure_memory_usage, datastore_pat
wait_for_all_checks(client)
# Create a full backup
client.get(url_for("backups.request_backup"), follow_redirects=True)
client.post(url_for("backups.request_backup"), follow_redirects=True)
time.sleep(4)
# Download the latest backup zip
@@ -493,7 +493,7 @@ def test_tag_mute_persists(client, live_server):
tag_uuid = datastore.add_tag('Test Tag')
# Mute the tag
response = client.get(url_for("tags.mute", uuid=tag_uuid))
response = client.post(url_for("tags.mute", uuid=tag_uuid))
assert response.status_code == 302 # Redirect
# Verify muted in memory
+21 -21
View File
@@ -11,7 +11,7 @@ def test_zh_TW(client, live_server, measure_memory_usage, datastore_path):
# Be sure we got a session cookie
res = client.get(url_for("watchlist.index"), follow_redirects=True)
res = client.get(
res = client.post(
url_for("set_language", locale="zh_Hant_TW"), # Traditional
follow_redirects=True
)
@@ -21,7 +21,7 @@ def test_zh_TW(client, live_server, measure_memory_usage, datastore_path):
assert '選擇語言'.encode() in res.data
# Check second set works
res = client.get(
res = client.post(
url_for("set_language", locale="en_GB"),
follow_redirects=True
)
@@ -30,7 +30,7 @@ def test_zh_TW(client, live_server, measure_memory_usage, datastore_path):
assert b"Select Language" in res.data, "Second set of language worked"
# Check arbitration between zh_Hant_TW<->zh
res = client.get(
res = client.post(
url_for("set_language", locale="zh"), # Simplified chinese
follow_redirects=True
)
@@ -89,7 +89,7 @@ def test_language_switching(client, live_server, measure_memory_usage, datastore
client.get(url_for("add_watch_ui.add_watch_ui_index"), follow_redirects=True)
# Step 1: Set the language to Italian using the /set-language endpoint
res = client.get(
res = client.post(
url_for("set_language", locale="it"),
follow_redirects=True
)
@@ -119,7 +119,7 @@ def test_language_switching(client, live_server, measure_memory_usage, datastore
# NB: use 'en_GB' not 'en' — only the variants are in language_codes; the
# plain 'en' code is silently rejected by set_language and the locale would
# remain at 'it', defeating the round-trip assertion below.
res = client.get(
res = client.post(
url_for("set_language", locale="en_GB"),
follow_redirects=True
)
@@ -152,7 +152,7 @@ def test_invalid_locale(client, live_server, measure_memory_usage, datastore_pat
# bare 'en' is NOT in language_codes and is silently rejected by
# set_language, so passing it here would leave the session locale unset
# and let the (unrelated) Accept-Language fallback decide what renders.
res = client.get(
res = client.post(
url_for("set_language", locale="en_GB"),
follow_redirects=True
)
@@ -160,7 +160,7 @@ def test_invalid_locale(client, live_server, measure_memory_usage, datastore_pat
assert res.status_code == 200
# Try to set an invalid locale
res = client.get(
res = client.post(
url_for("set_language", locale="invalid_locale_xyz"),
follow_redirects=True
)
@@ -190,7 +190,7 @@ def test_language_persistence_in_session(client, live_server, measure_memory_usa
client.get(url_for("watchlist.index"), follow_redirects=True)
# Set language to Italian
res = client.get(
res = client.post(
url_for("set_language", locale="it"),
follow_redirects=True
)
@@ -215,7 +215,7 @@ def test_language_persistence_in_session(client, live_server, measure_memory_usa
assert sess.get('locale') == 'it', "Locale should be set in session"
# Call auto-detect to clear the locale
res = client.get(
res = client.post(
url_for("ui.delete_locale_language_session_var_if_it_exists"),
follow_redirects=True
)
@@ -254,7 +254,7 @@ def test_set_language_with_redirect(client, live_server, measure_memory_usage, d
client.get(url_for("watchlist.index"), follow_redirects=True)
# Set language with a redirect parameter (simulating language change from /settings)
res = client.get(
res = client.post(
url_for("set_language", locale="de", redirect="/settings"),
follow_redirects=False
)
@@ -268,7 +268,7 @@ def test_set_language_with_redirect(client, live_server, measure_memory_usage, d
assert sess.get('locale') == 'de'
# Test with invalid locale (should still redirect safely)
res = client.get(
res = client.post(
url_for("set_language", locale="invalid_locale", redirect="/settings"),
follow_redirects=False
)
@@ -276,7 +276,7 @@ def test_set_language_with_redirect(client, live_server, measure_memory_usage, d
assert '/settings' in res.location
# Test with malicious redirect (should default to watchlist)
res = client.get(
res = client.post(
url_for("set_language", locale="en", redirect="https://evil.com"),
follow_redirects=False
)
@@ -296,7 +296,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
client.get(url_for("watchlist.index"), follow_redirects=True)
# Test Italian translations
res = client.get(url_for("set_language", locale="it"), follow_redirects=True)
res = client.post(url_for("set_language", locale="it"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -312,7 +312,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Korean translations
res = client.get(url_for("set_language", locale="ko"), follow_redirects=True)
res = client.post(url_for("set_language", locale="ko"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -332,7 +332,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Chinese Simplified translations
res = client.get(url_for("set_language", locale="zh"), follow_redirects=True)
res = client.post(url_for("set_language", locale="zh"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -348,7 +348,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test German translations
res = client.get(url_for("set_language", locale="de"), follow_redirects=True)
res = client.post(url_for("set_language", locale="de"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -363,7 +363,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Russian translations
res = client.get(url_for("set_language", locale="ru"), follow_redirects=True)
res = client.post(url_for("set_language", locale="ru"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -378,7 +378,7 @@ def test_time_unit_translations(client, live_server, measure_memory_usage, datas
assert b"Time Between Check" not in res.data, "Should not have English 'Time Between Check'"
# Test Traditional Chinese (zh_Hant_TW) translations
res = client.get(url_for("set_language", locale="zh_Hant_TW"), follow_redirects=True)
res = client.post(url_for("set_language", locale="zh_Hant_TW"), follow_redirects=True)
assert res.status_code == 200
res = client.get(url_for("settings.settings_page"), follow_redirects=True)
@@ -627,7 +627,7 @@ def test_session_locale_overrides_accept_language(client, live_server, measure_m
"Expected Taiwan flag 'fi fi-tw' from auto-detect"
# Step 2: User explicitly selects Korean language
res = client.get(
res = client.post(
url_for("set_language", locale="ko"),
headers={'Accept-Language': 'zh-TW,zh;q=0.9,en;q=0.8'}, # Browser still sends zh-TW
follow_redirects=True
@@ -700,7 +700,7 @@ def test_clear_history_translated_confirmation(client, live_server, measure_memo
wait_for_all_checks(client)
# Set language to German
res = client.get(
res = client.post(
url_for("set_language", locale="de"),
follow_redirects=True
)
@@ -726,7 +726,7 @@ def test_clear_history_translated_confirmation(client, live_server, measure_memo
"German confirmation word 'loschen' should be accepted (issue #3865)"
# Switch back to English and verify English word still works
res = client.get(
res = client.post(
url_for("set_language", locale="en_US"),
follow_redirects=True
)
@@ -25,7 +25,7 @@ def test_language_endpoints_work_for_anonymous_users(client, live_server, measur
follow_redirects=True)
assert res.status_code == 200
client.get(url_for("logout"), follow_redirects=True)
client.post(url_for("logout"), follow_redirects=True)
# Both language links are rendered on the login page, so both must be reachable
res = client.get(url_for("login"))
@@ -33,13 +33,13 @@ def test_language_endpoints_work_for_anonymous_users(client, live_server, measur
assert b'language-selector' in res.data, "Language modal trigger should render for anonymous users"
# Picking a specific language must not redirect to the login page
res = client.get(url_for("set_language", locale="de"), follow_redirects=False)
res = client.post(url_for("set_language", locale="de"), follow_redirects=False)
assert res.status_code == 302
assert '/login' not in res.headers.get("Location", ""), \
"set_language must not bounce anonymous users to /login"
# ...and neither must clearing it back to auto-detect
res = client.get(url_for("ui.delete_locale_language_session_var_if_it_exists"), follow_redirects=False)
res = client.post(url_for("ui.delete_locale_language_session_var_if_it_exists"), follow_redirects=False)
assert res.status_code == 302
assert '/login' not in res.headers.get("Location", ""), \
"Auto-detect must not bounce anonymous users to /login (it renders on the login page)"
@@ -393,12 +393,12 @@ def test_llm_models_endpoint_blocks_private_api_base(
def test_llm_test_endpoint_blocks_private_api_base(
client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""GET /settings/llm/test must refuse api_base pointing at private/loopback
"""POST /settings/llm/test must refuse api_base pointing at private/loopback
hosts and must never reach litellm.completion()."""
monkeypatch.delenv('ALLOW_IANA_RESTRICTED_ADDRESSES', raising=False)
for bad in _SSRF_PRIVATE_HOSTS:
res = client.get(
res = client.post(
url_for('settings.llm.llm_test'),
query_string={'model': 'openai/gpt-4', 'api_base': bad},
)
@@ -530,7 +530,7 @@ def test_llm_test_refuses_to_leak_stored_key_to_different_api_base(
monkeypatch.setattr(llm_client, 'completion',
lambda **kw: calls.append(kw) or ('', 0, 0, 0))
res = client.get(
res = client.post(
url_for('settings.llm.llm_test'),
query_string={
'model': 'gpt-4o-mini',
@@ -38,7 +38,7 @@ def test_rss_tag_feed_ignores_security_token(client, live_server, datastore_path
wait_for_all_checks(client)
# Logout
client.get(url_for("logout"), follow_redirects=True)
client.post(url_for("logout"), follow_redirects=True)
# Request the tag RSS feed WITH the token
res = client.get(
+5 -5
View File
@@ -440,7 +440,7 @@ def test_login_redirect_with_password(client, live_server, measure_memory_usage,
assert b"evil.com" not in res.data
# Logout for cleanup
client.get(url_for("logout"))
client.post(url_for("logout"))
# Test 5: Incorrect password with redirect should stay on login page
res = client.post(
@@ -483,7 +483,7 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
client.application.config['DATASTORE'].data['settings']['application']['password'] = salted_pass
# Logout to ensure we're not authenticated
client.get(url_for("logout"))
client.post(url_for("logout"))
# Try to access a protected page (edit page for first watch)
res = client.get(
@@ -524,7 +524,7 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
assert b'Edit' in res.data or b'Watching' in res.data
# Cleanup
client.get(url_for("logout"))
client.post(url_for("logout"))
del client.application.config['DATASTORE'].data['settings']['application']['password']
@@ -554,7 +554,7 @@ def test_logout_with_redirect(client, live_server, measure_memory_usage, datasto
assert res.status_code == 200
# Now logout with a redirect parameter (simulating logout from /settings)
res = client.get(
res = client.post(
url_for("logout", redirect="/settings"),
follow_redirects=False
)
@@ -961,7 +961,7 @@ def test_ghsa_8757_69j2_hx56_backup_restore_history_path_traversal(client, live_
wait_for_all_checks(client)
# Download a legitimate backup to use as a template
client.get(url_for("backups.request_backup"), follow_redirects=True)
client.post(url_for("backups.request_backup"), follow_redirects=True)
time.sleep(4)
res = client.get(url_for("backups.download_backup", filename="latest"), follow_redirects=True)
assert res.content_type == "application/zip"
+1 -1
View File
@@ -33,7 +33,7 @@ def test_share_watch(client, live_server, measure_memory_usage, datastore_path):
assert bytes(include_filters.encode('utf-8')) in res.data
# click share the link
res = client.get(
res = client.post(
url_for("ui.form_share_put_watch", uuid=uuid),
follow_redirects=True
)
@@ -0,0 +1,181 @@
"""Guards for the translation overlay layer (changedetectionio/translations_overlay).
The overlay is a second gettext tree merged on top of ``changedetectionio/translations``, letting a
deployment reword individual strings without editing the ``_()`` call site. See that directory's
README.md. Three separate things can break it, so there is a test for each.
1. Overlay entries key on the exact upstream msgid. When a string is reworded upstream the override
stops matching and silently reverts to upstream wording - no error, no log entry.
``test_overlay_catalogs_are_valid`` makes that a build failure.
2. The layering relies on Flask-Babel merging catalogs with ``dict.update`` semantics (later
directory wins per-message). Were a Flask-Babel upgrade to change that to an ``add_fallback``
chain, overrides would stop applying while everything still looked fine.
``test_overlay_overrides_a_string_in_a_rendered_page`` pins it against a real rendered page.
3. The directory has to actually reach ``BABEL_TRANSLATION_DIRECTORIES``.
``test_overlay_dir_*`` cover the wiring in flask_app.py, including the env var.
"""
import os
import subprocess
import sys
import pytest
from babel.messages.catalog import Catalog
from babel.messages.mofile import write_mo
from flask import url_for
PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
REPO_ROOT = os.path.dirname(PKG_DIR)
OVERLAY_DIR = os.path.join(PKG_DIR, 'translations_overlay')
BASE_DIR = os.path.join(PKG_DIR, 'translations')
MANAGE = os.path.join(OVERLAY_DIR, 'manage.py')
# A msgid that renders as a settings-page tab label. The test asserts it is present *before*
# overriding it, so a rename upstream fails loudly rather than making the test silently vacuous.
OVERRIDDEN_MSGID = 'Global Filters'
SENTINEL = 'zzOverlaySentinelFiltersZZ'
def _write_mo(root, locale, entries):
"""Write a compiled catalog at <root>/<locale>/LC_MESSAGES/messages.mo."""
catalog = Catalog(locale=locale, domain='messages')
for msgid, msgstr in entries.items():
catalog.add(msgid, msgstr)
mo_dir = os.path.join(root, locale, 'LC_MESSAGES')
os.makedirs(mo_dir, exist_ok=True)
with open(os.path.join(mo_dir, 'messages.mo'), 'wb') as fp:
write_mo(fp, catalog)
def _import_app_with(env_overlay_dir):
"""Import flask_app in a clean subprocess and report its BABEL_TRANSLATION_DIRECTORIES.
Has to be a subprocess: the config is built at module import time, so it cannot be re-evaluated
under a different environment once flask_app is already in sys.modules.
"""
env = dict(os.environ, TRANSLATION_OVERLAY_DIR=env_overlay_dir)
result = subprocess.run(
[
sys.executable,
'-c',
'from changedetectionio import flask_app;'
'print("DIRS=" + flask_app.app.config["BABEL_TRANSLATION_DIRECTORIES"])',
],
capture_output=True,
text=True,
env=env,
cwd=REPO_ROOT,
)
assert result.returncode == 0, f"importing flask_app failed:\n{result.stdout}\n{result.stderr}"
line = [l for l in result.stdout.splitlines() if l.startswith('DIRS=')]
assert line, f"no config line in output:\n{result.stdout}\n{result.stderr}"
return line[0][len('DIRS=') :].split(';')
# ---------------------------------------------------------------------------
# 1. The overlay catalogs shipped in this repo are internally consistent
# ---------------------------------------------------------------------------
@pytest.mark.skipif(
not os.path.isdir(OVERLAY_DIR), reason='no translation overlay in this deployment'
)
def test_overlay_catalogs_are_valid():
"""Every override must still match an upstream msgid, be non-empty, and be compiled.
A failure here usually means upstream edited a string the overlay overrides. Re-copy the new
msgid verbatim from translations/messages.pot into the overlay catalog, then recompile with
`python changedetectionio/translations_overlay/manage.py compile`.
"""
result = subprocess.run([sys.executable, MANAGE, 'check'], capture_output=True, text=True)
assert result.returncode == 0, (
f"translation overlay is invalid:\n{result.stdout}{result.stderr}"
)
@pytest.mark.skipif(
not os.path.isdir(OVERLAY_DIR), reason='no translation overlay in this deployment'
)
def test_overlay_locales_have_a_base_catalog():
"""An overlay for a locale the app does not ship never loads, so it is silently dead."""
for locale in sorted(os.listdir(OVERLAY_DIR)):
if not os.path.isfile(os.path.join(OVERLAY_DIR, locale, 'LC_MESSAGES', 'messages.po')):
continue
assert os.path.isdir(os.path.join(BASE_DIR, locale)), (
f"overlay locale {locale!r} has no base catalog in translations/"
)
# ---------------------------------------------------------------------------
# 2. The merge actually happens, end to end, on a real page
# ---------------------------------------------------------------------------
def test_overlay_overrides_a_string_in_a_rendered_page(client, live_server, tmp_path):
"""A real overlay catalog changes real rendered output, and only the string it names."""
app = client.application
# The rest of this test injects its own directory, which would still pass if flask_app.py had
# stopped configuring the real one. Tie the two together so that regression fails here too.
if os.path.isdir(OVERLAY_DIR):
configured = app.config['BABEL_TRANSLATION_DIRECTORIES'].split(';')
assert OVERLAY_DIR in configured, (
f"{OVERLAY_DIR} exists but is not in BABEL_TRANSLATION_DIRECTORIES ({configured})"
)
baseline = client.get(url_for('settings.settings_page'))
assert baseline.status_code == 200
assert OVERRIDDEN_MSGID.encode() in baseline.data, (
f"{OVERRIDDEN_MSGID!r} no longer renders on the settings page - this test needs a new msgid"
)
assert SENTINEL.encode() not in baseline.data
overlay = tmp_path / 'overlay'
# en_GB is BABEL_DEFAULT_LOCALE, and the test client sends no Accept-Language header
_write_mo(str(overlay), 'en_GB', {OVERRIDDEN_MSGID: SENTINEL})
# The default Domain delegates to the app-level directory list, and caches per (locale, domain),
# so both have to be touched for a new catalog to be picked up mid-process.
dirs = app.extensions['babel'].translation_directories
domain_cache = app.extensions['babel'].instance.domain_instance.cache
dirs.append(str(overlay))
domain_cache.clear()
try:
overridden = client.get(url_for('settings.settings_page'))
assert overridden.status_code == 200
assert SENTINEL.encode() in overridden.data, (
'overlay catalog did not override the base catalog'
)
assert OVERRIDDEN_MSGID.encode() not in overridden.data, 'base wording is still rendering'
# Neighbouring tab label, deliberately not in the overlay - merging must not drop it
assert b'UI Options' in overridden.data, (
'overlay replaced the catalog instead of merging into it'
)
finally:
dirs.remove(str(overlay))
domain_cache.clear()
restored = client.get(url_for('settings.settings_page'))
assert SENTINEL.encode() not in restored.data
assert OVERRIDDEN_MSGID.encode() in restored.data, 'base wording did not come back'
# ---------------------------------------------------------------------------
# 3. flask_app.py wires the directory up, and stays a no-op when there isn't one
# ---------------------------------------------------------------------------
def test_overlay_dir_from_env_var_is_used(tmp_path):
overlay = tmp_path / 'my-overlay'
overlay.mkdir()
dirs = _import_app_with(str(overlay))
assert dirs[-1] == str(overlay), f"TRANSLATION_OVERLAY_DIR not appended, got {dirs}"
assert dirs[0] == BASE_DIR, 'base catalog must stay first so the overlay wins on conflicts'
def test_missing_overlay_dir_is_a_noop(tmp_path):
"""No overlay directory means the config is exactly what it was before the feature existed."""
dirs = _import_app_with(str(tmp_path / 'does-not-exist'))
assert dirs == [BASE_DIR], f"expected only the base catalog, got {dirs}"
+222
View File
@@ -0,0 +1,222 @@
#!/usr/bin/env python3
"""PAGE_WATCH_LIMIT - the optional cap on how many watches one instance will hold.
The limit is enforced in datastore.add_watch(), which every add path funnels through, but each
surface has to report it in its own terms: a flash for the UI, a 429 for the API, one flash for
a whole file in the importers, and nothing at all (just a None) where there's no request context.
"""
import json
from flask import url_for
from .util import delete_all_watches
def test_watch_limit_absent_or_junk_means_unlimited(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""No env var, an empty one, or an unparseable one all leave the limit switched off."""
datastore = live_server.app.config['DATASTORE']
monkeypatch.delenv('PAGE_WATCH_LIMIT', raising=False)
assert datastore.watch_limit is None
assert datastore.watch_limit_reached() is False
# Junk must not block every add, and must not raise
monkeypatch.setenv('PAGE_WATCH_LIMIT', 'not-a-number')
assert datastore.watch_limit is None
assert datastore.watch_limit_reached() is False
# Set-but-empty is the same as unset
monkeypatch.setenv('PAGE_WATCH_LIMIT', '')
assert datastore.watch_limit is None
assert datastore.watch_limit_reached() is False
def test_api_create_watch_refused_at_limit(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
monkeypatch.setenv('PAGE_WATCH_LIMIT', '1')
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": test_url}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
)
assert res.status_code == 201
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": f"{test_url}?second=1"}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
)
assert res.status_code == 429
assert b'Watch limit reached (1/1 watches)' in res.data
assert len(datastore.data['watching']) == 1
delete_all_watches(client)
def test_api_import_refuses_the_whole_batch(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""A 429 from import always means nothing was created, so the same request can be retried."""
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
headers = {'x-api-key': api_key, 'content-type': 'text/plain'}
monkeypatch.setenv('PAGE_WATCH_LIMIT', '3')
res = client.post(url_for("import"), data=f"{test_url}?a=1\n{test_url}?a=2", headers=headers)
assert res.status_code == 200
assert len(res.json) == 2
# Two more would make four - refused whole rather than importing the one that fits
res = client.post(url_for("import"), data=f"{test_url}?a=3\n{test_url}?a=4", headers=headers)
assert res.status_code == 429
assert b'would exceed it' in res.data
assert len(datastore.data['watching']) == 2
# The one that does fit still goes in
res = client.post(url_for("import"), data=f"{test_url}?a=3", headers=headers)
assert res.status_code == 200
assert len(datastore.data['watching']) == 3
delete_all_watches(client)
def test_quick_watch_add_refused_at_limit(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
monkeypatch.setenv('PAGE_WATCH_LIMIT', '1')
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": test_url, 'tags': ''},
follow_redirects=True
)
assert b'Watch added' in res.data
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": f"{test_url}?second=1", 'tags': ''},
follow_redirects=True
)
assert b'Watch limit reached (1/1 watches)' in res.data
assert b'Watch added' not in res.data
assert len(datastore.data['watching']) == 1
delete_all_watches(client)
def test_ui_import_reports_limit_once_and_hands_back_the_rest(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""The importer stops at the limit instead of letting add_watch() flash per remaining row."""
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
monkeypatch.setenv('PAGE_WATCH_LIMIT', '2')
urls = "\n".join(f"{test_url}?i={i}" for i in range(5))
res = client.post(url_for("imports.import_page"), data={"urls": urls}, follow_redirects=True)
assert res.data.count(b'Watch limit reached') == 1, "The limit should be reported once for the file, not once per row"
assert len(datastore.data['watching']) == 2
# 3 unprocessed URLs come back in the textarea to retry once there's room
assert b'3 Skipped' in res.data
assert b'i=4' in res.data
delete_all_watches(client)
def test_clone_refused_at_limit(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""Clone used to raise KeyError(None) here, then redirect to an edit page for uuid=None."""
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
uuid = datastore.add_watch(url=test_url)
monkeypatch.setenv('PAGE_WATCH_LIMIT', '1')
res = client.post(url_for("ui.form_clone", uuid=uuid), follow_redirects=True)
assert res.status_code == 200
assert b'Watch limit reached (1/1 watches)' in res.data
assert b'Cloned' not in res.data
assert len(datastore.data['watching']) == 1
delete_all_watches(client)
def test_over_limit_instance_still_loads_and_edits(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""A limit set below what an install already holds must only block *new* watches.
Everything already there keeps loading from disk and stays editable - the limit is not
retroactive and never hides or drops a watch.
"""
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
monkeypatch.delenv('PAGE_WATCH_LIMIT', raising=False)
uuids = [datastore.add_watch(url=f"{test_url}?i={i}") for i in range(3)]
assert all(uuids)
# Now cap it well below what's already stored
monkeypatch.setenv('PAGE_WATCH_LIMIT', '1')
assert datastore.watch_limit_reached() is True
# Re-reading from disk is not gated by the limit
datastore._load_watches()
assert len(datastore.data['watching']) == 3
# Still all listed
assert client.get(url_for("watchlist.index")).status_code == 200
res = client.get(url_for("createwatch"), headers={'x-api-key': api_key})
assert len(res.json) == 3
# And still editable
res = client.put(
url_for("watch", uuid=uuids[0]),
data=json.dumps({"title": "Still editable"}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
)
assert res.status_code == 200
assert datastore.data['watching'][uuids[0]].get('title') == "Still editable"
# Only adding is refused
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": f"{test_url}?new=1"}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
)
assert res.status_code == 429
assert len(datastore.data['watching']) == 3
delete_all_watches(client)
def test_limit_shown_in_settings_info_tab(client, live_server, measure_memory_usage, datastore_path, monkeypatch):
"""The Info tab names the limit only when one is configured."""
monkeypatch.delenv('PAGE_WATCH_LIMIT', raising=False)
res = client.get(url_for("settings.settings_page"))
assert b'Maximum number of page watches' not in res.data
monkeypatch.setenv('PAGE_WATCH_LIMIT', '42')
res = client.get(url_for("settings.settings_page"))
assert b'Maximum number of page watches' in res.data
assert b'42' in res.data
def test_limit_reported_without_a_request_context(client, live_server, measure_memory_usage, datastore_path, monkeypatch, mocker):
"""The CLI (-u) and the API's background import thread call add_watch() with no request
context, where flash() raises RuntimeError instead of reporting anything."""
datastore = live_server.app.config['DATASTORE']
test_url = url_for('test_endpoint', _external=True)
monkeypatch.setenv('PAGE_WATCH_LIMIT', '1')
assert datastore.add_watch(url=test_url)
# pytest-flask pushes a request context around every test, so take it away
mocker.patch('changedetectionio.store.has_request_context', return_value=False)
assert datastore.add_watch(url=f"{test_url}?second=1") is None
assert len(datastore.data['watching']) == 1
delete_all_watches(client)
@@ -79,5 +79,69 @@ class TestTriggerConditions(unittest.TestCase):
self.assertTrue(result.get('result'))
def test_conditions_filter_complete_rules_with_zero_values(self):
from changedetectionio.conditions import filter_complete_rules
rules = [
{"operator": "==", "field": "word_count", "value": 0},
{"operator": "<=", "field": "levenshtein_distance", "value": 0.0},
{"operator": "==", "field": "price", "value": "0"},
{"operator": "==", "field": "empty_val", "value": ""},
{"operator": "==", "field": "none_val", "value": None},
{"operator": "==", "field": "str_none", "value": "None"},
{"operator": "", "field": "missing_op", "value": 0},
{"operator": "None", "field": "str_none_op", "value": 0},
{"operator": "==", "field": "", "value": 0},
]
complete = filter_complete_rules(rules)
self.assertEqual(len(complete), 3)
self.assertEqual(complete[0]["value"], 0)
self.assertEqual(complete[1]["value"], 0.0)
self.assertEqual(complete[2]["value"], "0")
def test_conditions_convert_to_jsonlogic_with_zero_values(self):
from json_logic import jsonLogic
from changedetectionio.conditions import convert_to_jsonlogic
rule_int_zero = [{"operator": "==", "field": "word_count", "value": 0}]
jl_int = convert_to_jsonlogic("and", rule_int_zero)
self.assertEqual(jl_int, {"==": [{"var": "word_count"}, 0]})
self.assertTrue(jsonLogic(jl_int, {"word_count": 0}))
self.assertFalse(jsonLogic(jl_int, {"word_count": 5}))
rule_float_zero = [{"operator": "<=", "field": "levenshtein_distance", "value": 0.0}]
jl_float = convert_to_jsonlogic("and", rule_float_zero)
self.assertEqual(jl_float, {"<=": [{"var": "levenshtein_distance"}, 0.0]})
self.assertTrue(jsonLogic(jl_float, {"levenshtein_distance": 0.0}))
self.assertFalse(jsonLogic(jl_float, {"levenshtein_distance": 2.5}))
def test_conditions_execution_zero_word_count(self):
# Test condition checking for empty page (word_count == 0)
self.store.data['watching'][self.watch_uuid].update(
{
"conditions_match_logic": "ALL",
"conditions": [
{"operator": "==", "field": "word_count", "value": 0},
],
}
)
# Empty text has word_count 0 -> condition should match (True)
res_empty = execute_ruleset_against_all_plugins(
current_watch_uuid=self.watch_uuid,
application_datastruct=self.store.data,
ephemeral_data={'text': ""},
)
self.assertTrue(res_empty.get('result'))
# Non-empty text has word_count > 0 -> condition should not match (False)
res_nonempty = execute_ruleset_against_all_plugins(
current_watch_uuid=self.watch_uuid,
application_datastruct=self.store.data,
ephemeral_data={'text': "Hello world"},
)
self.assertFalse(res_nonempty.get('result'))
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""Tests for the shared "may the server fetch this URL?" gate.
# run from dir above changedetectionio/ dir
# python3 -m unittest changedetectionio.tests.unit.test_fetch_url_gate
Every server-side fetch entry point routes through validate_url.is_fetch_url_allowed(). Before it
existed, the file:// and private-IP rules were enforced inline in call_browser() only, so any fetch
path that did not go through call_browser() was unprotected:
* a "Goto URL" browser step could read file:///etc/passwd (GHSA-hm22-wg2m-35v4)
* /add-watch-ui/snapshot url= could fetch internal hosts (GHSA-56fq-63vj-9992)
These tests pin the gate's rules AND the browser-step choke point, so a future fetch path that
forgets to call the gate is the only way to regress it.
"""
import asyncio
import unittest
from unittest.mock import patch
from changedetectionio.browser_steps.browser_steps import steppable_browser_interface
from changedetectionio.validate_url import (
is_fetch_url_allowed,
is_special_purpose_ip,
validate_fetch_url,
validate_fetch_url_async,
)
# tests/conftest.py sets ALLOW_IANA_RESTRICTED_ADDRESSES=true for the functional suite, so the
# locked-down default has to be re-asserted explicitly rather than assumed.
LOCKED_DOWN = {'ALLOW_IANA_RESTRICTED_ADDRESSES': 'false', 'ALLOW_FILE_URI': 'false'}
OPTED_IN = {'ALLOW_IANA_RESTRICTED_ADDRESSES': 'true', 'ALLOW_FILE_URI': 'true'}
class TestFetchUrlGate(unittest.TestCase):
def assertBlocked(self, url):
ok, reason = is_fetch_url_allowed(url)
self.assertFalse(ok, f"URL '{url}' should have been blocked")
self.assertTrue(reason, f"URL '{url}' was blocked without a reason to show the user")
def assertAllowed(self, url):
ok, reason = is_fetch_url_allowed(url)
self.assertTrue(ok, f"URL '{url}' should have been allowed, got: {reason}")
def test_file_uri_blocked_by_default(self):
with patch.dict('os.environ', LOCKED_DOWN):
# All the spellings that reach the same local file
for url in ('file:///etc/passwd', 'FILE:///etc/passwd', 'file:/etc/passwd', 'file://etc/passwd'):
with self.subTest(url=url):
self.assertBlocked(url)
def test_file_uri_allowed_when_operator_opts_in(self):
with patch.dict('os.environ', OPTED_IN):
self.assertAllowed('file:///etc/passwd')
def test_file_uri_blocked_even_if_safe_protocol_regex_was_loosened(self):
"""An operator who widens SAFE_PROTOCOL_REGEX for some other scheme must not get local
file reads thrown in for free - hence the explicit file: check ahead of is_safe_valid_url()."""
env = dict(LOCKED_DOWN, SAFE_PROTOCOL_REGEX='^(http|https|ftp|file):')
with patch.dict('os.environ', env):
self.assertBlocked('file:///etc/passwd')
def test_private_and_reserved_addresses_blocked_by_default(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('http://127.0.0.1:5000/',
'http://localhost/',
'http://169.254.169.254/latest/meta-data/', # cloud metadata
'http://192.168.1.1/',
'http://10.0.0.1/',
'http://[::1]/'):
with self.subTest(url=url):
self.assertBlocked(url)
def test_cgnat_and_other_non_global_addresses_blocked_by_default(self):
"""GHSA-gwph-fp79-379w - the 0.54.1 predicate only tested is_private/is_loopback/
is_link_local/is_reserved, none of which are True for RFC 6598 CGNAT space, so
100.64.0.0/10 (an ISP's other subscribers, CPE admin panels, CGNAT gateways) stayed
fetchable. These are IP literals, so no DNS is involved and CI cannot flake."""
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('http://100.64.0.1/', # RFC 6598 CGNAT, first usable
'http://100.127.255.254/', # RFC 6598 CGNAT, last usable
'http://100.100.100.100/', # inside CGNAT (Alibaba Cloud metadata)
'http://192.88.99.1/', # RFC 7526 deprecated 6to4 relay anycast
'http://224.0.0.1/', # IPv4 multicast all-hosts
'http://[ff02::1]/'): # IPv6 multicast all-nodes
with self.subTest(url=url):
self.assertBlocked(url)
def test_cgnat_allowed_when_operator_opts_in(self):
"""CGNAT is legitimate for operators monitoring their own carrier network, so the
opt-in has to release it the same way it releases 127.0.0.1."""
with patch.dict('os.environ', OPTED_IN):
self.assertAllowed('http://100.64.0.1/')
def test_special_purpose_ip_classification(self):
"""The predicate itself, without DNS - one place to pin what is and is not fetchable."""
for ip in ('100.64.0.1', '100.127.255.254', '192.88.99.1', '224.0.0.1', 'ff02::1',
'127.0.0.1', '10.0.0.1', '169.254.169.254', '192.168.1.1', '::1',
'0.0.0.0', '255.255.255.255', '198.18.0.1', 'fc00::1', 'fe80::1',
'::ffff:100.64.0.1', # CGNAT wrapped as an IPv4-mapped IPv6 address
'2002:6440:1::'): # CGNAT wrapped as a 6to4 address
with self.subTest(ip=ip):
blocked, why = is_special_purpose_ip(ip)
self.assertTrue(blocked, f"{ip} should be refused")
self.assertTrue(why, f"{ip} was refused without a stated reason")
for ip in ('1.1.1.1', '8.8.8.8', '93.184.216.34', '2606:4700:4700::1111'):
with self.subTest(ip=ip):
blocked, why = is_special_purpose_ip(ip)
self.assertFalse(blocked, f"public address {ip} was refused as '{why}'")
def test_cgnat_boundaries_are_exact(self):
"""100.64.0.0/10 ends at 100.127.255.255 - 100.63.x and 100.128.x are ordinary public
space and must not be collateral damage from a /8-sized over-block."""
for ip in ('100.63.255.255', '100.128.0.0'):
with self.subTest(ip=ip):
blocked, why = is_special_purpose_ip(ip)
self.assertFalse(blocked, f"public address {ip} was refused as '{why}'")
def test_private_addresses_allowed_when_operator_opts_in(self):
with patch.dict('os.environ', OPTED_IN):
self.assertAllowed('http://127.0.0.1:5000/')
def test_source_prefix_is_stripped_before_the_hostname_check(self):
"""Load-bearing, not cosmetic: urlparse('source:http://127.0.0.1/') reports NO hostname,
so leaving the prefix on would hand the private-IP check nothing to look at and let it pass."""
with patch.dict('os.environ', LOCKED_DOWN):
self.assertBlocked('source:http://127.0.0.1/')
self.assertBlocked('SOURCE:http://169.254.169.254/')
self.assertBlocked('source:file:///etc/passwd')
def test_jinja2_is_rendered_before_the_hostname_check(self):
"""The fetch uses the rendered URL, so the rendered URL is what must be judged - otherwise
a template expression hides the real target from the check."""
with patch.dict('os.environ', LOCKED_DOWN):
self.assertBlocked("http://{{ '127.0.0.1' }}/")
self.assertBlocked("http://{% if 1 %}127.0.0.1{% endif %}/")
def test_parser_differential_payload_always_rejected(self):
"""GHSA-rph4-96w6-q594: urlparse sees PUBLIC, urllib3 connects to INTERNAL. A backslash has
no legitimate use in a URL, so this is refused even with both opt-ins enabled."""
for env in (LOCKED_DOWN, OPTED_IN):
with self.subTest(env=env), patch.dict('os.environ', env):
self.assertBlocked('http://127.0.0.1:8888\\@example.com/')
def test_unsupported_schemes_rejected(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('javascript:alert(1)', 'data:text/html,<h1>x', 'chrome://version'):
with self.subTest(url=url):
self.assertBlocked(url)
def test_empty_input_rejected(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('', ' ', None):
with self.subTest(url=url):
self.assertBlocked(url)
def test_ordinary_public_urls_still_allowed(self):
# Unresolvable hostnames are allowed by design (DNS may be down, domain not yet live), so
# these pass with or without working DNS in CI.
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('https://example.com/',
'source:https://example.com/',
'https://example.com/path?a=b&c=d#frag'):
with self.subTest(url=url):
self.assertAllowed(url)
def test_validate_fetch_url_raises_with_the_reason(self):
with patch.dict('os.environ', LOCKED_DOWN):
with self.assertRaises(ValueError):
validate_fetch_url('file:///etc/passwd')
validate_fetch_url('https://example.com/') # must not raise
def test_validate_fetch_url_async_raises_with_the_reason(self):
with patch.dict('os.environ', LOCKED_DOWN):
with self.assertRaises(ValueError):
asyncio.run(validate_fetch_url_async('http://127.0.0.1/'))
asyncio.run(validate_fetch_url_async('https://example.com/')) # must not raise
class _RecordingPage:
"""Stands in for the Playwright page so we can assert navigation never happened."""
def __init__(self):
self.goto_calls = []
async def goto(self, url, **kwargs):
self.goto_calls.append(url)
return None
async def wait_for_timeout(self, ms):
return None
class TestBrowserStepGotoUrlGate(unittest.TestCase):
"""GHSA-hm22-wg2m-35v4 - browser step values are raw user input and were never validated.
action_goto_url() is the single choke point for every navigation we initiate (the "Goto URL"
step, "Goto site", the live Browser Steps UI and the Add Watch preview all land here), so the
assertion that matters is that page.goto() is never reached for a refused URL.
"""
def _interface(self, start_url='https://example.com/'):
interface = steppable_browser_interface(start_url=start_url)
interface.page = _RecordingPage()
return interface
def test_goto_url_step_cannot_read_local_files(self):
with patch.dict('os.environ', LOCKED_DOWN):
interface = self._interface()
with self.assertRaises(ValueError):
asyncio.run(interface.action_goto_url(value='file:///etc/passwd'))
self.assertEqual(interface.page.goto_calls, [], "Chromium was navigated to a refused URL")
def test_goto_url_step_cannot_reach_private_addresses(self):
with patch.dict('os.environ', LOCKED_DOWN):
for url in ('http://127.0.0.1:5000/', 'http://169.254.169.254/latest/meta-data/'):
with self.subTest(url=url):
interface = self._interface()
with self.assertRaises(ValueError):
asyncio.run(interface.action_goto_url(value=url))
self.assertEqual(interface.page.goto_calls, [])
def test_goto_site_step_validates_the_start_url_too(self):
with patch.dict('os.environ', LOCKED_DOWN):
interface = self._interface(start_url='source:http://127.0.0.1/')
with self.assertRaises(ValueError):
asyncio.run(interface.action_goto_site())
self.assertEqual(interface.page.goto_calls, [])
def test_permitted_url_still_navigates(self):
with patch.dict('os.environ', LOCKED_DOWN):
interface = self._interface()
asyncio.run(interface.action_goto_url(value='https://example.com/'))
self.assertEqual(interface.page.goto_calls, ['https://example.com/'])
if __name__ == '__main__':
unittest.main()
@@ -205,8 +205,8 @@ def test_browsersteps_edit_UI_startsession(client, live_server, measure_memory_u
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, extras={'fetch_backend': 'html_webdriver', 'paused': True})
# Test starting a browsersteps session
res = client.get(
# Test starting a browsersteps session (POST only - it spins up a real browser)
res = client.post(
url_for("browser_steps.browsersteps_start_session", uuid=uuid),
follow_redirects=True
)
@@ -1567,7 +1567,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} sledování otagováno"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Sledujte tuto adresu URL!"
@@ -1588,7 +1588,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} Überwachungen wurde markiert"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Überwachung nicht gefunden"
@@ -1561,7 +1561,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr ""
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr ""
@@ -1561,7 +1561,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr ""
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr ""
@@ -1608,7 +1608,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "Se etiquetaron {} monitores"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Monitor no encontrado"
@@ -1570,7 +1570,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr ""
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Surveillance non trouvée"
@@ -1563,7 +1563,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr ""
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Monitoraggio non trovato"
@@ -1569,7 +1569,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} 件のウォッチにタグを付けました"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "ウォッチが見つかりません"
@@ -1571,7 +1571,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{}개 모니터링에 태그를 추가했습니다."
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "모니터링을 찾을 수 없습니다."
+3 -3
View File
@@ -6,9 +6,9 @@
#, fuzzy
msgid ""
msgstr ""
"Project-Id-Version: changedetection.io 0.55.8\n"
"Project-Id-Version: changedetection.io 0.60.3\n"
"Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
"POT-Creation-Date: 2026-09-02 12:06+0200\n"
"POT-Creation-Date: 2026-09-04 14:11+0200\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
@@ -1560,7 +1560,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr ""
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr ""
@@ -1691,7 +1691,7 @@ msgstr "Wybrano nieprawidłowy serwer proxy"
msgid "{} watches were tagged"
msgstr "Oznaczono {} obserwacji"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Nie znaleziono obserwacji"
@@ -1593,7 +1593,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} monitoramentos foram tagueados"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Monitoramento não encontrado"
@@ -1657,7 +1657,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} часы были помечены"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Часы не найдены"
@@ -1600,7 +1600,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} izleyici etiketlendi"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "İzleyici bulunamadı"
@@ -1581,7 +1581,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} завдань було позначено тегами"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "Завдання не знайдено"
@@ -1566,7 +1566,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "已为 {} 个监控项打标签"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "未找到监控项"
@@ -1565,7 +1565,7 @@ msgstr ""
msgid "{} watches were tagged"
msgstr "{} 個監測任務已加上標籤"
#: changedetectionio/blueprint/ui/__init__.py
#: changedetectionio/blueprint/ui/__init__.py changedetectionio/blueprint/watchlist/__init__.py
msgid "Watch not found"
msgstr "找不到監測任務"
@@ -0,0 +1,72 @@
# Translation overlay
An optional second catalog layer that lets a deployment reword individual strings **without forking
any template or Python source**.
It exists because rewording a string normally means editing the `_()` call at its call site, and that
edit conflicts on every upstream merge, forever. The overlay moves the reworded text out of the
source tree and into a catalog keyed on the upstream `msgid`, so the call site keeps upstream's
string verbatim and carries no diff at all.
Typical uses: white-labelling, or suppressing instructions that don't apply to how you run the app —
e.g. telling users to set `WEBDRIVER_URL` when the operator, not the user, controls that.
## How it works
`BABEL_TRANSLATION_DIRECTORIES` is a `;` separated list. For each locale, Flask-Babel loads one
catalog per directory and merges them in order (`Domain.get_translations` → babel
`Translations.merge` → `dict.update`), so a **later directory overrides an earlier one per message**,
not per file.
```
changedetectionio/translations/ <- base, upstream
changedetectionio/translations_overlay/ <- this layer, merged on top
```
So for German, `translations/de` and `translations_overlay/de` become a single catalog: the msgids
listed in the overlay take the overlay's text, and the other ~580 keep their upstream German. A
language with no overlay catalog is completely unaffected, as is a msgid the overlay doesn't mention.
Overriding **English** works too. The `en_GB` base catalog has every `msgstr` empty, so English
currently renders straight from the msgids; an overlay `en_GB` entry fills that gap.
Wiring is in `changedetectionio/flask_app.py`. The directory is picked up only if it exists, and
`TRANSLATION_OVERLAY_DIR` can point somewhere else (e.g. a path mounted into a container). No
directory, or a directory with no overrides, means no behaviour change.
## The failure mode this needs guarding
An override matches on the **exact upstream msgid**. When upstream edits that string — even fixing a
typo — the override stops matching and the string silently reverts to upstream wording. It fails
soft, so nothing tells you.
`manage.py check` turns that into a hard failure, and `tests/test_translation_overlay.py` runs it in
CI. It also catches three other quiet ways an override does nothing:
| Problem | Why it bites |
|---|---|
| msgid no longer in `messages.pot` | upstream reworded it; your override is dead |
| empty `msgstr` | the compiler drops empty entries, so it silently does nothing |
| `Plural-Forms` differs from base | Flask-Babel copies `plural` from the last catalog that has one, so a wrong header here breaks plurals for **every** string in that language |
| `.mo` stale or missing | the `.mo` is what's loaded; `.po` edits alone have no effect |
## Workflow
```bash
# create a catalog for a language (inherits Plural-Forms from the base catalog)
python changedetectionio/translations_overlay/manage.py add de
# ... add msgid/msgstr pairs, copying the msgid verbatim from translations/messages.pot ...
python changedetectionio/translations_overlay/manage.py compile # .po -> .mo
python changedetectionio/translations_overlay/manage.py check # validate (also runs in CI)
```
Both `.po` and `.mo` are committed, matching how the base catalogs are handled.
Overriding a string in English only is fine — other languages keep their upstream translation of the
original wording. If the reworded English changes the *meaning* rather than the phrasing, add the
matching override per language, otherwise the translations will drift from what English now says.
Writing the override text itself follows the same rules as any other catalog entry: see
[`../translations/README.md`](../translations/README.md), especially "do not fragment msgids".
@@ -0,0 +1,198 @@
#!/usr/bin/env python3
"""Manage the deployment wording overlay catalogs.
The overlay is a second gettext tree merged on top of ``changedetectionio/translations`` at
runtime (see the BABEL_TRANSLATION_DIRECTORIES block in ``changedetectionio/flask_app.py``).
Flask-Babel merges the two catalogs per-locale, later directory winning per-message, so an
overlay catalog only needs the msgids whose wording this deployment changes.
The catch this tooling exists to manage: an override is keyed on the exact upstream msgid.
When upstream edits a string - even a typo fix - the override stops matching and that string
silently reverts to upstream wording. ``check`` turns that silent revert into a hard failure.
python changedetectionio/translations_overlay/manage.py check
python changedetectionio/translations_overlay/manage.py compile
python changedetectionio/translations_overlay/manage.py add de
"""
import argparse
import os
import sys
from babel.messages.mofile import read_mo, write_mo
from babel.messages.pofile import read_po, write_po
OVERLAY_DIR = os.path.dirname(os.path.abspath(__file__))
BASE_DIR = os.path.join(os.path.dirname(OVERLAY_DIR), 'translations')
POT_FILE = os.path.join(BASE_DIR, 'messages.pot')
def _po_path(root, locale):
return os.path.join(root, locale, 'LC_MESSAGES', 'messages.po')
def _mo_path(root, locale):
return os.path.join(root, locale, 'LC_MESSAGES', 'messages.mo')
def overlay_locales():
"""Locales that have an overlay catalog, sorted."""
return sorted(d for d in os.listdir(OVERLAY_DIR) if os.path.isfile(_po_path(OVERLAY_DIR, d)))
def _load(path):
with open(path, 'rb') as fp:
return read_po(fp)
def _message_ids(catalog):
"""Real (non-header, non-obsolete) message ids in a catalog."""
return {m.id for m in catalog if m.id}
def cmd_check(_args):
"""Validate every overlay catalog against the base catalogs. Returns an exit code."""
if not os.path.isfile(POT_FILE):
print(
f"error: {POT_FILE} missing - run `python setup.py extract_messages` first",
file=sys.stderr,
)
return 1
known_ids = _message_ids(_load(POT_FILE))
problems = []
for locale in overlay_locales():
overlay = _load(_po_path(OVERLAY_DIR, locale))
# 1. Every overridden msgid must still exist upstream, or the override is dead weight.
for message in overlay:
if message.id and message.id not in known_ids:
problems.append(
f"{locale}: msgid no longer exists in messages.pot, override is dead:\n"
f" {message.id!r}"
)
# 2. An override with an empty msgstr is dropped by the compiler, so it does nothing
# while looking like it does something.
for message in overlay:
if message.id and not message.string:
problems.append(
f"{locale}: empty msgstr, override will be a no-op:\n {message.id!r}"
)
base_po = _po_path(BASE_DIR, locale)
if not os.path.isfile(base_po):
problems.append(f"{locale}: no base catalog at {base_po} - is this a real locale?")
else:
# 3. Flask-Babel copies `plural` from the last catalog that has one, so a wrong
# Plural-Forms header here breaks plurals for every string in the language,
# not just the overridden ones.
base_plural = _load(base_po).plural_expr
if overlay.plural_expr != base_plural:
problems.append(
f"{locale}: Plural-Forms disagrees with the base catalog and would override it\n"
f" overlay: {overlay.plural_expr}\n"
f" base: {base_plural}"
)
# 4. The .mo is what actually gets loaded; a stale one means edits to the .po do nothing.
mo_file = _mo_path(OVERLAY_DIR, locale)
expected = {m.id for m in overlay if m.id and m.string}
if not expected:
if os.path.isfile(mo_file) and _message_ids(_read_mo(mo_file)):
problems.append(
f"{locale}: .po has no overrides but .mo still contains some - recompile"
)
elif not os.path.isfile(mo_file):
problems.append(
f"{locale}: {os.path.basename(mo_file)} missing - run `manage.py compile`"
)
elif _message_ids(_read_mo(mo_file)) != expected:
problems.append(f"{locale}: .mo is out of date with .po - run `manage.py compile`")
if problems:
print(f"{len(problems)} problem(s) in the translation overlay:\n", file=sys.stderr)
for problem in problems:
print(f" - {problem}", file=sys.stderr)
return 1
total = sum(
len({m.id for m in _load(_po_path(OVERLAY_DIR, loc)) if m.id}) for loc in overlay_locales()
)
print(
f"overlay ok: {total} override(s) across {len(overlay_locales())} locale(s): {', '.join(overlay_locales()) or '-'}"
)
return 0
def _read_mo(path):
with open(path, 'rb') as fp:
return read_mo(fp)
def cmd_compile(_args):
for locale in overlay_locales():
catalog = _load(_po_path(OVERLAY_DIR, locale))
mo_file = _mo_path(OVERLAY_DIR, locale)
with open(mo_file, 'wb') as fp:
write_mo(fp, catalog)
overrides = len({m.id for m in catalog if m.id and m.string})
print(
f"compiled {locale}: {overrides} override(s) -> {os.path.relpath(mo_file, os.getcwd())}"
)
return 0
def cmd_add(args):
"""Bootstrap an empty overlay catalog for a locale, inheriting the base catalog's headers."""
locale = args.locale
base_po = _po_path(BASE_DIR, locale)
if not os.path.isfile(base_po):
print(f"error: no base catalog for {locale!r} at {base_po}", file=sys.stderr)
return 1
target = _po_path(OVERLAY_DIR, locale)
if os.path.exists(target):
print(f"error: {target} already exists", file=sys.stderr)
return 1
base = _load(base_po)
# Take the language metadata (crucially Plural-Forms) from base, but none of the messages.
catalog = type(base)(
locale=base.locale,
domain='messages',
project=base.project,
msgid_bugs_address=base.msgid_bugs_address,
header_comment=(
f'# Deployment wording overlay for changedetection.io - {locale}\n'
'# Merged on top of the base catalog for this language; only list the msgids\n'
'# whose wording this deployment changes. See ../../README.md.\n'
),
)
os.makedirs(os.path.dirname(target), exist_ok=True)
with open(target, 'wb') as fp:
write_po(fp, catalog, width=120)
print(f"created {os.path.relpath(target, os.getcwd())}")
print("add your msgid/msgstr pairs, then run `manage.py compile`")
return 0
def main():
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
sub = parser.add_subparsers(dest='command', required=True)
sub.add_parser(
'check', help='validate overlay catalogs against the base catalogs'
).set_defaults(fn=cmd_check)
sub.add_parser('compile', help='compile overlay .po files to .mo').set_defaults(fn=cmd_compile)
add = sub.add_parser('add', help='create an overlay catalog for a locale')
add.add_argument('locale')
add.set_defaults(fn=cmd_add)
args = parser.parse_args()
sys.exit(args.fn(args))
if __name__ == '__main__':
main()
+48 -3
View File
@@ -28,7 +28,7 @@ info:
For example: `x-api-key: YOUR_API_KEY`
version: 0.1.8
version: 0.1.9
contact:
name: ChangeDetection.io
url: https://github.com/dgtlmoon/changedetection.io
@@ -299,13 +299,24 @@ components:
maxLength: 5000
tag:
type: string
description: Tag UUID to associate with this web page change monitor (watch)
deprecated: true
description: |
**Deprecated - use `tags` instead.** Kept working for API v1 and may be removed in v2.
Takes comma-separated tag *names* (not UUIDs, despite what earlier revisions of this
document said). Names are matched case-insensitively against existing tags and created
when they don't exist yet. An existing tag's UUID is also accepted and resolves to that
tag; a UUID matching no tag is ignored.
maxLength: 5000
example: "Production, Price tracking"
tags:
type: array
items:
type: string
description: Array of tag UUIDs
description: |
Array of tag UUIDs. Unlike `tag`, these are applied exactly as given - no tag is ever
created from this field. The UUIDs are not validated: an unknown one is stored but
resolves to no group.
paused:
type: boolean
description: Whether the web page change monitor (watch) is paused
@@ -1260,6 +1271,40 @@ paths:
'404':
description: Web page change monitor (watch) not found
delete:
operationId: deleteWatchHistory
tags: [Watch History]
summary: Delete watch history
description: |
Delete/clear all historical snapshots for a web page change monitor (watch), the watch itself is kept.
This also resets the "last checked"/"last changed" state, so the watch will be re-checked on the next cycle.
x-code-samples:
- lang: 'curl'
source: |
curl -X DELETE "http://localhost:5000/api/v1/watch/095be615-a8ad-4c33-8e9c-c7612fbf6c9f/history" \
-H "x-api-key: YOUR_API_KEY"
- lang: 'Python'
source: |
import requests
headers = {'x-api-key': 'YOUR_API_KEY'}
uuid = '095be615-a8ad-4c33-8e9c-c7612fbf6c9f'
response = requests.delete(f'http://localhost:5000/api/v1/watch/{uuid}/history', headers=headers)
print(response.status_code)
parameters:
- name: uuid
in: path
required: true
description: Web page change monitor (watch) unique ID
schema:
type: string
format: uuid
responses:
'204':
description: History deleted successfully
'404':
description: Web page change monitor (watch) not found
/watch/{uuid}/history/{timestamp}:
get:
operationId: getWatchSnapshot
+95 -26
View File
File diff suppressed because one or more lines are too long