Page MenuHomeSoftware Heritage
Differential D8753 Diff 31662 swh/loader/package/rpm/tests/test_tasks.py

Changeset View

Standalone View

View Options

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

  • This file was added.
import uuid
import pytest
from swh.scheduler.model import ListedOrigin, Lister
from swh.scheduler.utils import create_origin_task_dict
@pytest.fixture(autouse=True)
def celery_worker_and_swh_config(swh_scheduler_celery_worker, swh_config):
pass
@pytest.fixture
def rpm_lister():
return Lister(name="rpm-lister", instance_name="example", id=uuid.uuid4())
@pytest.fixture
def rpm_listed_origin(rpm_lister):
return ListedOrigin(
lister_id=rpm_lister.id,
url="https://archives.fedoraproject.org/package",
visit_type="rpm",
extra_loader_arguments={"packages": {}},
)
def test_tasks_rpm_loader(mocker, swh_scheduler_celery_app):
mock_load = mocker.patch("swh.loader.package.rpm.loader.RpmLoader.load")
mock_load.return_value = {"status": "eventful"}
res = swh_scheduler_celery_app.send_task(
"swh.loader.package.rpm.tasks.LoadRpm",
kwargs=dict(url="some-url", packages={}),
)
assert res
res.wait()
assert res.successful()
assert mock_load.called
assert res.result == {"status": "eventful"}
def test_tasks_rpm_loader_for_listed_origin(
mocker, swh_scheduler_celery_app, rpm_lister, rpm_listed_origin
):
mock_load = mocker.patch("swh.loader.package.rpm.loader.RpmLoader.load")
mock_load.return_value = {"status": "eventful"}
task_dict = create_origin_task_dict(rpm_listed_origin, rpm_lister)
res = swh_scheduler_celery_app.send_task(
"swh.loader.package.rpm.tasks.LoadRpm",
kwargs=task_dict["arguments"]["kwargs"],
)
assert res
res.wait()
assert res.successful()
assert mock_load.called
assert res.result == {"status": "eventful"}
anlambertUnsubmitted
Not Done
Inline Actions

Use this instead to test the celery task for the rpm loader:

# Copyright (C) 2022  The Software Heritage developers
# 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 uuid

import pytest

from swh.scheduler.model import ListedOrigin, Lister

NAMESPACE = "swh.loader.package.rpm"

RPM_ORIGIN_URL = "https://src.fedoraproject.org/rpms/0xFFFF"

RPM_PACKAGES = {
    "fedora36/everything/0.10-4": {
        "name": "0xFFFF",
        "version": "0.10-4",
        "release": 36,
        "edition": "Everything",
        "buildTime": "2022-01-19T19:13:53+00:00",
        "url": (
            "https://archives.fedoraproject.org/pub/archive/fedora/linux/releases/"
            "36/Everything/source/tree/Packages/0/0xFFFF-0.10-4.fc36.src.rpm"
        ),
        "checksums": {
            "sha256": "45eee8d990d502324ae665233c320b8a5469c25d735f1862e094c1878d6ff2cd"
        },
    }
}


@pytest.fixture
def fedora_lister():
    return Lister(name="fedora", instance_name="fedora", id=uuid.uuid4())


@pytest.fixture
def fedora_listed_origin(fedora_lister):
    return ListedOrigin(
        lister_id=fedora_lister.id,
        url=RPM_ORIGIN_URL,
        visit_type="rpm",
        extra_loader_arguments={
            "packages": RPM_PACKAGES,
        },
    )


def test_rpm_loader_task_for_listed_origin(
    loading_task_creation_for_listed_origin_test,
    fedora_lister,
    fedora_listed_origin,
):

    loading_task_creation_for_listed_origin_test(
        loader_class_name=f"{NAMESPACE}.loader.RpmLoader",
        task_function_name=f"{NAMESPACE}.tasks.LoadRpm",
        lister=fedora_lister,
        listed_origin=fedora_listed_origin,
    )
anlambert: Use this instead to test the celery task for the rpm loader: ```lang=python # Copyright (C)…
About · Developer information · Legal