Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 383 Lines • ▼ Show 20 Lines | def test_content_get_metadata_missing_sha1(self, swh_storage): | ||||
| # All the metadata keys are None | # All the metadata keys are None | ||||
| missing_cont.pop('data') | missing_cont.pop('data') | ||||
| for key in missing_cont: | for key in missing_cont: | ||||
| if key != 'sha1': | if key != 'sha1': | ||||
| missing_cont[key] = None | missing_cont[key] = None | ||||
| assert list(gen) == [missing_cont] | assert list(gen) == [missing_cont] | ||||
| def test_content_get_random(self, swh_storage): | |||||
| swh_storage.content_add([data.cont, data.cont2, data.cont3]) | |||||
| assert swh_storage.content_get_random() in { | |||||
| data.cont['sha1_git'], data.cont2['sha1_git'], | |||||
| data.cont3['sha1_git']} | |||||
| def test_directory_add(self, swh_storage): | def test_directory_add(self, swh_storage): | ||||
| init_missing = list(swh_storage.directory_missing([data.dir['id']])) | init_missing = list(swh_storage.directory_missing([data.dir['id']])) | ||||
| assert [data.dir['id']] == init_missing | assert [data.dir['id']] == init_missing | ||||
| actual_result = swh_storage.directory_add([data.dir]) | actual_result = swh_storage.directory_add([data.dir]) | ||||
| assert actual_result == {'directory:add': 1} | assert actual_result == {'directory:add': 1} | ||||
| assert list(swh_storage.journal_writer.objects) == \ | assert list(swh_storage.journal_writer.objects) == \ | ||||
| ▲ Show 20 Lines • Show All 174 Lines • ▼ Show 20 Lines | def test_directory_entry_get_by_path(self, swh_storage): | ||||
| # when (nothing should be found here since data.dir is not persisted.) | # when (nothing should be found here since data.dir is not persisted.) | ||||
| for entry in data.dir['entries']: | for entry in data.dir['entries']: | ||||
| actual_entry = swh_storage.directory_entry_get_by_path( | actual_entry = swh_storage.directory_entry_get_by_path( | ||||
| data.dir['id'], | data.dir['id'], | ||||
| [entry['name']]) | [entry['name']]) | ||||
| assert actual_entry is None | assert actual_entry is None | ||||
| def test_directory_get_random(self, swh_storage): | |||||
| swh_storage.directory_add([data.dir, data.dir2, data.dir3]) | |||||
| assert swh_storage.directory_get_random() in \ | |||||
| {data.dir['id'], data.dir2['id'], data.dir3['id']} | |||||
| def test_revision_add(self, swh_storage): | def test_revision_add(self, swh_storage): | ||||
| init_missing = swh_storage.revision_missing([data.revision['id']]) | init_missing = swh_storage.revision_missing([data.revision['id']]) | ||||
| assert list(init_missing) == [data.revision['id']] | assert list(init_missing) == [data.revision['id']] | ||||
| actual_result = swh_storage.revision_add([data.revision]) | actual_result = swh_storage.revision_add([data.revision]) | ||||
| assert actual_result == {'revision:add': 1} | assert actual_result == {'revision:add': 1} | ||||
| end_missing = swh_storage.revision_missing([data.revision['id']]) | end_missing = swh_storage.revision_missing([data.revision['id']]) | ||||
| ▲ Show 20 Lines • Show All 163 Lines • ▼ Show 20 Lines | class TestStorage: | ||||
| def test_revision_get_no_parents(self, swh_storage): | def test_revision_get_no_parents(self, swh_storage): | ||||
| swh_storage.revision_add([data.revision3]) | swh_storage.revision_add([data.revision3]) | ||||
| get = list(swh_storage.revision_get([data.revision3['id']])) | get = list(swh_storage.revision_get([data.revision3['id']])) | ||||
| assert len(get) == 1 | assert len(get) == 1 | ||||
| assert get[0]['parents'] == [] # no parents on this one | assert get[0]['parents'] == [] # no parents on this one | ||||
| def test_revision_get_random(self, swh_storage): | |||||
| swh_storage.revision_add( | |||||
| [data.revision, data.revision2, data.revision3]) | |||||
| assert swh_storage.revision_get_random() in \ | |||||
| {data.revision['id'], data.revision2['id'], data.revision3['id']} | |||||
| def test_release_add(self, swh_storage): | def test_release_add(self, swh_storage): | ||||
| init_missing = swh_storage.release_missing([data.release['id'], | init_missing = swh_storage.release_missing([data.release['id'], | ||||
| data.release2['id']]) | data.release2['id']]) | ||||
| assert [data.release['id'], data.release2['id']] == list(init_missing) | assert [data.release['id'], data.release2['id']] == list(init_missing) | ||||
| actual_result = swh_storage.release_add([data.release, data.release2]) | actual_result = swh_storage.release_add([data.release, data.release2]) | ||||
| assert actual_result == {'release:add': 2} | assert actual_result == {'release:add': 2} | ||||
| ▲ Show 20 Lines • Show All 94 Lines • ▼ Show 20 Lines | def test_release_get(self, swh_storage): | ||||
| normalize_entity(data.release), normalize_entity(data.release2)] \ | normalize_entity(data.release), normalize_entity(data.release2)] \ | ||||
| == [actual_releases[0], actual_releases[1]] | == [actual_releases[0], actual_releases[1]] | ||||
| unknown_releases = \ | unknown_releases = \ | ||||
| list(swh_storage.release_get([data.release3['id']])) | list(swh_storage.release_get([data.release3['id']])) | ||||
| assert unknown_releases[0] is None | assert unknown_releases[0] is None | ||||
| def test_release_get_random(self, swh_storage): | |||||
| swh_storage.release_add([data.release, data.release2, data.release3]) | |||||
| assert swh_storage.release_get_random() in \ | |||||
| {data.release['id'], data.release2['id'], data.release3['id']} | |||||
| def test_origin_add_one(self, swh_storage): | def test_origin_add_one(self, swh_storage): | ||||
| origin0 = swh_storage.origin_get(data.origin) | origin0 = swh_storage.origin_get(data.origin) | ||||
| assert origin0 is None | assert origin0 is None | ||||
| id = swh_storage.origin_add_one(data.origin) | id = swh_storage.origin_add_one(data.origin) | ||||
| actual_origin = swh_storage.origin_get({'url': data.origin['url']}) | actual_origin = swh_storage.origin_get({'url': data.origin['url']}) | ||||
| assert actual_origin['url'] == data.origin['url'] | assert actual_origin['url'] == data.origin['url'] | ||||
| ▲ Show 20 Lines • Show All 1,478 Lines • ▼ Show 20 Lines | def test_snapshot_get_latest__missing_snapshot(self, swh_storage): | ||||
| 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} \ | ||||
| == swh_storage.snapshot_get_latest(origin_url) | == swh_storage.snapshot_get_latest(origin_url) | ||||
| def test_snapshot_get_random(self, swh_storage): | |||||
| swh_storage.snapshot_add( | |||||
| [data.snapshot, data.empty_snapshot, data.complete_snapshot]) | |||||
| assert swh_storage.snapshot_get_random() in { | |||||
| data.snapshot['id'], data.empty_snapshot['id'], | |||||
| data.complete_snapshot['id']} | |||||
| def test_stat_counters(self, swh_storage): | def test_stat_counters(self, swh_storage): | ||||
| expected_keys = ['content', 'directory', | expected_keys = ['content', 'directory', | ||||
| 'origin', 'revision'] | 'origin', 'revision'] | ||||
| # Initially, all counters are 0 | # Initially, all counters are 0 | ||||
| swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
| counters = swh_storage.stat_counters() | counters = swh_storage.stat_counters() | ||||
| ▲ Show 20 Lines • Show All 1,134 Lines • Show Last 20 Lines | |||||