Spaces:
Running on Zero
Running on Zero
File size: 4,448 Bytes
c4cb115 db79f31 c4cb115 db79f31 c4cb115 db79f31 c4cb115 db79f31 c4cb115 db79f31 c4cb115 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | import sys
from datetime import datetime, timezone, timedelta
from types import SimpleNamespace
from unittest.mock import MagicMock
sys.path.insert(0, ".")
from logging_utils import (
_prune_old_files,
_file_date,
_make_path,
_maybe_squash_history,
)
NOW = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
KEEP_DAYS = 7
def _make_api(paths):
api = MagicMock()
api.list_repo_tree.return_value = [SimpleNamespace(path=p) for p in paths]
return api
def _file(days_ago):
return _make_path(NOW - timedelta(days=days_ago), "abcd1234")
def _deleted_paths(api):
return {c.kwargs["path_in_repo"] for c in api.delete_file.call_args_list}
def test_file_date_matches_make_path():
"""_file_date(_make_path(now, uid)) must return now's date."""
import uuid
path = _make_path(NOW, uuid.uuid4().hex[:8])
assert _file_date(path) == NOW.strftime("%Y-%m-%d")
def test_old_files_are_deleted():
api = _make_api([_file(8), _file(10)])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
assert _deleted_paths(api) == {_file(8), _file(10)}
def test_recent_files_are_kept():
api = _make_api([_file(6), _file(1), _file(0)])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
api.delete_file.assert_not_called()
def test_boundary_exactly_keep_days_old_is_kept():
api = _make_api([_file(KEEP_DAYS)])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
api.delete_file.assert_not_called()
def test_one_day_past_boundary_is_deleted():
api = _make_api([_file(KEEP_DAYS + 1)])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
assert _deleted_paths(api) == {_file(KEEP_DAYS + 1)}
def test_only_old_files_deleted_in_mixed_set():
api = _make_api([_file(8), _file(6), _file(10), _file(0)])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
assert _deleted_paths(api) == {_file(8), _file(10)}
def test_empty_directory_does_nothing():
api = _make_api([])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
api.delete_file.assert_not_called()
def test_keep_days_zero_skips_pruning():
api = _make_api([_file(8)])
_prune_old_files(api, "org/repo", keep_days=0, now=NOW)
api.list_repo_tree.assert_not_called()
api.delete_file.assert_not_called()
def test_non_parquet_files_are_ignored():
api = _make_api(["data/2020-01-01-README.md", _file(8)])
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW)
assert _deleted_paths(api) == {_file(8)}
def test_api_error_does_not_raise():
api = MagicMock()
api.list_repo_tree.side_effect = RuntimeError("network error")
_prune_old_files(api, "org/repo", KEEP_DAYS, NOW) # must not raise
# ── _maybe_squash_history ────────────────────────────────────────────────────
def _squash_api():
api = MagicMock()
api.token = "tok"
return api
def test_squash_runs_when_no_marker(monkeypatch):
monkeypatch.setattr(
"logging_utils.hf_hub_download",
MagicMock(side_effect=FileNotFoundError("no marker")),
)
api = _squash_api()
_maybe_squash_history(api, "org/repo", NOW)
api.super_squash_history.assert_called_once()
api.upload_file.assert_called_once()
def test_squash_skipped_when_marker_is_today(monkeypatch, tmp_path):
marker = tmp_path / "last_squash.txt"
marker.write_text(NOW.strftime("%Y-%m-%d"))
monkeypatch.setattr(
"logging_utils.hf_hub_download", MagicMock(return_value=str(marker))
)
api = _squash_api()
_maybe_squash_history(api, "org/repo", NOW)
api.super_squash_history.assert_not_called()
def test_squash_runs_when_marker_is_yesterday(monkeypatch, tmp_path):
marker = tmp_path / "last_squash.txt"
marker.write_text((NOW - timedelta(days=1)).strftime("%Y-%m-%d"))
monkeypatch.setattr(
"logging_utils.hf_hub_download", MagicMock(return_value=str(marker))
)
api = _squash_api()
_maybe_squash_history(api, "org/repo", NOW)
api.super_squash_history.assert_called_once()
def test_squash_error_does_not_raise(monkeypatch):
monkeypatch.setattr(
"logging_utils.hf_hub_download",
MagicMock(side_effect=FileNotFoundError("no marker")),
)
api = _squash_api()
api.super_squash_history.side_effect = RuntimeError("squash failed")
_maybe_squash_history(api, "org/repo", NOW) # must not raise
|