Coverage for tests/test_progress_file.py: 100%

1import io 7 ctx1hijklmn

2from datetime import datetime, timezone 7 ctx1hijklmn

3from unittest.mock import Mock, call 7 ctx1hijklmn

4 

5import time_machine 7 ctx1hijklmn

6 

7from fastapi_cloud_cli.utils.progress_file import ProgressFile 7 ctx1hijklmn

8 

9 

10def _make_file( 7 ctx1hijklmn

11 content: bytes = b"hello world", name: str = "test.tar.gz" 

12) -> io.BytesIO: 

13 f = io.BytesIO(content) 49 ctx1oCDapXQqEFbrYRsGHctZSuIJdv0TwKLex1UyMNfz2VAOPgB3W

14 f.name = name 49 ctx1oCDapXQqEFbrYRsGHctZSuIJdv0TwKLex1UyMNfz2VAOPgB3W

15 return f 49 ctx1oCDapXQqEFbrYRsGHctZSuIJdv0TwKLex1UyMNfz2VAOPgB3W

16 

17 

18def test_read_with_size() -> None: 7 ctx1hijklmn

19 file = _make_file(b"abcdef") 7 ctx1QRSTUVW

20 pf = ProgressFile(file, progress_callback=lambda _: None) 7 ctx1QRSTUVW

21 

22 assert pf.read(3) == b"abc" 7 ctx1QRSTUVW

23 assert pf.read(3) == b"def" 7 ctx1QRSTUVW

24 

25 

26def test_callback_not_called_within_interval() -> None: 7 ctx1hijklmn

27 file = _make_file(b"abcdef") 7 ctx1DFHJLNP

28 mock_callback = Mock() 7 ctx1DFHJLNP

29 pf = ProgressFile(file, progress_callback=mock_callback) 7 ctx1DFHJLNP

30 

31 pf.read(3) # Should trigger callback 7 ctx1DFHJLNP

32 pf.read(3) # Should NOT trigger 7 ctx1DFHJLNP

33 

34 mock_callback.assert_called_once_with(3) 7 ctx1DFHJLNP

35 

36 

37def test_callback_called_after_interval_elapses() -> None: 7 ctx1hijklmn

38 file = _make_file(b"abcdef") 7 ctx1oqsuwyA

39 mock_callback = Mock() 7 ctx1oqsuwyA

40 

41 with time_machine.travel( 7 ctx1oqsuwyA

42 datetime(2026, 1, 1, tzinfo=timezone.utc), tick=False 

43 ) as traveller: 

44 pf = ProgressFile(file, progress_callback=mock_callback) 7 ctx1oqsuwyA

45 

46 pf.read(3) 7 ctx1oqsuwyA

47 traveller.shift(0.6) 7 ctx1oqsuwyA

48 pf.read(3) 7 ctx1oqsuwyA

49 

50 mock_callback.assert_has_calls([call(3), call(6)]) 7 ctx1oqsuwyA

51 

52 

53def test_callback_tracks_cumulative_bytes() -> None: 7 ctx1hijklmn

54 file = _make_file(b"a" * 100) 7 ctx1abcdefg

55 mock_callback = Mock() 7 ctx1abcdefg

56 

57 with time_machine.travel( 7 ctx1abcdefg

58 datetime(2026, 1, 1, tzinfo=timezone.utc), tick=False 

59 ) as traveller: 

60 pf = ProgressFile(file, progress_callback=mock_callback) 7 ctx1abcdefg

61 

62 pf.read(10) # Should trigger callback with 10 bytes read 7 ctx1abcdefg

63 traveller.shift(0.1) 7 ctx1abcdefg

64 pf.read(10) 7 ctx1abcdefg

65 traveller.shift(0.5) 7 ctx1abcdefg

66 pf.read(10) # Should trigger callback with 10 + 10 + 10 = 30 bytes read 7 ctx1abcdefg

67 traveller.shift(0.6) 7 ctx1abcdefg

68 pf.read(10) # Should trigger callback with 30 + 10 = 40 bytes read 7 ctx1abcdefg

69 

70 mock_callback.assert_has_calls([call(10), call(30), call(40)]) 7 ctx1abcdefg

71 

72 

73def test_callback_called_on_eof() -> None: 7 ctx1hijklmn

74 file = _make_file(b"abcd") 7 ctx1CEGIKMO

75 mock_callback = Mock() 7 ctx1CEGIKMO

76 

77 pf = ProgressFile(file, progress_callback=mock_callback) 7 ctx1CEGIKMO

78 pf.read(3) 7 ctx1CEGIKMO

79 pf.read(3) 7 ctx1CEGIKMO

80 mock_callback.assert_has_calls([call(3), call(4)]) 7 ctx1CEGIKMO

81 

82 

83def test_name_property() -> None: 7 ctx1hijklmn

84 file = _make_file(name="test.tar.gz") 7 ctx1XYZ0123

85 pf = ProgressFile(file, progress_callback=lambda _: None) 7 ctx1XYZ0123

86 

87 assert pf.name == "test.tar.gz" 7 ctx1XYZ0123

88 

89 

90def test_callback_uses_current_file_position_after_seek() -> None: 7 ctx1hijklmn

91 file = _make_file(b"abcde") 7 ctx1prtvxzB

92 mock_callback = Mock() 7 ctx1prtvxzB

93 

94 pf = ProgressFile(file, progress_callback=mock_callback) 7 ctx1prtvxzB

95 

96 pf.read(3) 7 ctx1prtvxzB

97 

98 # Imitate retrying 

99 pf.seek(0) 7 ctx1prtvxzB

100 pf.read(3) 7 ctx1prtvxzB

101 

102 pf.read(3) 7 ctx1prtvxzB

103 

104 mock_callback.assert_has_calls([call(3), call(5)]) 7 ctx1prtvxzB