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