Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/data.py
Show First 20 Lines • Show All 184 Lines • ▼ Show 20 Lines | for origin in _TEST_ORIGINS: | ||||
origin.update(storage.origin_get(origin)) # add an 'id' key if enabled | origin.update(storage.origin_get(origin)) # add an 'id' key if enabled | ||||
contents = set() | contents = set() | ||||
directories = set() | directories = set() | ||||
revisions = set() | revisions = set() | ||||
releases = set() | releases = set() | ||||
snapshots = set() | snapshots = set() | ||||
persons = set() | |||||
content_path = {} | content_path = {} | ||||
# Get all objects loaded into the test archive | # Get all objects loaded into the test archive | ||||
for origin in _TEST_ORIGINS: | for origin in _TEST_ORIGINS: | ||||
snp = storage.snapshot_get_latest(origin['url']) | snp = storage.snapshot_get_latest(origin['url']) | ||||
snapshots.add(hash_to_hex(snp['id'])) | snapshots.add(hash_to_hex(snp['id'])) | ||||
for branch_name, branch_data in snp['branches'].items(): | for branch_name, branch_data in snp['branches'].items(): | ||||
if branch_data['target_type'] == 'revision': | if branch_data['target_type'] == 'revision': | ||||
revisions.add(branch_data['target']) | revisions.add(branch_data['target']) | ||||
elif branch_data['target_type'] == 'release': | elif branch_data['target_type'] == 'release': | ||||
release = next(storage.release_get([branch_data['target']])) | release = next(storage.release_get([branch_data['target']])) | ||||
revisions.add(release['target']) | revisions.add(release['target']) | ||||
releases.add(hash_to_hex(branch_data['target'])) | releases.add(hash_to_hex(branch_data['target'])) | ||||
persons.add(release['author']['id']) | |||||
for rev_log in storage.revision_shortlog(set(revisions)): | for rev_log in storage.revision_shortlog(set(revisions)): | ||||
rev_id = rev_log[0] | rev_id = rev_log[0] | ||||
revisions.add(rev_id) | revisions.add(rev_id) | ||||
for rev in storage.revision_get(revisions): | for rev in storage.revision_get(revisions): | ||||
dir_id = rev['directory'] | dir_id = rev['directory'] | ||||
persons.add(rev['author']['id']) | |||||
persons.add(rev['committer']['id']) | |||||
directories.add(hash_to_hex(dir_id)) | directories.add(hash_to_hex(dir_id)) | ||||
for entry in dir_iterator(storage, dir_id): | for entry in dir_iterator(storage, dir_id): | ||||
content_path[entry['sha1']] = '/'.join( | content_path[entry['sha1']] = '/'.join( | ||||
[hash_to_hex(dir_id), entry['path'].decode('utf-8')]) | [hash_to_hex(dir_id), entry['path'].decode('utf-8')]) | ||||
if entry['type'] == 'file': | if entry['type'] == 'file': | ||||
contents.add(entry['sha1']) | contents.add(entry['sha1']) | ||||
elif entry['type'] == 'dir': | elif entry['type'] == 'dir': | ||||
directories.add(hash_to_hex(entry['target'])) | directories.add(hash_to_hex(entry['target'])) | ||||
Show All 28 Lines | def _init_tests_data(): | ||||
# Return tests data | # Return tests data | ||||
return { | return { | ||||
'storage': storage, | 'storage': storage, | ||||
'idx_storage': idx_storage, | 'idx_storage': idx_storage, | ||||
'origins': _TEST_ORIGINS, | 'origins': _TEST_ORIGINS, | ||||
'contents': contents, | 'contents': contents, | ||||
'directories': list(directories), | 'directories': list(directories), | ||||
'persons': list(persons), | |||||
'releases': list(releases), | 'releases': list(releases), | ||||
'revisions': list(map(hash_to_hex, revisions)), | 'revisions': list(map(hash_to_hex, revisions)), | ||||
'snapshots': list(snapshots), | 'snapshots': list(snapshots), | ||||
'generated_checksums': set(), | 'generated_checksums': set(), | ||||
} | } | ||||
def _init_indexers(tests_data): | def _init_indexers(tests_data): | ||||
▲ Show 20 Lines • Show All 199 Lines • Show Last 20 Lines |