From b9a2f781acbf9a8ae4098a41baae814ffb1fa7f9 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Wed, 18 Feb 2026 17:29:45 +0100 Subject: [PATCH] Backups was missing tags --- .../blueprint/backups/__init__.py | 15 +++- changedetectionio/tests/test_backup.py | 83 ++++++++++++++++++- 2 files changed, 96 insertions(+), 2 deletions(-) diff --git a/changedetectionio/blueprint/backups/__init__.py b/changedetectionio/blueprint/backups/__init__.py index 6959c8c18..1eff4ff4c 100644 --- a/changedetectionio/blueprint/backups/__init__.py +++ b/changedetectionio/blueprint/backups/__init__.py @@ -13,11 +13,14 @@ from loguru import logger BACKUP_FILENAME_FORMAT = "changedetection-backup-{}.zip" -def create_backup(datastore_path, watches: dict): +def create_backup(datastore_path, watches: dict, tags: dict = None): logger.debug("Creating backup...") import zipfile from pathlib import Path + if tags is None: + tags = {} + # create a ZipFile object timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") backupname = BACKUP_FILENAME_FORMAT.format(timestamp) @@ -45,6 +48,15 @@ def create_backup(datastore_path, watches: dict): if os.path.isfile(secret_file): zipObj.write(secret_file, arcname="secret.txt") + # Add tag data directories (each tag has its own {uuid}/tag.json) + for uuid, tag in tags.items(): + for f in Path(tag.data_dir).glob('*'): + zipObj.write(f, + arcname=os.path.join(f.parts[-2], f.parts[-1]), + compress_type=zipfile.ZIP_DEFLATED, + compresslevel=8) + logger.debug(f"Added tag '{tag.get('title')}' ({uuid}) to backup") + # Add any data in the watch data directory. for uuid, w in watches.items(): for f in Path(w.data_dir).glob('*'): @@ -109,6 +121,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): zip_thread = threading.Thread( target=create_backup, args=(datastore.datastore_path, datastore.data.get("watching")), + kwargs={'tags': datastore.data['settings']['application'].get('tags', {})}, daemon=True, name="BackupCreator" ) diff --git a/changedetectionio/tests/test_backup.py b/changedetectionio/tests/test_backup.py index d5da8c15b..245446afa 100644 --- a/changedetectionio/tests/test_backup.py +++ b/changedetectionio/tests/test_backup.py @@ -6,6 +6,7 @@ import io from zipfile import ZipFile import re import time +from changedetectionio.model import Watch, Tag def test_backup(client, live_server, measure_memory_usage, datastore_path): @@ -112,4 +113,84 @@ def test_watch_data_package_download(client, live_server, measure_memory_usage, # Should contain history/snapshot files uuid4hex_txt = re.compile(f'^{re.escape(uuid)}/.*\\.txt', re.I) txt_files = list(filter(uuid4hex_txt.match, files)) - assert len(txt_files) > 0, f"Should have at least one .txt file (history/snapshot), got: {files}" \ No newline at end of file + assert len(txt_files) > 0, f"Should have at least one .txt file (history/snapshot), got: {files}" + + +def test_backup_restore(client, live_server, measure_memory_usage, datastore_path): + """Test that a full backup zip can be restored — watches and tags survive a round-trip.""" + + set_original_response(datastore_path=datastore_path) + + datastore = live_server.app.config['DATASTORE'] + watch_url = url_for('test_endpoint', _external=True) + + # Set up: one watch and two tags + uuid = datastore.add_watch(url=watch_url) + tag_uuid = datastore.add_tag(title="Tasty backup tag") + tag_uuid2 = datastore.add_tag(title="Tasty backup tag number two") + + client.get(url_for("ui.form_watch_checknow"), follow_redirects=True) + wait_for_all_checks(client) + + # Create a full backup + client.get(url_for("backups.request_backup"), follow_redirects=True) + time.sleep(4) + + # Download the latest backup zip + res = client.get(url_for("backups.download_backup", filename="latest"), follow_redirects=True) + assert res.content_type == "application/zip" + zip_data = res.data + + # Confirm the zip contains both watch.json and tag.json entries + backup = ZipFile(io.BytesIO(zip_data)) + names = backup.namelist() + assert f"{uuid}/watch.json" in names, f"watch.json missing from backup: {names}" + assert f"{tag_uuid}/tag.json" in names, f"tag.json for tag 1 missing from backup: {names}" + assert f"{tag_uuid2}/tag.json" in names, f"tag.json for tag 2 missing from backup: {names}" + + # --- Wipe everything --- + datastore.delete('all') + client.get(url_for("tags.delete_all"), follow_redirects=True) + + assert uuid not in datastore.data['watching'], "Watch should be gone after delete" + assert tag_uuid not in datastore.data['settings']['application']['tags'], "Tag 1 should be gone after delete" + assert tag_uuid2 not in datastore.data['settings']['application']['tags'], "Tag 2 should be gone after delete" + + # --- Restore from the backup zip --- + res = client.post( + url_for("backups.restore.backups_restore_start"), + data={ + 'zip_file': (io.BytesIO(zip_data), 'backup.zip'), + 'include_groups': 'y', + 'include_groups_replace_existing': 'y', + 'include_watches': 'y', + 'include_watches_replace_existing': 'y', + }, + content_type='multipart/form-data', + follow_redirects=True + ) + assert res.status_code == 200 + + # --- Watch checks --- + restored_watch = datastore.data['watching'].get(uuid) + assert restored_watch is not None, f"Watch {uuid} not found after restore" + assert restored_watch['url'] == watch_url, "Restored watch URL does not match" + assert isinstance(restored_watch, Watch.model), \ + f"Watch not properly rehydrated, got {type(restored_watch)}" + assert restored_watch.history_n >= 1, \ + f"Restored watch should have at least 1 history entry, got {restored_watch.history_n}" + + # --- Tag checks --- + restored_tags = datastore.data['settings']['application']['tags'] + + restored_tag = restored_tags.get(tag_uuid) + assert restored_tag is not None, f"Tag {tag_uuid} not found after restore" + assert restored_tag['title'] == "Tasty backup tag", "Restored tag 1 title does not match" + assert isinstance(restored_tag, Tag.model), \ + f"Tag 1 not properly rehydrated, got {type(restored_tag)}" + + restored_tag2 = restored_tags.get(tag_uuid2) + assert restored_tag2 is not None, f"Tag {tag_uuid2} not found after restore" + assert restored_tag2['title'] == "Tasty backup tag number two", "Restored tag 2 title does not match" + assert isinstance(restored_tag2, Tag.model), \ + f"Tag 2 not properly rehydrated, got {type(restored_tag2)}" \ No newline at end of file