Differential D3959 Diff 13938 swh/storage/tests/migrate_extrinsic_metadata/test_pypi.py

Changeset View

Standalone View

swh/storage/tests/migrate_extrinsic_metadata/test_pypi.py

Show All 13 Lines

from swh.model.identifiers import parse_swhid		from swh.model.identifiers import parse_swhid
from swh.model.model import (		from swh.model.model import (
MetadataAuthority,		MetadataAuthority,
MetadataAuthorityType,		MetadataAuthorityType,
MetadataFetcher,		MetadataFetcher,
MetadataTargetType,		MetadataTargetType,
Origin,		Origin,
		OriginVisit,
		OriginVisitStatus,
RawExtrinsicMetadata,		RawExtrinsicMetadata,
		Snapshot,
		SnapshotBranch,
		TargetType,
)		)

from swh.storage import get_storage		from swh.storage import get_storage
from swh.storage.interface import PagedResult		from swh.storage.interface import PagedResult
from swh.storage.migrate_extrinsic_metadata import (		from swh.storage.migrate_extrinsic_metadata import (
handle_row,		handle_row,
pypi_project_from_filename,		pypi_project_from_filename,
)		)


FETCHER = MetadataFetcher(		FETCHER = MetadataFetcher(
name="migrate-extrinsic-metadata-from-revisions", version="0.0.1",		name="migrate-extrinsic-metadata-from-revisions", version="0.0.1",
)		)
PYPI_AUTHORITY = MetadataAuthority(		PYPI_AUTHORITY = MetadataAuthority(
type=MetadataAuthorityType.FORGE, url="https://pypi.org/",		type=MetadataAuthorityType.FORGE, url="https://pypi.org/",
)		)
SWH_AUTHORITY = MetadataAuthority(		SWH_AUTHORITY = MetadataAuthority(
type=MetadataAuthorityType.REGISTRY, url="https://softwareheritage.org/",		type=MetadataAuthorityType.REGISTRY, url="https://softwareheritage.org/",
)		)


		def now():
		return datetime.datetime.now(tz=datetime.timezone.utc)


def test_pypi_project_from_filename():		def test_pypi_project_from_filename():
files = [		files = [
("django-agent-trust-0.1.8.tar.gz", "django-agent-trust"),		("django-agent-trust-0.1.8.tar.gz", "django-agent-trust"),
("python_test-1.0.1.zip", "python_test"),		("python_test-1.0.1.zip", "python_test"),
("py-evm-0.2.0a9.tar.gz", "py-evm"),		("py-evm-0.2.0a9.tar.gz", "py-evm"),
("collective.texttospeech-1.0rc1.tar.gz", "collective.texttospeech"),		("collective.texttospeech-1.0rc1.tar.gz", "collective.texttospeech"),
("flatland-fork-0.4.post1.dev40550160.zip", "flatland-fork"),		("flatland-fork-0.4.post1.dev40550160.zip", "flatland-fork"),
]		]
▲ Show 20 Lines • Show All 313 Lines • ▼ Show 20 Lines	) == PagedResult(
fetcher=FETCHER,		fetcher=FETCHER,
format="original-artifacts-json",		format="original-artifacts-json",
metadata=json.dumps(dest_original_artifacts).encode(),		metadata=json.dumps(dest_original_artifacts).encode(),
origin=None,		origin=None,
),		),
],		],
next_page_token=None,		next_page_token=None,
)		)


		def test_pypi_good_origin():
		"""Tests loading a revision generated by a vert old PyPI loader that
		ardumontUnsubmitted Not Done Inline Actions very* ardumont: very*
		does not have a provider orhas 'project' metadata."""
		ardumontUnsubmitted Not Done Inline Actions `or has` ardumont: `or has`

		source_original_artifact = {
		"url": "https://files.pythonhosted.org/packages/34/4f/30087f22eaae8ad7077a28ce157342745a2977e264b8a8e4e7f804a8aa5e/PyPDFLite-0.1.32.tar.gz",
		"date": "2014-05-07T22:03:00",
		"sha1": "3289269f75b4111dd00eaea53e00330db9a1db12",
		"size": 46644,
		"sha256": "911497d655cf7ef6530c5b57773dad7da97e21cf4d608ad9ad1e38bd7bec7824",
		"filename": "PyPDFLite-0.1.32.tar.gz",
		"sha1_git": "1e5c38014731242cfa8594839bcba8a0c4e158c5",
		"blake2s256": "45792e57873f56d385c694e36c98a580cbba60d5ea91eb6fd0a2d1c71c1fb385",
		"archive_type": "tar",
		}

		dest_original_artifacts = [
		{
		"url": "https://files.pythonhosted.org/packages/34/4f/30087f22eaae8ad7077a28ce157342745a2977e264b8a8e4e7f804a8aa5e/PyPDFLite-0.1.32.tar.gz",
		"filename": "PyPDFLite-0.1.32.tar.gz",
		"archive_type": "tar",
		"length": 46644,
		"checksums": {
		"sha1": "3289269f75b4111dd00eaea53e00330db9a1db12",
		"sha256": "911497d655cf7ef6530c5b57773dad7da97e21cf4d608ad9ad1e38bd7bec7824",
		"sha1_git": "1e5c38014731242cfa8594839bcba8a0c4e158c5",
		"blake2s256": "45792e57873f56d385c694e36c98a580cbba60d5ea91eb6fd0a2d1c71c1fb385",
		},
		}
		]

		revision_id = b"N\xa9\x91\|\xdfS\xcd\x13SJ\x04.N\xb3x{\x86\xc84\xd2"
		row = {
		"id": revision_id,
		"date": datetime.datetime(2014, 5, 7, 22, 3, tzinfo=datetime.timezone.utc),
		"committer_date": datetime.datetime(
		2014, 5, 7, 22, 3, tzinfo=datetime.timezone.utc
		),
		"type": "tar",
		"message": b"0.1.32",
		"metadata": {"original_artifact": source_original_artifact},
		}

		origin_url = "https://pypi.org/project/PyPDFLite/"

		storage = get_storage("memory")

		snapshot_id = b"42" * 10
		storage.origin_add([Origin(url=origin_url)])
		storage.origin_visit_add(
		[OriginVisit(origin=origin_url, visit=1, date=now(), type="pypi")]
		)
		storage.origin_visit_status_add(
		[
		OriginVisitStatus(
		origin=origin_url,
		visit=1,
		date=now(),
		status="partial",
		snapshot=snapshot_id,
		)
		]
		)
		storage.snapshot_add(
		[
		Snapshot(
		id=snapshot_id,
		branches={
		b"foo": SnapshotBranch(
		target_type=TargetType.REVISION, target=revision_id,
		)
		},
		)
		]
		)
		storage.metadata_authority_add(
		[
		attr.evolve(PYPI_AUTHORITY, metadata={}),
		attr.evolve(SWH_AUTHORITY, metadata={}),
		]
		)
		storage.metadata_fetcher_add([FETCHER])
		deposit_cur = None
		handle_row(copy.deepcopy(row), storage, deposit_cur, dry_run=False)

		revision_swhid = parse_swhid("swh:1:rev:4ea9917cdf53cd13534a042e4eb3787b86c834d2")

		assert storage.raw_extrinsic_metadata_get(
		MetadataTargetType.REVISION, revision_swhid, authority=PYPI_AUTHORITY,
		) == PagedResult(results=[], next_page_token=None,)
		assert storage.raw_extrinsic_metadata_get(
		MetadataTargetType.REVISION, revision_swhid, authority=SWH_AUTHORITY,
		) == PagedResult(
		results=[
		RawExtrinsicMetadata(
		type=MetadataTargetType.REVISION,
		id=revision_swhid,
		discovery_date=datetime.datetime(
		2014, 5, 7, 22, 3, tzinfo=datetime.timezone.utc,
		),
		authority=SWH_AUTHORITY,
		fetcher=FETCHER,
		format="original-artifacts-json",
		metadata=json.dumps(dest_original_artifacts).encode(),
		origin=origin_url,
		),
		],
		next_page_token=None,
		)