Coverage for cli / tests / test_upload_files.py: 100%

1""" 

2Tests for `_upload_files` - concurrent S3 uploads via aiobotocore. 

3""" 

4 

5import asyncio 

6from collections.abc import Iterator 

7from contextlib import contextmanager 

8from pathlib import Path 

9from typing import Any 

10from unittest.mock import AsyncMock, MagicMock, patch 

11 

12import pytest 

13 

14from covered.cli import _upload_files 

15 

16BUCKET = "test-bucket" 

17SITE_ID = "site-abc" 

18SESSION = { 

19 "site_id": SITE_ID, 

20 "bucket": BUCKET, 

21 "region": "us-east-1", 

22 "access_key_id": "testing-key-id", 

23 "secret_access_key": "testing-secret", 

24 "session_token": "test-token", 

25} 

26 

27 

28class FakeS3Client: 

29 """ 

30 Aiobotocore-compatible stub: its own async context manager, exposes `put_object`. 

31 """ 

32 

33 def __init__(self, put_object: Any) -> None: 

34 self.put_object = put_object 

35 

36 async def __aenter__(self) -> "FakeS3Client": 

37 return self 

38 

39 async def __aexit__(self, *exc: object) -> None: 

40 return None 

41 

42 

43@contextmanager 

44def patched_aiobotocore(put_object: Any) -> Iterator[MagicMock]: 

45 """ 

46 Patch `covered.cli.get_session` so the s3 client's `put_object` is `put_object`. 

47 Yields the session mock so callers can inspect `create_client` args. 

48 """ 

49 fake_session = MagicMock() 

50 fake_session.create_client = MagicMock( 

51 return_value=FakeS3Client(put_object=put_object) 

52 ) 

53 with patch("covered.cli.get_session", return_value=fake_session): 

54 yield fake_session 

55 

56 

57async def test_upload_files_uploads_every_file_recursively(tmp_path: Path): 

58 """ 

59 Every regular file in a nested directory tree is uploaded to S3. 

60 """ 

61 (tmp_path / "a.txt").write_text("aaa") 

62 (tmp_path / "sub" / "deep").mkdir(parents=True) 

63 (tmp_path / "sub" / "deep" / "b.txt").write_text("bbb") 

64 (tmp_path / "sub" / "c.txt").write_text("ccc") 

65 

66 put_object = AsyncMock(return_value={}) 

67 with patched_aiobotocore(put_object): 

68 count = await _upload_files(tmp_path, SESSION, concurrency=2) 

69 

70 assert count == 3 

71 keys = {c.kwargs["Key"] for c in put_object.call_args_list} 

72 assert keys == { 

73 f"sites/{SITE_ID}/a.txt", 

74 f"sites/{SITE_ID}/sub/deep/b.txt", 

75 f"sites/{SITE_ID}/sub/c.txt", 

76 } 

77 

78 

79async def test_upload_files_skips_directories(tmp_path: Path): 

80 """ 

81 Directory entries are not uploaded as objects (only files are). 

82 """ 

83 (tmp_path / "empty_subdir").mkdir() 

84 (tmp_path / "another").mkdir() 

85 (tmp_path / "another" / "file.txt").write_text("x") 

86 

87 put_object = AsyncMock(return_value={}) 

88 with patched_aiobotocore(put_object): 

89 count = await _upload_files(tmp_path, SESSION, concurrency=2) 

90 

91 assert count == 1 

92 keys = {c.kwargs["Key"] for c in put_object.call_args_list} 

93 assert keys == {f"sites/{SITE_ID}/another/file.txt"} 

94 

95 

96async def test_upload_files_uses_relative_key_under_site_prefix(tmp_path: Path): 

97 """ 

98 S3 object key is `sites/{site_id}/{path-relative-to-directory}`. 

99 """ 

100 (tmp_path / "report.html").write_text("html") 

101 

102 put_object = AsyncMock(return_value={}) 

103 with patched_aiobotocore(put_object): 

104 await _upload_files(tmp_path, SESSION, concurrency=1) 

105 

106 put_object.assert_called_once_with( 

107 Bucket=BUCKET, 

108 Key=f"sites/{SITE_ID}/report.html", 

109 Body=b"html", 

110 ) 

111 

112 

113async def test_upload_files_returns_uploaded_count(tmp_path: Path): 

114 """ 

115 Return value equals the number of files in the tree. 

116 """ 

117 for i in range(5): 

118 (tmp_path / f"f{i}.txt").write_text(str(i)) 

119 

120 put_object = AsyncMock(return_value={}) 

121 with patched_aiobotocore(put_object): 

122 count = await _upload_files(tmp_path, SESSION, concurrency=2) 

123 

124 assert count == 5 

125 assert put_object.call_count == 5 

126 

127 

128async def test_upload_files_empty_directory_returns_zero(tmp_path: Path): 

129 """ 

130 Empty directory results in no S3 calls and a return value of 0. 

131 """ 

132 put_object = AsyncMock(return_value={}) 

133 with patched_aiobotocore(put_object): 

134 count = await _upload_files(tmp_path, SESSION, concurrency=2) 

135 

136 assert count == 0 

137 put_object.assert_not_called() 

138 

139 

140async def test_upload_files_preserves_file_bytes(tmp_path: Path): 

141 """ 

142 The body sent to S3 matches `file_path.read_bytes()` exactly (binary-safe). 

143 """ 

144 binary = bytes(range(256)) 

145 (tmp_path / "blob.bin").write_bytes(binary) 

146 

147 put_object = AsyncMock(return_value={}) 

148 with patched_aiobotocore(put_object): 

149 await _upload_files(tmp_path, SESSION, concurrency=1) 

150 

151 put_object.assert_called_once_with( 

152 Bucket=BUCKET, 

153 Key=f"sites/{SITE_ID}/blob.bin", 

154 Body=binary, 

155 ) 

156 

157 

158async def test_upload_files_passes_session_credentials_to_s3_client(tmp_path: Path): 

159 """ 

160 Region, access key, secret, token come from session dict. 

161 """ 

162 (tmp_path / "f.txt").write_text("x") 

163 

164 put_object = AsyncMock(return_value={}) 

165 with patched_aiobotocore(put_object) as fake_session: 

166 await _upload_files(tmp_path, SESSION, concurrency=1) 

167 

168 fake_session.create_client.assert_called_once_with( 

169 "s3", 

170 region_name="us-east-1", 

171 aws_access_key_id="testing-key-id", 

172 aws_secret_access_key="testing-secret", 

173 aws_session_token="test-token", 

174 ) 

175 put_object.assert_called_once_with( 

176 Bucket=BUCKET, 

177 Key=f"sites/{SITE_ID}/f.txt", 

178 Body=b"x", 

179 ) 

180 

181 

182async def test_upload_files_respects_concurrency_limit(tmp_path: Path): 

183 """ 

184 The semaphore caps the number of in-flight uploads at the configured concurrency. 

185 """ 

186 n_files = 10 

187 for i in range(n_files): 

188 (tmp_path / f"f{i}.txt").write_text(str(i)) 

189 

190 in_flight = 0 

191 peak = 0 

192 

193 async def put_object(**kwargs: Any) -> dict: 

194 nonlocal in_flight, peak 

195 in_flight += 1 

196 peak = max(peak, in_flight) 

197 await asyncio.sleep(0.01) 

198 in_flight -= 1 

199 return {} 

200 

201 with patched_aiobotocore(put_object): 

202 await _upload_files(tmp_path, SESSION, concurrency=3) 

203 

204 assert peak == 3 

205 

206 

207async def test_upload_files_propagates_s3_errors(tmp_path: Path): 

208 """ 

209 If `put_object` raises, the exception is surfaced to the caller. 

210 """ 

211 (tmp_path / "f1.txt").write_text("x") 

212 (tmp_path / "f2.txt").write_text("y") 

213 

214 async def put_object(**kwargs: Any) -> dict: 

215 raise RuntimeError("upload failed") 

216 

217 with ( 

218 patched_aiobotocore(put_object), 

219 pytest.raises(RuntimeError, match="upload failed"), 

220 ): 

221 await _upload_files(tmp_path, SESSION, concurrency=2)