Page MenuHomeSoftware Heritage
Differential D2893 Diff 10297 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 1,864 Lines • ▼ Show 20 Lines def test_origin_visit_update_missing_snapshot(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
origin_visit = swh_storage.origin_visit_add( origin_visit = swh_storage.origin_visit_add(
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_visit1, type=data.type_visit1)
# when # when
swh_storage.origin_visit_update( swh_storage.origin_visit_update(
origin_url, origin_url,
origin_visit.visit, origin_visit.visit,
snapshot=data.snapshot['id']) status='ongoing',
snapshot=data.snapshot['id']) # snapshot does not exist yet
# 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])
assert actual_origin_visit['snapshot'] == data.snapshot['id'] assert actual_origin_visit['snapshot'] == data.snapshot['id']
def test_origin_visit_get_by(self, swh_storage): def test_origin_visit_get_by(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
origin_url2 = swh_storage.origin_add_one(data.origin2) origin_url2 = swh_storage.origin_add_one(data.origin2)
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit2, type=data.type_visit2) origin_url, date=data.date_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_url, origin_url,
origin_visit1.visit, origin_visit1.visit,
status='ongoing',
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_url, date=data.date_visit3, type=data.type_visit3) origin_url, date=data.date_visit3, type=data.type_visit3)
swh_storage.origin_visit_add( swh_storage.origin_visit_add(
origin_url2, date=data.date_visit3, type=data.type_visit3) origin_url2, date=data.date_visit3, type=data.type_visit3)
▲ Show 20 Lines • Show All 186 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_url, ov1.visit, snapshot=data.complete_snapshot['id']) origin_url, ov1.visit,
status='ongoing',
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
Show All 10 Lines def test_origin_visit_get_latest(self, swh_storage):
'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_url, ov2.visit, snapshot=data.empty_snapshot['id']) origin_url, ov2.visit,
status='ongoing',
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_url, ov3.visit, snapshot=data.complete_snapshot['id']) origin_url, ov3.visit,
status='ongoing',
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 All 37 Lines def test_snapshot_add_get_empty(self, swh_storage):
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_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_url, origin_visit1.visit, origin_url, origin_visit1.visit,
status='ongoing',
snapshot=data.empty_snapshot['id']) 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}
Show All 26 Lines def test_snapshot_add_get_complete(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_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_url, origin_visit1.visit, origin_url, origin_visit1.visit,
status='ongoing',
snapshot=data.complete_snapshot['id']) 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 138 Lines • ▼ Show 20 Lines
def test_snapshot_add_get_filtered(self, swh_storage): def test_snapshot_add_get_filtered(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_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_url, origin_visit1.visit, origin_url, origin_visit1.visit,
status='ongoing',
snapshot=data.complete_snapshot['id']) 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'])
▲ Show 20 Lines • Show All 99 Lines • ▼ Show 20 Lines
def test_snapshot_add_get(self, swh_storage): def test_snapshot_add_get(self, swh_storage):
origin_url = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_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_url, origin_visit1.visit, snapshot=data.snapshot['id']) origin_url, origin_visit1.visit,
status='ongoing',
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 = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
# unknown visit # unknown visit
visit_id = 54164461156 visit_id = 54164461156
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_url, visit_id, snapshot=data.snapshot['id']) origin_url, visit_id,
status='ongoing',
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 = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_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_url, origin_visit1.visit, snapshot=data.snapshot['id']) origin_url, origin_visit1.visit,
status='ongoing',
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_url, date=data.date_visit2, type=data.type_visit2) origin_url, date=data.date_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_url, origin_visit2.visit, snapshot=data.snapshot['id']) origin_url, origin_visit2.visit,
status='ongoing',
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 49 Lines • ▼ Show 20 Lines def test_snapshot_get_latest(self, swh_storage):
origin_url, date=data.date_visit2, type=data.type_visit3) origin_url, date=data.date_visit2, type=data.type_visit3)
# 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_url, origin_visit1.visit, origin_url, origin_visit1.visit,
status='ongoing',
snapshot=data.complete_snapshot['id']) 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_url, origin_visit1.visit, status='full') origin_url, origin_visit1.visit, 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_url, origin_visit2.visit, origin_url, origin_visit2.visit,
status='ongoing',
snapshot=data.empty_snapshot['id']) 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_url, origin_visit3.visit, origin_url, origin_visit3.visit,
status='ongoing',
snapshot=data.complete_snapshot['id']) 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_url = swh_storage.origin_add_one(data.origin) origin_url = swh_storage.origin_add_one(data.origin)
assert swh_storage.snapshot_get_latest(origin_url) is None assert swh_storage.snapshot_get_latest(origin_url) is None
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit1, type=data.type_visit1) origin_url, date=data.date_visit1, type=data.type_visit1)
origin_visit2 = swh_storage.origin_visit_add( origin_visit2 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit2, type=data.type_visit2) origin_url, date=data.date_visit2, type=data.type_visit2)
# 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_url, origin_visit1.visit, origin_url, origin_visit1.visit,
status='ongoing',
snapshot=data.complete_snapshot['id']) 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(
Show All 11 Lines def test_snapshot_get_latest__missing_snapshot(self, swh_storage):
# 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_url, origin_visit2.visit, snapshot=data.snapshot['id']) origin_url, origin_visit2.visit,
status='ongoing',
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 45 Lines • ▼ Show 20 Lines def test_stat_counters(self, swh_storage):
# Add other objects. Check their counter increased as well. # Add other objects. Check their counter increased as well.
origin_url = swh_storage.origin_add_one(data.origin2) origin_url = swh_storage.origin_add_one(data.origin2)
origin_visit1 = swh_storage.origin_visit_add( origin_visit1 = swh_storage.origin_visit_add(
origin_url, date=data.date_visit2, type=data.type_visit2) origin_url, date=data.date_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_url, origin_visit1.visit, snapshot=data.snapshot['id']) origin_url, origin_visit1.visit,
status='ongoing',
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 874 Lines • ▼ Show 20 Lines def test_origin_count_with_visit_with_visits_and_snapshot(
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_url = 'https://github.com/user1/repo1' origin_url = 'https://github.com/user1/repo1'
visit = swh_storage.origin_visit_add( visit = swh_storage.origin_visit_add(
origin_url, date=now, type='git') origin_url, date=now, type='git')
swh_storage.origin_visit_update( swh_storage.origin_visit_update(
origin_url, visit.visit, snapshot=data.snapshot['id']) origin_url, visit.visit,
status='ongoing',
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 198 Lines • Show Last 20 Lines
About · Developer information · Legal