Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 952 Lines • ▼ Show 20 Lines | def test_origin_add_one(self): | ||||
| 'type': self.origin['type']}) | 'type': self.origin['type']}) | ||||
| self.assertEqual(actual_origin['id'], id) | self.assertEqual(actual_origin['id'], id) | ||||
| id2 = self.storage.origin_add_one(self.origin) | id2 = self.storage.origin_add_one(self.origin) | ||||
| self.assertEqual(id, id2) | self.assertEqual(id, id2) | ||||
| def test_origin_add(self): | def test_origin_add(self): | ||||
| origin0 = self.storage.origin_get(self.origin) | origin0 = self.storage.origin_get([self.origin])[0] | ||||
| self.assertIsNone(origin0) | self.assertIsNone(origin0) | ||||
| origin1, origin2 = self.storage.origin_add([self.origin, self.origin2]) | origin1, origin2 = self.storage.origin_add([self.origin, self.origin2]) | ||||
| actual_origin = self.storage.origin_get({ | actual_origin = self.storage.origin_get([{ | ||||
| 'url': self.origin['url'], | 'url': self.origin['url'], | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| }) | }])[0] | ||||
| self.assertEqual(actual_origin['id'], origin1['id']) | self.assertEqual(actual_origin['id'], origin1['id']) | ||||
| actual_origin2 = self.storage.origin_get({ | actual_origin2 = self.storage.origin_get([{ | ||||
| 'url': self.origin2['url'], | 'url': self.origin2['url'], | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| }) | }])[0] | ||||
| self.assertEqual(actual_origin2['id'], origin2['id']) | self.assertEqual(actual_origin2['id'], origin2['id']) | ||||
| def test_origin_add_twice(self): | def test_origin_add_twice(self): | ||||
| add1 = self.storage.origin_add([self.origin, self.origin2]) | add1 = self.storage.origin_add([self.origin, self.origin2]) | ||||
| add2 = self.storage.origin_add([self.origin, self.origin2]) | add2 = self.storage.origin_add([self.origin, self.origin2]) | ||||
| self.assertEqual(add1, add2) | self.assertEqual(add1, add2) | ||||
| def test_origin_get(self): | def test_origin_get_legacy(self): | ||||
| self.assertIsNone(self.storage.origin_get(self.origin)) | self.assertIsNone(self.storage.origin_get(self.origin)) | ||||
| id = self.storage.origin_add_one(self.origin) | id = self.storage.origin_add_one(self.origin) | ||||
| # lookup per type and url (returns id) | # lookup per type and url (returns id) | ||||
| actual_origin0 = self.storage.origin_get({'url': self.origin['url'], | actual_origin0 = self.storage.origin_get( | ||||
| 'type': self.origin['type']}) | {'url': self.origin['url'], 'type': self.origin['type']}) | ||||
| self.assertEqual(actual_origin0['id'], id) | self.assertEqual(actual_origin0['id'], id) | ||||
| # lookup per id (returns dict) | # lookup per id (returns dict) | ||||
| actual_origin1 = self.storage.origin_get({'id': id}) | actual_origin1 = self.storage.origin_get({'id': id}) | ||||
| self.assertEqual(actual_origin1, {'id': id, | self.assertEqual(actual_origin1, {'id': id, | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'url': self.origin['url']}) | 'url': self.origin['url']}) | ||||
| def test_origin_get(self): | |||||
| self.assertIsNone(self.storage.origin_get(self.origin)) | |||||
| id = self.storage.origin_add_one(self.origin) | |||||
anlambert: s/id/origin_id/ | |||||
| # lookup per type and url (returns id) | |||||
| actual_origin0 = self.storage.origin_get( | |||||
| [{'url': self.origin['url'], 'type': self.origin['type']}]) | |||||
| self.assertEqual(len(actual_origin0), 1, actual_origin0) | |||||
| self.assertEqual(actual_origin0[0]['id'], id) | |||||
| # lookup per id (returns dict) | |||||
| actual_origin1 = self.storage.origin_get([{'id': id}]) | |||||
| self.assertEqual(len(actual_origin1), 1, actual_origin1) | |||||
| self.assertEqual(actual_origin1[0], {'id': id, | |||||
| 'type': self.origin['type'], | |||||
| 'url': self.origin['url']}) | |||||
| def test_origin_get_consistency(self): | |||||
| self.assertIsNone(self.storage.origin_get(self.origin)) | |||||
| id = self.storage.origin_add_one(self.origin) | |||||
| with self.assertRaises(ValueError): | |||||
| self.storage.origin_get([ | |||||
| {'url': self.origin['url'], 'type': self.origin['type']}, | |||||
| {'id': id}]) | |||||
| def test_origin_search(self): | def test_origin_search(self): | ||||
| found_origins = list(self.storage.origin_search(self.origin['url'])) | found_origins = list(self.storage.origin_search(self.origin['url'])) | ||||
| self.assertEqual(len(found_origins), 0) | self.assertEqual(len(found_origins), 0) | ||||
| found_origins = list(self.storage.origin_search(self.origin['url'], | found_origins = list(self.storage.origin_search(self.origin['url'], | ||||
| regexp=True)) | regexp=True)) | ||||
| self.assertEqual(len(found_origins), 0) | self.assertEqual(len(found_origins), 0) | ||||
| ▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | def test_origin_search(self): | ||||
| self.assertEqual(found_origins[0], origin2_data) | self.assertEqual(found_origins[0], origin2_data) | ||||
| found_origins = list(self.storage.origin_search('.*/.*', offset=1, limit=1, regexp=True)) # noqa | found_origins = list(self.storage.origin_search('.*/.*', offset=1, limit=1, regexp=True)) # noqa | ||||
| self.assertEqual(len(found_origins), 1) | self.assertEqual(len(found_origins), 1) | ||||
| self.assertEqual(found_origins[0], origin2_data) | self.assertEqual(found_origins[0], origin2_data) | ||||
| def test_origin_visit_add(self): | def test_origin_visit_add(self): | ||||
| # given | # given | ||||
| self.assertIsNone(self.storage.origin_get(self.origin2)) | self.assertIsNone(self.storage.origin_get([self.origin2])[0]) | ||||
| origin_id = self.storage.origin_add_one(self.origin2) | origin_id = self.storage.origin_add_one(self.origin2) | ||||
| self.assertIsNotNone(origin_id) | self.assertIsNotNone(origin_id) | ||||
| # when | # when | ||||
| origin_visit1 = self.storage.origin_visit_add( | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id, | ||||
| date=self.date_visit2) | date=self.date_visit2) | ||||
| ▲ Show 20 Lines • Show All 1,071 Lines • ▼ Show 20 Lines | def test_generate_content_get_range_limit(self, contents): | ||||
| self.assertEqual(1, len(actual_contents2)) | self.assertEqual(1, len(actual_contents2)) | ||||
| self.assertIsNone(actual_next2) | self.assertIsNone(actual_next2) | ||||
| self.assert_contents_ok([contents_map[actual_next]], actual_contents2, | self.assert_contents_ok([contents_map[actual_next]], actual_contents2, | ||||
| keys_to_check) | keys_to_check) | ||||
| def test_origin_get_invalid_id(self): | def test_origin_get_invalid_id_legacy(self): | ||||
| invalid_origin_id = 1 | invalid_origin_id = 1 | ||||
| origin_info = self.storage.origin_get({'id': invalid_origin_id}) | origin_info = self.storage.origin_get({'id': invalid_origin_id}) | ||||
| self.assertIsNone(origin_info) | self.assertIsNone(origin_info) | ||||
| origin_visits = list(self.storage.origin_visit_get(invalid_origin_id)) | origin_visits = list(self.storage.origin_visit_get( | ||||
| invalid_origin_id)) | |||||
| self.assertEqual(origin_visits, []) | |||||
| def test_origin_get_invalid_id(self): | |||||
| origin_info = self.storage.origin_get([{'id': 1}, {'id': 2}]) | |||||
| self.assertEqual(origin_info, [None, None]) | |||||
| origin_visits = list(self.storage.origin_visit_get(1)) | |||||
| self.assertEqual(origin_visits, []) | self.assertEqual(origin_visits, []) | ||||
| @given(gen_origins(min_size=100, max_size=100)) | @given(gen_origins(min_size=100, max_size=100)) | ||||
| def test_origin_get_range(self, new_origins): | def test_origin_get_range(self, new_origins): | ||||
| nb_origins = len(new_origins) | nb_origins = len(new_origins) | ||||
| self.storage.origin_add(new_origins) | self.storage.origin_add(new_origins) | ||||
| ▲ Show 20 Lines • Show All 138 Lines • Show Last 20 Lines | |||||
s/id/origin_id/