mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
f15e54c383
The Windows installer's embedded Python doesn't carry an IANA tz
database, and the stdlib zoneinfo has no system DB to read on Windows.
ZoneInfo("UTC") raises ZoneInfoNotFoundError on those installs, and
the new /api/local-backup/status endpoint 500s on the resulting
uncaught exception. Surfaced via a Windows traceback from a user's log:
File "...\backend\app\services\local_backup.py", line 32, in _local_zone
return ZoneInfo("UTC")
zoneinfo._common.ZoneInfoNotFoundError: 'No time zone found with key UTC'
_local_zone()'s try/except only covered the TZ-env branch — both
fallbacks unconditionally called ZoneInfo("UTC") and re-raised.
Fix (two parts):
1. services/local_backup.py — return type widened from ZoneInfo to
tzinfo, the UTC fallback is wrapped in its own try, and the
last-resort fallback returns datetime.timezone.utc (stdlib, no
IANA DB needed). str(timezone.utc) == "UTC" so the response shape
on /api/local-backup/status is unchanged. The astimezone call in
_calculate_next_run accepts any tzinfo — no other call sites
affected.
2. requirements.txt — pin tzdata>=2024.1; sys_platform == "win32" so
the next Windows installer build ships the IANA DB, and any non-
UTC TZ value (e.g. Europe/Berlin) resolves correctly. The stdlib
fallback can only ever give UTC. Linux/macOS unaffected by the
platform marker — they already have the system tz database.
327 lines
14 KiB
Python
327 lines
14 KiB
Python
"""Unit tests for scheduled local backup service (#884)."""
|
|
|
|
import tempfile
|
|
import zipfile
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.local_backup import LocalBackupService
|
|
|
|
|
|
class TestCalculateNextRun:
|
|
"""Tests for _calculate_next_run scheduling logic.
|
|
|
|
The HH:MM picker is interpreted in the container's local timezone (TZ env
|
|
var, UTC fallback). Each test pins TZ so the assertions don't depend on
|
|
the test runner's environment.
|
|
"""
|
|
|
|
def test_hourly_returns_next_full_hour(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 14, 30, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("hourly", "03:00")
|
|
assert result.hour == 15
|
|
assert result.minute == 0
|
|
|
|
def test_daily_before_target_time_schedules_today_utc(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "03:00")
|
|
assert result == datetime(2026, 4, 12, 3, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_daily_after_target_time_schedules_tomorrow_utc(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 4, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "03:00")
|
|
assert result == datetime(2026, 4, 13, 3, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_weekly_adds_full_week_utc(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("weekly", "03:00")
|
|
assert result == datetime(2026, 4, 19, 3, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_weekly_after_target_time_adds_full_week_from_tomorrow_utc(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 4, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("weekly", "03:00")
|
|
assert result == datetime(2026, 4, 20, 3, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_invalid_time_defaults_to_0300(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "invalid")
|
|
assert result.hour == 3
|
|
assert result.minute == 0
|
|
|
|
def test_unknown_schedule_type_defaults_to_daily(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "UTC")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("every_5_min", "03:00")
|
|
# Should fall through to daily behavior (time-based)
|
|
assert result.hour == 3
|
|
|
|
def test_daily_berlin_local_time_converts_to_utc(self, monkeypatch):
|
|
"""User in Europe/Berlin entering 21:00 should run at 19:00 UTC (CEST/UTC+2)."""
|
|
monkeypatch.setenv("TZ", "Europe/Berlin")
|
|
service = LocalBackupService()
|
|
# Mid-June: Europe/Berlin is CEST (+02:00)
|
|
now = datetime(2026, 6, 15, 10, 0, 0, tzinfo=timezone.utc) # 12:00 Berlin
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "21:00")
|
|
# 21:00 Berlin (CEST, +02:00) on 2026-06-15 == 19:00 UTC same day
|
|
assert result == datetime(2026, 6, 15, 19, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_daily_istanbul_local_time_converts_to_utc(self, monkeypatch):
|
|
"""The #1602 reporter: UTC+3 user entering 21:00 should run at 18:00 UTC."""
|
|
monkeypatch.setenv("TZ", "Europe/Istanbul")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 6, 15, 10, 0, 0, tzinfo=timezone.utc) # 13:00 Istanbul
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "21:00")
|
|
assert result == datetime(2026, 6, 15, 18, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_no_tz_env_falls_back_to_utc(self, monkeypatch):
|
|
monkeypatch.delenv("TZ", raising=False)
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 6, 15, 10, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "21:00")
|
|
# No TZ → behaves as UTC: 21:00 today is in the future of 10:00, so today
|
|
assert result == datetime(2026, 6, 15, 21, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_unrecognised_tz_falls_back_to_utc(self, monkeypatch):
|
|
monkeypatch.setenv("TZ", "Not/A_Real_Zone")
|
|
service = LocalBackupService()
|
|
now = datetime(2026, 6, 15, 10, 0, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
result = service._calculate_next_run("daily", "21:00")
|
|
assert result == datetime(2026, 6, 15, 21, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_zoneinfo_completely_unavailable_falls_back_to_stdlib_utc(self, monkeypatch):
|
|
"""Windows installer ships an embedded Python without the IANA tz DB
|
|
(no system tzdata, no ``tzdata`` PyPI package). Even ``ZoneInfo("UTC")``
|
|
raises ``ZoneInfoNotFoundError`` then, and /api/local-backup/status
|
|
500s. The fallback must catch that and return ``datetime.timezone.utc``
|
|
so scheduling still works without the DB.
|
|
"""
|
|
from zoneinfo import ZoneInfoNotFoundError
|
|
|
|
from backend.app.services import local_backup as lb_module
|
|
|
|
monkeypatch.delenv("TZ", raising=False)
|
|
|
|
def _always_missing(_key):
|
|
raise ZoneInfoNotFoundError("no tz database on this platform")
|
|
|
|
monkeypatch.setattr(lb_module, "ZoneInfo", _always_missing)
|
|
assert lb_module._local_zone() is timezone.utc
|
|
|
|
def test_dst_spring_forward_gap_does_not_crash(self, monkeypatch):
|
|
"""Europe/Berlin spring-forward 2026-03-29 jumps 02:00 → 03:00 local;
|
|
02:30 wall-clock does not exist. ``replace(hour=2, minute=30)`` should
|
|
still normalise to a valid UTC instant via astimezone, not raise.
|
|
"""
|
|
monkeypatch.setenv("TZ", "Europe/Berlin")
|
|
service = LocalBackupService()
|
|
# 2026-03-29 00:30 UTC == 01:30 Berlin (CET, just before the gap)
|
|
now = datetime(2026, 3, 29, 0, 30, 0, tzinfo=timezone.utc)
|
|
with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
mock_dt.now.return_value = now
|
|
mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
# 02:30 local is in the non-existent gap on that day.
|
|
result = service._calculate_next_run("daily", "02:30")
|
|
# Result must be a UTC-aware datetime — exact value depends on
|
|
# zoneinfo's gap normalisation; we just guarantee no crash and that
|
|
# the run is in the future of ``now``.
|
|
assert result.tzinfo == timezone.utc
|
|
assert result > now
|
|
|
|
|
|
class TestPruneBackups:
|
|
"""Tests for backup retention pruning."""
|
|
|
|
def test_prune_keeps_retention_count(self, tmp_path):
|
|
service = LocalBackupService()
|
|
# Create 5 backup files
|
|
for i in range(5):
|
|
f = tmp_path / f"bambuddy-backup-20260412-{i:06d}.zip"
|
|
f.write_text(f"backup{i}")
|
|
service._prune_backups(tmp_path, retention=3)
|
|
remaining = list(tmp_path.glob("bambuddy-backup-*.zip"))
|
|
assert len(remaining) == 3
|
|
|
|
def test_prune_noop_when_under_retention(self, tmp_path):
|
|
service = LocalBackupService()
|
|
for i in range(2):
|
|
f = tmp_path / f"bambuddy-backup-20260412-{i:06d}.zip"
|
|
f.write_text(f"backup{i}")
|
|
service._prune_backups(tmp_path, retention=5)
|
|
remaining = list(tmp_path.glob("bambuddy-backup-*.zip"))
|
|
assert len(remaining) == 2
|
|
|
|
def test_prune_only_touches_matching_files(self, tmp_path):
|
|
service = LocalBackupService()
|
|
# Create backup files and a non-backup file
|
|
for i in range(3):
|
|
f = tmp_path / f"bambuddy-backup-20260412-{i:06d}.zip"
|
|
f.write_text(f"backup{i}")
|
|
other = tmp_path / "other_file.txt"
|
|
other.write_text("keep me")
|
|
service._prune_backups(tmp_path, retention=1)
|
|
assert other.exists()
|
|
remaining = list(tmp_path.glob("bambuddy-backup-*.zip"))
|
|
assert len(remaining) == 1
|
|
|
|
|
|
class TestResolveBackupFile:
|
|
"""Tests for backup file resolution with path traversal protection."""
|
|
|
|
def test_valid_filename(self, tmp_path):
|
|
service = LocalBackupService()
|
|
f = tmp_path / "bambuddy-backup-20260412-120000.zip"
|
|
f.write_text("data")
|
|
result = service.resolve_backup_file(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
assert result == f
|
|
|
|
def test_path_traversal_blocked(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.resolve_backup_file(str(tmp_path), "../etc/passwd")
|
|
assert result is None
|
|
|
|
def test_backslash_blocked(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.resolve_backup_file(str(tmp_path), "..\\etc\\passwd")
|
|
assert result is None
|
|
|
|
def test_dotdot_blocked(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.resolve_backup_file(str(tmp_path), "..bambuddy-backup.zip")
|
|
assert result is None
|
|
|
|
def test_wrong_prefix_blocked(self, tmp_path):
|
|
service = LocalBackupService()
|
|
f = tmp_path / "evil-file.zip"
|
|
f.write_text("data")
|
|
result = service.resolve_backup_file(str(tmp_path), "evil-file.zip")
|
|
assert result is None
|
|
|
|
def test_nonexistent_file(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.resolve_backup_file(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
assert result is None
|
|
|
|
|
|
class TestDeleteBackup:
|
|
"""Tests for backup deletion."""
|
|
|
|
def test_delete_valid_backup(self, tmp_path):
|
|
service = LocalBackupService()
|
|
f = tmp_path / "bambuddy-backup-20260412-120000.zip"
|
|
f.write_text("data")
|
|
result = service.delete_backup(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
assert result["success"] is True
|
|
assert not f.exists()
|
|
|
|
def test_delete_nonexistent_backup(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.delete_backup(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
assert result["success"] is False
|
|
|
|
def test_delete_path_traversal_blocked(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.delete_backup(str(tmp_path), "../important.zip")
|
|
assert result["success"] is False
|
|
|
|
|
|
class TestListBackups:
|
|
"""Tests for backup listing."""
|
|
|
|
def test_list_empty_dir(self, tmp_path):
|
|
service = LocalBackupService()
|
|
result = service.list_backups(str(tmp_path))
|
|
assert result == []
|
|
|
|
def test_list_nonexistent_dir(self):
|
|
service = LocalBackupService()
|
|
result = service.list_backups("/nonexistent/path/12345")
|
|
assert result == []
|
|
|
|
def test_list_only_matching_files(self, tmp_path):
|
|
service = LocalBackupService()
|
|
(tmp_path / "bambuddy-backup-20260412-120000.zip").write_text("a")
|
|
(tmp_path / "bambuddy-backup-20260412-130000.zip").write_text("bb")
|
|
(tmp_path / "other-file.txt").write_text("ccc")
|
|
result = service.list_backups(str(tmp_path))
|
|
assert len(result) == 2
|
|
assert all(r["filename"].startswith("bambuddy-backup-") for r in result)
|
|
|
|
def test_list_sorted_newest_first(self, tmp_path):
|
|
import time
|
|
|
|
service = LocalBackupService()
|
|
f1 = tmp_path / "bambuddy-backup-20260412-120000.zip"
|
|
f1.write_text("a")
|
|
time.sleep(0.05)
|
|
f2 = tmp_path / "bambuddy-backup-20260412-130000.zip"
|
|
f2.write_text("b")
|
|
result = service.list_backups(str(tmp_path))
|
|
assert result[0]["filename"] == "bambuddy-backup-20260412-130000.zip"
|
|
|
|
def test_list_includes_size(self, tmp_path):
|
|
service = LocalBackupService()
|
|
(tmp_path / "bambuddy-backup-20260412-120000.zip").write_bytes(b"x" * 1024)
|
|
result = service.list_backups(str(tmp_path))
|
|
assert result[0]["size"] == 1024
|
|
|
|
|
|
class TestGetStatus:
|
|
"""Tests for status reporting."""
|
|
|
|
def test_initial_status(self):
|
|
service = LocalBackupService()
|
|
status = service.get_status()
|
|
assert status["is_running"] is False
|
|
assert status["last_backup_at"] is None
|
|
assert status["last_status"] is None
|
|
assert status["next_run"] is None
|