Differential D8753 Diff 31722 swh/loader/package/rpm/tests/test_rpm.py

Changeset View

Standalone View

swh/loader/package/rpm/tests/test_rpm.py

This file was added.

				import os
				anlambertUnsubmitted Not Done Inline Actions (C) 2022 anlambert: (C) 2022
				import tempfile

				import pytest

				from swh.loader.package.rpm.loader import RpmLoader, extract_rpm_package
				from swh.loader.package.utils import download
				from swh.loader.tests import assert_last_visit_matches, check_snapshot, get_stats
				from swh.model.hashutil import hash_to_bytes
				from swh.model.model import ObjectType, Release, Snapshot, SnapshotBranch, TargetType

				ORIGIN = "rpm://Fedora/packages/nginx"

				PACKAGES = {
				"34/Everything/1.18.0": {
				"name": "nginx",
				"version": "1.18.0",
				"release": 34,
				"buildTime": "2022-11-01T12:00:55.764371",
				"url": "https://archives.fedoraproject.org/nginx-1.18.0-5.fc34.src.rpm",
				}
				}


				def test_download_and_extract_rpm_package(requests_mock_datadir):
				rpm_url = "https://archives.fedoraproject.org/nginx-1.18.0-5.fc34.src.rpm"

				with tempfile.TemporaryDirectory() as tmpdir:
				rpm_path, _ = download(rpm_url, tmpdir)
				anlambertUnsubmitted Not Done Inline Actions `fedora34/everything/1.18.0-5` instead to match the lister output anlambert: `fedora34/everything/1.18.0-5` instead to match the lister output
				extract_rpm_package(rpm_path, tmpdir)

				anlambertUnsubmitted Not Done Inline Actions `1.18.0-5` to match the lister output anlambert: `1.18.0-5` to match the lister output
				assert os.path.exists(f"{tmpdir}/extracted/nginx.spec")
				assert os.path.exists(f"{tmpdir}/extracted/nginx-1.18.0/README")

				with open(f"{tmpdir}/extract.log", "r") as f:
				logs = f.read()
				assert logs.startswith("404.html\n")


				def test_extract_non_rpm_package(requests_mock_datadir):
				rpm_url = "https://archives.fedoraproject.org/nginx-1.18.0-5.fc34.src.rpm"

				with tempfile.TemporaryDirectory() as tmpdir:
				rpm_path, _ = download(rpm_url, tmpdir)
				extract_rpm_package(rpm_path, tmpdir)

				with pytest.raises(ValueError):
				extract_rpm_package(f"{tmpdir}/extracted/nginx.spec", tmpdir)


				def test_extract_non_existent_rpm_package():

				with tempfile.TemporaryDirectory() as tmpdir:
				with pytest.raises(FileNotFoundError) as e:
				extract_rpm_package(f"{tmpdir}/non-existent.src.rpm", tmpdir)
				assert f"RPM package {tmpdir}/non-existent.src.rpm not found" in str(e)
				anlambertUnsubmitted Not Done Inline Actions same here anlambert: same here


				def test_rpm_first_visit(swh_storage, requests_mock_datadir): #
				loader = RpmLoader(swh_storage, ORIGIN, packages=PACKAGES)

				actual_load_status = loader.load()
				expected_snapshot_id = "53868f8d97a1389f14f446602de4cbed2c563e21"
				assert actual_load_status == {
				"status": "eventful",
				"snapshot_id": expected_snapshot_id,
				}

				assert_last_visit_matches(
				swh_storage,
				ORIGIN,
				status="full",
				type="rpm",
				snapshot=hash_to_bytes(expected_snapshot_id),
				)

				release_id = hash_to_bytes("c5907200204f373a511c6d6b5ccb67585426b131")

				expected_snapshot = Snapshot(
				id=hash_to_bytes(expected_snapshot_id),
				branches={
				b"34/Everything/1.18.0": SnapshotBranch(
				target=release_id,
				target_type=TargetType.RELEASE,
				)
				},
				)

				check_snapshot(expected_snapshot, swh_storage)

				assert swh_storage.release_get([release_id])[0] == Release(
				id=release_id,
				name=b"nginx",
				message=(
				b"Synthetic release for Rpm source package "
				b"nginx version 34/Everything/1.18.0\n"
				),
				target=hash_to_bytes("454fa893059134281b76fc6502ca9f82bac7e4de"),
				target_type=ObjectType.DIRECTORY,
				synthetic=True,
				)

				stats = get_stats(swh_storage)
				assert {
				"content": 420,
				"directory": 39,
				"origin": 1,
				"origin_visit": 1,
				"release": 1,
				"revision": 0,
				"skipped_content": 0,
				"snapshot": 1,
				} == stats


				def test_rpm_multiple_visits(swh_storage, requests_mock_datadir):
				loader = RpmLoader(swh_storage, ORIGIN, packages=PACKAGES)

				actual_load_status = loader.load()
				expected_snapshot_id = "53868f8d97a1389f14f446602de4cbed2c563e21"
				assert actual_load_status == {
				"status": "eventful",
				"snapshot_id": expected_snapshot_id,
				}

				actual_load_status2 = loader.load()
				assert actual_load_status2 == {
				"status": "eventful",
				"snapshot_id": expected_snapshot_id,
				} # FIXME Shouldn't this be uneventful?

				assert_last_visit_matches(
				swh_storage,
				ORIGIN,
				status="full",
				KShivenduAuthorUnsubmitted Done Inline Actions The content hasn't changed so it should be uneventful, right? KShivendu: The content hasn't changed so it should be uneventful, right?
				type="rpm",
				snapshot=hash_to_bytes(expected_snapshot_id),
				)

				stats2 = get_stats(swh_storage)
				assert {
				"content": 420 + 0,
				"directory": 39 + 0,
				"origin": 1,
				"origin_visit": 1 + 1, # a new visit occurred
				"release": 1,
				"revision": 0,
				"skipped_content": 0,
				"snapshot": 1, # same snapshot across 2 visits
				} == stats2

				requested_urls = [
				m.url
				for m in requests_mock_datadir.request_history
				if m.url.startswith("https://archives.fedoraproject.org")
				]
				assert (
				requested_urls
				== ["https://archives.fedoraproject.org/nginx-1.18.0-5.fc34.src.rpm"] * 2
				)


				# TODO: maybe add test for kojipkgs .rpms as well?