test_downloads.py
1 """Tests for the `downloads` module.""" 2 3 from __future__ import annotations 4 5 import datetime 6 from datetime import timezone 7 from pathlib import Path 8 9 import pytest 10 11 from aria2p import API, BitTorrent, ClientException, Download, File, Options 12 from tests.conftest import Aria2Server 13 14 15 class TestBitTorrentClass: 16 def setup_method(self) -> None: 17 self.bittorrent = BitTorrent( 18 {"announceList": [], "comment": "", "creationDate": 10, "mode": "single", "info": {"name": ""}}, 19 ) 20 21 def test_init_method(self) -> None: 22 assert BitTorrent(self.bittorrent._struct) 23 24 def test_str_method(self) -> None: 25 assert self.bittorrent.info 26 assert str(self.bittorrent) == self.bittorrent.info["name"] 27 28 def test_announce_list_property(self) -> None: 29 assert self.bittorrent.announce_list == [] 30 31 def test_comment_property(self) -> None: 32 assert self.bittorrent.comment == "" 33 34 def test_creation_date_property(self) -> None: 35 assert self.bittorrent.creation_date == datetime.datetime.fromtimestamp(10, tz=timezone.utc) 36 37 def test_mode_property(self) -> None: 38 assert self.bittorrent.mode == "single" 39 40 def test_info_property(self) -> None: 41 assert self.bittorrent.info == {"name": ""} 42 43 44 class TestDownloadClass: 45 def setup_method(self) -> None: 46 self.bitfield = ( 47 "9e7ef5fbdefdffffdfffffffffffffdf7fff5dfefff7feffefffdff7fffffffbfeffbffddf7de9f7eefffffffeffe77bb" 48 "f8fdbdcffef7fffffbefad7ffffffbf55bff7edfedfeff7ffff7ffffff3ffff7d3ffbfffddddffe7ffffffffffdedf7fd" 49 "fef62fbdfffffbffbffdcfaffffafebeff3ebff7d5fffbbb2bafef77ffaffe7d37fbffffffb6dfffffffedfebbecbbffe" 50 "bdefffffffff977f7ffdffee7fffffffdfeb3f67dddffeedfffffbfffffdaffbfeedfadef6dfd9d2df9fb7f689ffcff3f" 51 "fbfebbfdbd7fcddfa77dfddffdfe327fdcbf77fad87ffff6ff7ffebfefdfbbffdefdafeefed7fefffe7ffad9ffdcffefc" 52 "ffbbf3c07ef7fc0" 53 ) 54 self.download = Download( 55 API(), 56 { 57 "bitfield": self.bitfield, 58 "bittorrent": {}, 59 "completedLength": "889592532", 60 "connections": "44", 61 "dir": "/home/pawamoy/Downloads/torrents/tmp", 62 "downloadSpeed": "745509", 63 "files": [ 64 { 65 "completedLength": "58132", 66 "index": "1", 67 "length": "58132", 68 "path": "/home/pawamoy/Downloads/torrents/tmp/dl/logo.jpg", 69 "selected": "true", 70 "uris": [], 71 }, 72 { 73 "completedLength": "885864384", 74 "index": "2", 75 "length": "1045247936", 76 "path": "/home/pawamoy/Downloads/torrents/tmp/dl/video.mp4", 77 "selected": "true", 78 "uris": [], 79 }, 80 ], 81 "following": "a89a9c5ac990e6ef", 82 "gid": "0a6635c602761000", 83 "infoHash": "4f1da018803b65f61ed76612af9ad00d4373a771", 84 "numPieces": "1994", 85 "numSeeders": "12", 86 "pieceLength": "524288", 87 "seeder": "false", 88 "status": "active", 89 "totalLength": "1045306068", 90 "uploadLength": "97207027", 91 "uploadSpeed": "11032", 92 }, 93 ) 94 95 def test_eq_method(self) -> None: 96 assert self.download == Download(API(), {"gid": "0a6635c602761000"}) 97 98 def test_init_method(self) -> None: 99 assert Download(API(), {}) 100 101 def test_str_method(self) -> None: 102 assert str(self.download) == self.download.name 103 104 def test_update_method(self, tmp_path: Path, port: int) -> None: 105 with Aria2Server(tmp_path, port, session="1-dl-paused.txt") as server: 106 download = server.api.get_download("0000000000000001") 107 download.update() 108 109 def test_belongs_to(self, server: Aria2Server) -> None: 110 self.download.api = server.api 111 assert self.download.belongs_to is None 112 113 def test_belongs_to_id(self) -> None: 114 assert self.download.belongs_to_id is None 115 116 def test_bitfield(self) -> None: 117 assert self.download.bitfield == self.bitfield 118 119 def test_bittorrent(self) -> None: 120 assert self.download.bittorrent 121 122 # assert value was cached 123 self.download._struct["bittorrent"] = {"mode": "single"} 124 assert self.download.bittorrent.mode is None 125 126 # assert is None when "bittorrent" key is not in struct 127 del self.download._struct["bittorrent"] 128 self.download._bittorrent = None 129 assert self.download.bittorrent is None 130 131 def test_completed_length(self) -> None: 132 assert self.download.completed_length == 889_592_532 133 134 def test_completed_length_string(self) -> None: 135 assert self.download.completed_length_string() == "848.38 MiB" 136 assert self.download.completed_length_string(human_readable=False) == "889592532 B" 137 138 def test_connections(self) -> None: 139 assert self.download.connections == 44 140 141 def test_dir(self) -> None: 142 assert self.download.dir == Path("/home/pawamoy/Downloads/torrents/tmp") 143 144 def test_download_speed(self) -> None: 145 assert self.download.download_speed == 745_509 146 147 def test_download_speed_string(self) -> None: 148 assert self.download.download_speed_string() == "728.04 KiB/s" 149 assert self.download.download_speed_string(human_readable=False) == "745509 B/s" 150 151 def test_error_code(self) -> None: 152 assert self.download.error_code is None 153 154 def test_error_message(self) -> None: 155 assert self.download.error_message is None 156 157 def test_eta(self) -> None: 158 assert isinstance(self.download.eta, datetime.timedelta) 159 self.download._struct["downloadSpeed"] = "0" 160 assert self.download.eta == datetime.timedelta.max 161 162 def test_eta_string(self) -> None: 163 assert self.download.eta_string() == "3m28s" 164 self.download._struct["downloadSpeed"] = "0" 165 assert self.download.eta_string() == "-" 166 167 def test_files(self) -> None: 168 assert len(self.download.files) == 2 169 170 # assert value was cached 171 self.download._struct["files"] = [] 172 assert len(self.download.files) == 2 173 174 def test_followed_by(self, server: Aria2Server) -> None: 175 self.download.api = server.api 176 assert self.download.followed_by == [] 177 178 def test_followed_by_ids(self, server: Aria2Server) -> None: 179 self.download.api = server.api 180 assert self.download.followed_by_ids == [] 181 182 def test_following(self, server: Aria2Server) -> None: 183 self.download.api = server.api 184 assert self.download.following is None 185 186 def test_following_id(self) -> None: 187 assert self.download.following_id == "a89a9c5ac990e6ef" 188 189 def test_gid(self) -> None: 190 assert self.download.gid == "0a6635c602761000" 191 192 def test_has_failed(self) -> None: 193 assert self.download.has_failed is False 194 195 def test_info_hash(self) -> None: 196 assert self.download.info_hash == "4f1da018803b65f61ed76612af9ad00d4373a771" 197 198 def test_is_active(self) -> None: 199 assert self.download.is_active is True 200 201 def test_is_complete(self) -> None: 202 assert self.download.is_complete is False 203 204 def test_is_paused(self) -> None: 205 assert self.download.is_paused is False 206 207 def test_is_removed(self) -> None: 208 assert self.download.is_removed is False 209 210 def test_is_waiting(self) -> None: 211 assert self.download.is_waiting is False 212 213 def test_move(self, server: Aria2Server) -> None: 214 self.download.api = server.api 215 with pytest.raises(ClientException): 216 self.download.move(2) 217 218 def test_move_down(self, server: Aria2Server) -> None: 219 self.download.api = server.api 220 with pytest.raises(ClientException): 221 self.download.move_down(1) 222 223 def test_move_to(self, server: Aria2Server) -> None: 224 self.download.api = server.api 225 with pytest.raises(ClientException): 226 self.download.move_to(10) 227 228 def test_move_to_bottom(self, server: Aria2Server) -> None: 229 self.download.api = server.api 230 with pytest.raises(ClientException): 231 self.download.move_to_bottom() 232 233 def test_move_to_top(self, server: Aria2Server) -> None: 234 self.download.api = server.api 235 with pytest.raises(ClientException): 236 self.download.move_to_top() 237 238 def test_move_up(self, server: Aria2Server) -> None: 239 self.download.api = server.api 240 with pytest.raises(ClientException): 241 self.download.move_up() 242 243 def test_name(self) -> None: 244 assert self.download.name == "dl" 245 246 def test_name_bittorrent(self) -> None: 247 # TODO 248 pass 249 250 def test_name_metadata(self) -> None: 251 # TODO 252 pass 253 254 def test_name_filepath(self) -> None: 255 # TODO 256 pass 257 258 def test_name_uri(self) -> None: 259 # TODO 260 pass 261 262 def test_num_pieces(self) -> None: 263 assert self.download.num_pieces == 1994 264 265 def test_num_seeders(self) -> None: 266 assert self.download.num_seeders == 12 267 268 def test_options(self, server: Aria2Server) -> None: 269 self.download.api = server.api 270 with pytest.raises(ClientException): 271 print(self.download.options) # noqa: T201 272 273 def test_options2(self) -> None: 274 witness = [] 275 276 def mocked() -> None: 277 witness.append(0) 278 self.download._options = True # type: ignore[assignment] 279 280 self.download.update_options = mocked # type: ignore[method-assign] 281 assert self.download.options is True 282 assert witness == [0] 283 assert self.download.options is True 284 assert witness == [0] 285 286 def test_set_options(self) -> None: 287 options = Options(self.download.api, {}) 288 self.download.options = options 289 assert self.download.options is options 290 291 def test_pause(self, server: Aria2Server) -> None: 292 self.download.api = server.api 293 with pytest.raises(ClientException): 294 self.download.pause() 295 296 def test_piece_length(self) -> None: 297 assert self.download.piece_length == 524_288 298 299 def test_piece_length_string(self) -> None: 300 assert self.download.piece_length_string() == "512.00 KiB" 301 assert self.download.piece_length_string(human_readable=False) == "524288 B" 302 303 def test_progress(self) -> None: 304 assert self.download.progress == self.download.completed_length / self.download.total_length * 100 305 self.download._struct["totalLength"] = 0 306 assert self.download.progress == 0.0 307 308 def test_progress_string(self) -> None: 309 progress = self.download.completed_length / self.download.total_length * 100 310 assert self.download.progress_string() == f"{progress:.2f}%" 311 self.download._struct["totalLength"] = 0 312 assert self.download.progress_string() == "0.00%" 313 314 def test_remove(self, server: Aria2Server) -> None: 315 self.download.api = server.api 316 with pytest.raises(ClientException): 317 self.download.remove() 318 319 def test_resume(self, server: Aria2Server) -> None: 320 self.download.api = server.api 321 with pytest.raises(ClientException): 322 self.download.resume() 323 324 def test_seeder(self) -> None: 325 assert self.download.seeder is False 326 327 def test_status(self) -> None: 328 assert self.download.status == "active" 329 330 def test_total_length(self) -> None: 331 assert self.download.total_length == 1_045_306_068 332 333 def test_total_length_string(self) -> None: 334 assert self.download.total_length_string() == "996.88 MiB" 335 assert self.download.total_length_string(human_readable=False) == "1045306068 B" 336 337 def test_update(self, server: Aria2Server) -> None: 338 self.download.api = server.api 339 with pytest.raises(ClientException): 340 self.download.update() 341 342 def test_update_options(self, server: Aria2Server) -> None: 343 self.download.api = server.api 344 with pytest.raises(ClientException): 345 self.download.update_options() 346 347 def test_upload_length(self) -> None: 348 assert self.download.upload_length == 97_207_027 349 350 def test_upload_length_string(self) -> None: 351 assert self.download.upload_length_string() == "92.70 MiB" 352 assert self.download.upload_length_string(human_readable=False) == "97207027 B" 353 354 def test_upload_speed(self) -> None: 355 assert self.download.upload_speed == 11032 356 357 def test_upload_speed_string(self) -> None: 358 assert self.download.upload_speed_string() == "10.77 KiB/s" 359 assert self.download.upload_speed_string(human_readable=False) == "11032 B/s" 360 361 def test_verified_length(self) -> None: 362 assert self.download.verified_length == 0 363 364 def test_verified_length_string(self) -> None: 365 assert self.download.verified_length_string() == "0.00 B" 366 assert self.download.verified_length_string(human_readable=False) == "0 B" 367 368 def test_verify_integrity_pending(self) -> None: 369 assert self.download.verify_integrity_pending is None 370 371 def test_is_torrent(self) -> None: 372 assert self.download.is_torrent 373 del self.download._struct["bittorrent"] 374 assert not self.download.is_torrent 375 376 377 class TestFileClass: 378 def setup_method(self) -> None: 379 self.file = File( 380 { 381 "index": "1", 382 "path": "/some/file/path.txt", 383 "length": "2097152", 384 "completedLength": "2048", 385 "selected": "true", 386 "uris": [{"uri": "some uri", "status": "used"}, {"uri": "some other uri", "status": "waiting"}], 387 }, 388 ) 389 390 def test_eq_method(self) -> None: 391 assert self.file == File({"path": "/some/file/path.txt"}) 392 393 def test_init_method(self) -> None: 394 assert File({}) 395 396 def test_str_method(self) -> None: 397 assert str(self.file) == str(self.file.path) 398 399 def test_completed_length(self) -> None: 400 assert self.file.completed_length == 2048 401 402 def test_completed_length_string(self) -> None: 403 assert self.file.completed_length_string() == "2.00 KiB" 404 assert self.file.completed_length_string(human_readable=False) == "2048 B" 405 406 def test_index(self) -> None: 407 assert self.file.index == 1 408 409 def test_length(self) -> None: 410 assert self.file.length == 2_097_152 411 412 def test_length_string(self) -> None: 413 assert self.file.length_string() == "2.00 MiB" 414 assert self.file.length_string(human_readable=False) == "2097152 B" 415 416 def test_path(self) -> None: 417 assert self.file.path == Path("/some/file/path.txt") 418 419 def test_selected(self) -> None: 420 assert self.file.selected is True 421 422 def test_uris(self) -> None: 423 assert self.file.uris == [{"uri": "some uri", "status": "used"}, {"uri": "some other uri", "status": "waiting"}]