Page Menu
Home
Software Heritage
Search
Configure Global Search
Log In
Files
F7123471
D8613.id31110.diff
No One
Temporary
Actions
View File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Flag For Later
Size
39 KB
Subscribers
None
D8613.id31110.diff
View Options
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
Details
Attached
Mime Type
text/plain
Expires
Thu, Dec 19, 9:30 AM (15 h, 47 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3224422
Attached To
D8613: test: Deduplicate tests implementation for loader tasks creation
Event Timeline
Log In to Comment