Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,595 Lines • ▼ Show 20 Lines | def test_origin_visit_add(self, use_url): | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| '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', data)]) | ('origin_visit', data)]) | ||||
| def test_origin_visit_add_default_type(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(10))) | |||||
| @given(strategies.booleans()) | |||||
| def test_origin_visit_add_default_type(self, use_url): | |||||
| 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 | |||||
| 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_or_url, | ||||
anlambert: Same comment as in D1664, using a variable to store the origin identifier or url would improve… | |||||
| date=self.date_visit2) | date=self.date_visit2) | ||||
| origin_visit2 = self.storage.origin_visit_add( | origin_visit2 = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id_or_url, | ||||
| date='2018-01-01 23:00:00+00') | date='2018-01-01 23:00:00+00') | ||||
| # then | # then | ||||
| self.assertEqual(origin_visit1['origin'], origin_id) | self.assertEqual(origin_visit1['origin'], origin_id) | ||||
| self.assertIsNotNone(origin_visit1['visit']) | self.assertIsNotNone(origin_visit1['visit']) | ||||
| actual_origin_visits = list(self.storage.origin_visit_get(origin_id)) | actual_origin_visits = list(self.storage.origin_visit_get( | ||||
| origin_id_or_url)) | |||||
| self.assertEqual(actual_origin_visits, [ | self.assertEqual(actual_origin_visits, [ | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': 'hg', | 'type': 'hg', | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| Show All 29 Lines | def test_origin_visit_add_default_type(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)]) | ||||
| def test_origin_visit_update(self): | @given(strategies.booleans()) | ||||
| def test_origin_visit_update(self, use_url): | |||||
| 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 | |||||
| origin_id_or_url = self.origin['url'] if use_url else origin_id | |||||
| origin_visit1 = self.storage.origin_visit_add( | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id_or_url, | ||||
| date=self.date_visit2) | date=self.date_visit2) | ||||
| origin_visit2 = self.storage.origin_visit_add( | origin_visit2 = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id_or_url, | ||||
| date=self.date_visit3) | date=self.date_visit3) | ||||
| origin_visit3 = self.storage.origin_visit_add( | origin_visit3 = self.storage.origin_visit_add( | ||||
| origin_id2, | origin2_id_or_url, | ||||
| date=self.date_visit3) | date=self.date_visit3) | ||||
| # when | # when | ||||
| visit1_metadata = { | visit1_metadata = { | ||||
| 'contents': 42, | 'contents': 42, | ||||
| 'directories': 22, | 'directories': 22, | ||||
| } | } | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, origin_visit1['visit'], status='full', | origin_id_or_url, | ||||
| origin_visit1['visit'], status='full', | |||||
| metadata=visit1_metadata) | metadata=visit1_metadata) | ||||
| self.storage.origin_visit_update(origin_id2, origin_visit3['visit'], | self.storage.origin_visit_update( | ||||
| status='partial') | origin2_id_or_url, | ||||
| origin_visit3['visit'], status='partial') | |||||
| # then | # then | ||||
| actual_origin_visits = list(self.storage.origin_visit_get(origin_id)) | actual_origin_visits = list(self.storage.origin_visit_get( | ||||
| origin_id_or_url)) | |||||
| self.assertEqual(actual_origin_visits, [{ | self.assertEqual(actual_origin_visits, [{ | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_visit2['origin'], | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, { | }, { | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_visit2['origin'], | ||||
| 'date': self.date_visit3, | 'date': self.date_visit3, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }]) | }]) | ||||
| actual_origin_visits_bis = list(self.storage.origin_visit_get( | actual_origin_visits_bis = list(self.storage.origin_visit_get( | ||||
| origin_id, limit=1)) | origin_id_or_url, | ||||
| limit=1)) | |||||
| self.assertEqual(actual_origin_visits_bis, | self.assertEqual(actual_origin_visits_bis, | ||||
| [{ | [{ | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_visit2['origin'], | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }]) | }]) | ||||
| actual_origin_visits_ter = list(self.storage.origin_visit_get( | actual_origin_visits_ter = list(self.storage.origin_visit_get( | ||||
| origin_id, last_visit=origin_visit1['visit'])) | origin_id_or_url, | ||||
| last_visit=origin_visit1['visit'])) | |||||
| self.assertEqual(actual_origin_visits_ter, | self.assertEqual(actual_origin_visits_ter, | ||||
| [{ | [{ | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_visit2['origin'], | ||||
| 'date': self.date_visit3, | 'date': self.date_visit3, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }]) | }]) | ||||
| actual_origin_visits2 = list(self.storage.origin_visit_get(origin_id2)) | actual_origin_visits2 = list(self.storage.origin_visit_get( | ||||
| origin2_id_or_url)) | |||||
| self.assertEqual(actual_origin_visits2, | self.assertEqual(actual_origin_visits2, | ||||
| [{ | [{ | ||||
| 'origin': origin_visit3['origin'], | 'origin': origin_visit3['origin'], | ||||
| 'date': self.date_visit3, | 'date': self.date_visit3, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'status': 'partial', | 'status': 'partial', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| ▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Lines | def test_origin_visit_find_by_date(self): | ||||
| self.origin['url'], self.date_visit3) | self.origin['url'], self.date_visit3) | ||||
| self.assertEqual(visit['visit'], origin_visit2['visit']) | self.assertEqual(visit['visit'], origin_visit2['visit']) | ||||
| # 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_update_missing_snapshot(self): | def test_origin_visit_find_by_date__unknown_origin(self): | ||||
| self.storage.origin_visit_find_by_date('foo', self.date_visit2) | |||||
| @given(strategies.booleans()) | |||||
| def test_origin_visit_update_missing_snapshot(self, use_url): | |||||
| 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_visit = self.storage.origin_visit_add( | origin_visit = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id_or_url, | ||||
| date=self.date_visit1) | date=self.date_visit1) | ||||
| # when | # when | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, origin_visit['visit'], | origin_id_or_url, | ||||
| origin_visit['visit'], | |||||
| snapshot=self.snapshot['id']) | snapshot=self.snapshot['id']) | ||||
| # then | # then | ||||
| actual_origin_visit = self.storage.origin_visit_get_by( | actual_origin_visit = self.storage.origin_visit_get_by( | ||||
| origin_visit['origin'], origin_visit['visit']) | origin_id_or_url, | ||||
| 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']) | ||||
| def test_origin_visit_get_by(self): | @given(strategies.booleans()) | ||||
| def test_origin_visit_get_by(self, use_url): | |||||
| 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 | |||||
| origin2_id_or_url = self.origin2['url'] if use_url else origin_id2 | |||||
| origin_visit1 = self.storage.origin_visit_add( | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id, | origin_id_or_url, | ||||
| date=self.date_visit2) | date=self.date_visit2) | ||||
| self.storage.snapshot_add(origin_id, origin_visit1['visit'], | self.storage.snapshot_add(origin_id, origin_visit1['visit'], | ||||
| self.snapshot) | self.snapshot) | ||||
| # Add some other {origin, visit} entries | # Add some other {origin, visit} entries | ||||
| self.storage.origin_visit_add(origin_id, date=self.date_visit3) | self.storage.origin_visit_add( | ||||
| self.storage.origin_visit_add(origin_id2, date=self.date_visit3) | origin_id_or_url, | ||||
| date=self.date_visit3) | |||||
| self.storage.origin_visit_add( | |||||
| origin2_id_or_url, | |||||
| date=self.date_visit3) | |||||
| # when | # when | ||||
| visit1_metadata = { | visit1_metadata = { | ||||
| 'contents': 42, | 'contents': 42, | ||||
| 'directories': 22, | 'directories': 22, | ||||
| } | } | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, origin_visit1['visit'], status='full', | origin_id_or_url, | ||||
| origin_visit1['visit'], status='full', | |||||
| metadata=visit1_metadata) | metadata=visit1_metadata) | ||||
| expected_origin_visit = origin_visit1.copy() | expected_origin_visit = origin_visit1.copy() | ||||
| expected_origin_visit.update({ | expected_origin_visit.update({ | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'type': self.origin['type'], | 'type': self.origin['type'], | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'snapshot': self.snapshot['id'], | 'snapshot': self.snapshot['id'], | ||||
| }) | }) | ||||
| # when | # when | ||||
| actual_origin_visit1 = self.storage.origin_visit_get_by( | actual_origin_visit1 = self.storage.origin_visit_get_by( | ||||
| origin_visit1['origin'], origin_visit1['visit']) | origin_id_or_url, | ||||
| 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_upsert_new(self): | def test_origin_visit_get_by__unknown_origin(self): | ||||
| self.assertIsNone(self.storage.origin_visit_get_by(2, 10)) | |||||
| self.assertIsNone(self.storage.origin_visit_get_by('foo', 10)) | |||||
| @given(strategies.booleans()) | |||||
| def test_origin_visit_upsert_new(self, use_url): | |||||
| 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 | |||||
| self.assertIsNotNone(origin_id) | self.assertIsNotNone(origin_id) | ||||
| # when | # when | ||||
| self.storage.origin_visit_upsert([ | self.storage.origin_visit_upsert([ | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id_or_url, | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': 123, | 'visit': 123, | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, | }, | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id_or_url, | ||||
| 'date': '2018-01-01 23:00:00+00', | 'date': '2018-01-01 23:00:00+00', | ||||
| 'visit': 1234, | 'visit': 1234, | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, | }, | ||||
| ]) | ]) | ||||
| # then | # then | ||||
| actual_origin_visits = list(self.storage.origin_visit_get(origin_id)) | actual_origin_visits = list(self.storage.origin_visit_get( | ||||
| origin_id_or_url)) | |||||
| self.assertEqual(actual_origin_visits, [ | self.assertEqual(actual_origin_visits, [ | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': 123, | 'visit': 123, | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| Show All 29 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)]) | ||||
| def test_origin_visit_upsert_existing(self): | @given(strategies.booleans()) | ||||
| def test_origin_visit_upsert_existing(self, use_url): | |||||
| 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 | |||||
| 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_or_url, | ||||
| date=self.date_visit2) | date=self.date_visit2) | ||||
| self.storage.origin_visit_upsert([{ | self.storage.origin_visit_upsert([{ | ||||
| 'origin': origin_id, | 'origin': origin_id_or_url, | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }]) | }]) | ||||
| # then | # then | ||||
| self.assertEqual(origin_visit1['origin'], origin_id) | self.assertEqual(origin_visit1['origin'], origin_id) | ||||
| self.assertIsNotNone(origin_visit1['visit']) | self.assertIsNotNone(origin_visit1['visit']) | ||||
| actual_origin_visits = list(self.storage.origin_visit_get(origin_id)) | actual_origin_visits = list(self.storage.origin_visit_get( | ||||
| origin_id_or_url)) | |||||
| self.assertEqual(actual_origin_visits, | self.assertEqual(actual_origin_visits, | ||||
| [{ | [{ | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': self.date_visit2, | 'date': self.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': self.origin2['type'], | 'type': self.origin2['type'], | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| Show All 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): | ||||
| # No result | |||||
| 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.storage.origin_add([self.origin]) | |||||
| actual_origin_visit = self.storage.origin_visit_get_by( | |||||
| 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) | ||||
| self.storage.revision_add([self.revision]) | self.storage.revision_add([self.revision]) | ||||
| rev = list(self.storage.revision_get([self.revision['id']]))[0] | rev = list(self.storage.revision_get([self.revision['id']]))[0] | ||||
| id0 = rev['committer']['id'] | id0 = rev['committer']['id'] | ||||
| Show All 24 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) | ||||
| def test_origin_visit_get_latest(self): | @given(strategies.booleans()) | ||||
| def test_origin_visit_get_latest(self, use_url): | |||||
| 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_url = self.origin['url'] | origin_url = self.origin['url'] | ||||
| origin_visit1 = self.storage.origin_visit_add(origin_id, | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id_or_url, | |||||
| 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_or_url, | |||||
| self.date_visit2) | self.date_visit2) | ||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
| origin_visit3 = self.storage.origin_visit_add(origin_id, | origin_visit3 = self.storage.origin_visit_add( | ||||
| origin_id_or_url, | |||||
| self.date_visit2) | self.date_visit2) | ||||
| visit3_id = origin_visit3['visit'] | visit3_id = origin_visit3['visit'] | ||||
| origin_visit1 = self.storage.origin_visit_get_by(origin_url, visit1_id) | origin_visit1 = self.storage.origin_visit_get_by(origin_url, visit1_id) | ||||
| origin_visit2 = self.storage.origin_visit_get_by(origin_url, visit2_id) | origin_visit2 = self.storage.origin_visit_get_by(origin_url, visit2_id) | ||||
| origin_visit3 = self.storage.origin_visit_get_by(origin_url, visit3_id) | origin_visit3 = self.storage.origin_visit_get_by(origin_url, visit3_id) | ||||
| # Two visits, both with no snapshot | # Two visits, both with no snapshot | ||||
| self.assertEqual( | self.assertEqual( | ||||
| origin_visit3, | origin_visit3, | ||||
| self.storage.origin_visit_get_latest(origin_url)) | self.storage.origin_visit_get_latest(origin_url)) | ||||
| self.assertIsNone( | self.assertIsNone( | ||||
| self.storage.origin_visit_get_latest(origin_url, | self.storage.origin_visit_get_latest(origin_url, | ||||
| require_snapshot=True)) | require_snapshot=True)) | ||||
| # Add snapshot to visit1; require_snapshot=True makes it return | # Add snapshot to visit1; require_snapshot=True makes it return | ||||
| # visit1 and require_snapshot=False still returns visit2 | # visit1 and require_snapshot=False still returns visit2 | ||||
| self.storage.snapshot_add([self.complete_snapshot]) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit1_id, snapshot=self.complete_snapshot['id']) | origin_id_or_url, | ||||
| visit1_id, snapshot=self.complete_snapshot['id']) | |||||
| self.assertEqual( | self.assertEqual( | ||||
| {**origin_visit1, 'snapshot': self.complete_snapshot['id']}, | {**origin_visit1, 'snapshot': self.complete_snapshot['id']}, | ||||
| self.storage.origin_visit_get_latest( | self.storage.origin_visit_get_latest( | ||||
| origin_url, require_snapshot=True) | origin_url, require_snapshot=True) | ||||
| ) | ) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| origin_visit3, | origin_visit3, | ||||
| self.storage.origin_visit_get_latest(origin_url) | self.storage.origin_visit_get_latest(origin_url) | ||||
| ) | ) | ||||
| # Status filter: all three visits are status=ongoing, so no visit | # Status filter: all three visits are status=ongoing, so no visit | ||||
| # returned | # returned | ||||
| self.assertIsNone( | self.assertIsNone( | ||||
| self.storage.origin_visit_get_latest( | self.storage.origin_visit_get_latest( | ||||
| origin_url, allowed_statuses=['full']) | origin_url, allowed_statuses=['full']) | ||||
| ) | ) | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| self.storage.origin_visit_update(origin_id, visit1_id, status='full') | self.storage.origin_visit_update( | ||||
| origin_id_or_url, | |||||
| visit1_id, status='full') | |||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': self.complete_snapshot['id'], | 'snapshot': self.complete_snapshot['id'], | ||||
| 'status': 'full'}, | 'status': 'full'}, | ||||
| self.storage.origin_visit_get_latest( | self.storage.origin_visit_get_latest( | ||||
| origin_url, allowed_statuses=['full']), | origin_url, allowed_statuses=['full']), | ||||
| ) | ) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| origin_visit3, | origin_visit3, | ||||
| self.storage.origin_visit_get_latest(origin_url), | self.storage.origin_visit_get_latest(origin_url), | ||||
| ) | ) | ||||
| # Add snapshot to visit2 and check that the new snapshot is returned | # Add snapshot to visit2 and check that the new snapshot is returned | ||||
| self.storage.snapshot_add([self.empty_snapshot]) | self.storage.snapshot_add([self.empty_snapshot]) | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit2_id, snapshot=self.empty_snapshot['id']) | origin_id_or_url, | ||||
| visit2_id, snapshot=self.empty_snapshot['id']) | |||||
| self.assertEqual( | self.assertEqual( | ||||
| {**origin_visit2, 'snapshot': self.empty_snapshot['id']}, | {**origin_visit2, 'snapshot': self.empty_snapshot['id']}, | ||||
| self.storage.origin_visit_get_latest( | self.storage.origin_visit_get_latest( | ||||
| origin_url, require_snapshot=True), | origin_url, require_snapshot=True), | ||||
| ) | ) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| origin_visit3, | origin_visit3, | ||||
| self.storage.origin_visit_get_latest(origin_url), | self.storage.origin_visit_get_latest(origin_url), | ||||
| ) | ) | ||||
| # Check that the status filter is still working | # Check that the status filter is still working | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': self.complete_snapshot['id'], | 'snapshot': self.complete_snapshot['id'], | ||||
| 'status': 'full'}, | 'status': 'full'}, | ||||
| self.storage.origin_visit_get_latest( | self.storage.origin_visit_get_latest( | ||||
| origin_url, allowed_statuses=['full']), | origin_url, allowed_statuses=['full']), | ||||
| ) | ) | ||||
| # Add snapshot to visit3 (same date as visit2) | # Add snapshot to visit3 (same date as visit2) | ||||
| self.storage.snapshot_add([self.complete_snapshot]) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit3_id, snapshot=self.complete_snapshot['id']) | origin_id_or_url, | ||||
| visit3_id, snapshot=self.complete_snapshot['id']) | |||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': self.complete_snapshot['id'], | 'snapshot': self.complete_snapshot['id'], | ||||
| 'status': 'full'}, | 'status': 'full'}, | ||||
| self.storage.origin_visit_get_latest( | self.storage.origin_visit_get_latest( | ||||
| origin_url, allowed_statuses=['full']), | origin_url, allowed_statuses=['full']), | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 639 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest(self, use_url): | ||||
| # the new snapshot is returned | # the new snapshot is returned | ||||
| self.storage.snapshot_add([self.complete_snapshot]) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| 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)) | ||||
| def test_snapshot_get_latest__missing_snapshot(self): | @given(strategies.booleans()) | ||||
| def test_snapshot_get_latest__missing_snapshot(self, use_url): | |||||
| self.reset_storage() | |||||
| # Origin does not exist | |||||
| self.assertIsNone(self.storage.snapshot_get_latest( | |||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin_visit1 = self.storage.origin_visit_add( | |||||
| origin_id_or_url, | |||||
| 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_or_url, | |||||
| self.date_visit2) | self.date_visit2) | ||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| self.assertIsNone(self.storage.snapshot_get_latest(origin_id)) | self.assertIsNone(self.storage.snapshot_get_latest( | ||||
| origin_id_or_url)) | |||||
| # Add unknown snapshot to visit1, check that the inconsistency is | # Add unknown snapshot to visit1, check that the inconsistency is | ||||
| # detected | # detected | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit1_id, snapshot=self.complete_snapshot['id']) | origin_id_or_url, | ||||
| visit1_id, snapshot=self.complete_snapshot['id']) | |||||
| with self.assertRaises(ValueError): | with self.assertRaises(ValueError): | ||||
| self.storage.snapshot_get_latest(origin_id) | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url) | |||||
| # Status filter: both visits are status=ongoing, so no snapshot | # Status filter: both visits are status=ongoing, so no snapshot | ||||
| # returned | # returned | ||||
| self.assertIsNone( | self.assertIsNone( | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']) | allowed_statuses=['full']) | ||||
| ) | ) | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| self.storage.origin_visit_update(origin_id, visit1_id, status='full') | self.storage.origin_visit_update( | ||||
| origin_id_or_url, | |||||
| visit1_id, status='full') | |||||
| with self.assertRaises(ValueError): | with self.assertRaises(ValueError): | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']), | allowed_statuses=['full']), | ||||
| # Actually add the snapshot and check status filter again | # Actually add the snapshot and check status filter again | ||||
| self.storage.snapshot_add([self.complete_snapshot]) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| self.complete_snapshot, | self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id) | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url) | |||||
| ) | ) | ||||
| # Add unknown snapshot to visit2 and check that the inconsistency | # Add unknown snapshot to visit2 and check that the inconsistency | ||||
| # is detected | # is detected | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit2_id, snapshot=self.snapshot['id']) | origin_id_or_url, | ||||
| visit2_id, snapshot=self.snapshot['id']) | |||||
| with self.assertRaises(ValueError): | with self.assertRaises(ValueError): | ||||
| self.storage.snapshot_get_latest(origin_id) | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url) | |||||
| # Actually add that snapshot and check that the new one is returned | # Actually add that snapshot and check that the new one is returned | ||||
| self.storage.snapshot_add([self.snapshot]) | self.storage.snapshot_add([self.snapshot]) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| self.snapshot, | self.snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id) | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url) | |||||
| ) | ) | ||||
| def test_snapshot_get_latest__legacy_add(self): | @given(strategies.booleans()) | ||||
| def test_snapshot_get_latest__legacy_add(self, use_url): | |||||
| self.reset_storage() | |||||
| origin_id = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| origin_visit1 = self.storage.origin_visit_add(origin_id, | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin_visit1 = self.storage.origin_visit_add( | |||||
| origin_id_or_url, | |||||
| 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_or_url, | |||||
| self.date_visit2) | self.date_visit2) | ||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
| origin_visit3 = self.storage.origin_visit_add(origin_id, | origin_visit3 = self.storage.origin_visit_add( | ||||
| origin_id_or_url, | |||||
| self.date_visit2) | self.date_visit2) | ||||
| visit3_id = origin_visit3['visit'] | visit3_id = origin_visit3['visit'] | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| self.assertIsNone(self.storage.snapshot_get_latest(origin_id)) | self.assertIsNone(self.storage.snapshot_get_latest( | ||||
| origin_id_or_url)) | |||||
| # Add snapshot to visit1, latest snapshot = visit 1 snapshot | # Add snapshot to visit1, latest snapshot = visit 1 snapshot | ||||
| self.storage.snapshot_add(origin_id, visit1_id, self.complete_snapshot) | self.storage.snapshot_add( | ||||
| origin_id_or_url, | |||||
| visit1_id, self.complete_snapshot) | |||||
| self.assertEqual(self.complete_snapshot, | self.assertEqual(self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id)) | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url)) | |||||
| # Status filter: both visits are status=ongoing, so no snapshot | # Status filter: both visits are status=ongoing, so no snapshot | ||||
| # returned | # returned | ||||
| self.assertIsNone( | self.assertIsNone( | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']) | allowed_statuses=['full']) | ||||
| ) | ) | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| self.storage.origin_visit_update(origin_id, visit1_id, status='full') | self.storage.origin_visit_update( | ||||
| origin_id_or_url, | |||||
| visit1_id, status='full') | |||||
| self.assertEqual( | self.assertEqual( | ||||
| self.complete_snapshot, | self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']), | allowed_statuses=['full']), | ||||
| ) | ) | ||||
| # Add snapshot to visit2 and check that the new snapshot is returned | # Add snapshot to visit2 and check that the new snapshot is returned | ||||
| self.storage.snapshot_add(origin_id, visit2_id, self.empty_snapshot) | self.storage.snapshot_add( | ||||
| origin_id_or_url, | |||||
| visit2_id, self.empty_snapshot) | |||||
| self.assertEqual(self.empty_snapshot, | self.assertEqual(self.empty_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id)) | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url)) | |||||
| # Check that the status filter is still working | # Check that the status filter is still working | ||||
| self.assertEqual( | self.assertEqual( | ||||
| self.complete_snapshot, | self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id, | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']), | allowed_statuses=['full']), | ||||
| ) | ) | ||||
| # Add snapshot to visit3 (same date as visit2) and check that | # Add snapshot to visit3 (same date as visit2) and check that | ||||
| # the new snapshot is returned | # the new snapshot is returned | ||||
| self.storage.snapshot_add(origin_id, visit3_id, self.complete_snapshot) | self.storage.snapshot_add( | ||||
| origin_id_or_url, | |||||
| visit3_id, self.complete_snapshot) | |||||
| self.assertEqual(self.complete_snapshot, | self.assertEqual(self.complete_snapshot, | ||||
| self.storage.snapshot_get_latest(origin_id)) | self.storage.snapshot_get_latest( | ||||
| 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', 'person', 'revision'] | ||||
| # Initially, all counters are 0 | # Initially, all counters are 0 | ||||
| self.storage.refresh_stat_counters() | self.storage.refresh_stat_counters() | ||||
| Show All 12 Lines | def test_stat_counters(self): | ||||
| self.assertTrue(set(expected_keys) <= set(counters)) | self.assertTrue(set(expected_keys) <= set(counters)) | ||||
| for key in expected_keys: | for key in expected_keys: | ||||
| if key != 'content': | if key != 'content': | ||||
| self.assertEqual(counters[key], 0) | self.assertEqual(counters[key], 0) | ||||
| self.assertEqual(counters['content'], 1) | self.assertEqual(counters['content'], 1) | ||||
| # Add other objects. Check their counter increased as well. | # Add other objects. Check their counter increased as well. | ||||
| origin_id = self.storage.origin_add_one(self.origin2) | self.storage.origin_add_one(self.origin2) | ||||
| origin_visit1 = self.storage.origin_visit_add( | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id, | self.origin2['url'], date=self.date_visit2) | ||||
| date=self.date_visit2) | self.storage.snapshot_add( | ||||
| self.storage.snapshot_add(origin_id, origin_visit1['visit'], | self.origin2['url'], origin_visit1['visit'], self.snapshot) | ||||
| self.snapshot) | |||||
| self.storage.directory_add([self.dir]) | self.storage.directory_add([self.dir]) | ||||
| self.storage.revision_add([self.revision]) | self.storage.revision_add([self.revision]) | ||||
| self.storage.release_add([self.release]) | self.storage.release_add([self.release]) | ||||
| self.storage.refresh_stat_counters() | self.storage.refresh_stat_counters() | ||||
| counters = self.storage.stat_counters() | counters = self.storage.stat_counters() | ||||
| self.assertEqual(counters['content'], 1) | self.assertEqual(counters['content'], 1) | ||||
| self.assertEqual(counters['directory'], 1) | self.assertEqual(counters['directory'], 1) | ||||
| ▲ Show 20 Lines • Show All 409 Lines • ▼ Show 20 Lines | def test_metadata_provider_get_by(self): | ||||
| 'provider_url': self.provider['url'] | 'provider_url': self.provider['url'] | ||||
| }) | }) | ||||
| # then | # then | ||||
| self.assertTrue(provider_id, actual_provider['id']) | self.assertTrue(provider_id, actual_provider['id']) | ||||
| def test_origin_metadata_add(self): | def test_origin_metadata_add(self): | ||||
| # given | # given | ||||
| origin_id = self.storage.origin_add([self.origin])[0]['id'] | origin_id = self.storage.origin_add([self.origin])[0]['id'] | ||||
| origin_metadata0 = list(self.storage.origin_metadata_get_by(origin_id)) | origin_metadata0 = list(self.storage.origin_metadata_get_by( | ||||
| origin_id)) | |||||
| self.assertTrue(len(origin_metadata0) == 0) | self.assertTrue(len(origin_metadata0) == 0) | ||||
| tools = self.storage.tool_add([self.metadata_tool]) | tools = self.storage.tool_add([self.metadata_tool]) | ||||
| tool = tools[0] | tool = tools[0] | ||||
| self.storage.metadata_provider_add( | self.storage.metadata_provider_add( | ||||
| self.provider['name'], | self.provider['name'], | ||||
| self.provider['type'], | self.provider['type'], | ||||
| Show All 12 Lines | def test_origin_metadata_add(self): | ||||
| tool['id'], | tool['id'], | ||||
| self.origin_metadata['metadata']) | self.origin_metadata['metadata']) | ||||
| self.storage.origin_metadata_add( | self.storage.origin_metadata_add( | ||||
| origin_id, | origin_id, | ||||
| '2015-01-01 23:00:00+00', | '2015-01-01 23:00:00+00', | ||||
| provider['id'], | provider['id'], | ||||
| tool['id'], | tool['id'], | ||||
| self.origin_metadata2['metadata']) | self.origin_metadata2['metadata']) | ||||
| actual_om = list(self.storage.origin_metadata_get_by(origin_id)) | actual_om = list(self.storage.origin_metadata_get_by( | ||||
| origin_id)) | |||||
| # then | # then | ||||
| self.assertCountEqual( | self.assertCountEqual( | ||||
| [item['origin_id'] for item in actual_om], | [item['origin_id'] for item in actual_om], | ||||
| [origin_id, origin_id]) | [origin_id, origin_id]) | ||||
| def test_origin_metadata_get(self): | def test_origin_metadata_get(self): | ||||
| # given | # given | ||||
| origin_id = self.storage.origin_add([self.origin])[0]['id'] | origin_id = self.storage.origin_add([self.origin])[0]['id'] | ||||
| Show All 22 Lines | def test_origin_metadata_get(self): | ||||
| tool['id'], | tool['id'], | ||||
| self.origin_metadata2['metadata']) | self.origin_metadata2['metadata']) | ||||
| self.storage.origin_metadata_add( | self.storage.origin_metadata_add( | ||||
| origin_id, | origin_id, | ||||
| self.origin_metadata2['discovery_date'], | self.origin_metadata2['discovery_date'], | ||||
| provider['id'], | provider['id'], | ||||
| tool['id'], | tool['id'], | ||||
| self.origin_metadata2['metadata']) | self.origin_metadata2['metadata']) | ||||
| all_metadatas = list(self.storage.origin_metadata_get_by(origin_id)) | all_metadatas = list(self.storage.origin_metadata_get_by( | ||||
| origin_id)) | |||||
| metadatas_for_origin2 = list(self.storage.origin_metadata_get_by( | metadatas_for_origin2 = list(self.storage.origin_metadata_get_by( | ||||
| origin_id2)) | origin_id2)) | ||||
| expected_results = [{ | expected_results = [{ | ||||
| 'origin_id': origin_id, | 'origin_id': origin_id, | ||||
| 'discovery_date': datetime.datetime( | 'discovery_date': datetime.datetime( | ||||
| 2017, 1, 1, 23, 0, | 2017, 1, 1, 23, 0, | ||||
| tzinfo=datetime.timezone.utc), | tzinfo=datetime.timezone.utc), | ||||
| 'metadata': { | 'metadata': { | ||||
| 'name': 'test_origin_metadata', | 'name': 'test_origin_metadata', | ||||
| 'version': '0.0.1' | 'version': '0.0.1' | ||||
| ▲ Show 20 Lines • Show All 76 Lines • ▼ Show 20 Lines | def test_origin_metadata_get_by_provider_type(self): | ||||
| self.storage.metadata_provider_get(provider2_id)) | self.storage.metadata_provider_get(provider2_id)) | ||||
| # using the only tool now inserted in the data.sql, but for this | # using the only tool now inserted in the data.sql, but for this | ||||
| # provider should be a crawler tool (not yet implemented) | # provider should be a crawler tool (not yet implemented) | ||||
| tool = self.storage.tool_add([self.metadata_tool])[0] | tool = self.storage.tool_add([self.metadata_tool])[0] | ||||
| # when adding for the same origin 2 metadatas | # when adding for the same origin 2 metadatas | ||||
| self.storage.origin_metadata_add( | self.storage.origin_metadata_add( | ||||
| origin_id, | origin_id, | ||||
| self.origin_metadata['discovery_date'], | self.origin_metadata['discovery_date'], | ||||
| provider1['id'], | provider1['id'], | ||||
| tool['id'], | tool['id'], | ||||
| self.origin_metadata['metadata']) | self.origin_metadata['metadata']) | ||||
| self.storage.origin_metadata_add( | self.storage.origin_metadata_add( | ||||
| origin_id2, | origin_id2, | ||||
| self.origin_metadata2['discovery_date'], | self.origin_metadata2['discovery_date'], | ||||
| provider2['id'], | provider2['id'], | ||||
| tool['id'], | tool['id'], | ||||
| self.origin_metadata2['metadata']) | self.origin_metadata2['metadata']) | ||||
| provider_type = 'registry' | provider_type = 'registry' | ||||
| m_by_provider = list(self.storage. | m_by_provider = list(self.storage.origin_metadata_get_by( | ||||
| origin_metadata_get_by( | |||||
| origin_id2, | origin_id2, | ||||
| provider_type)) | provider_type)) | ||||
| for item in m_by_provider: | for item in m_by_provider: | ||||
| if 'id' in item: | if 'id' in item: | ||||
| del item['id'] | del item['id'] | ||||
| expected_results = [{ | expected_results = [{ | ||||
| 'origin_id': origin_id2, | 'origin_id': origin_id2, | ||||
| 'discovery_date': datetime.datetime( | 'discovery_date': datetime.datetime( | ||||
| 2017, 1, 1, 23, 0, | 2017, 1, 1, 23, 0, | ||||
| tzinfo=datetime.timezone.utc), | tzinfo=datetime.timezone.utc), | ||||
| ▲ Show 20 Lines • Show All 276 Lines • ▼ Show 20 Lines | class TestLocalStorage(CommonTestStorage, StorageTestDbFixture, | ||||
| """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): | ||||
| self.reset_storage() | self.reset_storage() | ||||
| origin = self.storage.origin_add_one(self.origin) | origin_id = self.storage.origin_add_one(self.origin) | ||||
| if use_url: | origin_id_or_url = self.origin['url'] if use_url else origin_id | ||||
| origin_id = self.origin['url'] | |||||
| else: | |||||
| origin_id = origin | |||||
| 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(origin_id) | fetch_history_id = self.storage.fetch_history_start( | ||||
| origin_id_or_url) | |||||
| datetime.datetime.now.assert_called_with(tz=datetime.timezone.utc) | datetime.datetime.now.assert_called_with(tz=datetime.timezone.utc) | ||||
| with patch('datetime.datetime'): | with patch('datetime.datetime'): | ||||
| datetime.datetime.now.return_value = self.fetch_history_end | datetime.datetime.now.return_value = self.fetch_history_end | ||||
| self.storage.fetch_history_end(fetch_history_id, | self.storage.fetch_history_end(fetch_history_id, | ||||
| self.fetch_history_data) | self.fetch_history_data) | ||||
| fetch_history = self.storage.fetch_history_get(fetch_history_id) | fetch_history = self.storage.fetch_history_get(fetch_history_id) | ||||
| expected_fetch_history = self.fetch_history_data.copy() | expected_fetch_history = self.fetch_history_data.copy() | ||||
| expected_fetch_history['id'] = fetch_history_id | expected_fetch_history['id'] = fetch_history_id | ||||
| expected_fetch_history['origin'] = origin | expected_fetch_history['origin'] = origin_id | ||||
| expected_fetch_history['date'] = self.fetch_history_date | expected_fetch_history['date'] = self.fetch_history_date | ||||
| expected_fetch_history['duration'] = self.fetch_history_duration | expected_fetch_history['duration'] = self.fetch_history_duration | ||||
| self.assertEqual(expected_fetch_history, fetch_history) | self.assertEqual(expected_fetch_history, fetch_history) | ||||
| # This test is only relevant on the local storage, with an actual | # This test is only relevant on the local storage, with an actual | ||||
| # objstorage raising an exception | # objstorage raising an exception | ||||
| def test_content_add_objstorage_exception(self): | def test_content_add_objstorage_exception(self): | ||||
| ▲ Show 20 Lines • Show All 82 Lines • Show Last 20 Lines | |||||
Same comment as in D1664, using a variable to store the origin identifier or url would improve code readability.