Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,389 Lines • ▼ Show 20 Lines | def test_origin_visit_get_random(self, swh_storage): | ||||
| visit_type = 'git' | visit_type = 'git' | ||||
| # Add visits to those origins | # Add visits to those origins | ||||
| for origin in data.origins: | for origin in data.origins: | ||||
| origin = Origin.from_dict(origin) | origin = Origin.from_dict(origin) | ||||
| for date_visit in visits: | for date_visit in visits: | ||||
| visit = swh_storage.origin_visit_add( | visit = swh_storage.origin_visit_add( | ||||
| origin, date=date_visit, type=visit_type) | origin, date=date_visit, type=visit_type) | ||||
| swh_storage.origin_visit_update(origin, visit, status='full') | swh_storage.origin_visit_update(visit, status='full') | ||||
| swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
| stats = swh_storage.stat_counters() | stats = swh_storage.stat_counters() | ||||
| assert stats['origin'] == len(data.origins) | assert stats['origin'] == len(data.origins) | ||||
| assert stats['origin_visit'] == len(data.origins) * len(visits) | assert stats['origin_visit'] == len(data.origins) * len(visits) | ||||
| random_origin_visit = swh_storage.origin_visit_get_random(visit_type) | random_origin_visit = swh_storage.origin_visit_get_random(visit_type) | ||||
| assert random_origin_visit | assert random_origin_visit | ||||
| assert random_origin_visit['origin'] is not None | assert random_origin_visit['origin'] is not None | ||||
| original_urls = [o['url'] for o in data.origins] | original_urls = [o['url'] for o in data.origins] | ||||
| assert random_origin_visit['origin'] in original_urls | assert random_origin_visit['origin'] in original_urls | ||||
| def test_origin_visit_get_random_nothing_found(self, swh_storage): | def test_origin_visit_get_random_nothing_found(self, swh_storage): | ||||
| swh_storage.origin_add(data.origins) | swh_storage.origin_add(data.origins) | ||||
| visit_type = 'hg' | visit_type = 'hg' | ||||
| # Add some visits outside of the random generation selection so nothing | # Add some visits outside of the random generation selection so nothing | ||||
| # will be found by the random selection | # will be found by the random selection | ||||
| visits = self._generate_random_visits(nb_visits=3, start=13, end=24) | visits = self._generate_random_visits(nb_visits=3, start=13, end=24) | ||||
| for origin in data.origins: | for origin in data.origins: | ||||
| origin = Origin.from_dict(origin) | origin = Origin.from_dict(origin) | ||||
| for date_visit in visits: | for date_visit in visits: | ||||
| visit = swh_storage.origin_visit_add( | visit = swh_storage.origin_visit_add( | ||||
| origin, date=date_visit, type=visit_type) | origin, date=date_visit, type=visit_type) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update(visit, status='full') | ||||
| origin, visit=visit, status='full') | |||||
| random_origin_visit = swh_storage.origin_visit_get_random(visit_type) | random_origin_visit = swh_storage.origin_visit_get_random(visit_type) | ||||
| assert random_origin_visit is None | assert random_origin_visit is None | ||||
| def test_origin_get_by_sha1(self, swh_storage): | def test_origin_get_by_sha1(self, swh_storage): | ||||
| assert swh_storage.origin_get(data.origin) is None | assert swh_storage.origin_get(data.origin) is None | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| ▲ Show 20 Lines • Show All 269 Lines • ▼ Show 20 Lines | def test_origin_visit_update(self, swh_storage): | ||||
| ) | ) | ||||
| # when | # when | ||||
| visit1_metadata = { | visit1_metadata = { | ||||
| 'contents': 42, | 'contents': 42, | ||||
| 'directories': 22, | 'directories': 22, | ||||
| } | } | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, status='full', | origin_visit1, status='full', metadata=visit1_metadata) | ||||
| metadata=visit1_metadata) | swh_storage.origin_visit_update(origin_visit3, status='partial') | ||||
| swh_storage.origin_visit_update( | |||||
| origin2, origin_visit3, status='partial') | |||||
| # then | # then | ||||
| actual_origin_visits = list(swh_storage.origin_visit_get( | actual_origin_visits = list(swh_storage.origin_visit_get( | ||||
| origin.url)) | origin.url)) | ||||
| expected_visits = [{ | expected_visits = [{ | ||||
| 'origin': origin.url, | 'origin': origin.url, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1.visit, | 'visit': origin_visit1.visit, | ||||
| ▲ Show 20 Lines • Show All 111 Lines • ▼ Show 20 Lines | def test_origin_visit_update_validation(self, swh_storage): | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin_visit = swh_storage.origin_visit_add( | origin_visit = swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit2, | date=data.date_visit2, | ||||
| type=data.type_visit2, | type=data.type_visit2, | ||||
| ) | ) | ||||
| with pytest.raises(StorageArgumentException, match='status') as cm: | with pytest.raises(StorageArgumentException, match='status') as cm: | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update(origin_visit, status='foobar') | ||||
| origin, origin_visit, status='foobar') | |||||
| if type(cm.value) == psycopg2.DataError: | if type(cm.value) == psycopg2.DataError: | ||||
| assert cm.value.pgcode == \ | assert cm.value.pgcode == \ | ||||
| psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION | psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION | ||||
| def test_origin_visit_find_by_date(self, swh_storage): | def test_origin_visit_find_by_date(self, swh_storage): | ||||
| # given | # given | ||||
| origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
| Show All 38 Lines | def test_origin_visit_update_missing_snapshot(self, swh_storage): | ||||
| origin_visit = swh_storage.origin_visit_add( | origin_visit = swh_storage.origin_visit_add( | ||||
| Origin(url=origin_url), | Origin(url=origin_url), | ||||
| date=data.date_visit1, | date=data.date_visit1, | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| # when | # when | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, | origin_visit, snapshot=data.snapshot['id']) | ||||
| origin_visit, | |||||
| snapshot=data.snapshot['id']) | |||||
| # then | # then | ||||
| actual_origin_visit = swh_storage.origin_visit_get_by( | actual_origin_visit = swh_storage.origin_visit_get_by( | ||||
| origin.url, origin_visit.visit) | origin.url, origin_visit.visit) | ||||
| assert actual_origin_visit['snapshot'] == data.snapshot['id'] | assert actual_origin_visit['snapshot'] == data.snapshot['id'] | ||||
| # when | # when | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| Show All 9 Lines | def test_origin_visit_get_by(self, swh_storage): | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit2, | date=data.date_visit2, | ||||
| type=data.type_visit2, | type=data.type_visit2, | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, | |||||
| origin_visit1, | origin_visit1, | ||||
| snapshot=data.snapshot['id']) | snapshot=data.snapshot['id']) | ||||
| # Add some other {origin, visit} entries | # Add some other {origin, visit} entries | ||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit3, | date=data.date_visit3, | ||||
| type=data.type_visit3, | type=data.type_visit3, | ||||
| ) | ) | ||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| origin2, | origin2, | ||||
| date=data.date_visit3, | date=data.date_visit3, | ||||
| type=data.type_visit3, | type=data.type_visit3, | ||||
| ) | ) | ||||
| # when | # when | ||||
| visit1_metadata = { | visit1_metadata = { | ||||
| 'contents': 42, | 'contents': 42, | ||||
| 'directories': 22, | 'directories': 22, | ||||
| } | } | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, | origin_visit1, status='full', metadata=visit1_metadata) | ||||
| status='full', metadata=visit1_metadata) | |||||
| expected_origin_visit = origin_visit1.to_dict() | expected_origin_visit = origin_visit1.to_dict() | ||||
| expected_origin_visit.update({ | expected_origin_visit.update({ | ||||
| 'origin': origin.url, | 'origin': origin.url, | ||||
| 'visit': origin_visit1.visit, | 'visit': origin_visit1.visit, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| ▲ Show 20 Lines • Show All 187 Lines • ▼ Show 20 Lines | def test_origin_visit_get_latest(self, swh_storage): | ||||
| assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
| assert swh_storage.origin_visit_get_latest( | assert swh_storage.origin_visit_get_latest( | ||||
| origin.url, require_snapshot=True) is None | origin.url, require_snapshot=True) is None | ||||
| # 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 | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, ov1, snapshot=data.complete_snapshot['id']) | ov1, snapshot=data.complete_snapshot['id']) | ||||
| assert {**origin_visit1, | assert {**origin_visit1, | ||||
| 'snapshot': data.complete_snapshot['id']} \ | 'snapshot': data.complete_snapshot['id']} \ | ||||
| == swh_storage.origin_visit_get_latest( | == swh_storage.origin_visit_get_latest( | ||||
| origin.url, require_snapshot=True) | origin.url, require_snapshot=True) | ||||
| assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | assert origin_visit3 == swh_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 | ||||
| assert swh_storage.origin_visit_get_latest( | assert swh_storage.origin_visit_get_latest( | ||||
| origin.url, allowed_statuses=['full']) is None | origin.url, allowed_statuses=['full']) is None | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update(ov1, status='full') | ||||
| origin, ov1, status='full') | |||||
| assert { | assert { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': data.complete_snapshot['id'], | 'snapshot': data.complete_snapshot['id'], | ||||
| 'status': 'full'} == swh_storage.origin_visit_get_latest( | 'status': 'full'} == swh_storage.origin_visit_get_latest( | ||||
| origin.url, allowed_statuses=['full']) | origin.url, allowed_statuses=['full']) | ||||
| assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | assert origin_visit3 == swh_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 | ||||
| swh_storage.snapshot_add([data.empty_snapshot]) | swh_storage.snapshot_add([data.empty_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, ov2, snapshot=data.empty_snapshot['id']) | ov2, snapshot=data.empty_snapshot['id']) | ||||
| assert {**origin_visit2, 'snapshot': data.empty_snapshot['id']} == \ | assert {**origin_visit2, 'snapshot': data.empty_snapshot['id']} == \ | ||||
| swh_storage.origin_visit_get_latest( | swh_storage.origin_visit_get_latest( | ||||
| origin.url, require_snapshot=True) | origin.url, require_snapshot=True) | ||||
| assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
| # Check that the status filter is still working | # Check that the status filter is still working | ||||
| assert { | assert { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': data.complete_snapshot['id'], | 'snapshot': data.complete_snapshot['id'], | ||||
| 'status': 'full'} == swh_storage.origin_visit_get_latest( | 'status': 'full'} == swh_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) | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, ov3, snapshot=data.complete_snapshot['id']) | ov3, snapshot=data.complete_snapshot['id']) | ||||
| assert { | assert { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': data.complete_snapshot['id'], | 'snapshot': data.complete_snapshot['id'], | ||||
| 'status': 'full'} == swh_storage.origin_visit_get_latest( | 'status': 'full'} == swh_storage.origin_visit_get_latest( | ||||
| origin.url, allowed_statuses=['full']) | origin.url, allowed_statuses=['full']) | ||||
| assert { | assert { | ||||
| **origin_visit1, | **origin_visit1, | ||||
| 'snapshot': data.complete_snapshot['id'], | 'snapshot': data.complete_snapshot['id'], | ||||
| ▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_empty(self, swh_storage): | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| visit_id = origin_visit1.visit | visit_id = origin_visit1.visit | ||||
| actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | ||||
| assert actual_result == {'snapshot:add': 1} | assert actual_result == {'snapshot:add': 1} | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.empty_snapshot['id']) | origin_visit1, snapshot=data.empty_snapshot['id']) | ||||
| by_id = swh_storage.snapshot_get(data.empty_snapshot['id']) | by_id = swh_storage.snapshot_get(data.empty_snapshot['id']) | ||||
| assert by_id == {**data.empty_snapshot, 'next_branch': None} | assert by_id == {**data.empty_snapshot, 'next_branch': None} | ||||
| by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | ||||
| assert by_ov == {**data.empty_snapshot, 'next_branch': None} | assert by_ov == {**data.empty_snapshot, 'next_branch': None} | ||||
| data1 = { | data1 = { | ||||
| Show All 28 Lines | def test_snapshot_add_get_complete(self, swh_storage): | ||||
| origin, | origin, | ||||
| date=data.date_visit1, | date=data.date_visit1, | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| visit_id = origin_visit1.visit | visit_id = origin_visit1.visit | ||||
| actual_result = swh_storage.snapshot_add([data.complete_snapshot]) | actual_result = swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.complete_snapshot['id']) | origin_visit1, snapshot=data.complete_snapshot['id']) | ||||
| assert actual_result == {'snapshot:add': 1} | assert actual_result == {'snapshot:add': 1} | ||||
| by_id = swh_storage.snapshot_get(data.complete_snapshot['id']) | by_id = swh_storage.snapshot_get(data.complete_snapshot['id']) | ||||
| assert by_id == {**data.complete_snapshot, 'next_branch': None} | assert by_id == {**data.complete_snapshot, 'next_branch': None} | ||||
| by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | ||||
| assert by_ov == {**data.complete_snapshot, 'next_branch': None} | assert by_ov == {**data.complete_snapshot, 'next_branch': None} | ||||
| ▲ Show 20 Lines • Show All 141 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_filtered(self, swh_storage): | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit1, | date=data.date_visit1, | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.complete_snapshot['id']) | origin_visit1, snapshot=data.complete_snapshot['id']) | ||||
| snp_id = data.complete_snapshot['id'] | snp_id = data.complete_snapshot['id'] | ||||
| branches = data.complete_snapshot['branches'] | branches = data.complete_snapshot['branches'] | ||||
| snapshot = swh_storage.snapshot_get_branches( | snapshot = swh_storage.snapshot_get_branches( | ||||
| snp_id, target_types=['release', 'revision']) | snp_id, target_types=['release', 'revision']) | ||||
| expected_snapshot = { | expected_snapshot = { | ||||
| ▲ Show 20 Lines • Show All 103 Lines • ▼ Show 20 Lines | def test_snapshot_add_get(self, swh_storage): | ||||
| origin, | origin, | ||||
| date=data.date_visit1, | date=data.date_visit1, | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| visit_id = origin_visit1.visit | visit_id = origin_visit1.visit | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.snapshot['id']) | origin_visit1, snapshot=data.snapshot['id']) | ||||
| by_id = swh_storage.snapshot_get(data.snapshot['id']) | by_id = swh_storage.snapshot_get(data.snapshot['id']) | ||||
| assert by_id == {**data.snapshot, 'next_branch': None} | assert by_id == {**data.snapshot, 'next_branch': None} | ||||
| by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | ||||
| assert by_ov == {**data.snapshot, 'next_branch': None} | assert by_ov == {**data.snapshot, 'next_branch': None} | ||||
| origin_visit_info = swh_storage.origin_visit_get_by( | origin_visit_info = swh_storage.origin_visit_get_by( | ||||
| origin_url, visit_id) | origin_url, visit_id) | ||||
| assert origin_visit_info['snapshot'] == data.snapshot['id'] | assert origin_visit_info['snapshot'] == data.snapshot['id'] | ||||
| def test_snapshot_add_nonexistent_visit(self, swh_storage): | def test_snapshot_add_nonexistent_visit(self, swh_storage): | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin = Origin(url=origin_url) | |||||
| # unknown visit | # unknown visit | ||||
| visit = OriginVisit( | visit = OriginVisit( | ||||
| origin=origin, | origin=origin_url, | ||||
| visit=54164461156, | visit=54164461156, | ||||
| date=data.date_visit1, | date=data.date_visit1, | ||||
| status='ongoing', | status='ongoing', | ||||
| type='git', | type='git', | ||||
| snapshot=None | snapshot=None | ||||
| ) | ) | ||||
| swh_storage.journal_writer.journal.objects[:] = [] | swh_storage.journal_writer.journal.objects[:] = [] | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| with pytest.raises(StorageArgumentException): | with pytest.raises(StorageArgumentException): | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, visit, snapshot=data.snapshot['id']) | visit, snapshot=data.snapshot['id']) | ||||
| assert list(swh_storage.journal_writer.journal.objects) == [ | assert list(swh_storage.journal_writer.journal.objects) == [ | ||||
| ('snapshot', data.snapshot)] | ('snapshot', data.snapshot)] | ||||
| def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin = Origin(url=origin_url) | origin = Origin(url=origin_url) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit1, | date=data.date_visit1, | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| visit1_id = origin_visit1.visit | visit1_id = origin_visit1.visit | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.snapshot['id']) | origin_visit1, snapshot=data.snapshot['id']) | ||||
| by_ov1 = swh_storage.snapshot_get_by_origin_visit( | by_ov1 = swh_storage.snapshot_get_by_origin_visit( | ||||
| origin_url, visit1_id) | origin_url, visit1_id) | ||||
| assert by_ov1 == {**data.snapshot, 'next_branch': None} | assert by_ov1 == {**data.snapshot, 'next_branch': None} | ||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit2, | date=data.date_visit2, | ||||
| type=data.type_visit2, | type=data.type_visit2, | ||||
| ) | ) | ||||
| visit2_id = origin_visit2.visit | visit2_id = origin_visit2.visit | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit2, snapshot=data.snapshot['id']) | origin_visit2, snapshot=data.snapshot['id']) | ||||
| by_ov2 = swh_storage.snapshot_get_by_origin_visit( | by_ov2 = swh_storage.snapshot_get_by_origin_visit( | ||||
| origin_url, visit2_id) | origin_url, visit2_id) | ||||
| assert by_ov2 == {**data.snapshot, 'next_branch': None} | assert by_ov2 == {**data.snapshot, 'next_branch': None} | ||||
| data1 = { | data1 = { | ||||
| 'origin': origin_url, | 'origin': origin_url, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| ▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest(self, swh_storage): | ||||
| ) | ) | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| assert swh_storage.snapshot_get_latest(origin_url) is None | assert swh_storage.snapshot_get_latest(origin_url) is None | ||||
| # Add snapshot to visit1, latest snapshot = visit 1 snapshot | # Add snapshot to visit1, latest snapshot = visit 1 snapshot | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.complete_snapshot['id']) | origin_visit1, snapshot=data.complete_snapshot['id']) | ||||
| assert {**data.complete_snapshot, 'next_branch': None} \ | assert {**data.complete_snapshot, 'next_branch': None} \ | ||||
| == swh_storage.snapshot_get_latest(origin_url) | == swh_storage.snapshot_get_latest(origin_url) | ||||
| # Status filter: all three visits are status=ongoing, so no snapshot | # Status filter: all three visits are status=ongoing, so no snapshot | ||||
| # returned | # returned | ||||
| assert swh_storage.snapshot_get_latest( | assert swh_storage.snapshot_get_latest( | ||||
| origin_url, | origin_url, | ||||
| allowed_statuses=['full']) is None | allowed_statuses=['full']) is None | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update(origin_visit1, status='full') | ||||
| origin, origin_visit1, status='full') | |||||
| assert {**data.complete_snapshot, 'next_branch': None} \ | assert {**data.complete_snapshot, 'next_branch': None} \ | ||||
| == swh_storage.snapshot_get_latest( | == swh_storage.snapshot_get_latest( | ||||
| origin_url, | origin_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 | ||||
| swh_storage.snapshot_add([data.empty_snapshot]) | swh_storage.snapshot_add([data.empty_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit2, snapshot=data.empty_snapshot['id']) | origin_visit2, snapshot=data.empty_snapshot['id']) | ||||
| assert {**data.empty_snapshot, 'next_branch': None} \ | assert {**data.empty_snapshot, 'next_branch': None} \ | ||||
| == swh_storage.snapshot_get_latest(origin_url) | == swh_storage.snapshot_get_latest(origin_url) | ||||
| # Check that the status filter is still working | # Check that the status filter is still working | ||||
| assert {**data.complete_snapshot, 'next_branch': None} \ | assert {**data.complete_snapshot, 'next_branch': None} \ | ||||
| == swh_storage.snapshot_get_latest( | == swh_storage.snapshot_get_latest( | ||||
| origin_url, | origin_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 | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit3, snapshot=data.complete_snapshot['id']) | origin_visit3, snapshot=data.complete_snapshot['id']) | ||||
| assert {**data.complete_snapshot, 'next_branch': None} \ | assert {**data.complete_snapshot, 'next_branch': None} \ | ||||
| == swh_storage.snapshot_get_latest(origin_url) | == swh_storage.snapshot_get_latest(origin_url) | ||||
| def test_snapshot_get_latest__missing_snapshot(self, swh_storage): | def test_snapshot_get_latest__missing_snapshot(self, swh_storage): | ||||
| # Origin does not exist | # Origin does not exist | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| assert swh_storage.snapshot_get_latest(origin_url) is None | assert swh_storage.snapshot_get_latest(origin_url) is None | ||||
| Show All 12 Lines | def test_snapshot_get_latest__missing_snapshot(self, swh_storage): | ||||
| ) | ) | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| assert swh_storage.snapshot_get_latest(origin_url) is None | assert swh_storage.snapshot_get_latest(origin_url) is None | ||||
| # Add unknown snapshot to visit1, check that the inconsistency is | # Add unknown snapshot to visit1, check that the inconsistency is | ||||
| # detected | # detected | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.complete_snapshot['id']) | origin_visit1, snapshot=data.complete_snapshot['id']) | ||||
| with pytest.raises(Exception): | with pytest.raises(Exception): | ||||
| # XXX: should the exception be more specific than this? | # XXX: should the exception be more specific than this? | ||||
| swh_storage.snapshot_get_latest(origin_url) | swh_storage.snapshot_get_latest(origin_url) | ||||
| # Status filter: both visits are status=ongoing, so no snapshot | # Status filter: both visits are status=ongoing, so no snapshot | ||||
| # returned | # returned | ||||
| assert swh_storage.snapshot_get_latest( | assert swh_storage.snapshot_get_latest( | ||||
| origin_url, allowed_statuses=['full']) is None | origin_url, allowed_statuses=['full']) is None | ||||
| # Mark the first visit as completed and check status filter again | # Mark the first visit as completed and check status filter again | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update(origin_visit1, status='full') | ||||
| origin, origin_visit1, status='full') | |||||
| with pytest.raises(Exception): | with pytest.raises(Exception): | ||||
| # XXX: should the exception be more specific than this? | # XXX: should the exception be more specific than this? | ||||
| swh_storage.snapshot_get_latest( | swh_storage.snapshot_get_latest( | ||||
| origin_url, | origin_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 | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| assert {**data.complete_snapshot, 'next_branch': None} \ | assert {**data.complete_snapshot, 'next_branch': None} \ | ||||
| == swh_storage.snapshot_get_latest(origin_url) | == swh_storage.snapshot_get_latest(origin_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 | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit2, snapshot=data.snapshot['id']) | origin_visit2, snapshot=data.snapshot['id']) | ||||
| with pytest.raises(Exception): | with pytest.raises(Exception): | ||||
| # XXX: should the exception be more specific than this? | # XXX: should the exception be more specific than this? | ||||
| swh_storage.snapshot_get_latest( | swh_storage.snapshot_get_latest( | ||||
| origin_url) | origin_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 | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| assert{**data.snapshot, 'next_branch': None} \ | assert{**data.snapshot, 'next_branch': None} \ | ||||
| ▲ Show 20 Lines • Show All 49 Lines • ▼ Show 20 Lines | def test_stat_counters(self, swh_storage): | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin, | origin, | ||||
| date=data.date_visit2, | date=data.date_visit2, | ||||
| type=data.type_visit2, | type=data.type_visit2, | ||||
| ) | ) | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin, origin_visit1, snapshot=data.snapshot['id']) | origin_visit1, snapshot=data.snapshot['id']) | ||||
| swh_storage.directory_add([data.dir]) | swh_storage.directory_add([data.dir]) | ||||
| swh_storage.revision_add([data.revision]) | swh_storage.revision_add([data.revision]) | ||||
| swh_storage.release_add([data.release]) | swh_storage.release_add([data.release]) | ||||
| swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
| counters = swh_storage.stat_counters() | counters = swh_storage.stat_counters() | ||||
| assert counters['content'] == 1 | assert counters['content'] == 1 | ||||
| assert counters['directory'] == 1 | assert counters['directory'] == 1 | ||||
| ▲ Show 20 Lines • Show All 873 Lines • ▼ Show 20 Lines | def test_origin_count_with_visit_with_visits_and_snapshot( | ||||
| self, swh_storage): | self, swh_storage): | ||||
| swh_storage.origin_add([{'url': url} for url in self.ORIGINS]) | swh_storage.origin_add([{'url': url} for url in self.ORIGINS]) | ||||
| now = datetime.datetime.now(tz=datetime.timezone.utc) | now = datetime.datetime.now(tz=datetime.timezone.utc) | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| origin = Origin(url='https://github.com/user1/repo1') | origin = Origin(url='https://github.com/user1/repo1') | ||||
| visit = swh_storage.origin_visit_add( | visit = swh_storage.origin_visit_add( | ||||
| origin, date=now, type='git') | origin, date=now, type='git') | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update(visit, snapshot=data.snapshot['id']) | ||||
| origin, visit, snapshot=data.snapshot['id']) | |||||
| assert swh_storage.origin_count('github', with_visit=False) == 3 | assert swh_storage.origin_count('github', with_visit=False) == 3 | ||||
| # github/user1 has a visit and a snapshot, so with_visit=True => 1 | # github/user1 has a visit and a snapshot, so with_visit=True => 1 | ||||
| assert swh_storage.origin_count('github', with_visit=True) == 1 | assert swh_storage.origin_count('github', with_visit=True) == 1 | ||||
| assert swh_storage.origin_count('github.*user1', regexp=True, | assert swh_storage.origin_count('github.*user1', regexp=True, | ||||
| with_visit=False) == 1 | with_visit=False) == 1 | ||||
| assert swh_storage.origin_count('github.*user1', regexp=True, | assert swh_storage.origin_count('github.*user1', regexp=True, | ||||
| ▲ Show 20 Lines • Show All 199 Lines • Show Last 20 Lines | |||||