1import copy 2 3from flexget.api.app import base_message 4from flexget.components.managed_lists.lists.pending_list.api import ObjectsContainer as OC 5from flexget.components.managed_lists.lists.pending_list.db import ( 6 PendingListEntry, 7 PendingListList, 8) 9from flexget.entry import Entry 10from flexget.manager import Session 11from flexget.utils import json 12 13 14class TestPendingListAPI: 15 config = 'tasks: {}' 16 17 def test_pending_list_list(self, api_client, schema_match): 18 # No params 19 rsp = api_client.get('/pending_list/') 20 assert rsp.status_code == 200 21 22 data = json.loads(rsp.get_data(as_text=True)) 23 24 errors = schema_match(OC.pending_list_return_lists, data) 25 assert not errors 26 27 payload = {'name': 'test_list'} 28 29 # Create list 30 rsp = api_client.json_post('/pending_list/', data=json.dumps(payload)) 31 assert rsp.status_code == 201 32 data = json.loads(rsp.get_data(as_text=True)) 33 34 errors = schema_match(OC.pending_list_base_object, data) 35 assert not errors 36 37 for field, value in payload.items(): 38 assert data.get(field) == value 39 40 # Named param 41 rsp = api_client.get('/pending_list/?name=test_list') 42 assert rsp.status_code == 200 43 data = json.loads(rsp.get_data(as_text=True)) 44 45 errors = schema_match(OC.pending_list_return_lists, data) 46 assert not errors 47 48 for field, value in payload.items(): 49 assert data[0].get(field) == value 50 51 # Try to Create list again 52 rsp = api_client.json_post('/pending_list/', data=json.dumps(payload)) 53 assert rsp.status_code == 409 54 data = json.loads(rsp.get_data(as_text=True)) 55 56 errors = schema_match(base_message, data) 57 assert not errors 58 59 def test_pending_list_list_id(self, api_client, schema_match): 60 payload = {'name': 'test_list'} 61 62 # Create list 63 rsp = api_client.json_post('/pending_list/', data=json.dumps(payload)) 64 assert rsp.status_code == 201 65 data = json.loads(rsp.get_data(as_text=True)) 66 67 errors = schema_match(OC.pending_list_base_object, data) 68 assert not errors 69 70 for field, value in payload.items(): 71 assert data.get(field) == value 72 73 # Get list 74 rsp = api_client.get('/pending_list/1/') 75 assert rsp.status_code == 200 76 data = json.loads(rsp.get_data(as_text=True)) 77 78 errors = schema_match(OC.pending_list_base_object, data) 79 assert not errors 80 81 for field, value in payload.items(): 82 assert data.get(field) == value 83 84 # Delete list 85 rsp = api_client.delete('/pending_list/1/') 86 assert rsp.status_code == 200 87 data = json.loads(rsp.get_data(as_text=True)) 88 89 errors = schema_match(base_message, data) 90 assert not errors 91 92 # Try to get list 93 rsp = api_client.get('/pending_list/1/') 94 assert rsp.status_code == 404 95 data = json.loads(rsp.get_data(as_text=True)) 96 97 errors = schema_match(base_message, data) 98 assert not errors 99 100 # Try to Delete list 101 rsp = api_client.delete('/pending_list/1/') 102 assert rsp.status_code == 404 103 data = json.loads(rsp.get_data(as_text=True)) 104 105 errors = schema_match(base_message, data) 106 assert not errors 107 108 def test_pending_list_entries(self, api_client, schema_match): 109 # Get non existent list 110 rsp = api_client.get('/pending_list/1/entries/') 111 assert rsp.status_code == 404 112 data = json.loads(rsp.get_data(as_text=True)) 113 114 errors = schema_match(base_message, data) 115 assert not errors 116 117 payload = {'name': 'test_list'} 118 119 # Create list 120 rsp = api_client.json_post('/pending_list/', data=json.dumps(payload)) 121 assert rsp.status_code == 201 122 data = json.loads(rsp.get_data(as_text=True)) 123 124 errors = schema_match(OC.pending_list_base_object, data) 125 assert not errors 126 127 for field, value in payload.items(): 128 assert data.get(field) == value 129 130 entry_data = {'title': 'title', 'original_url': 'http://test.com'} 131 132 # Add entry to list 133 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(entry_data)) 134 assert rsp.status_code == 201 135 data = json.loads(rsp.get_data(as_text=True)) 136 137 errors = schema_match(OC.pending_list_entry_base_object, data) 138 assert not errors 139 140 for field, value in entry_data.items(): 141 assert data.get(field) == value 142 143 # Get entries from list 144 rsp = api_client.get('/pending_list/1/entries/') 145 assert rsp.status_code == 200 146 data = json.loads(rsp.get_data(as_text=True)) 147 148 errors = schema_match(OC.pending_lists_entries_return_object, data) 149 assert not errors 150 151 for field, value in entry_data.items(): 152 assert data[0].get(field) == value 153 154 # Try to re-add entry to list 155 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(entry_data)) 156 assert rsp.status_code == 409 157 data = json.loads(rsp.get_data(as_text=True)) 158 159 errors = schema_match(OC.pending_list_entry_base_object, data) 160 assert not errors 161 162 # Try to post to non existing list 163 rsp = api_client.json_post('/pending_list/10/entries/', data=json.dumps(entry_data)) 164 assert rsp.status_code == 404 165 data = json.loads(rsp.get_data(as_text=True)) 166 167 errors = schema_match(OC.pending_list_entry_base_object, data) 168 assert not errors 169 170 def test_pending_list_entries_batch_operation(self, api_client, schema_match): 171 payload = {'name': 'test_list'} 172 173 # Create list 174 api_client.json_post('/pending_list/', data=json.dumps(payload)) 175 176 # Add 3 entries to list 177 for i in range(3): 178 payload = {'title': f'title {i}', 'original_url': f'http://{i}test.com'} 179 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(payload)) 180 assert rsp.status_code == 201 181 182 payload = {'operation': 'approve', 'ids': [1, 2, 3]} 183 184 # Approve several entries 185 rsp = api_client.json_put('/pending_list/1/entries/batch/', data=json.dumps(payload)) 186 assert rsp.status_code == 201 187 data = json.loads(rsp.get_data(as_text=True)) 188 189 errors = schema_match(OC.pending_lists_entries_return_object, data) 190 assert not errors 191 assert len(data) == 3 192 193 assert all((item['approved'] for item in data)) 194 195 # get entries is correct 196 rsp = api_client.get('/pending_list/1/entries/') 197 assert rsp.status_code == 200 198 data = json.loads(rsp.get_data(as_text=True)) 199 200 errors = schema_match(OC.pending_lists_entries_return_object, data) 201 assert not errors 202 assert len(data) == 3 203 204 for item in data: 205 assert item.get('approved') 206 207 payload['operation'] = 'reject' 208 209 # reject several entries 210 rsp = api_client.json_put('pending_list/1/entries/batch', data=json.dumps(payload)) 211 assert rsp.status_code == 201 212 data = json.loads(rsp.get_data(as_text=True)) 213 214 errors = schema_match(OC.pending_lists_entries_return_object, data) 215 assert not errors 216 assert len(data) == 3 217 218 for item in data: 219 assert not item.get('approved') 220 221 rsp = api_client.get('/pending_list/1/entries/') 222 assert rsp.status_code == 200 223 data = json.loads(rsp.get_data(as_text=True)) 224 225 errors = schema_match(OC.pending_lists_entries_return_object, data) 226 assert not errors 227 assert len(data) == 3 228 229 for item in data: 230 assert not item.get('approved') 231 232 def test_pending_list_entries_batch_remove(self, api_client, schema_match): 233 payload = {'name': 'test_list'} 234 235 # Create list 236 api_client.json_post('/pending_list/', data=json.dumps(payload)) 237 238 # Add 3 entries to list 239 for i in range(3): 240 payload = {'title': f'title {i}', 'original_url': f'http://{i}test.com'} 241 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(payload)) 242 assert rsp.status_code == 201 243 244 # get entries is correct 245 rsp = api_client.get('/pending_list/1/entries/') 246 assert rsp.status_code == 200 247 data = json.loads(rsp.get_data(as_text=True)) 248 249 errors = schema_match(OC.pending_lists_entries_return_object, data) 250 assert not errors 251 assert len(data) == 3 252 253 payload = {'ids': [1, 2, 3]} 254 255 rsp = api_client.json_delete('pending_list/1/entries/batch', data=json.dumps(payload)) 256 assert rsp.status_code == 204 257 258 rsp = api_client.get('/pending_list/1/entries/') 259 assert rsp.status_code == 200 260 data = json.loads(rsp.get_data(as_text=True)) 261 262 errors = schema_match(OC.pending_lists_entries_return_object, data) 263 assert not errors 264 assert not data 265 266 def test_pending_list_entry(self, api_client, schema_match): 267 payload = {'name': 'test_list'} 268 269 # Create list 270 rsp = api_client.json_post('/pending_list/', data=json.dumps(payload)) 271 assert rsp.status_code == 201 272 data = json.loads(rsp.get_data(as_text=True)) 273 274 errors = schema_match(OC.pending_list_base_object, data) 275 assert not errors 276 277 for field, value in payload.items(): 278 assert data.get(field) == value 279 280 entry_data = {'title': 'title', 'original_url': 'http://test.com'} 281 282 # Add entry to list 283 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(entry_data)) 284 assert rsp.status_code == 201 285 data = json.loads(rsp.get_data(as_text=True)) 286 287 errors = schema_match(OC.pending_list_entry_base_object, data) 288 assert not errors 289 290 for field, value in entry_data.items(): 291 assert data.get(field) == value 292 293 # Get entries from list 294 rsp = api_client.get('/pending_list/1/entries/') 295 assert rsp.status_code == 200 296 data = json.loads(rsp.get_data(as_text=True)) 297 298 errors = schema_match(OC.pending_lists_entries_return_object, data) 299 assert not errors 300 301 for field, value in entry_data.items(): 302 assert data[0].get(field) == value 303 304 # Get specific entry from list 305 rsp = api_client.get('/pending_list/1/entries/1/') 306 assert rsp.status_code == 200 307 data = json.loads(rsp.get_data(as_text=True)) 308 309 errors = schema_match(OC.pending_list_entry_base_object, data) 310 assert not errors 311 312 for field, value in entry_data.items(): 313 assert data.get(field) == value 314 315 new_entry_data = {'operation': 'approve'} 316 317 # Change specific entry from list 318 rsp = api_client.json_put('/pending_list/1/entries/1/', data=json.dumps(new_entry_data)) 319 assert rsp.status_code == 201 320 data = json.loads(rsp.get_data(as_text=True)) 321 322 errors = schema_match(OC.pending_list_entry_base_object, data) 323 assert not errors 324 325 assert data['approved'] 326 327 # Try to change non-existent entry from list 328 rsp = api_client.json_put('/pending_list/1/entries/10/', data=json.dumps(new_entry_data)) 329 assert rsp.status_code == 404 330 data = json.loads(rsp.get_data(as_text=True)) 331 332 errors = schema_match(base_message, data) 333 assert not errors 334 335 # Delete specific entry from list 336 rsp = api_client.delete('/pending_list/1/entries/1/') 337 assert rsp.status_code == 200 338 data = json.loads(rsp.get_data(as_text=True)) 339 340 errors = schema_match(base_message, data) 341 assert not errors 342 343 # Get non existent entry from list 344 rsp = api_client.get('/pending_list/1/entries/1/') 345 assert rsp.status_code == 404 346 data = json.loads(rsp.get_data(as_text=True)) 347 348 errors = schema_match(base_message, data) 349 assert not errors 350 351 # Delete non existent entry from list 352 rsp = api_client.delete('/pending_list/1/entries/1/') 353 assert rsp.status_code == 404 354 data = json.loads(rsp.get_data(as_text=True)) 355 356 errors = schema_match(base_message, data) 357 assert not errors 358 359 def test_pending_list_filter_entries(self, api_client, schema_match): 360 payload = {'name': 'test_list'} 361 362 rsp = api_client.json_post('/pending_list/', data=json.dumps(payload)) 363 assert rsp.status_code == 201 364 365 entry_1_data = {'title': 'Foobaz', 'original_url': 'http://test1.com'} 366 entry_2_data = {'title': 'fooBar', 'original_url': 'http://test2.com'} 367 368 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(entry_1_data)) 369 assert rsp.status_code == 201 370 371 rsp = api_client.json_post('/pending_list/1/entries/', data=json.dumps(entry_2_data)) 372 assert rsp.status_code == 201 373 374 rsp = api_client.get('/pending_list/1/entries?filter=bar') 375 assert rsp.status_code == 200 376 data = json.loads(rsp.get_data(as_text=True)) 377 378 errors = schema_match(OC.pending_lists_entries_return_object, data) 379 assert not errors 380 381 assert len(data) == 1 382 assert data[0]['title'] == 'fooBar' 383 384 rsp = api_client.get('/pending_list/1/entries?filter=foo') 385 assert rsp.status_code == 200 386 data = json.loads(rsp.get_data(as_text=True)) 387 388 assert len(data) == 2 389 390 rsp = api_client.get('/pending_list/1/entries?filter=bla') 391 assert rsp.status_code == 200 392 data = json.loads(rsp.get_data(as_text=True)) 393 394 assert not data 395 396 397class TestPendingListPagination: 398 config = 'tasks: {}' 399 400 def test_pending_list_pagination(self, api_client, link_headers): 401 base_entry = dict(title='test_title_', original_url='url_') 402 number_of_entries = 200 403 404 with Session() as session: 405 pending_list = PendingListList(name='test list') 406 session.add(pending_list) 407 408 for i in range(number_of_entries): 409 entry = copy.deepcopy(base_entry) 410 for k, v in entry.items(): 411 entry[k] = v + str(i) 412 e = Entry(entry) 413 pending_list.entries.append(PendingListEntry(e, pending_list.id)) 414 415 # Default values 416 rsp = api_client.get('/pending_list/1/entries/') 417 assert rsp.status_code == 200 418 data = json.loads(rsp.get_data(as_text=True)) 419 420 assert len(data) == 50 421 assert int(rsp.headers['total-count']) == 200 422 assert int(rsp.headers['count']) == 50 423 424 links = link_headers(rsp) 425 assert links['last']['page'] == 4 426 assert links['next']['page'] == 2 427 428 # Change page size 429 rsp = api_client.get('/pending_list/1/entries/?per_page=100') 430 assert rsp.status_code == 200 431 data = json.loads(rsp.get_data(as_text=True)) 432 433 assert len(data) == 100 434 assert int(rsp.headers['total-count']) == 200 435 assert int(rsp.headers['count']) == 100 436 437 links = link_headers(rsp) 438 assert links['last']['page'] == 2 439 assert links['next']['page'] == 2 440 441 # Get different page 442 rsp = api_client.get('/pending_list/1/entries/?page=2') 443 assert rsp.status_code == 200 444 data = json.loads(rsp.get_data(as_text=True)) 445 446 assert len(data) == 50 447 assert int(rsp.headers['total-count']) == 200 448 assert int(rsp.headers['count']) == 50 449 450 links = link_headers(rsp) 451 assert links['last']['page'] == 4 452 assert links['next']['page'] == 3 453 assert links['prev']['page'] == 1 454 455 def test_pending_list_sorting(self, api_client): 456 base_entry_1 = dict(title='test_title_1', original_url='url_c') 457 base_entry_2 = dict(title='test_title_2', original_url='url_b') 458 base_entry_3 = dict(title='test_title_3', original_url='url_a') 459 460 with Session() as session: 461 pending_list = PendingListList(name='test list') 462 session.add(pending_list) 463 464 e1 = Entry(base_entry_1) 465 e2 = Entry(base_entry_2) 466 e3 = Entry(base_entry_3) 467 468 pending_list.entries.append(PendingListEntry(e1, pending_list.id)) 469 pending_list.entries.append(PendingListEntry(e2, pending_list.id)) 470 pending_list.entries.append(PendingListEntry(e3, pending_list.id)) 471 472 # Sort by title 473 rsp = api_client.get('/pending_list/1/entries/?sort_by=title') 474 assert rsp.status_code == 200 475 data = json.loads(rsp.get_data(as_text=True)) 476 477 assert data[0]['title'] == 'test_title_3' 478 479 rsp = api_client.get('/pending_list/1/entries/?sort_by=title&order=asc') 480 assert rsp.status_code == 200 481 data = json.loads(rsp.get_data(as_text=True)) 482 483 assert data[0]['title'] == 'test_title_1' 484 485 # Sort by original url 486 rsp = api_client.get('/pending_list/1/entries/?sort_by=original_url') 487 assert rsp.status_code == 200 488 data = json.loads(rsp.get_data(as_text=True)) 489 490 assert data[0]['original_url'] == 'url_c' 491 492 rsp = api_client.get('/pending_list/1/entries/?sort_by=original_url&order=asc') 493 assert rsp.status_code == 200 494 data = json.loads(rsp.get_data(as_text=True)) 495 496 assert data[0]['original_url'] == 'url_a' 497 498 # Combine sorting and pagination 499 rsp = api_client.get('/pending_list/1/entries/?sort_by=title&per_page=2&page=2') 500 assert rsp.status_code == 200 501 data = json.loads(rsp.get_data(as_text=True)) 502 503 assert data[0]['title'] == 'test_title_1' 504