tests/test_logging_utils.py

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