Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,905 Lines • ▼ Show 20 Lines | def test_origin_visit_get_by(self, use_url): | ||||
| 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 | ||||
| origin_visit1 = self.storage.origin_visit_add( | origin_visit1 = self.storage.origin_visit_add( | ||||
| origin_id_or_url, | 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([self.snapshot]) | ||||
| self.snapshot) | self.storage.origin_visit_update( | ||||
| origin_id_or_url, | |||||
| origin_visit1['visit'], | |||||
| snapshot=self.snapshot['id']) | |||||
| # Add some other {origin, visit} entries | # Add some other {origin, visit} entries | ||||
| self.storage.origin_visit_add( | self.storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| date=self.date_visit3) | date=self.date_visit3) | ||||
| self.storage.origin_visit_add( | self.storage.origin_visit_add( | ||||
| origin2_id_or_url, | origin2_id_or_url, | ||||
| date=self.date_visit3) | date=self.date_visit3) | ||||
| ▲ Show 20 Lines • Show All 260 Lines • ▼ Show 20 Lines | def test_origin_visit_get_latest(self, use_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_or_url, | origin_id_or_url, visit1_id, | ||||
| visit1_id, snapshot=self.complete_snapshot['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) | ||||
| Show All 21 Lines | def test_origin_visit_get_latest(self, use_url): | ||||
| 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_or_url, | origin_id_or_url, visit2_id, | ||||
| visit2_id, snapshot=self.empty_snapshot['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_or_url, | origin_id_or_url, visit3_id, snapshot=self.complete_snapshot['id']) | ||||
| 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 85 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_empty(self): | ||||
| 'snapshot': self.empty_snapshot['id'], | 'snapshot': self.empty_snapshot['id'], | ||||
| } | } | ||||
| 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.empty_snapshot), | ('snapshot', self.empty_snapshot), | ||||
| ('origin_visit', data2)]) | ('origin_visit', data2)]) | ||||
| def test_snapshot_add_get_empty__legacy_add(self): | |||||
| origin_id = self.storage.origin_add_one(self.origin) | |||||
| origin_visit1 = self.storage.origin_visit_add(origin_id, | |||||
| self.date_visit1) | |||||
| visit_id = origin_visit1['visit'] | |||||
| self.storage.snapshot_add(origin_id, visit_id, self.empty_snapshot) | |||||
| by_id = self.storage.snapshot_get(self.empty_snapshot['id']) | |||||
| self.assertEqual(by_id, self.empty_snapshot) | |||||
| by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id) | |||||
| self.assertEqual(by_ov, self.empty_snapshot) | |||||
| expected_origin = self.origin.copy() | |||||
| data1 = { | |||||
| 'origin': expected_origin, | |||||
| 'date': self.date_visit1, | |||||
| 'visit': origin_visit1['visit'], | |||||
| 'type': self.origin['type'], | |||||
| 'status': 'ongoing', | |||||
| 'metadata': None, | |||||
| 'snapshot': None, | |||||
| } | |||||
| data2 = { | |||||
| 'origin': expected_origin, | |||||
| 'date': self.date_visit1, | |||||
| 'visit': origin_visit1['visit'], | |||||
| 'type': self.origin['type'], | |||||
| 'status': 'ongoing', | |||||
| 'metadata': None, | |||||
| 'snapshot': self.empty_snapshot['id'], | |||||
| } | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('origin', expected_origin), | |||||
| ('origin_visit', data1), | |||||
| ('snapshot', self.empty_snapshot), | |||||
| ('origin_visit', data2)]) | |||||
| def test_snapshot_add_get_complete(self): | def test_snapshot_add_get_complete(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| actual_result = self.storage.snapshot_add( | actual_result = self.storage.snapshot_add([self.complete_snapshot]) | ||||
| origin_id, visit_id, self.complete_snapshot) | self.storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=self.complete_snapshot['id']) | |||||
| self.assertEqual(actual_result, {'snapshot:add': 1}) | self.assertEqual(actual_result, {'snapshot:add': 1}) | ||||
| by_id = self.storage.snapshot_get(self.complete_snapshot['id']) | by_id = self.storage.snapshot_get(self.complete_snapshot['id']) | ||||
| self.assertEqual(by_id, self.complete_snapshot) | self.assertEqual(by_id, self.complete_snapshot) | ||||
| by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id) | by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id) | ||||
| self.assertEqual(by_ov, self.complete_snapshot) | self.assertEqual(by_ov, self.complete_snapshot) | ||||
| Show All 27 Lines | def test_snapshot_add_many_incremental(self): | ||||
| self.storage.snapshot_get(self.snapshot['id'])) | self.storage.snapshot_get(self.snapshot['id'])) | ||||
| def test_snapshot_add_count_branches(self): | def test_snapshot_add_count_branches(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| actual_result = self.storage.snapshot_add( | actual_result = self.storage.snapshot_add([self.complete_snapshot]) | ||||
| origin_id, visit_id, self.complete_snapshot) | self.storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=self.complete_snapshot['id']) | |||||
| self.assertEqual(actual_result, {'snapshot:add': 1}) | self.assertEqual(actual_result, {'snapshot:add': 1}) | ||||
| snp_id = self.complete_snapshot['id'] | snp_id = self.complete_snapshot['id'] | ||||
| snp_size = self.storage.snapshot_count_branches(snp_id) | snp_size = self.storage.snapshot_count_branches(snp_id) | ||||
| expected_snp_size = { | expected_snp_size = { | ||||
| 'alias': 1, | 'alias': 1, | ||||
| 'content': 1, | 'content': 1, | ||||
| 'directory': 2, | 'directory': 2, | ||||
| 'release': 1, | 'release': 1, | ||||
| 'revision': 1, | 'revision': 1, | ||||
| 'snapshot': 1, | 'snapshot': 1, | ||||
| None: 1 | None: 1 | ||||
| } | } | ||||
| self.assertEqual(snp_size, expected_snp_size) | self.assertEqual(snp_size, expected_snp_size) | ||||
| def test_snapshot_add_get_paginated(self): | def test_snapshot_add_get_paginated(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| self.storage.snapshot_add(origin_id, visit_id, self.complete_snapshot) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.storage.origin_visit_update( | |||||
| origin_id, visit_id, | |||||
| snapshot=self.complete_snapshot['id']) | |||||
| snp_id = self.complete_snapshot['id'] | snp_id = self.complete_snapshot['id'] | ||||
| branches = self.complete_snapshot['branches'] | branches = self.complete_snapshot['branches'] | ||||
| branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
| # Test branch_from | # Test branch_from | ||||
| snapshot = self.storage.snapshot_get_branches(snp_id, | snapshot = self.storage.snapshot_get_branches(snp_id, | ||||
| ▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_paginated(self): | ||||
| self.assertEqual(snapshot, expected_snapshot) | self.assertEqual(snapshot, expected_snapshot) | ||||
| def test_snapshot_add_get_filtered(self): | def test_snapshot_add_get_filtered(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| self.storage.snapshot_add(origin_id, visit_id, self.complete_snapshot) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.storage.origin_visit_update( | |||||
| origin_id, visit_id, snapshot=self.complete_snapshot['id']) | |||||
| snp_id = self.complete_snapshot['id'] | snp_id = self.complete_snapshot['id'] | ||||
| branches = self.complete_snapshot['branches'] | branches = self.complete_snapshot['branches'] | ||||
| snapshot = self.storage.snapshot_get_branches( | snapshot = self.storage.snapshot_get_branches( | ||||
| snp_id, target_types=['release', 'revision']) | snp_id, target_types=['release', 'revision']) | ||||
| expected_snapshot = { | expected_snapshot = { | ||||
| Show All 24 Lines | def test_snapshot_add_get_filtered(self): | ||||
| self.assertEqual(snapshot, expected_snapshot) | self.assertEqual(snapshot, expected_snapshot) | ||||
| def test_snapshot_add_get_filtered_and_paginated(self): | def test_snapshot_add_get_filtered_and_paginated(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| self.storage.snapshot_add(origin_id, visit_id, self.complete_snapshot) | self.storage.snapshot_add([self.complete_snapshot]) | ||||
| self.storage.origin_visit_update( | |||||
| origin_id, visit_id, snapshot=self.complete_snapshot['id']) | |||||
| snp_id = self.complete_snapshot['id'] | snp_id = self.complete_snapshot['id'] | ||||
| branches = self.complete_snapshot['branches'] | branches = self.complete_snapshot['branches'] | ||||
| branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
| # Test branch_from | # Test branch_from | ||||
| snapshot = self.storage.snapshot_get_branches( | snapshot = self.storage.snapshot_get_branches( | ||||
| ▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_filtered_and_paginated(self): | ||||
| self.assertEqual(snapshot, expected_snapshot) | self.assertEqual(snapshot, expected_snapshot) | ||||
| def test_snapshot_add_get(self): | def test_snapshot_add_get(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| self.storage.snapshot_add(origin_id, visit_id, self.snapshot) | self.storage.snapshot_add([self.snapshot]) | ||||
| self.storage.origin_visit_update( | |||||
| origin_id, visit_id, snapshot=self.snapshot['id']) | |||||
| by_id = self.storage.snapshot_get(self.snapshot['id']) | by_id = self.storage.snapshot_get(self.snapshot['id']) | ||||
| self.assertEqual(by_id, self.snapshot) | self.assertEqual(by_id, self.snapshot) | ||||
| by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id) | by_ov = self.storage.snapshot_get_by_origin_visit(origin_id, visit_id) | ||||
| self.assertEqual(by_ov, self.snapshot) | self.assertEqual(by_ov, self.snapshot) | ||||
| origin_visit_info = self.storage.origin_visit_get_by(origin_id, | origin_visit_info = self.storage.origin_visit_get_by(origin_id, | ||||
| Show All 10 Lines | def test_snapshot_add_nonexistent_visit(self): | ||||
| with self.assertRaises(ValueError): | with self.assertRaises(ValueError): | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=self.snapshot['id']) | origin_id, visit_id, snapshot=self.snapshot['id']) | ||||
| self.assertEqual(list(self.journal_writer.objects), [ | self.assertEqual(list(self.journal_writer.objects), [ | ||||
| ('snapshot', self.snapshot)]) | ('snapshot', self.snapshot)]) | ||||
| def test_snapshot_add_nonexistent_visit__legacy_add(self): | |||||
| origin_id = self.storage.origin_add_one(self.origin) | |||||
| visit_id = 54164461156 | |||||
| self.journal_writer.objects[:] = [] | |||||
| with self.assertRaises(ValueError): | |||||
| self.storage.snapshot_add(origin_id, visit_id, self.snapshot) | |||||
| # Note: the actual legacy behavior was to abort before adding | |||||
| # the snapshot; but delaying non-existence checks makes the | |||||
| # compatibility code simpler | |||||
| self.assertEqual(list(self.journal_writer.objects), [ | |||||
| ('snapshot', self.snapshot)]) | |||||
| def test_snapshot_add_twice(self): | def test_snapshot_add_twice(self): | ||||
| 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_visit1 = self.storage.origin_visit_add(origin_id, | ||||
| self.date_visit1) | self.date_visit1) | ||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| self.storage.snapshot_add([self.snapshot]) | self.storage.snapshot_add([self.snapshot]) | ||||
| self.storage.origin_visit_update( | self.storage.origin_visit_update( | ||||
| origin_id, visit1_id, snapshot=self.snapshot['id']) | origin_id, visit1_id, snapshot=self.snapshot['id']) | ||||
| ▲ Show 20 Lines • Show All 54 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)]) | ||||
| def test_snapshot_add_twice__legacy_add(self): | |||||
| origin_id = self.storage.origin_add_one(self.origin) | |||||
| origin_visit1 = self.storage.origin_visit_add(origin_id, | |||||
| self.date_visit1) | |||||
| visit1_id = origin_visit1['visit'] | |||||
| self.storage.snapshot_add(origin_id, visit1_id, self.snapshot) | |||||
| by_ov1 = self.storage.snapshot_get_by_origin_visit(origin_id, | |||||
| visit1_id) | |||||
| self.assertEqual(by_ov1, self.snapshot) | |||||
| origin_visit2 = self.storage.origin_visit_add(origin_id, | |||||
| self.date_visit2) | |||||
| visit2_id = origin_visit2['visit'] | |||||
| self.storage.snapshot_add(origin_id, visit2_id, self.snapshot) | |||||
| by_ov2 = self.storage.snapshot_get_by_origin_visit(origin_id, | |||||
| visit2_id) | |||||
| self.assertEqual(by_ov2, self.snapshot) | |||||
| expected_origin = self.origin.copy() | |||||
| data1 = { | |||||
| 'origin': expected_origin, | |||||
| 'date': self.date_visit1, | |||||
| 'visit': origin_visit1['visit'], | |||||
| 'type': self.origin['type'], | |||||
| 'status': 'ongoing', | |||||
| 'metadata': None, | |||||
| 'snapshot': None, | |||||
| } | |||||
| data2 = { | |||||
| 'origin': expected_origin, | |||||
| 'date': self.date_visit1, | |||||
| 'visit': origin_visit1['visit'], | |||||
| 'type': self.origin['type'], | |||||
| 'status': 'ongoing', | |||||
| 'metadata': None, | |||||
| 'snapshot': self.snapshot['id'], | |||||
| } | |||||
| data3 = { | |||||
| 'origin': expected_origin, | |||||
| 'date': self.date_visit2, | |||||
| 'visit': origin_visit2['visit'], | |||||
| 'type': self.origin['type'], | |||||
| 'status': 'ongoing', | |||||
| 'metadata': None, | |||||
| 'snapshot': None, | |||||
| } | |||||
| data4 = { | |||||
| 'origin': expected_origin, | |||||
| 'date': self.date_visit2, | |||||
| 'visit': origin_visit2['visit'], | |||||
| 'type': self.origin['type'], | |||||
| 'status': 'ongoing', | |||||
| 'metadata': None, | |||||
| 'snapshot': self.snapshot['id'], | |||||
| } | |||||
| self.assertEqual(list(self.journal_writer.objects), | |||||
| [('origin', expected_origin), | |||||
| ('origin_visit', data1), | |||||
| ('snapshot', self.snapshot), | |||||
| ('origin_visit', data2), | |||||
| ('origin_visit', data3), | |||||
| ('origin_visit', data4)]) | |||||
| def test_snapshot_get_nonexistent(self): | |||||
| bogus_snapshot_id = b'bogus snapshot id 00' | |||||
| bogus_origin_id = 1 | |||||
| bogus_visit_id = 1 | |||||
| by_id = self.storage.snapshot_get(bogus_snapshot_id) | |||||
| self.assertIsNone(by_id) | |||||
| by_ov = self.storage.snapshot_get_by_origin_visit(bogus_origin_id, | |||||
| bogus_visit_id) | |||||
| self.assertIsNone(by_ov) | |||||
| @given(strategies.booleans()) | @given(strategies.booleans()) | ||||
| def test_snapshot_get_latest(self, use_url): | def test_snapshot_get_latest(self, use_url): | ||||
| 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) | ||||
| ▲ Show 20 Lines • Show All 129 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest__missing_snapshot(self, use_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( | self.storage.snapshot_get_latest( | ||||
| origin_id_or_url) | origin_id_or_url) | ||||
| ) | ) | ||||
| @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_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) | |||||
| visit1_id = origin_visit1['visit'] | |||||
| origin_visit2 = self.storage.origin_visit_add( | |||||
| origin_id_or_url, | |||||
| self.date_visit2) | |||||
| visit2_id = origin_visit2['visit'] | |||||
| # Add a visit with the same date as the previous one | |||||
| origin_visit3 = self.storage.origin_visit_add( | |||||
| origin_id_or_url, | |||||
| self.date_visit2) | |||||
| visit3_id = origin_visit3['visit'] | |||||
| # Two visits, both with no snapshot: latest snapshot is None | |||||
| self.assertIsNone(self.storage.snapshot_get_latest( | |||||
| origin_id_or_url)) | |||||
| # Add snapshot to visit1, latest snapshot = visit 1 snapshot | |||||
| self.storage.snapshot_add( | |||||
| origin_id_or_url, | |||||
| visit1_id, self.complete_snapshot) | |||||
| self.assertEqual(self.complete_snapshot, | |||||
| self.storage.snapshot_get_latest( | |||||
| origin_id_or_url)) | |||||
| # Status filter: both visits are status=ongoing, so no snapshot | |||||
| # returned | |||||
| self.assertIsNone( | |||||
| self.storage.snapshot_get_latest( | |||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']) | |||||
| ) | |||||
| # Mark the first visit as completed and check status filter again | |||||
| self.storage.origin_visit_update( | |||||
| origin_id_or_url, | |||||
| visit1_id, status='full') | |||||
| self.assertEqual( | |||||
| self.complete_snapshot, | |||||
| self.storage.snapshot_get_latest( | |||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']), | |||||
| ) | |||||
| # Add snapshot to visit2 and check that the new snapshot is returned | |||||
| self.storage.snapshot_add( | |||||
| origin_id_or_url, | |||||
| visit2_id, self.empty_snapshot) | |||||
| self.assertEqual(self.empty_snapshot, | |||||
| self.storage.snapshot_get_latest( | |||||
| origin_id_or_url)) | |||||
| # Check that the status filter is still working | |||||
| self.assertEqual( | |||||
| self.complete_snapshot, | |||||
| self.storage.snapshot_get_latest( | |||||
| origin_id_or_url, | |||||
| allowed_statuses=['full']), | |||||
| ) | |||||
| # Add snapshot to visit3 (same date as visit2) and check that | |||||
| # the new snapshot is returned | |||||
| self.storage.snapshot_add( | |||||
| origin_id_or_url, | |||||
| visit3_id, self.complete_snapshot) | |||||
| self.assertEqual(self.complete_snapshot, | |||||
| 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() | ||||
| counters = self.storage.stat_counters() | counters = self.storage.stat_counters() | ||||
| Show All 14 Lines | def test_stat_counters(self): | ||||
| 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. | ||||
| 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( | ||||
| self.origin2['url'], date=self.date_visit2) | self.origin2['url'], date=self.date_visit2) | ||||
| self.storage.snapshot_add( | self.storage.snapshot_add([self.snapshot]) | ||||
| self.origin2['url'], origin_visit1['visit'], self.snapshot) | self.storage.origin_visit_update( | ||||
| self.origin2['url'], origin_visit1['visit'], | |||||
| snapshot=self.snapshot['id']) | |||||
| 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 998 Lines • Show Last 20 Lines | |||||