Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 549 Lines • ▼ Show 20 Lines | class CommonTestStorage(TestStorageData): | ||||
| test_remote_storage.py. | test_remote_storage.py. | ||||
| We need to have the two classes inherit from this base class | We need to have the two classes inherit from this base class | ||||
| separately to avoid nosetests running the tests from the base | separately to avoid nosetests running the tests from the base | ||||
| class twice. | class twice. | ||||
| """ | """ | ||||
| maxDiff = None | maxDiff = None | ||||
| _test_origin_ids = True | |||||
| @staticmethod | @staticmethod | ||||
| def normalize_entity(entity): | def normalize_entity(entity): | ||||
| entity = copy.deepcopy(entity) | entity = copy.deepcopy(entity) | ||||
| for key in ('date', 'committer_date'): | for key in ('date', 'committer_date'): | ||||
| if key in entity: | if key in entity: | ||||
| entity[key] = identifiers.normalize_timestamp(entity[key]) | entity[key] = identifiers.normalize_timestamp(entity[key]) | ||||
| ▲ Show 20 Lines • Show All 280 Lines • ▼ Show 20 Lines | def test_skipped_content_add(self): | ||||
| 'skipped_content:add': 2, | 'skipped_content:add': 2, | ||||
| }) | }) | ||||
| missing = list(self.storage.skipped_content_missing([cont, cont2])) | missing = list(self.storage.skipped_content_missing([cont, cont2])) | ||||
| self.assertEqual(missing, []) | self.assertEqual(missing, []) | ||||
| @pytest.mark.property_based | @pytest.mark.property_based | ||||
| @settings(deadline=None) # this test is very slow | |||||
| @given(strategies.sets( | @given(strategies.sets( | ||||
| elements=strategies.sampled_from( | elements=strategies.sampled_from( | ||||
| ['sha256', 'sha1_git', 'blake2s256']), | ['sha256', 'sha1_git', 'blake2s256']), | ||||
| min_size=0)) | min_size=0)) | ||||
| def test_content_missing(self, algos): | def test_content_missing(self, algos): | ||||
| algos |= {'sha1'} | algos |= {'sha1'} | ||||
| cont2 = self.cont2 | cont2 = self.cont2 | ||||
| missing_cont = self.missing_cont | missing_cont = self.missing_cont | ||||
| ▲ Show 20 Lines • Show All 491 Lines • ▼ Show 20 Lines | class CommonTestStorage(TestStorageData): | ||||
| def test_origin_add_one(self): | def test_origin_add_one(self): | ||||
| origin0 = self.storage.origin_get(self.origin) | origin0 = self.storage.origin_get(self.origin) | ||||
| self.assertIsNone(origin0) | self.assertIsNone(origin0) | ||||
| id = self.storage.origin_add_one(self.origin) | id = self.storage.origin_add_one(self.origin) | ||||
| actual_origin = self.storage.origin_get({'url': self.origin['url'], | actual_origin = self.storage.origin_get({'url': self.origin['url'], | ||||
| 'type': self.origin['type']}) | 'type': self.origin['type']}) | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual(actual_origin['id'], id) | self.assertEqual(actual_origin['id'], id) | ||||
| self.assertEqual(actual_origin['url'], self.origin['url']) | |||||
| 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])[0] | 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] | }])[0] | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual(actual_origin['id'], origin1['id']) | self.assertEqual(actual_origin['id'], origin1['id']) | ||||
| self.assertEqual(actual_origin['url'], origin1['url']) | |||||
| 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] | }])[0] | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual(actual_origin2['id'], origin2['id']) | self.assertEqual(actual_origin2['id'], origin2['id']) | ||||
| self.assertEqual(actual_origin2['url'], origin2['url']) | |||||
| if 'id' in actual_origin: | |||||
| del actual_origin['id'] | del actual_origin['id'] | ||||
| del actual_origin2['id'] | del actual_origin2['id'] | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', actual_origin), | [('origin', actual_origin), | ||||
| ('origin', actual_origin2)]) | ('origin', actual_origin2)]) | ||||
| 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_without_type(self): | def test_origin_get_without_type(self): | ||||
| origin0 = self.storage.origin_get([self.origin])[0] | origin0 = self.storage.origin_get([self.origin])[0] | ||||
| self.assertIsNone(origin0) | self.assertIsNone(origin0) | ||||
| origin3 = self.origin2.copy() | origin3 = self.origin2.copy() | ||||
| origin3['type'] += 'foo' | origin3['type'] += 'foo' | ||||
| origin1, origin2, origin3 = self.storage.origin_add( | origin1, origin2, origin3 = self.storage.origin_add( | ||||
| [self.origin, self.origin2, origin3]) | [self.origin, self.origin2, origin3]) | ||||
| actual_origin = self.storage.origin_get([{ | actual_origin = self.storage.origin_get([{ | ||||
| 'url': self.origin['url'], | 'url': self.origin['url'], | ||||
| }])[0] | }])[0] | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual(actual_origin['id'], origin1['id']) | self.assertEqual(actual_origin['id'], origin1['id']) | ||||
| self.assertEqual(actual_origin['url'], origin1['url']) | |||||
| actual_origin_2_or_3 = self.storage.origin_get([{ | actual_origin_2_or_3 = self.storage.origin_get([{ | ||||
| 'url': self.origin2['url'], | 'url': self.origin2['url'], | ||||
| }])[0] | }])[0] | ||||
| if self._test_origin_ids: | |||||
| self.assertIn( | self.assertIn( | ||||
| actual_origin_2_or_3['id'], | actual_origin_2_or_3['id'], | ||||
| [origin2['id'], origin3['id']]) | [origin2['id'], origin3['id']]) | ||||
| self.assertIn( | |||||
| actual_origin_2_or_3['url'], | |||||
| [origin2['url'], origin3['url']]) | |||||
| if 'id' in actual_origin: | |||||
| del actual_origin['id'] | del actual_origin['id'] | ||||
| del actual_origin_2_or_3['id'] | del actual_origin_2_or_3['id'] | ||||
| del origin3['id'] | del origin3['id'] | ||||
| objects = list(self.journal_writer.objects) | objects = list(self.journal_writer.objects) | ||||
| if len(objects) == 3: | if len(objects) == 3: | ||||
| # current behavior of the pg storage, where 'type' is part of | # current behavior of the pg storage, where 'type' is part of | ||||
| # the primary key | # the primary key | ||||
| self.assertEqual(objects, | self.assertEqual(objects, | ||||
| [('origin', self.origin), | [('origin', self.origin), | ||||
| ('origin', self.origin2), | ('origin', self.origin2), | ||||
| ('origin', origin3)]) | ('origin', origin3)]) | ||||
| else: | else: | ||||
| # current behavior of the in-mem storage, and future behavior | # current behavior of the in-mem storage, and future behavior | ||||
| # of the pg storage, where 'type' is not part of the PK | # of the pg storage, where 'type' is not part of the PK | ||||
| self.assertEqual(objects, | self.assertEqual(objects, | ||||
| [('origin', self.origin), | [('origin', self.origin), | ||||
| ('origin', self.origin2)]) | ('origin', self.origin2)]) | ||||
| def test_origin_get_legacy(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( | actual_origin0 = self.storage.origin_get( | ||||
| {'url': self.origin['url'], 'type': self.origin['type']}) | {'url': self.origin['url'], 'type': self.origin['type']}) | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual(actual_origin0['id'], id) | self.assertEqual(actual_origin0['id'], id) | ||||
| self.assertEqual(actual_origin0['url'], self.origin['url']) | |||||
| # lookup per id (returns dict) | # lookup per id (returns dict) | ||||
| if self._test_origin_ids: | |||||
| 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): | def test_origin_get(self): | ||||
| self.assertIsNone(self.storage.origin_get(self.origin)) | self.assertIsNone(self.storage.origin_get(self.origin)) | ||||
| origin_id = self.storage.origin_add_one(self.origin) | 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( | actual_origin0 = self.storage.origin_get( | ||||
| [{'url': self.origin['url'], 'type': self.origin['type']}]) | [{'url': self.origin['url'], 'type': self.origin['type']}]) | ||||
| self.assertEqual(len(actual_origin0), 1, actual_origin0) | self.assertEqual(len(actual_origin0), 1, actual_origin0) | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual(actual_origin0[0]['id'], origin_id) | self.assertEqual(actual_origin0[0]['id'], origin_id) | ||||
| self.assertEqual(actual_origin0[0]['url'], self.origin['url']) | |||||
| if self._test_origin_ids: | |||||
| # lookup per id (returns dict) | # lookup per id (returns dict) | ||||
| actual_origin1 = self.storage.origin_get([{'id': origin_id}]) | actual_origin1 = self.storage.origin_get([{'id': origin_id}]) | ||||
| self.assertEqual(len(actual_origin1), 1, actual_origin1) | self.assertEqual(len(actual_origin1), 1, actual_origin1) | ||||
| self.assertEqual(actual_origin1[0], {'id': origin_id, | self.assertEqual(actual_origin1[0], {'id': origin_id, | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'url': self.origin['url']}) | 'url': self.origin['url']}) | ||||
| def test_origin_get_consistency(self): | def test_origin_get_consistency(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) | ||||
| with self.assertRaises(ValueError): | with self.assertRaises(ValueError): | ||||
| self.storage.origin_get([ | self.storage.origin_get([ | ||||
| {'url': self.origin['url'], 'type': self.origin['type']}, | {'url': self.origin['url'], 'type': self.origin['type']}, | ||||
| {'id': id}]) | {'id': id}]) | ||||
| def test_origin_search(self): | def test_origin_search_single_result(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) | ||||
| id = self.storage.origin_add_one(self.origin) | self.storage.origin_add_one(self.origin) | ||||
| origin_data = {'id': id, | origin_data = { | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'url': self.origin['url']} | 'url': self.origin['url']} | ||||
| 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), 1) | self.assertEqual(len(found_origins), 1) | ||||
| if 'id' in found_origins[0]: | |||||
| del found_origins[0]['id'] | |||||
| self.assertEqual(found_origins[0], origin_data) | self.assertEqual(found_origins[0], origin_data) | ||||
| found_origins = list(self.storage.origin_search( | found_origins = list(self.storage.origin_search( | ||||
| '.' + self.origin['url'][1:-1] + '.', regexp=True)) | '.' + self.origin['url'][1:-1] + '.', regexp=True)) | ||||
| self.assertEqual(len(found_origins), 1) | self.assertEqual(len(found_origins), 1) | ||||
| if 'id' in found_origins[0]: | |||||
| del found_origins[0]['id'] | |||||
| self.assertEqual(found_origins[0], origin_data) | self.assertEqual(found_origins[0], origin_data) | ||||
| id2 = self.storage.origin_add_one(self.origin2) | self.storage.origin_add_one(self.origin2) | ||||
| origin2_data = {'id': id2, | origin2_data = { | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'url': self.origin2['url']} | 'url': self.origin2['url']} | ||||
| found_origins = list(self.storage.origin_search(self.origin2['url'])) | found_origins = list(self.storage.origin_search(self.origin2['url'])) | ||||
| self.assertEqual(len(found_origins), 1) | self.assertEqual(len(found_origins), 1) | ||||
| if 'id' in found_origins[0]: | |||||
| del found_origins[0]['id'] | |||||
| self.assertEqual(found_origins[0], origin2_data) | self.assertEqual(found_origins[0], origin2_data) | ||||
| found_origins = list(self.storage.origin_search( | found_origins = list(self.storage.origin_search( | ||||
| '.' + self.origin2['url'][1:-1] + '.', regexp=True)) | '.' + self.origin2['url'][1:-1] + '.', regexp=True)) | ||||
| self.assertEqual(len(found_origins), 1) | self.assertEqual(len(found_origins), 1) | ||||
| if 'id' in found_origins[0]: | |||||
| del found_origins[0]['id'] | |||||
| self.assertEqual(found_origins[0], origin2_data) | self.assertEqual(found_origins[0], origin2_data) | ||||
| # Search / (regexp=False) | def test_origin_search_no_regexp(self): | ||||
| self.storage.origin_add_one(self.origin) | |||||
| self.storage.origin_add_one(self.origin2) | |||||
| origin = self.storage.origin_get({'url': self.origin['url']}) | |||||
| origin2 = self.storage.origin_get({'url': self.origin2['url']}) | |||||
| # no pagination | |||||
| found_origins = list(self.storage.origin_search('/')) | found_origins = list(self.storage.origin_search('/')) | ||||
| self.assertEqual(len(found_origins), 2) | self.assertEqual(len(found_origins), 2) | ||||
| # offset=0 | |||||
| found_origins0 = list(self.storage.origin_search('/', offset=0, limit=1)) # noqa | found_origins0 = list(self.storage.origin_search('/', offset=0, limit=1)) # noqa | ||||
| self.assertEqual(len(found_origins0), 1) | self.assertEqual(len(found_origins0), 1) | ||||
| self.assertIn(found_origins0[0], [origin_data, origin2_data]) | self.assertIn(found_origins0[0], [origin, origin2]) | ||||
| # offset=1 | |||||
| found_origins1 = list(self.storage.origin_search('/', offset=1, limit=1)) # noqa | found_origins1 = list(self.storage.origin_search('/', offset=1, limit=1)) # noqa | ||||
| self.assertEqual(len(found_origins1), 1) | self.assertEqual(len(found_origins1), 1) | ||||
| self.assertIn(found_origins1[0], [origin_data, origin2_data]) | self.assertIn(found_origins1[0], [origin, origin2]) | ||||
| # check both origins were returned | |||||
| self.assertCountEqual(found_origins0 + found_origins1, | self.assertCountEqual(found_origins0 + found_origins1, | ||||
| [origin_data, origin2_data]) | [origin, origin2]) | ||||
| def test_origin_search_regexp_substring(self): | |||||
| self.storage.origin_add_one(self.origin) | |||||
| self.storage.origin_add_one(self.origin2) | |||||
| origin = self.storage.origin_get({'url': self.origin['url']}) | |||||
| origin2 = self.storage.origin_get({'url': self.origin2['url']}) | |||||
| # Search / (regexp=True) | # no pagination | ||||
| found_origins = list(self.storage.origin_search('/', regexp=True)) | found_origins = list(self.storage.origin_search('/', regexp=True)) | ||||
| self.assertEqual(len(found_origins), 2) | self.assertEqual(len(found_origins), 2) | ||||
| # offset=0 | |||||
| found_origins0 = list(self.storage.origin_search('/', offset=0, limit=1, regexp=True)) # noqa | found_origins0 = list(self.storage.origin_search('/', offset=0, limit=1, regexp=True)) # noqa | ||||
| self.assertEqual(len(found_origins0), 1) | self.assertEqual(len(found_origins0), 1) | ||||
| self.assertIn(found_origins0[0], [origin_data, origin2_data]) | self.assertIn(found_origins0[0], [origin, origin2]) | ||||
| # offset=1 | |||||
| found_origins1 = list(self.storage.origin_search('/', offset=1, limit=1, regexp=True)) # noqa | found_origins1 = list(self.storage.origin_search('/', offset=1, limit=1, regexp=True)) # noqa | ||||
| self.assertEqual(len(found_origins1), 1) | self.assertEqual(len(found_origins1), 1) | ||||
| self.assertIn(found_origins1[0], [origin_data, origin2_data]) | self.assertIn(found_origins1[0], [origin, origin2]) | ||||
| # check both origins were returned | |||||
| self.assertCountEqual(found_origins0 + found_origins1, | self.assertCountEqual(found_origins0 + found_origins1, | ||||
| [origin_data, origin2_data]) | [origin, origin2]) | ||||
| def test_origin_search_regexp_fullstring(self): | |||||
| self.storage.origin_add_one(self.origin) | |||||
| self.storage.origin_add_one(self.origin2) | |||||
| origin = self.storage.origin_get({'url': self.origin['url']}) | |||||
| origin2 = self.storage.origin_get({'url': self.origin2['url']}) | |||||
| # Search .*/.* (regexp=True) | # no pagination | ||||
| found_origins = list(self.storage.origin_search('.*/.*', regexp=True)) | found_origins = list(self.storage.origin_search('.*/.*', regexp=True)) | ||||
| self.assertEqual(len(found_origins), 2) | self.assertEqual(len(found_origins), 2) | ||||
| found_origins = list(self.storage.origin_search('/', offset=0, limit=1)) # noqa | # offset=0 | ||||
| self.assertEqual(len(found_origins), 1) | |||||
| self.assertEqual(found_origins[0], origin_data) | |||||
vlorentz: note that I dropped these three lines [...] | |||||
| found_origins = list(self.storage.origin_search('.*/.*', offset=0, limit=1, regexp=True)) # noqa | found_origins0 = list(self.storage.origin_search('.*/.*', offset=0, limit=1, regexp=True)) # noqa | ||||
| self.assertEqual(len(found_origins), 1) | self.assertEqual(len(found_origins0), 1) | ||||
| self.assertEqual(found_origins[0], origin_data) | self.assertIn(found_origins0[0], [origin, origin2]) | ||||
| found_origins = list(self.storage.origin_search('/', offset=1, limit=1)) # noqa | # offset=1 | ||||
| self.assertEqual(len(found_origins), 1) | |||||
| self.assertEqual(found_origins[0], origin2_data) | |||||
Done Inline Actions[...] and these three, because they are duplicates of lines above. vlorentz: [...] and these three, because they are duplicates of lines above. | |||||
| found_origins = list(self.storage.origin_search('.*/.*', offset=1, limit=1, regexp=True)) # noqa | found_origins1 = list(self.storage.origin_search('.*/.*', offset=1, limit=1, regexp=True)) # noqa | ||||
| self.assertEqual(len(found_origins), 1) | self.assertEqual(len(found_origins1), 1) | ||||
| self.assertEqual(found_origins[0], origin2_data) | self.assertIn(found_origins1[0], [origin, origin2]) | ||||
| # check both origins were returned | |||||
| self.assertCountEqual( | |||||
| found_origins0 + found_origins1, | |||||
| [origin, origin2]) | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_add(self, use_url): | def test_origin_visit_add(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # given | # given | ||||
| self.assertIsNone(self.storage.origin_get([self.origin2])[0]) | 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) | ||||
| Show All 29 Lines | def test_origin_visit_add(self, use_url): | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin_visit', data)]) | ('origin_visit', data)]) | ||||
| def test_origin_visit_get__unknown_origin(self): | def test_origin_visit_get__unknown_origin(self): | ||||
| self.assertEqual([], list(self.storage.origin_visit_get('foo'))) | self.assertEqual([], list(self.storage.origin_visit_get('foo'))) | ||||
| if self._test_origin_ids: | |||||
| self.assertEqual([], list(self.storage.origin_visit_get(10))) | self.assertEqual([], list(self.storage.origin_visit_get(10))) | ||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_add_default_type(self, use_url): | def test_origin_visit_add_default_type(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # given | # given | ||||
| self.assertIsNone(self.storage.origin_get([self.origin2])[0]) | 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) | ||||
| origin_id_or_url = self.origin2['url'] if use_url else origin_id | origin_id_or_url = self.origin2['url'] if use_url else origin_id | ||||
| self.assertIsNotNone(origin_id) | self.assertIsNotNone(origin_id) | ||||
| ▲ Show 20 Lines • Show All 54 Lines • ▼ Show 20 Lines | def test_origin_visit_add_default_type(self, use_url): | ||||
| } | } | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)]) | ('origin_visit', data2)]) | ||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_update(self, use_url): | def test_origin_visit_update(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # given | # given | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id2 = self.storage.origin_add_one(self.origin2) | origin_id2 = self.storage.origin_add_one(self.origin2) | ||||
| origin2_id_or_url = self.origin2['url'] if use_url else origin_id2 | origin2_id_or_url = self.origin2['url'] if use_url else origin_id2 | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| ▲ Show 20 Lines • Show All 165 Lines • ▼ Show 20 Lines | def test_origin_visit_find_by_date(self): | ||||
| # There are two visits at the same date, the latest must be returned | # There are two visits at the same date, the latest must be returned | ||||
| visit = self.storage.origin_visit_find_by_date( | visit = self.storage.origin_visit_find_by_date( | ||||
| self.origin['url'], self.date_visit2) | self.origin['url'], self.date_visit2) | ||||
| self.assertEqual(visit['visit'], origin_visit3['visit']) | self.assertEqual(visit['visit'], origin_visit3['visit']) | ||||
| def test_origin_visit_find_by_date__unknown_origin(self): | def test_origin_visit_find_by_date__unknown_origin(self): | ||||
| self.storage.origin_visit_find_by_date('foo', self.date_visit2) | self.storage.origin_visit_find_by_date('foo', self.date_visit2) | ||||
| @settings(deadline=None) | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_update_missing_snapshot(self, use_url): | def test_origin_visit_update_missing_snapshot(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # given | # given | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin_visit = self.storage.origin_visit_add( | origin_visit = self.storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| Show All 10 Lines | def test_origin_visit_update_missing_snapshot(self, use_url): | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| origin_visit['visit']) | origin_visit['visit']) | ||||
| self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id']) | self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id']) | ||||
| # when | # when | ||||
| self.storage.snapshot_add([self.snapshot]) | self.storage.snapshot_add([self.snapshot]) | ||||
| self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id']) | self.assertEqual(actual_origin_visit['snapshot'], self.snapshot['id']) | ||||
| @settings(deadline=None) | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_get_by(self, use_url): | def test_origin_visit_get_by(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id2 = self.storage.origin_add_one(self.origin2) | origin_id2 = self.storage.origin_add_one(self.origin2) | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin2_id_or_url = self.origin2['url'] if use_url else origin_id2 | origin2_id_or_url = self.origin2['url'] if use_url else origin_id2 | ||||
| ▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines | def test_origin_visit_get_by(self, use_url): | ||||
| actual_origin_visit1 = self.storage.origin_visit_get_by( | actual_origin_visit1 = self.storage.origin_visit_get_by( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| origin_visit1['visit']) | origin_visit1['visit']) | ||||
| # then | # then | ||||
| self.assertEqual(actual_origin_visit1, expected_origin_visit) | self.assertEqual(actual_origin_visit1, expected_origin_visit) | ||||
| def test_origin_visit_get_by__unknown_origin(self): | def test_origin_visit_get_by__unknown_origin(self): | ||||
| if self._test_origin_ids: | |||||
| self.assertIsNone(self.storage.origin_visit_get_by(2, 10)) | self.assertIsNone(self.storage.origin_visit_get_by(2, 10)) | ||||
| self.assertIsNone(self.storage.origin_visit_get_by('foo', 10)) | self.assertIsNone(self.storage.origin_visit_get_by('foo', 10)) | ||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_upsert_new(self, use_url): | def test_origin_visit_upsert_new(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # given | # given | ||||
| self.assertIsNone(self.storage.origin_get([self.origin2])[0]) | 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) | ||||
| origin_id_or_url = self.origin2['url'] if use_url else origin_id | origin_id_or_url = self.origin2['url'] if use_url else origin_id | ||||
| self.assertIsNotNone(origin_id) | self.assertIsNotNone(origin_id) | ||||
| ▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines | def test_origin_visit_upsert_new(self, use_url): | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)]) | ('origin_visit', data2)]) | ||||
| @settings(deadline=None) | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_upsert_existing(self, use_url): | def test_origin_visit_upsert_existing(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # given | # given | ||||
| self.assertIsNone(self.storage.origin_get([self.origin2])[0]) | 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) | ||||
| origin_id_or_url = self.origin2['url'] if use_url else origin_id | origin_id_or_url = self.origin2['url'] if use_url else origin_id | ||||
| self.assertIsNotNone(origin_id) | self.assertIsNotNone(origin_id) | ||||
| ▲ Show 20 Lines • Show All 49 Lines • ▼ Show 20 Lines | def test_origin_visit_upsert_existing(self, use_url): | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)]) | ('origin_visit', data2)]) | ||||
| def test_origin_visit_get_by_no_result(self): | def test_origin_visit_get_by_no_result(self): | ||||
| if self._test_origin_ids: | |||||
| actual_origin_visit = self.storage.origin_visit_get_by( | actual_origin_visit = self.storage.origin_visit_get_by( | ||||
| 10, 999) | 10, 999) | ||||
| self.assertIsNone(actual_origin_visit) | self.assertIsNone(actual_origin_visit) | ||||
| self.storage.origin_add([self.origin]) | self.storage.origin_add([self.origin]) | ||||
| actual_origin_visit = self.storage.origin_visit_get_by( | actual_origin_visit = self.storage.origin_visit_get_by( | ||||
| self.origin['url'], 999) | self.origin['url'], 999) | ||||
| self.assertIsNone(actual_origin_visit) | self.assertIsNone(actual_origin_visit) | ||||
| def test_person_get(self): | def test_person_get(self): | ||||
| # given (person injection through revision for example) | # given (person injection through revision for example) | ||||
| Show All 28 Lines | def test_person_get(self): | ||||
| # when | # when | ||||
| actual_persons = self.storage.person_get([id1, id0]) | actual_persons = self.storage.person_get([id1, id0]) | ||||
| # then | # then | ||||
| expected_persons.reverse() | expected_persons.reverse() | ||||
| self.assertEqual(list(actual_persons), expected_persons) | self.assertEqual(list(actual_persons), expected_persons) | ||||
| @settings(deadline=None) # this test is very slow | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_origin_visit_get_latest(self, use_url): | def test_origin_visit_get_latest(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin_url = self.origin['url'] | origin_url = self.origin['url'] | ||||
| origin_visit1 = self.storage.origin_visit_add( | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| ▲ Show 20 Lines • Show All 548 Lines • ▼ Show 20 Lines | def test_snapshot_add_twice(self): | ||||
| self.assertEqual(list(self.journal_writer.objects), | self.assertEqual(list(self.journal_writer.objects), | ||||
| [('origin', expected_origin), | [('origin', expected_origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('snapshot', self.snapshot), | ('snapshot', self.snapshot), | ||||
| ('origin_visit', data2), | ('origin_visit', data2), | ||||
| ('origin_visit', data3), | ('origin_visit', data3), | ||||
| ('origin_visit', data4)]) | ('origin_visit', data4)]) | ||||
| @settings(deadline=None) # this test is very slow | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_snapshot_get_latest(self, use_url): | def test_snapshot_get_latest(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin_visit1 = self.storage.origin_visit_add(origin_id, | origin_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| origin_visit2 = self.storage.origin_visit_add(origin_id, | origin_visit2 = self.storage.origin_visit_add(origin_id, | ||||
| ▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest(self, use_url): | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit3_id, snapshot=self.complete_snapshot['id']) | origin_id, visit3_id, snapshot=self.complete_snapshot['id']) | ||||
| self.assertEqual(self.complete_snapshot, | self.assertEqual(self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest( | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url)) | origin_id_or_url)) | ||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_snapshot_get_latest__missing_snapshot(self, use_url): | def test_snapshot_get_latest__missing_snapshot(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| # Origin does not exist | # Origin does not exist | ||||
| self.assertIsNone(self.storage.snapshot_get_latest( | self.assertIsNone(self.storage.snapshot_get_latest( | ||||
| self.origin['url'] if use_url else 999)) | self.origin['url'] if use_url else 999)) | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| ▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest__missing_snapshot(self, use_url): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| self.snapshot, | self.snapshot, | ||||
| self.storage.snapshot_get_latest( | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url) | origin_id_or_url) | ||||
| ) | ) | ||||
| def test_stat_counters(self): | def test_stat_counters(self): | ||||
| expected_keys = ['content', 'directory', | expected_keys = ['content', 'directory', | ||||
| 'origin', 'person', 'revision'] | 'origin', 'revision'] | ||||
| # Initially, all counters are 0 | # Initially, all counters are 0 | ||||
| self.storage.refresh_stat_counters() | self.storage.refresh_stat_counters() | ||||
| counters = self.storage.stat_counters() | counters = self.storage.stat_counters() | ||||
| self.assertTrue(set(expected_keys) <= set(counters)) | self.assertTrue(set(expected_keys) <= set(counters)) | ||||
| for key in expected_keys: | for key in expected_keys: | ||||
| self.assertEqual(counters[key], 0) | self.assertEqual(counters[key], 0) | ||||
| Show All 29 Lines | def test_stat_counters(self): | ||||
| self.assertEqual(counters['content'], 1) | self.assertEqual(counters['content'], 1) | ||||
| self.assertEqual(counters['directory'], 1) | self.assertEqual(counters['directory'], 1) | ||||
| self.assertEqual(counters['snapshot'], 1) | self.assertEqual(counters['snapshot'], 1) | ||||
| self.assertEqual(counters['origin'], 1) | self.assertEqual(counters['origin'], 1) | ||||
| self.assertEqual(counters['origin_visit'], 1) | self.assertEqual(counters['origin_visit'], 1) | ||||
| self.assertEqual(counters['revision'], 1) | self.assertEqual(counters['revision'], 1) | ||||
| self.assertEqual(counters['release'], 1) | self.assertEqual(counters['release'], 1) | ||||
| self.assertEqual(counters['snapshot'], 1) | self.assertEqual(counters['snapshot'], 1) | ||||
| if 'person' in counters: | |||||
| self.assertEqual(counters['person'], 3) | self.assertEqual(counters['person'], 3) | ||||
| def test_content_find_ctime(self): | def test_content_find_ctime(self): | ||||
| cont = self.cont.copy() | cont = self.cont.copy() | ||||
| del cont['data'] | del cont['data'] | ||||
| now = datetime.datetime.now(tz=datetime.timezone.utc) | now = datetime.datetime.now(tz=datetime.timezone.utc) | ||||
| cont['ctime'] = now | cont['ctime'] = now | ||||
| self.storage.content_add_metadata([cont]) | self.storage.content_add_metadata([cont]) | ||||
| actually_present = self.storage.content_find({'sha1': cont['sha1']}) | actually_present = self.storage.content_find({'sha1': cont['sha1']}) | ||||
| # check ctime up to one second | |||||
| dt = actually_present[0]['ctime'] - now | |||||
| self.assertLessEqual(abs(dt.total_seconds()), 1, dt) | |||||
| del actually_present[0]['ctime'] | |||||
| self.assertEqual(actually_present[0], { | self.assertEqual(actually_present[0], { | ||||
| 'ctime': now, | |||||
| 'sha1': cont['sha1'], | 'sha1': cont['sha1'], | ||||
| 'sha256': cont['sha256'], | 'sha256': cont['sha256'], | ||||
| 'sha1_git': cont['sha1_git'], | 'sha1_git': cont['sha1_git'], | ||||
| 'blake2s256': cont['blake2s256'], | 'blake2s256': cont['blake2s256'], | ||||
| 'length': cont['length'], | 'length': cont['length'], | ||||
| 'status': 'visible' | 'status': 'visible' | ||||
| }) | }) | ||||
| ▲ Show 20 Lines • Show All 244 Lines • ▼ Show 20 Lines | def test_object_find_by_sha1_git(self): | ||||
| 'sha1_git': self.release['id'], | 'sha1_git': self.release['id'], | ||||
| 'type': 'release', | 'type': 'release', | ||||
| 'id': self.release['id'], | 'id': self.release['id'], | ||||
| }] | }] | ||||
| ret = self.storage.object_find_by_sha1_git(sha1_gits) | ret = self.storage.object_find_by_sha1_git(sha1_gits) | ||||
| for val in ret.values(): | for val in ret.values(): | ||||
| for obj in val: | for obj in val: | ||||
| if 'object_id' in obj: | |||||
| del obj['object_id'] | del obj['object_id'] | ||||
| self.assertEqual(expected, ret) | self.assertEqual(expected, ret) | ||||
| def test_tool_add(self): | def test_tool_add(self): | ||||
| tool = { | tool = { | ||||
| 'name': 'some-unknown-tool', | 'name': 'some-unknown-tool', | ||||
| 'version': 'some-version', | 'version': 'some-version', | ||||
| 'configuration': {"debian-package": "some-package"}, | 'configuration': {"debian-package": "some-package"}, | ||||
| ▲ Show 20 Lines • Show All 327 Lines • ▼ Show 20 Lines | def test_origin_metadata_get_by_provider_type(self): | ||||
| }] | }] | ||||
| # then | # then | ||||
| self.assertEqual(len(m_by_provider), 1) | self.assertEqual(len(m_by_provider), 1) | ||||
| self.assertEqual(m_by_provider, expected_results) | self.assertEqual(m_by_provider, expected_results) | ||||
| class CommonPropTestStorage: | class CommonPropTestStorage: | ||||
| _test_origin_ids = True | |||||
| def assert_contents_ok(self, expected_contents, actual_contents, | def assert_contents_ok(self, expected_contents, actual_contents, | ||||
| keys_to_check={'sha1', 'data'}): | keys_to_check={'sha1', 'data'}): | ||||
| """Assert that a given list of contents matches on a given set of keys. | """Assert that a given list of contents matches on a given set of keys. | ||||
| """ | """ | ||||
| for k in keys_to_check: | for k in keys_to_check: | ||||
| expected_list = sorted([c[k] for c in expected_contents]) | expected_list = sorted([c[k] for c in expected_contents]) | ||||
| actual_list = sorted([c[k] for c in actual_contents]) | actual_list = sorted([c[k] for c in actual_contents]) | ||||
| ▲ Show 20 Lines • Show All 131 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_legacy(self): | def test_origin_get_invalid_id_legacy(self): | ||||
| if self._test_origin_ids: | |||||
| 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( | origin_visits = list(self.storage.origin_visit_get( | ||||
| invalid_origin_id)) | invalid_origin_id)) | ||||
| self.assertEqual(origin_visits, []) | self.assertEqual(origin_visits, []) | ||||
| def test_origin_get_invalid_id(self): | def test_origin_get_invalid_id(self): | ||||
| if self._test_origin_ids: | |||||
| origin_info = self.storage.origin_get([{'id': 1}, {'id': 2}]) | origin_info = self.storage.origin_get([{'id': 1}, {'id': 2}]) | ||||
| self.assertEqual(origin_info, [None, None]) | self.assertEqual(origin_info, [None, None]) | ||||
| origin_visits = list(self.storage.origin_visit_get(1)) | origin_visits = list(self.storage.origin_visit_get(1)) | ||||
| self.assertEqual(origin_visits, []) | self.assertEqual(origin_visits, []) | ||||
| @given(strategies.sets(origins().map(lambda x: tuple(x.to_dict().items())), | @given(strategies.sets(origins().map(lambda x: tuple(x.to_dict().items())), | ||||
| min_size=6, max_size=15)) | min_size=6, max_size=15)) | ||||
| def test_origin_get_range(self, new_origins): | def test_origin_get_range(self, new_origins): | ||||
| self.reset_storage() | self.reset_storage() | ||||
| new_origins = list(map(dict, new_origins)) | new_origins = list(map(dict, new_origins)) | ||||
| nb_origins = len(new_origins) | nb_origins = len(new_origins) | ||||
| ▲ Show 20 Lines • Show All 87 Lines • ▼ Show 20 Lines | |||||
| class TestLocalStorage(CommonTestStorage, StorageTestDbFixture, | class TestLocalStorage(CommonTestStorage, StorageTestDbFixture, | ||||
| unittest.TestCase): | unittest.TestCase): | ||||
| """Test the local storage""" | """Test the local storage""" | ||||
| # Can only be tested with local storage as you can't mock | # Can only be tested with local storage as you can't mock | ||||
| # datetimes for the remote server | # datetimes for the remote server | ||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_fetch_history(self, use_url): | def test_fetch_history(self, use_url): | ||||
| if not self._test_origin_ids and not use_url: | |||||
| return | |||||
| self.reset_storage() | self.reset_storage() | ||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| with patch('datetime.datetime'): | with patch('datetime.datetime'): | ||||
| datetime.datetime.now.return_value = self.fetch_history_date | datetime.datetime.now.return_value = self.fetch_history_date | ||||
| fetch_history_id = self.storage.fetch_history_start( | fetch_history_id = self.storage.fetch_history_start( | ||||
| origin_id_or_url) | origin_id_or_url) | ||||
| ▲ Show 20 Lines • Show All 102 Lines • Show Last 20 Lines | |||||
note that I dropped these three lines [...]