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