1import copy
2
3from flexget.api.app import base_message
4from flexget.components.seen.api import ObjectsContainer as OC
5from flexget.components.seen.db import SeenEntry, SeenField
6from flexget.manager import Session
7from flexget.utils import json
8
9
10class TestSeenAPI:
11    config = "{'tasks': {}}"
12
13    def add_seen_entries(self):
14        seen_entry_1 = dict(title='test_title', reason='test_reason', task='test_task')
15        field_1 = dict(field='test_field_1', value='test_value_1')
16        field_2 = dict(field='test_field_2', value='test_value_2')
17        seen_entry_2 = dict(
18            title='test_title_2', reason='test_reason_2', task='test_task_2', local=True
19        )
20        field_3 = dict(field='test_field_3', value='test_value_3')
21        field_4 = dict(field='test_field_4', value='test_value_4')
22
23        entries = sorted([seen_entry_1, seen_entry_2], key=lambda entry: entry['title'])
24
25        with Session() as session:
26            seen_db_1 = SeenEntry(**seen_entry_1)
27            seen_db_1.fields = [SeenField(**field_1), SeenField(**field_2)]
28            session.add(seen_db_1)
29            seen_db_2 = SeenEntry(**seen_entry_2)
30            seen_db_2.fields = [SeenField(**field_3), SeenField(**field_4)]
31            session.add(seen_db_2)
32
33        return entries
34
35    def test_seen_get_all(self, api_client, schema_match):
36        rsp = api_client.get('/seen/')
37        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
38        data = json.loads(rsp.get_data(as_text=True))
39
40        errors = schema_match(OC.seen_search_object, data)
41        assert not errors
42
43        entries = self.add_seen_entries()
44
45        rsp = api_client.get('/seen/')
46        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
47        data = json.loads(rsp.get_data(as_text=True))
48
49        errors = schema_match(OC.seen_search_object, data)
50        assert not errors
51
52        for idx, value in enumerate(sorted(data, key=lambda entry: entry['title'])):
53            for k, v in entries[idx].items():
54                assert value[k] == v
55
56        assert len(data) == 2
57
58        rsp = api_client.get('/seen/?local=true')
59        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
60        data = json.loads(rsp.get_data(as_text=True))
61
62        errors = schema_match(OC.seen_search_object, data)
63        assert not errors
64
65        assert len(data) == 1
66
67        rsp = api_client.get('/seen/?value=test_value_2')
68        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
69        data = json.loads(rsp.get_data(as_text=True))
70
71        errors = schema_match(OC.seen_search_object, data)
72        assert not errors
73
74        assert len(data) == 1
75
76        rsp = api_client.get('/seen/?value=bla')
77        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
78        data = json.loads(rsp.get_data(as_text=True))
79
80        errors = schema_match(OC.seen_search_object, data)
81        assert not errors
82
83        assert len(data) == 0
84
85    def test_seen_delete_all(self, api_client, schema_match):
86        entries = self.add_seen_entries()
87
88        rsp = api_client.get('/seen/')
89        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
90        data = json.loads(rsp.get_data(as_text=True))
91
92        errors = schema_match(OC.seen_search_object, data)
93        assert not errors
94
95        for idx, value in enumerate(sorted(data, key=lambda entry: entry['title'])):
96            for k, v in entries[idx].items():
97                assert value[k] == v
98
99        rsp = api_client.delete('/seen/')
100        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
101        data = json.loads(rsp.get_data(as_text=True))
102
103        errors = schema_match(base_message, data)
104        assert not errors
105
106        assert data['message'] == 'successfully deleted 2 entries'
107
108        self.add_seen_entries()
109
110        rsp = api_client.delete('/seen/?local=true')
111        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
112        data = json.loads(rsp.get_data(as_text=True))
113
114        errors = schema_match(base_message, data)
115        assert not errors
116
117        assert data['message'] == 'successfully deleted 1 entries'
118
119        rsp = api_client.delete('/seen/?value=test_value_2')
120        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
121        data = json.loads(rsp.get_data(as_text=True))
122
123        errors = schema_match(base_message, data)
124        assert not errors
125
126        assert data['message'] == 'successfully deleted 1 entries'
127
128        rsp = api_client.get('/seen/')
129        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
130        data = json.loads(rsp.get_data(as_text=True))
131
132        errors = schema_match(OC.seen_search_object, data)
133        assert not errors
134
135        assert data == []
136
137    def test_seen_get_by_id(self, api_client, schema_match):
138        entries = self.add_seen_entries()
139
140        rsp = api_client.get('/seen/1/')
141        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
142        data = json.loads(rsp.get_data(as_text=True))
143
144        errors = schema_match(OC.seen_object, data)
145        assert not errors
146
147        for key, value in entries[0].items():
148            assert data[key] == value
149
150        rsp = api_client.get('/seen/10/')
151        assert rsp.status_code == 404, 'Response code is %s' % rsp.status_code
152        data = json.loads(rsp.get_data(as_text=True))
153
154        errors = schema_match(base_message, data)
155        assert not errors
156
157    def test_seen_get_delete_id(self, api_client, schema_match):
158        self.add_seen_entries()
159
160        rsp = api_client.delete('/seen/1/')
161        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
162        data = json.loads(rsp.get_data(as_text=True))
163
164        errors = schema_match(base_message, data)
165        assert not errors
166
167        rsp = api_client.get('/seen/')
168        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
169        data = json.loads(rsp.get_data(as_text=True))
170
171        errors = schema_match(OC.seen_search_object, data)
172        assert not errors
173
174        assert len(data) == 1
175
176        rsp = api_client.delete('/seen/10/')
177        assert rsp.status_code == 404, 'Response code is %s' % rsp.status_code
178        data = json.loads(rsp.get_data(as_text=True))
179
180        errors = schema_match(base_message, data)
181        assert not errors
182
183
184class TestSeenPagination:
185    config = 'tasks: {}'
186
187    def test_seen_pagination(self, api_client, link_headers):
188        base_seen_entry = dict(title='test_title_', task='test_task_', reason='test_reason_')
189        base_seen_field = dict(field='test_field_', value='test_value_')
190        number_of_entries = 200
191
192        with Session() as session:
193            for i in range(number_of_entries):
194                entry = copy.deepcopy(base_seen_entry)
195                field = copy.deepcopy(base_seen_field)
196
197                for key, value in entry.items():
198                    entry[key] = value + str(i)
199
200                for key, value in field.items():
201                    field[key] = value + str(i)
202
203                seen_entry = SeenEntry(**entry)
204                session.add(seen_entry)
205                seen_entry.fields = [SeenField(**field)]
206
207        # Default values
208        rsp = api_client.get('/seen/')
209        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
210        data = json.loads(rsp.get_data(as_text=True))
211
212        assert len(data) == 50
213        assert int(rsp.headers['total-count']) == 200
214        assert int(rsp.headers['count']) == 50
215
216        links = link_headers(rsp)
217        assert links['last']['page'] == 4
218        assert links['next']['page'] == 2
219
220        # Change page size
221        rsp = api_client.get('/seen/?per_page=100')
222        assert rsp.status_code == 200
223        data = json.loads(rsp.get_data(as_text=True))
224
225        assert len(data) == 100
226        assert int(rsp.headers['total-count']) == 200
227        assert int(rsp.headers['count']) == 100
228
229        links = link_headers(rsp)
230        assert links['last']['page'] == 2
231        assert links['next']['page'] == 2
232
233        # Get different page
234        rsp = api_client.get('/seen/?page=2')
235        assert rsp.status_code == 200
236        data = json.loads(rsp.get_data(as_text=True))
237
238        assert len(data) == 50
239        assert int(rsp.headers['total-count']) == 200
240        assert int(rsp.headers['count']) == 50
241
242        links = link_headers(rsp)
243        assert links['last']['page'] == 4
244        assert links['next']['page'] == 3
245        assert links['prev']['page'] == 1
246
247    def test_seen_sorting(self, api_client):
248        seen_entry_1 = dict(
249            title='test_title_1', reason='test_reason_c', task='test_task_2', local=True
250        )
251        field_1 = dict(field='test_field_1', value='test_value_1')
252        field_2 = dict(field='test_field_2', value='test_value_2')
253        seen_entry_2 = dict(
254            title='test_title_2', reason='test_reason_b', task='test_task_3', local=True
255        )
256        field_3 = dict(field='test_field_3', value='test_value_3')
257        field_4 = dict(field='test_field_4', value='test_value_4')
258        seen_entry_3 = dict(
259            title='test_title_3', reason='test_reason_a', task='test_task_1', local=False
260        )
261        field_5 = dict(field='test_field_3', value='test_value_3')
262        field_6 = dict(field='test_field_4', value='test_value_4')
263
264        with Session() as session:
265            seen_db_1 = SeenEntry(**seen_entry_1)
266            session.add(seen_db_1)
267            seen_db_1.fields = [SeenField(**field_1), SeenField(**field_2)]
268
269            seen_db_2 = SeenEntry(**seen_entry_2)
270            session.add(seen_db_2)
271            seen_db_2.fields = [SeenField(**field_3), SeenField(**field_4)]
272
273            seen_db_3 = SeenEntry(**seen_entry_3)
274            session.add(seen_db_3)
275            seen_db_3.fields = [SeenField(**field_5), SeenField(**field_6)]
276
277        # Sort by title
278        rsp = api_client.get('/seen/?sort_by=title')
279        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
280        data = json.loads(rsp.get_data(as_text=True))
281
282        assert data[0]['title'] == 'test_title_3'
283
284        rsp = api_client.get('/seen/?sort_by=title&order=asc')
285        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
286        data = json.loads(rsp.get_data(as_text=True))
287
288        assert data[0]['title'] == 'test_title_1'
289
290        # Sort by task
291        rsp = api_client.get('/seen/?sort_by=task')
292        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
293        data = json.loads(rsp.get_data(as_text=True))
294
295        assert data[0]['task'] == 'test_task_3'
296
297        rsp = api_client.get('/seen/?sort_by=task&order=asc')
298        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
299        data = json.loads(rsp.get_data(as_text=True))
300
301        assert data[0]['task'] == 'test_task_1'
302
303        # Sort by reason
304        rsp = api_client.get('/seen/?sort_by=reason')
305        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
306        data = json.loads(rsp.get_data(as_text=True))
307
308        assert data[0]['reason'] == 'test_reason_c'
309
310        rsp = api_client.get('/seen/?sort_by=reason&order=asc')
311        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
312        data = json.loads(rsp.get_data(as_text=True))
313
314        assert data[0]['reason'] == 'test_reason_a'
315
316        # Sort by local
317        rsp = api_client.get('/seen/?sort_by=local')
318        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
319        data = json.loads(rsp.get_data(as_text=True))
320
321        assert data[0]['local'] == True
322
323        rsp = api_client.get('/seen/?sort_by=local&order=asc')
324        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
325        data = json.loads(rsp.get_data(as_text=True))
326
327        assert data[0]['local'] == False
328
329        # Combine sorting and pagination
330        rsp = api_client.get('/seen/?sort_by=reason&per_page=2&page=2')
331        assert rsp.status_code == 200, 'Response code is %s' % rsp.status_code
332        data = json.loads(rsp.get_data(as_text=True))
333
334        assert data[0]['reason'] == 'test_reason_a'
335