Page MenuHomeSoftware Heritage

D8613.diff
No OneTemporary

D8613.diff

diff --git a/swh/loader/core/tests/test_tasks.py b/swh/loader/core/tests/test_tasks.py
--- a/swh/loader/core/tests/test_tasks.py
+++ b/swh/loader/core/tests/test_tasks.py
@@ -8,7 +8,6 @@
import pytest
from swh.scheduler.model import ListedOrigin, Lister
-from swh.scheduler.utils import create_origin_task_dict
NAMESPACE = "swh.loader.core"
@@ -19,16 +18,11 @@
@pytest.mark.parametrize("loader_name", ["Content", "Directory"])
-def test_tasks_loader_for_listed_origin(
- mocker,
- swh_scheduler_celery_app,
- swh_scheduler_celery_worker,
- swh_config,
+def test_loader_tasks_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
nixguix_lister,
loader_name,
):
- mock_load = mocker.patch(f"{NAMESPACE}.loader.{loader_name}Loader.load")
- mock_load.return_value = {"status": "eventful"}
listed_origin = ListedOrigin(
lister_id=nixguix_lister.id,
@@ -40,14 +34,9 @@
},
)
- task_dict = create_origin_task_dict(listed_origin, nixguix_lister)
-
- res = swh_scheduler_celery_app.send_task(
- f"{NAMESPACE}.tasks.Load{loader_name}",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.{loader_name}Loader",
+ task_function_name=f"{NAMESPACE}.tasks.Load{loader_name}",
+ lister=nixguix_lister,
+ listed_origin=listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/arch/tests/test_tasks.py b/swh/loader/package/arch/tests/test_tasks.py
--- a/swh/loader/package/arch/tests/test_tasks.py
+++ b/swh/loader/package/arch/tests/test_tasks.py
@@ -3,18 +3,28 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_arch_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
-):
- mock_load = mocker.patch("swh.loader.package.arch.loader.ArchLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.arch.tasks.LoadArch",
- kwargs=dict(
- url="some-url/packages/s/some-package",
- artifacts=[
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.arch"
+
+
+@pytest.fixture
+def arch_lister():
+ return Lister(name="arch", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def arch_listed_origin(arch_lister):
+ return ListedOrigin(
+ lister_id=arch_lister.id,
+ url="some-url/packages/s/some-package",
+ visit_type="arch",
+ extra_loader_arguments={
+ "artifacts": [
{
"version": "0.0.1",
"url": "https://somewhere/some-package-0.0.1.pkg.xz",
@@ -22,7 +32,7 @@
"length": 42,
}
],
- arch_metadata=[
+ "arch_metadata": [
{
"version": "0.0.1",
"arch": "aarch64",
@@ -31,10 +41,19 @@
"last_modified": "1970-01-01T21:08:14",
}
],
- ),
+ },
+ )
+
+
+def test_arch_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ arch_lister,
+ arch_listed_origin,
+):
+
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.ArchLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadArch",
+ lister=arch_lister,
+ listed_origin=arch_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/archive/tests/test_tasks.py b/swh/loader/package/archive/tests/test_tasks.py
--- a/swh/loader/package/archive/tests/test_tasks.py
+++ b/swh/loader/package/archive/tests/test_tasks.py
@@ -8,12 +8,8 @@
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
+NAMESPACE = "swh.loader.package.archive"
@pytest.fixture
@@ -28,65 +24,29 @@
url="https://example.org/archives",
visit_type="tar",
extra_loader_arguments={
- "artifacts": [],
+ "artifacts": [
+ {
+ "time": "2010-08-14T01:41:56",
+ "url": "https://example.org/archives/project-v1.0.0.tar.gz",
+ "filename": "project-v1.0.0.tar.gz",
+ "version": "1.0.0",
+ "length": 2500,
+ }
+ ],
"snapshot_append": True,
},
)
-def test_tasks_archive_loader(
- mocker,
- swh_scheduler_celery_app,
-):
- mock_load = mocker.patch("swh.loader.package.archive.loader.ArchiveLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.archive.tasks.LoadArchive",
- kwargs=dict(url="https://gnu.org/", artifacts=[]),
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_archive_loader_snapshot_append(
- mocker,
- swh_scheduler_celery_app,
-):
- mock_load = mocker.patch("swh.loader.package.archive.loader.ArchiveLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.archive.tasks.LoadArchive",
- kwargs=dict(url="https://gnu.org/", artifacts=[], snapshot_append=True),
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_archive_loader_for_listed_origin(
- mocker,
- swh_scheduler_celery_app,
+def test_archive_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
archive_lister,
archive_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.archive.loader.ArchiveLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(archive_listed_origin, archive_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.archive.tasks.LoadArchive",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.ArchiveLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadArchive",
+ lister=archive_lister,
+ listed_origin=archive_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/aur/tests/test_tasks.py b/swh/loader/package/aur/tests/test_tasks.py
--- a/swh/loader/package/aur/tests/test_tasks.py
+++ b/swh/loader/package/aur/tests/test_tasks.py
@@ -3,25 +3,35 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_aur_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
-):
- mock_load = mocker.patch("swh.loader.package.aur.loader.AurLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.aur.tasks.LoadAur",
- kwargs=dict(
- url="https://somewhere/some-package.git",
- artifacts=[
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.aur"
+
+
+@pytest.fixture
+def aur_lister():
+ return Lister(name="aur", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def aur_listed_origin(aur_lister):
+ return ListedOrigin(
+ lister_id=aur_lister.id,
+ url="https://somewhere/some-package.git",
+ visit_type="aur",
+ extra_loader_arguments={
+ "artifacts": [
{
"filename": "some-package.tar.gz",
"url": "https://somewhere/some-package.tar.gz",
"version": "0.0.1",
}
],
- aur_metadata=[
+ "aur_metadata": [
{
"version": "0.0.1",
"project_url": "https://somewhere/some-package",
@@ -29,10 +39,19 @@
"pkgname": "some-package",
}
],
- ),
+ },
+ )
+
+
+def test_aur_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ aur_lister,
+ aur_listed_origin,
+):
+
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.AurLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadAur",
+ lister=aur_lister,
+ listed_origin=aur_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/cpan/tests/test_tasks.py b/swh/loader/package/cpan/tests/test_tasks.py
--- a/swh/loader/package/cpan/tests/test_tasks.py
+++ b/swh/loader/package/cpan/tests/test_tasks.py
@@ -3,21 +3,38 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_cpan_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.cpan"
+
+
+@pytest.fixture
+def cpan_lister():
+ return Lister(name="cpan", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def cpan_listed_origin(cpan_lister):
+ return ListedOrigin(
+ lister_id=cpan_lister.id,
+ url="https://metacpan.org/dist/Software-Packager",
+ visit_type="cpan",
+ )
+
+
+def test_cpan_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ cpan_lister,
+ cpan_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.cpan.loader.CpanLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.cpan.tasks.LoadCpan",
- kwargs=dict(
- url="some-url/api/packages/some-package",
- ),
+
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.CpanLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadCpan",
+ lister=cpan_lister,
+ listed_origin=cpan_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/cran/tests/test_tasks.py b/swh/loader/package/cran/tests/test_tasks.py
--- a/swh/loader/package/cran/tests/test_tasks.py
+++ b/swh/loader/package/cran/tests/test_tasks.py
@@ -8,17 +8,13 @@
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
+NAMESPACE = "swh.loader.package.cran"
@pytest.fixture
def cran_lister():
- return Lister(name="cran-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="cran", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -33,44 +29,15 @@
)
-def test_tasks_cran_loader(
- mocker,
- swh_scheduler_celery_app,
-):
- mock_load = mocker.patch("swh.loader.package.cran.loader.CRANLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.cran.tasks.LoadCRAN",
- kwargs=dict(
- url="some-url",
- artifacts=[{"version": "1.2.3", "url": "artifact-url"}],
- ),
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_cran_loader_for_listed_origin(
- mocker,
- swh_scheduler_celery_app,
+def test_cran_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
cran_lister,
cran_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.cran.loader.CRANLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(cran_listed_origin, cran_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.cran.tasks.LoadCRAN",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.CRANLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadCRAN",
+ lister=cran_lister,
+ listed_origin=cran_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/crates/tests/test_tasks.py b/swh/loader/package/crates/tests/test_tasks.py
--- a/swh/loader/package/crates/tests/test_tasks.py
+++ b/swh/loader/package/crates/tests/test_tasks.py
@@ -3,22 +3,41 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_crates_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.crates"
+
+
+@pytest.fixture
+def crates_lister():
+ return Lister(name="crates", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def crates_listed_origin(crates_lister):
+ return ListedOrigin(
+ lister_id=crates_lister.id,
+ url="some-url/api/v1/crates/some-package",
+ visit_type="crates",
+ extra_loader_arguments={
+ "artifacts": [{"version": "0.0.1", "url": "some-package-0.0.1.crate"}],
+ },
+ )
+
+
+def test_crates_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ crates_lister,
+ crates_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.crates.loader.CratesLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.crates.tasks.LoadCrates",
- kwargs=dict(
- url="some-url/api/v1/crates/some-package",
- artifacts=[{"version": "0.0.1", "url": "some-package-0.0.1.crate"}],
- ),
+
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.CratesLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadCrates",
+ lister=crates_lister,
+ listed_origin=crates_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/debian/tests/test_tasks.py b/swh/loader/package/debian/tests/test_tasks.py
--- a/swh/loader/package/debian/tests/test_tasks.py
+++ b/swh/loader/package/debian/tests/test_tasks.py
@@ -8,17 +8,13 @@
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
+NAMESPACE = "swh.loader.package.debian"
@pytest.fixture
def debian_lister():
- return Lister(name="debian-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="debian", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -31,35 +27,15 @@
)
-def test_tasks_debian_loader(mocker, swh_scheduler_celery_app):
- mock_load = mocker.patch("swh.loader.package.debian.loader.DebianLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.debian.tasks.LoadDebian",
- 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_debian_loader_for_listed_origin(
- mocker, swh_scheduler_celery_app, debian_lister, debian_listed_origin
+def test_debian_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ debian_lister,
+ debian_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.debian.loader.DebianLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(debian_listed_origin, debian_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.debian.tasks.LoadDebian",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.DebianLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadDebian",
+ lister=debian_lister,
+ listed_origin=debian_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/deposit/tests/test_tasks.py b/swh/loader/package/deposit/tests/test_tasks.py
--- a/swh/loader/package/deposit/tests/test_tasks.py
+++ b/swh/loader/package/deposit/tests/test_tasks.py
@@ -8,17 +8,13 @@
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
+NAMESPACE = "swh.loader.package.deposit"
@pytest.fixture
def deposit_lister():
- return Lister(name="deposit-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="deposit", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -31,50 +27,15 @@
)
-def test_tasks_deposit_loader(
- mocker,
- swh_scheduler_celery_app,
-):
- mock_loader = mocker.patch(
- "swh.loader.package.deposit.loader.DepositLoader.from_configfile"
- )
- mock_loader.return_value = mock_loader
- mock_loader.load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.deposit.tasks.LoadDeposit",
- kwargs=dict(
- url="some-url",
- deposit_id="some-d-id",
- ),
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_loader.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_deposit_loader_for_listed_origin(
- mocker,
- swh_scheduler_celery_app,
+def test_deposit_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
deposit_lister,
deposit_listed_origin,
):
- mock_loader = mocker.patch(
- "swh.loader.package.deposit.loader.DepositLoader.from_configfile"
- )
- mock_loader.return_value = mock_loader
- mock_loader.load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(deposit_listed_origin, deposit_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.deposit.tasks.LoadDeposit",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.DepositLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadDeposit",
+ lister=deposit_lister,
+ listed_origin=deposit_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_loader.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/golang/tests/test_tasks.py b/swh/loader/package/golang/tests/test_tasks.py
--- a/swh/loader/package/golang/tests/test_tasks.py
+++ b/swh/loader/package/golang/tests/test_tasks.py
@@ -3,19 +3,38 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_golang_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.golang"
+
+
+@pytest.fixture
+def golang_lister():
+ return Lister(name="golang", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def golang_listed_origin(golang_lister):
+ return ListedOrigin(
+ lister_id=golang_lister.id,
+ url="https://pkg.go.dev/golang.org/whatever/package",
+ visit_type="golang",
+ )
+
+
+def test_golang_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ golang_lister,
+ golang_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.golang.loader.GolangLoader.load")
- mock_load.return_value = {"status": "eventful"}
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.golang.tasks.LoadGolang",
- kwargs={"url": "https://pkg.go.dev/golang.org/whatever/package"},
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.GolangLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadGolang",
+ lister=golang_lister,
+ listed_origin=golang_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/maven/tests/test_tasks.py b/swh/loader/package/maven/tests/test_tasks.py
--- a/swh/loader/package/maven/tests/test_tasks.py
+++ b/swh/loader/package/maven/tests/test_tasks.py
@@ -8,7 +8,8 @@
import pytest
from swh.scheduler.model import ListedOrigin, Lister
-from swh.scheduler.utils import create_origin_task_dict
+
+NAMESPACE = "swh.loader.package.maven"
MVN_ARTIFACTS = [
{
@@ -24,14 +25,9 @@
]
-@pytest.fixture(autouse=True)
-def celery_worker_and_swh_config(swh_scheduler_celery_worker, swh_config):
- pass
-
-
@pytest.fixture
def maven_lister():
- return Lister(name="maven-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="maven", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -46,41 +42,15 @@
)
-def test_tasks_maven_loader(
- mocker,
- swh_scheduler_celery_app,
+def test_maven_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ maven_lister,
+ maven_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.maven.loader.MavenLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.maven.tasks.LoadMaven",
- kwargs=dict(
- url=MVN_ARTIFACTS[0]["url"],
- artifacts=MVN_ARTIFACTS,
- ),
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_maven_loader_for_listed_origin(
- mocker, swh_scheduler_celery_app, maven_lister, maven_listed_origin
-):
- mock_load = mocker.patch("swh.loader.package.maven.loader.MavenLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(maven_listed_origin, maven_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.maven.tasks.LoadMaven",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.MavenLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadMaven",
+ lister=maven_lister,
+ listed_origin=maven_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/nixguix/tests/test_tasks.py b/swh/loader/package/nixguix/tests/test_tasks.py
--- a/swh/loader/package/nixguix/tests/test_tasks.py
+++ b/swh/loader/package/nixguix/tests/test_tasks.py
@@ -8,17 +8,13 @@
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
+NAMESPACE = "swh.loader.package.nixguix"
@pytest.fixture
def nixguix_lister():
- return Lister(name="nixguix-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="nixguix", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -30,43 +26,15 @@
)
-def test_tasks_nixguix_loader(
- mocker,
- swh_scheduler_celery_app,
+def test_nixguix_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ nixguix_lister,
+ nixguix_listed_origin,
):
- mock_loader = mocker.patch(
- "swh.loader.package.nixguix.loader.NixGuixLoader.from_configfile"
- )
- mock_loader.return_value = mock_loader
- mock_loader.load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.nixguix.tasks.LoadNixguix", kwargs=dict(url="some-url")
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_loader.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_nixguix_loader_for_listed_origin(
- mocker, swh_scheduler_celery_app, nixguix_lister, nixguix_listed_origin
-):
- mock_loader = mocker.patch(
- "swh.loader.package.nixguix.loader.NixGuixLoader.from_configfile"
- )
- mock_loader.return_value = mock_loader
- mock_loader.load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(nixguix_listed_origin, nixguix_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.nixguix.tasks.LoadNixguix",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.NixGuixLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadNixguix",
+ lister=nixguix_lister,
+ listed_origin=nixguix_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_loader.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/npm/tests/test_tasks.py b/swh/loader/package/npm/tests/test_tasks.py
--- a/swh/loader/package/npm/tests/test_tasks.py
+++ b/swh/loader/package/npm/tests/test_tasks.py
@@ -8,17 +8,13 @@
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
+NAMESPACE = "swh.loader.package.npm"
@pytest.fixture
def npm_lister():
- return Lister(name="npm-lister", instance_name="npm", id=uuid.uuid4())
+ return Lister(name="npm", instance_name="npm", id=uuid.uuid4())
@pytest.fixture
@@ -30,38 +26,15 @@
)
-def test_tasks_npm_loader(
- mocker,
- swh_scheduler_celery_app,
+def test_npm_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ npm_lister,
+ npm_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.npm.loader.NpmLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.npm.tasks.LoadNpm",
- kwargs=dict(url="https://www.npmjs.com/package/some-package"),
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_npm_loader_for_listed_origin(
- mocker, swh_scheduler_celery_app, npm_lister, npm_listed_origin
-):
- mock_load = mocker.patch("swh.loader.package.npm.loader.NpmLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(npm_listed_origin, npm_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.npm.tasks.LoadNpm",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.NpmLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadNpm",
+ lister=npm_lister,
+ listed_origin=npm_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/opam/tests/test_tasks.py b/swh/loader/package/opam/tests/test_tasks.py
--- a/swh/loader/package/opam/tests/test_tasks.py
+++ b/swh/loader/package/opam/tests/test_tasks.py
@@ -8,7 +8,8 @@
import pytest
from swh.scheduler.model import ListedOrigin, Lister
-from swh.scheduler.utils import create_origin_task_dict
+
+NAMESPACE = "swh.loader.package.opam"
OPAM_LOADER_ARGS = {
"url": "opam+https://opam.ocaml.org/packages/agrid",
@@ -19,14 +20,9 @@
}
-@pytest.fixture(autouse=True)
-def celery_worker_and_swh_config(swh_scheduler_celery_worker, swh_config):
- pass
-
-
@pytest.fixture
def opam_lister():
- return Lister(name="opam-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="opam", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -41,38 +37,15 @@
)
-def test_tasks_opam_loader(
- mocker,
- swh_scheduler_celery_app,
+def test_opam_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ opam_lister,
+ opam_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.opam.loader.OpamLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.opam.tasks.LoadOpam",
- kwargs=OPAM_LOADER_ARGS,
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_opam_loader_for_listed_origin(
- mocker, swh_scheduler_celery_app, opam_lister, opam_listed_origin
-):
- mock_load = mocker.patch("swh.loader.package.opam.loader.OpamLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(opam_listed_origin, opam_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.opam.tasks.LoadOpam",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.OpamLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadOpam",
+ lister=opam_lister,
+ listed_origin=opam_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/pubdev/tests/test_tasks.py b/swh/loader/package/pubdev/tests/test_tasks.py
--- a/swh/loader/package/pubdev/tests/test_tasks.py
+++ b/swh/loader/package/pubdev/tests/test_tasks.py
@@ -3,21 +3,38 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_pubdev_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.pubdev"
+
+
+@pytest.fixture
+def pubdev_lister():
+ return Lister(name="pubdev", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def pubdev_listed_origin(pubdev_lister):
+ return ListedOrigin(
+ lister_id=pubdev_lister.id,
+ url="https://pub.dev/packages/some-package",
+ visit_type="pubdev",
+ )
+
+
+def test_pubdev_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ pubdev_lister,
+ pubdev_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.pubdev.loader.PubDevLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.pubdev.tasks.LoadPubDev",
- kwargs=dict(
- url="https://pub.dev/packages/some-package",
- ),
+
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.PubDevLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadPubDev",
+ lister=pubdev_lister,
+ listed_origin=pubdev_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/puppet/tests/test_tasks.py b/swh/loader/package/puppet/tests/test_tasks.py
--- a/swh/loader/package/puppet/tests/test_tasks.py
+++ b/swh/loader/package/puppet/tests/test_tasks.py
@@ -3,29 +3,48 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+import uuid
-def test_tasks_puppet_loader(
- mocker, swh_scheduler_celery_app, swh_scheduler_celery_worker, swh_config
-):
- mock_load = mocker.patch("swh.loader.package.puppet.loader.PuppetLoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.puppet.tasks.LoadPuppet",
- kwargs=dict(
- url="some-url/api/packages/some-package",
- artifacts={
+import pytest
+
+from swh.scheduler.model import ListedOrigin, Lister
+
+NAMESPACE = "swh.loader.package.puppet"
+
+
+@pytest.fixture
+def puppet_lister():
+ return Lister(name="puppet", instance_name="example", id=uuid.uuid4())
+
+
+@pytest.fixture
+def puppet_listed_origin(puppet_lister):
+ return ListedOrigin(
+ lister_id=puppet_lister.id,
+ url="some-url/api/packages/some-package",
+ visit_type="aur",
+ extra_loader_arguments={
+ "artifacts": {
"1.0.0": {
"url": "https://domain/some-package-1.0.0.tar.gz",
"version": "1.0.0",
"filename": "some-module-1.0.0.tar.gz",
"last_update": "2011-11-20T13:40:30-08:00",
},
- },
- ),
+ }
+ },
+ )
+
+
+def test_puppet_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
+ puppet_lister,
+ puppet_listed_origin,
+):
+
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.PuppetLoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadPuppet",
+ lister=puppet_lister,
+ listed_origin=puppet_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/package/pypi/tests/test_tasks.py b/swh/loader/package/pypi/tests/test_tasks.py
--- a/swh/loader/package/pypi/tests/test_tasks.py
+++ b/swh/loader/package/pypi/tests/test_tasks.py
@@ -8,17 +8,13 @@
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
+NAMESPACE = "swh.loader.package.pypi"
@pytest.fixture
def pypi_lister():
- return Lister(name="pypi-lister", instance_name="example", id=uuid.uuid4())
+ return Lister(name="pypi", instance_name="example", id=uuid.uuid4())
@pytest.fixture
@@ -30,40 +26,15 @@
)
-def test_tasks_pypi_loader(
- mocker,
- swh_scheduler_celery_app,
-):
- mock_load = mocker.patch("swh.loader.package.pypi.loader.PyPILoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.pypi.tasks.LoadPyPI", kwargs=dict(url="some-url")
- )
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
-
-
-def test_tasks_pypi_loader_for_listed_origin(
- mocker,
- swh_scheduler_celery_app,
+def test_pypi_loader_task_for_listed_origin(
+ loading_task_creation_for_listed_origin_test,
pypi_lister,
pypi_listed_origin,
):
- mock_load = mocker.patch("swh.loader.package.pypi.loader.PyPILoader.load")
- mock_load.return_value = {"status": "eventful"}
-
- task_dict = create_origin_task_dict(pypi_listed_origin, pypi_lister)
- res = swh_scheduler_celery_app.send_task(
- "swh.loader.package.pypi.tasks.LoadPyPI",
- kwargs=task_dict["arguments"]["kwargs"],
+ loading_task_creation_for_listed_origin_test(
+ loader_class_name=f"{NAMESPACE}.loader.PyPILoader",
+ task_function_name=f"{NAMESPACE}.tasks.LoadPyPI",
+ lister=pypi_lister,
+ listed_origin=pypi_listed_origin,
)
- assert res
- res.wait()
- assert res.successful()
- assert mock_load.called
- assert res.result == {"status": "eventful"}
diff --git a/swh/loader/pytest_plugin.py b/swh/loader/pytest_plugin.py
--- a/swh/loader/pytest_plugin.py
+++ b/swh/loader/pytest_plugin.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2019-2021 The Software Heritage developers
+# Copyright (C) 2019-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
@@ -9,6 +9,9 @@
import pytest
import yaml
+from swh.scheduler.model import ListedOrigin, Lister
+from swh.scheduler.utils import create_origin_task_dict
+
@pytest.fixture
def swh_storage_backend_config(swh_storage_postgresql) -> Dict[str, Any]:
@@ -52,3 +55,34 @@
"""
os.environ["http_proxy"] = "http://localhost:999"
os.environ["https_proxy"] = "http://localhost:999"
+
+
+@pytest.fixture
+def loading_task_creation_for_listed_origin_test(
+ mocker,
+ swh_scheduler_celery_app,
+ swh_scheduler_celery_worker,
+ swh_config,
+):
+ def test_implementation(
+ loader_class_name: str,
+ task_function_name: str,
+ lister: Lister,
+ listed_origin: ListedOrigin,
+ ):
+ mock_load = mocker.patch(f"{loader_class_name}.load")
+ mock_load.return_value = {"status": "eventful"}
+
+ task_dict = create_origin_task_dict(listed_origin, lister)
+
+ res = swh_scheduler_celery_app.send_task(
+ task_function_name,
+ kwargs=task_dict["arguments"]["kwargs"],
+ )
+ assert res
+ res.wait()
+ assert res.successful()
+ assert mock_load.called
+ assert res.result == {"status": "eventful"}
+
+ return test_implementation
diff --git a/swh/loader/tests/conftest.py b/swh/loader/tests/conftest.py
--- a/swh/loader/tests/conftest.py
+++ b/swh/loader/tests/conftest.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2019-2021 The Software Heritage developers
+# Copyright (C) 2019-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

File Metadata

Mime Type
text/plain
Expires
Thu, Dec 19, 6:38 AM (5 d, 45 m ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3224422

Event Timeline