Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/data.py
Show All 14 Lines | |||||
from swh.counters import get_counters | from swh.counters import get_counters | ||||
from swh.indexer.ctags import CtagsIndexer | from swh.indexer.ctags import CtagsIndexer | ||||
from swh.indexer.fossology_license import FossologyLicenseIndexer | from swh.indexer.fossology_license import FossologyLicenseIndexer | ||||
from swh.indexer.mimetype import MimetypeIndexer | from swh.indexer.mimetype import MimetypeIndexer | ||||
from swh.indexer.storage import get_indexer_storage | from swh.indexer.storage import get_indexer_storage | ||||
from swh.indexer.storage.model import OriginIntrinsicMetadataRow | from swh.indexer.storage.model import OriginIntrinsicMetadataRow | ||||
from swh.loader.git.from_disk import GitLoaderFromArchive | from swh.loader.git.from_disk import GitLoaderFromArchive | ||||
from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_hex | from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_hex | ||||
from swh.model.identifiers import CoreSWHID, ObjectType, QualifiedSWHID | |||||
from swh.model.model import ( | from swh.model.model import ( | ||||
Content, | Content, | ||||
Directory, | Directory, | ||||
Origin, | Origin, | ||||
OriginVisit, | OriginVisit, | ||||
OriginVisitStatus, | OriginVisitStatus, | ||||
Snapshot, | Snapshot, | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 241 Lines • ▼ Show 20 Lines | for i in range(250): | ||||
visit_type="tar", | visit_type="tar", | ||||
) | ) | ||||
sha1s: Set[Sha1] = set() | sha1s: Set[Sha1] = set() | ||||
directories = set() | directories = set() | ||||
revisions = set() | revisions = set() | ||||
releases = set() | releases = set() | ||||
snapshots = set() | snapshots = set() | ||||
swhids = [] | |||||
content_path = {} | content_path = {} | ||||
# Get all objects loaded into the test archive | # Get all objects loaded into the test archive | ||||
common_metadata = {ORIGIN_METADATA_KEY: ORIGIN_METADATA_VALUE} | common_metadata = {ORIGIN_METADATA_KEY: ORIGIN_METADATA_VALUE} | ||||
for origin in _TEST_ORIGINS: | for origin in _TEST_ORIGINS: | ||||
origin_revisions = set() | |||||
snp = snapshot_get_latest(storage, origin["url"]) | snp = snapshot_get_latest(storage, origin["url"]) | ||||
swhids.append( | |||||
QualifiedSWHID( | |||||
object_type=ObjectType.SNAPSHOT, object_id=snp.id, origin=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(): | ||||
target_type = branch_data.target_type.value | target_type = branch_data.target_type.value | ||||
if target_type == "revision": | if target_type == "revision": | ||||
revisions.add(branch_data.target) | origin_revisions.add(branch_data.target) | ||||
swhids.append( | |||||
QualifiedSWHID( | |||||
object_type=ObjectType.REVISION, | |||||
object_id=branch_data.target, | |||||
origin=origin["url"], | |||||
visit=CoreSWHID( | |||||
object_type=ObjectType.SNAPSHOT, object_id=snp.id | |||||
), | |||||
) | |||||
) | |||||
if b"master" in branch_name: | if b"master" in branch_name: | ||||
# Add some origin intrinsic metadata for tests | # Add some origin intrinsic metadata for tests | ||||
metadata = common_metadata | metadata = common_metadata | ||||
metadata.update(origin.get("metadata", {})) | metadata.update(origin.get("metadata", {})) | ||||
origin_metadata = OriginIntrinsicMetadataRow( | origin_metadata = OriginIntrinsicMetadataRow( | ||||
id=origin["url"], | id=origin["url"], | ||||
from_revision=branch_data.target, | from_revision=branch_data.target, | ||||
indexer_configuration_id=idx_tool["id"], | indexer_configuration_id=idx_tool["id"], | ||||
metadata=metadata, | metadata=metadata, | ||||
mappings=[], | mappings=[], | ||||
) | ) | ||||
idx_storage.origin_intrinsic_metadata_add([origin_metadata]) | idx_storage.origin_intrinsic_metadata_add([origin_metadata]) | ||||
search.origin_update( | search.origin_update( | ||||
[{"url": origin["url"], "intrinsic_metadata": metadata}] | [{"url": origin["url"], "intrinsic_metadata": metadata}] | ||||
) | ) | ||||
ORIGIN_MASTER_REVISION[origin["url"]] = hash_to_hex( | ORIGIN_MASTER_REVISION[origin["url"]] = hash_to_hex( | ||||
branch_data.target | branch_data.target | ||||
) | ) | ||||
elif target_type == "release": | elif target_type == "release": | ||||
release = storage.release_get([branch_data.target])[0] | release = storage.release_get([branch_data.target])[0] | ||||
revisions.add(release.target) | origin_revisions.add(release.target) | ||||
releases.add(hash_to_hex(branch_data.target)) | releases.add(hash_to_hex(branch_data.target)) | ||||
swhids.append( | |||||
QualifiedSWHID( | |||||
object_type=ObjectType.RELEASE, | |||||
object_id=branch_data.target, | |||||
origin=origin["url"], | |||||
visit=CoreSWHID( | |||||
object_type=ObjectType.SNAPSHOT, object_id=snp.id | |||||
), | |||||
) | |||||
) | |||||
for rev_log in storage.revision_shortlog(set(revisions)): | for rev_log in storage.revision_shortlog(origin_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(origin_revisions): | ||||
if rev is None: | if rev is None: | ||||
continue | continue | ||||
dir_id = rev.directory | dir_id = rev.directory | ||||
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): | ||||
if entry["type"] == "file": | if entry["type"] == "file": | ||||
sha1s.add(entry["sha1"]) | sha1s.add(entry["sha1"]) | ||||
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")] | ||||
) | ) | ||||
swhids.append( | |||||
QualifiedSWHID( | |||||
object_type=ObjectType.CONTENT, | |||||
object_id=entry["sha1_git"], | |||||
origin=origin["url"], | |||||
visit=CoreSWHID( | |||||
object_type=ObjectType.SNAPSHOT, object_id=snp.id | |||||
), | |||||
anchor=CoreSWHID( | |||||
object_type=ObjectType.REVISION, object_id=rev.id | |||||
), | |||||
path=b"/" + entry["path"], | |||||
) | |||||
) | |||||
elif entry["type"] == "dir": | elif entry["type"] == "dir": | ||||
directories.add(hash_to_hex(entry["target"])) | directories.add(hash_to_hex(entry["target"])) | ||||
swhids.append( | |||||
QualifiedSWHID( | |||||
object_type=ObjectType.DIRECTORY, | |||||
object_id=entry["target"], | |||||
origin=origin["url"], | |||||
visit=CoreSWHID( | |||||
object_type=ObjectType.SNAPSHOT, object_id=snp.id | |||||
), | |||||
anchor=CoreSWHID( | |||||
object_type=ObjectType.REVISION, object_id=rev.id | |||||
), | |||||
path=b"/" + entry["path"] + b"/", | |||||
) | |||||
) | |||||
_add_extra_contents(storage, sha1s) | _add_extra_contents(storage, sha1s) | ||||
# Get all checksums for each content | # Get all checksums for each content | ||||
result: List[Optional[Content]] = storage.content_get(list(sha1s)) | result: List[Optional[Content]] = storage.content_get(list(sha1s)) | ||||
contents: List[Dict] = [] | contents: List[Dict] = [] | ||||
for content in result: | for content in result: | ||||
▲ Show 20 Lines • Show All 67 Lines • ▼ Show 20 Lines | return { | ||||
"idx_storage": idx_storage, | "idx_storage": idx_storage, | ||||
"counters": counters, | "counters": counters, | ||||
"origins": _TEST_ORIGINS, | "origins": _TEST_ORIGINS, | ||||
"contents": contents, | "contents": contents, | ||||
"directories": list(directories), | "directories": list(directories), | ||||
"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(), | "swhids": swhids, | ||||
} | } | ||||
def _init_indexers(tests_data): | def _init_indexers(tests_data): | ||||
# Instantiate content indexers that will be used in tests | # Instantiate content indexers that will be used in tests | ||||
# and force them to use the memory storages | # and force them to use the memory storages | ||||
indexers = {} | indexers = {} | ||||
for idx_name, idx_class, idx_config in ( | for idx_name, idx_class, idx_config in ( | ||||
▲ Show 20 Lines • Show All 64 Lines • Show Last 20 Lines |