linux observer
test(sync): freeze datetime in retention tests to fix date-fragility
The zero-byte quarantine test hardcoded day 20260410 and asserted the
.failed dir survived cleanup. Once wall-clock time passed 2026-04-17
that day fell outside the 7-day retention window and cleanup correctly
deleted it, flipping the test red. Freeze sync.datetime.now() to keep
20260410 inside retention, and add a companion test in
TestCleanupFailedSegments covering the kept-within-retention case.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+32
-4
+32
-4
tests/test_sync.py
+32
-4
tests/test_sync.py
···
5
5
import json
6
6
import os
7
7
import time
8
+
from datetime import datetime
8
9
from pathlib import Path
9
10
from unittest.mock import AsyncMock, patch
10
11
···
409
410
"""A segment with all zero-byte files is renamed to .failed before upload."""
410
411
sync = self._make_sync(tmp_path)
411
412
captures = sync._config.captures_dir
413
+
fake_now = datetime(2026, 4, 11, 12, 0, 0)
412
414
413
415
seg = self._create_zero_byte_segment(
414
416
captures, "20260410", "archon", "120000_300"
415
417
)
416
418
server_response = []
417
419
418
-
with patch(
419
-
"asyncio.to_thread", new_callable=AsyncMock, return_value=server_response
420
-
):
421
-
await sync._sync()
420
+
with patch("solstone_linux.sync.datetime", wraps=datetime) as mock_datetime:
421
+
mock_datetime.now.return_value = fake_now
422
+
with patch(
423
+
"asyncio.to_thread",
424
+
new_callable=AsyncMock,
425
+
return_value=server_response,
426
+
):
427
+
await sync._sync()
422
428
423
429
assert not seg.exists()
424
430
assert seg.with_name("120000_300.failed").exists()
···
806
812
807
813
assert (captures / "20260101" / "archon" / "120000_300.failed").exists()
808
814
mock_thread.assert_not_called()
815
+
816
+
@pytest.mark.asyncio
817
+
async def test_failed_segments_kept_within_retention(self, tmp_path: Path):
818
+
""".failed segments are kept when the synced day is still within retention."""
819
+
sync = self._make_sync(tmp_path, retention=7)
820
+
captures = sync._config.captures_dir
821
+
fake_now = datetime(2026, 1, 8, 12, 0, 0)
822
+
823
+
seg = self._create_segment(captures, "20260107", "archon", "120000_300.failed")
824
+
sync._synced_days.add("20260107")
825
+
826
+
server_response = []
827
+
with patch("solstone_linux.sync.datetime", wraps=datetime) as mock_datetime:
828
+
mock_datetime.now.return_value = fake_now
829
+
with patch(
830
+
"asyncio.to_thread",
831
+
new_callable=AsyncMock,
832
+
return_value=server_response,
833
+
):
834
+
await sync._cleanup_synced_segments()
835
+
836
+
assert seg.exists()
809
837
810
838
@pytest.mark.asyncio
811
839
async def test_incomplete_still_skipped(self, tmp_path: Path):