music on atproto
plyr.fm
zzstoatzz.io
1"""Tests for progress tracking utilities."""
2
3import pytest
4
5from backend.storage.r2 import UploadProgressTracker
6
7
8class TestUploadProgressTracker:
9 """Tests for UploadProgressTracker bytes-to-percentage conversion."""
10
11 def test_converts_bytes_to_percentage(self):
12 """Callback receives bytes, passes percentage to inner callback."""
13 received_percentages: list[float] = []
14
15 def capture_pct(pct: float) -> None:
16 received_percentages.append(pct)
17
18 tracker = UploadProgressTracker(
19 total_size=1000,
20 callback=capture_pct,
21 min_bytes_between_updates=100, # Update every 100 bytes
22 min_time_between_updates=0, # No time throttling for test
23 )
24
25 # Simulate boto3 calling with bytes
26 tracker(100) # 10%
27 tracker(200) # 30%
28 tracker(300) # 60%
29 tracker(400) # 100%
30
31 # Should have received percentage values, not bytes
32 assert len(received_percentages) == 4
33 assert received_percentages[0] == pytest.approx(10.0)
34 assert received_percentages[1] == pytest.approx(30.0)
35 assert received_percentages[2] == pytest.approx(60.0)
36 assert received_percentages[3] == pytest.approx(100.0)
37
38 def test_throttles_by_bytes_and_time(self):
39 """Updates are throttled based on bytes OR time threshold (OR logic)."""
40 received_percentages: list[float] = []
41
42 def capture_pct(pct: float) -> None:
43 received_percentages.append(pct)
44
45 # With min_time=0, every call triggers based on time threshold
46 # This matches the OR logic in UploadProgressTracker
47 tracker = UploadProgressTracker(
48 total_size=1000,
49 callback=capture_pct,
50 min_bytes_between_updates=500,
51 min_time_between_updates=0, # Time threshold always met
52 )
53
54 # Every call triggers because time threshold (0) is always satisfied
55 tracker(100)
56 tracker(100)
57 assert len(received_percentages) == 2
58
59 def test_always_emits_near_completion(self):
60 """Always emits update when near 100% completion."""
61 received_percentages: list[float] = []
62
63 def capture_pct(pct: float) -> None:
64 received_percentages.append(pct)
65
66 tracker = UploadProgressTracker(
67 total_size=100,
68 callback=capture_pct,
69 min_bytes_between_updates=1000, # Very high threshold
70 min_time_between_updates=1000, # Very high threshold
71 )
72
73 # Near completion (99.9%+) should always emit
74 tracker(100) # 100%
75 assert len(received_percentages) == 1
76 assert received_percentages[0] == pytest.approx(100.0)