Differential D3679 Diff 12939 swh/loader/package/npm/tests/test_npm.py

Changeset View

Standalone View

swh/loader/package/npm/tests/test_npm.py

# Copyright (C) 2019-2020 The Software Heritage developers		# Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import json		import json
import os		import os
import pytest		import pytest

from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes, hash_to_hex
from swh.model.model import Person, Snapshot, SnapshotBranch, TargetType		from swh.model.identifiers import SWHID
		from swh.model.model import (
		MetadataAuthority,
		MetadataAuthorityType,
		MetadataFetcher,
		MetadataTargetType,
		Person,
		RawExtrinsicMetadata,
		Snapshot,
		SnapshotBranch,
		TargetType,
		)

		from swh.storage.interface import PagedResult

		from swh.loader.package import __version__
from swh.loader.package.npm.loader import (		from swh.loader.package.npm.loader import (
_author_str,		_author_str,
NpmLoader,		NpmLoader,
extract_npm_package_author,		extract_npm_package_author,
artifact_to_revision_id,		artifact_to_revision_id,
)		)
from swh.loader.package.tests.common import check_metadata_paths		from swh.loader.package.tests.common import check_metadata_paths
from swh.loader.tests import (		from swh.loader.tests import (
assert_last_visit_matches,		assert_last_visit_matches,
check_snapshot,		check_snapshot,
get_stats,		get_stats,
)		)


		@pytest.fixture
		def org_api_info(datadir) -> bytes:
		with open(os.path.join(datadir, "https_replicate.npmjs.com", "org"), "rb",) as f:
		return f.read()


def test_npm_author_str():		def test_npm_author_str():
for author, expected_author in [		for author, expected_author in [
("author", "author"),		("author", "author"),
(		(
["Al from quantum leap", "hal from 2001 space odyssey"],		["Al from quantum leap", "hal from 2001 space odyssey"],
"Al from quantum leap",		"Al from quantum leap",
),		),
([], ""),		([], ""),
▲ Show 20 Lines • Show All 282 Lines • ▼ Show 20 Lines	def test_revision_metadata_structure(swh_config, requests_mock_datadir):

for original_artifact in revision["metadata"]["original_artifact"]:		for original_artifact in revision["metadata"]["original_artifact"]:
check_metadata_paths(		check_metadata_paths(
original_artifact,		original_artifact,
paths=[("filename", str), ("length", int), ("checksums", dict),],		paths=[("filename", str), ("length", int), ("checksums", dict),],
)		)


def test_npm_loader_first_visit(swh_config, requests_mock_datadir):		def test_npm_loader_first_visit(swh_config, requests_mock_datadir, org_api_info):
package = "org"		package = "org"
url = package_url(package)		url = package_url(package)
loader = NpmLoader(url)		loader = NpmLoader(url)

actual_load_status = loader.load()		actual_load_status = loader.load()
expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d")		expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d")
assert actual_load_status == {		assert actual_load_status == {
"status": "eventful",		"status": "eventful",
▲ Show 20 Lines • Show All 48 Lines • ▼ Show 20 Lines	expected_snapshot = Snapshot(
b"releases/0.0.4": SnapshotBranch(		b"releases/0.0.4": SnapshotBranch(
target=hash_to_bytes("ba019b192bdb94bd0b5bd68b3a5f92b5acc2239a"),		target=hash_to_bytes("ba019b192bdb94bd0b5bd68b3a5f92b5acc2239a"),
target_type=TargetType.REVISION,		target_type=TargetType.REVISION,
),		),
},		},
)		)
check_snapshot(expected_snapshot, loader.storage)		check_snapshot(expected_snapshot, loader.storage)

		snapshot_swhid = SWHID(
		object_type="snapshot", object_id=hash_to_hex(expected_snapshot_id)
		)
		metadata_authority = MetadataAuthority(
		type=MetadataAuthorityType.FORGE, url="https://npmjs.com/",
		)
		expected_metadata = [
		RawExtrinsicMetadata(
		type=MetadataTargetType.SNAPSHOT,
		id=snapshot_swhid,
		authority=metadata_authority,
		fetcher=MetadataFetcher(
		name="swh.loader.package.npm.loader.NpmLoader", version=__version__,
		),
		discovery_date=loader.visit_date,
		format="replicate-npm-package-json",
		metadata=org_api_info,
		origin="https://www.npmjs.com/package/org",
		)
		]
		assert loader.storage.raw_extrinsic_metadata_get(
		type=MetadataTargetType.SNAPSHOT,
		id=snapshot_swhid,
		authority=metadata_authority,
		) == PagedResult(next_page_token=None, results=expected_metadata,)


def test_npm_loader_incremental_visit(swh_config, requests_mock_datadir_visits):		def test_npm_loader_incremental_visit(swh_config, requests_mock_datadir_visits):
package = "org"		package = "org"
url = package_url(package)		url = package_url(package)
loader = NpmLoader(url)		loader = NpmLoader(url)

expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d")		expected_snapshot_id = hash_to_bytes("d0587e1195aed5a8800411a008f2f2d627f18e2d")
actual_load_status = loader.load()		actual_load_status = loader.load()
Show All 14 Lines	assert {
"origin_visit": 1,		"origin_visit": 1,
"person": 2,		"person": 2,
"release": 0,		"release": 0,
"revision": len(_expected_new_revisions_first_visit),		"revision": len(_expected_new_revisions_first_visit),
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 1,		"snapshot": 1,
} == stats		} == stats

loader._info = None # reset loader internal state		# reset loader internal state
		del loader._cached_info
		del loader._cached__raw_info

actual_load_status2 = loader.load()		actual_load_status2 = loader.load()
assert actual_load_status2["status"] == "eventful"		assert actual_load_status2["status"] == "eventful"
snap_id2 = actual_load_status2["snapshot_id"]		snap_id2 = actual_load_status2["snapshot_id"]
assert snap_id2 is not None		assert snap_id2 is not None
assert snap_id2 != actual_load_status["snapshot_id"]		assert snap_id2 != actual_load_status["snapshot_id"]

assert_last_visit_matches(loader.storage, url, status="full", type="npm")		assert_last_visit_matches(loader.storage, url, status="full", type="npm")

▲ Show 20 Lines • Show All 231 Lines • Show Last 20 Lines