Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/rpm/tests/test_rpm.py
- This file was added.
# Copyright (C) 2019-2021 The Software Heritage developers | |||||
anlambert: (C) 2022 | |||||
# See the AUTHORS file at the top-level directory of this distribution | |||||
# License: GNU General Public License version 3, or any later version | |||||
# See top-level LICENSE file for more information | |||||
import os | |||||
import tempfile | |||||
import pytest | |||||
from swh.loader.package.rpm.loader import RpmLoader, extract_rpm_package | |||||
from swh.loader.package.utils import EMPTY_AUTHOR, 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, | |||||
TimestampWithTimezone, | |||||
) | |||||
ORIGIN = "rpm://Fedora/packages/nginx" | |||||
RPM_URL = "https://archives.fedoraproject.org/nginx-1.18.0-5.fc34.src.rpm" | |||||
PACKAGES = { | |||||
"34/Everything/1.18.0": { | |||||
Not Done Inline Actionsfedora34/everything/1.18.0-5 instead to match the lister output anlambert: `fedora34/everything/1.18.0-5` instead to match the lister output | |||||
"name": "nginx", | |||||
"version": "1.18.0", | |||||
Not Done Inline Actions1.18.0-5 to match the lister output anlambert: `1.18.0-5` to match the lister output | |||||
"release": 34, | |||||
"edition": "Everything", | |||||
"buildTime": "2022-11-01T12:00:55+00:00", | |||||
"url": RPM_URL, | |||||
} | |||||
} | |||||
NEW_PACKAGES = { | |||||
**PACKAGES, | |||||
"35/Everything/1.20.0": { | |||||
# using the same .rpm file but for a new branch | |||||
"name": "nginx", | |||||
"version": "1.20.0", | |||||
"release": 35, | |||||
"edition": "Everything", | |||||
"buildTime": "2022-11-01T12:00:55+00:00", | |||||
"url": RPM_URL, | |||||
}, | |||||
} | |||||
@pytest.fixture() | |||||
def expected_stats(): | |||||
return { | |||||
"content": 420, | |||||
Not Done Inline Actionssame here anlambert: same here | |||||
"directory": 39, | |||||
"origin": 1, | |||||
"origin_visit": 1, | |||||
"release": 1, | |||||
"revision": 0, | |||||
"skipped_content": 0, | |||||
"snapshot": 1, | |||||
} | |||||
snapshot_id = "75b5c5cbab49f3e732817dad625aa5d7b86ce0c5" | |||||
release_id = hash_to_bytes("4391fd5d78a90635737171dc43d591bffd77922a") | |||||
snapshot = Snapshot( | |||||
id=hash_to_bytes(snapshot_id), | |||||
branches={ | |||||
b"34/Everything/1.18.0": SnapshotBranch( | |||||
target=release_id, | |||||
target_type=TargetType.RELEASE, | |||||
) | |||||
}, | |||||
) | |||||
release = Release( | |||||
id=release_id, | |||||
name=b"nginx", | |||||
author=EMPTY_AUTHOR, | |||||
date=TimestampWithTimezone.from_iso8601("2022-11-01T12:00:55+00:00"), | |||||
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, | |||||
) | |||||
new_snapshot_id = "2057cadc0bd95220f6208904bc1f5d8ad98c39d7" | |||||
new_release_id = hash_to_bytes("ec87e339c7514e73a871a2e2dfbfcf5dc6a89ae4") | |||||
new_snapshot = Snapshot( | |||||
id=hash_to_bytes(new_snapshot_id), | |||||
branches={ | |||||
b"34/Everything/1.18.0": SnapshotBranch( | |||||
target=release_id, | |||||
target_type=TargetType.RELEASE, | |||||
), | |||||
b"35/Everything/1.20.0": SnapshotBranch( | |||||
target=new_release_id, | |||||
target_type=TargetType.RELEASE, | |||||
), | |||||
}, | |||||
) | |||||
new_release = Release( | |||||
id=new_release_id, | |||||
name=b"nginx", | |||||
author=EMPTY_AUTHOR, | |||||
date=TimestampWithTimezone.from_iso8601("2022-11-01T12:00:55+00:00"), | |||||
message=( | |||||
b"Synthetic release for Rpm source package " | |||||
b"nginx version 35/Everything/1.20.0\n" | |||||
), | |||||
target=hash_to_bytes("454fa893059134281b76fc6502ca9f82bac7e4de"), | |||||
target_type=ObjectType.DIRECTORY, | |||||
synthetic=True, | |||||
) | |||||
def test_download_and_extract_rpm_package(requests_mock_datadir): | |||||
rpm_url = RPM_URL | |||||
with tempfile.TemporaryDirectory() as tmpdir: | |||||
rpm_path, _ = download(rpm_url, tmpdir) | |||||
extract_rpm_package(rpm_path, tmpdir) | |||||
# .spec should get extracted from .rpm | |||||
assert os.path.exists(f"{tmpdir}/extracted/nginx.spec") | |||||
# README should get extracted from .tar (which itself is extracted from .rpm) | |||||
assert os.path.exists(f"{tmpdir}/extracted/nginx-1.18.0/README") | |||||
Done Inline ActionsThe content hasn't changed so it should be uneventful, right? KShivendu: The content hasn't changed so it should be uneventful, right? | |||||
with open(f"{tmpdir}/extract.log", "r") as f: | |||||
logs = f.read() | |||||
assert logs.startswith("404.html") | |||||
def test_extract_non_rpm_package(requests_mock_datadir): | |||||
rpm_url = RPM_URL | |||||
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) | |||||
def assert_stored(swh_storage, release: Release, snapshot: Snapshot, stats: dict): | |||||
assert_last_visit_matches( | |||||
swh_storage, | |||||
ORIGIN, | |||||
status="full", | |||||
type="rpm", | |||||
snapshot=hash_to_bytes(snapshot.id), | |||||
) | |||||
check_snapshot(snapshot, swh_storage) | |||||
assert swh_storage.release_get([release.id])[0] == release | |||||
assert get_stats(swh_storage) == stats | |||||
def test_rpm_first_visit(swh_storage, requests_mock_datadir, expected_stats): | |||||
loader = RpmLoader(swh_storage, ORIGIN, packages=PACKAGES) | |||||
actual_load_status = loader.load() | |||||
assert actual_load_status == {"status": "eventful", "snapshot_id": snapshot_id} | |||||
assert [m.url for m in requests_mock_datadir.request_history] == [RPM_URL] | |||||
assert_stored(swh_storage, release, snapshot, expected_stats) | |||||
def test_rpm_multiple_visits(swh_storage, requests_mock_datadir, expected_stats): | |||||
loader = RpmLoader(swh_storage, ORIGIN, packages=PACKAGES) | |||||
# First run: Discovered exactly 1 package | |||||
load_status = loader.load() | |||||
assert load_status == {"status": "eventful", "snapshot_id": snapshot_id} | |||||
# Second run: No updates | |||||
load_status = loader.load() | |||||
expected_stats["origin_visit"] += 1 # a new visit occurred but no new snapshot | |||||
assert load_status == {"status": "uneventful", "snapshot_id": snapshot_id} | |||||
assert [m.url for m in requests_mock_datadir.request_history] == [RPM_URL] | |||||
assert_stored(swh_storage, release, snapshot, expected_stats) | |||||
# Third run: New release (Updated snapshot) | |||||
loader.packages = NEW_PACKAGES | |||||
load_status = loader.load() | |||||
expected_stats["origin_visit"] += 1 # same rpm:// origin | |||||
expected_stats["release"] += 1 # new release (35/Everything/1.20.0) | |||||
expected_stats["snapshot"] += 1 # updated metadata (`packages` param) | |||||
assert load_status == {"status": "eventful", "snapshot_id": new_snapshot_id} | |||||
assert [m.url for m in requests_mock_datadir.request_history] == [RPM_URL, RPM_URL] | |||||
assert_stored(swh_storage, new_release, new_snapshot, expected_stats) |
(C) 2022