Convert more test fixtures to helpers

Author: medihack

adit/batch_query/tests/integration/test_batch_query.py

@@ -6,8 +6,10 @@
 from adit_radis_shared.common.utils.auth_utils import add_permission, add_user_to_group
 from adit_radis_shared.common.utils.worker_utils import run_worker_once
 from playwright.sync_api import Locator, Page, expect
+from pytest_django.live_server_helper import LiveServer
 
 from adit.batch_query.models import BatchQueryJob
+from adit.batch_query.utils.testing_helpers import create_batch_query_group
 from adit.core.utils.auth_utils import grant_access
 from adit.core.utils.testing_helpers import (
     create_excel_file,
@@ -19,11 +21,7 @@
 @pytest.mark.integration
 @pytest.mark.django_db(transaction=True)
 def test_urgent_batch_query_with_dimse_server(
-    page: Page,
-    poll: Callable[[Locator], Locator],
-    live_server,
-    create_and_login_user,
-    batch_query_group,
+    page: Page, poll: Callable[[Locator], Locator], live_server: LiveServer, create_and_login_user
 ):
     # Arrange
     df = pd.DataFrame(
@@ -33,12 +31,13 @@ def test_urgent_batch_query_with_dimse_server(
     batch_file = create_excel_file(df, "batch_file.xlsx")
 
     user: User = create_and_login_user(live_server.url)
-    add_user_to_group(user, batch_query_group)
-    add_permission(batch_query_group, BatchQueryJob, "can_process_urgently")
+    group = create_batch_query_group()
+    add_user_to_group(user, group)
+    add_permission(group, BatchQueryJob, "can_process_urgently")
 
     orthancs = setup_dimse_orthancs()
-    grant_access(batch_query_group, orthancs[0], source=True)
-    grant_access(batch_query_group, orthancs[1], destination=True)
+    grant_access(group, orthancs[0], source=True)
+    grant_access(group, orthancs[1], destination=True)
 
     # Act
     page.goto(live_server.url + "/batch-query/jobs/new/")
@@ -58,11 +57,7 @@ def test_urgent_batch_query_with_dimse_server(
 @pytest.mark.integration
 @pytest.mark.django_db(transaction=True)
 def test_urgent_batch_query_with_dicomweb_server(
-    page: Page,
-    poll: Callable[[Locator], Locator],
-    live_server,
-    create_and_login_user,
-    batch_query_group,
+    page: Page, poll: Callable[[Locator], Locator], live_server: LiveServer, create_and_login_user
 ):
     # Arrange
     df = pd.DataFrame(
@@ -72,12 +67,13 @@ def test_urgent_batch_query_with_dicomweb_server(
     batch_file = create_excel_file(df, "batch_file.xlsx")
 
     user: User = create_and_login_user(live_server.url)
-    add_user_to_group(user, batch_query_group)
-    add_permission(batch_query_group, BatchQueryJob, "can_process_urgently")
+    group = create_batch_query_group()
+    add_user_to_group(user, group)
+    add_permission(group, BatchQueryJob, "can_process_urgently")
 
     orthancs = setup_dicomweb_orthancs()
-    grant_access(batch_query_group, orthancs[0], source=True)
-    grant_access(batch_query_group, orthancs[1], destination=True)
+    grant_access(group, orthancs[0], source=True)
+    grant_access(group, orthancs[1], destination=True)
 
     # Act
     page.goto(live_server.url + "/batch-query/jobs/new/")

adit/batch_query/tests/conftest.py adit/batch_query/utils/testing_helpers.pyadit/batch_query/tests/conftest.py renamed to adit/batch_query/utils/testing_helpers.py

@@ -1,10 +1,8 @@
-import pytest
 from adit_radis_shared.accounts.factories import GroupFactory
 from adit_radis_shared.common.utils.auth_utils import add_permission
 
 
-@pytest.fixture
-def batch_query_group(db):
+def create_batch_query_group():
     group = GroupFactory.create(name="Radiologists")
     add_permission(group, "batch_query", "add_batchqueryjob")
     add_permission(group, "batch_query", "view_batchqueryjob")

adit/batch_transfer/tests/conftest.py

@@ -6,8 +6,10 @@
 from adit_radis_shared.common.utils.auth_utils import add_permission, add_user_to_group
 from adit_radis_shared.common.utils.worker_utils import run_worker_once
 from playwright.sync_api import Locator, Page, expect
+from pytest_django.live_server_helper import LiveServer
 
 from adit.batch_transfer.models import BatchTransferJob
+from adit.batch_transfer.utils.testing_helpers import create_batch_transfer_group
 from adit.core.utils.auth_utils import grant_access
 from adit.core.utils.testing_helpers import (
     create_excel_file,
@@ -19,11 +21,7 @@
 @pytest.mark.integration
 @pytest.mark.django_db(transaction=True)
 def test_unpseudonymized_urgent_batch_transfer_with_dimse_server(
-    page: Page,
-    poll: Callable[[Locator], Locator],
-    live_server,
-    create_and_login_user,
-    batch_transfer_group,
+    page: Page, poll: Callable[[Locator], Locator], live_server: LiveServer, create_and_login_user
 ):
     # Arrange
     df = pd.DataFrame(
@@ -33,13 +31,14 @@ def test_unpseudonymized_urgent_batch_transfer_with_dimse_server(
     batch_file = create_excel_file(df, "batch_file.xlsx")
 
     user: User = create_and_login_user(live_server.url)
-    add_user_to_group(user, batch_transfer_group)
-    add_permission(batch_transfer_group, BatchTransferJob, "can_process_urgently")
-    add_permission(batch_transfer_group, BatchTransferJob, "can_transfer_unpseudonymized")
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
+    add_permission(group, BatchTransferJob, "can_process_urgently")
+    add_permission(group, BatchTransferJob, "can_transfer_unpseudonymized")
 
     orthancs = setup_dimse_orthancs()
-    grant_access(batch_transfer_group, orthancs[0], source=True)
-    grant_access(batch_transfer_group, orthancs[1], destination=True)
+    grant_access(group, orthancs[0], source=True)
+    grant_access(group, orthancs[1], destination=True)
 
     # Act
     page.goto(live_server.url + "/batch-transfer/jobs/new/")
@@ -61,11 +60,7 @@ def test_unpseudonymized_urgent_batch_transfer_with_dimse_server(
 @pytest.mark.integration
 @pytest.mark.django_db(transaction=True)
 def test_unpseudonymized_urgent_batch_transfer_with_dicomweb_server(
-    page: Page,
-    poll: Callable[[Locator], Locator],
-    live_server,
-    create_and_login_user,
-    batch_transfer_group,
+    page: Page, poll: Callable[[Locator], Locator], live_server: LiveServer, create_and_login_user
 ):
     # Arrange
     df = pd.DataFrame(
@@ -75,13 +70,14 @@ def test_unpseudonymized_urgent_batch_transfer_with_dicomweb_server(
     batch_file = create_excel_file(df, "batch_file.xlsx")
 
     user: User = create_and_login_user(live_server.url)
-    add_user_to_group(user, batch_transfer_group)
-    add_permission(batch_transfer_group, BatchTransferJob, "can_process_urgently")
-    add_permission(batch_transfer_group, BatchTransferJob, "can_transfer_unpseudonymized")
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
+    add_permission(group, BatchTransferJob, "can_process_urgently")
+    add_permission(group, BatchTransferJob, "can_transfer_unpseudonymized")
 
     orthancs = setup_dicomweb_orthancs()
-    grant_access(batch_transfer_group, orthancs[0], source=True)
-    grant_access(batch_transfer_group, orthancs[1], destination=True)
+    grant_access(group, orthancs[0], source=True)
+    grant_access(group, orthancs[1], destination=True)
 
     # Act
     page.goto(live_server.url + "/batch-transfer/jobs/new/")

adit/batch_transfer/tests/integration/test_batch_transfer.py

@@ -4,6 +4,7 @@
 from django.core.files import File
 from pytest_mock import MockerFixture
 
+from adit.batch_transfer.utils.testing_helpers import create_batch_transfer_group
 from adit.core.factories import DicomServerFactory
 from adit.core.utils.auth_utils import grant_access
 
@@ -52,12 +53,13 @@ def test_field_labels(mocker: MockerFixture):
 
 
 @pytest.mark.django_db
-def test_with_valid_data(mocker: MockerFixture, batch_transfer_group, data_dict, file_dict):
+def test_with_valid_data(mocker: MockerFixture, data_dict, file_dict):
     # Arrange
     user = UserFactory.create()
-    add_user_to_group(user, batch_transfer_group)
-    grant_access(batch_transfer_group, data_dict["source"], source=True)
-    grant_access(batch_transfer_group, data_dict["destination"], destination=True)
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
+    grant_access(group, data_dict["source"], source=True)
+    grant_access(group, data_dict["destination"], destination=True)
     parse_mock = mocker.patch(
         "adit.batch_transfer.forms.BatchTransferFileParser.parse", autospec=True
     )

adit/batch_transfer/tests/test_forms.py

@@ -9,6 +9,7 @@
 from procrastinate.contrib.django.models import ProcrastinateJob
 from pytest_django.asserts import assertTemplateUsed
 
+from adit.batch_transfer.utils.testing_helpers import create_batch_transfer_group
 from adit.core.factories import DicomServerFactory
 from adit.core.models import DicomServer
 from adit.core.utils.auth_utils import grant_access
@@ -64,27 +65,27 @@ def test_user_must_have_permission_to_access_view(client):
 
 
 @pytest.mark.django_db
-def test_logged_in_user_with_permission_can_access_form(client, batch_transfer_group):
+def test_logged_in_user_with_permission_can_access_form(client):
     user = UserFactory.create()
-    add_user_to_group(user, batch_transfer_group)
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
     client.force_login(user)
     response = client.get(reverse("batch_transfer_job_create"))
     assert response.status_code == 200
     assertTemplateUsed(response, "batch_transfer/batch_transfer_job_form.html")
 
 
-def test_batch_job_created_and_enqueued_with_auto_verify(
-    client, settings, batch_transfer_group, form_data
-):
+def test_batch_job_created_and_enqueued_with_auto_verify(client, settings, form_data):
     settings.START_BATCH_TRANSFER_UNVERIFIED = True
 
     user = UserFactory.create()
-    add_user_to_group(user, batch_transfer_group)
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
 
     source_server = DicomServer.objects.get(pk=form_data["source"])
     destination_server = DicomServer.objects.get(pk=form_data["destination"])
-    grant_access(batch_transfer_group, source_server, source=True)
-    grant_access(batch_transfer_group, destination_server, destination=True)
+    grant_access(group, source_server, source=True)
+    grant_access(group, destination_server, destination=True)
 
     client.force_login(user)
     client.post(reverse("batch_transfer_job_create"), form_data)
@@ -94,18 +95,17 @@ def test_batch_job_created_and_enqueued_with_auto_verify(
     assert ProcrastinateJob.objects.count() == 3
 
 
-def test_batch_job_created_and_not_enqueued_without_auto_verify(
-    client, settings, batch_transfer_group, form_data
-):
+def test_batch_job_created_and_not_enqueued_without_auto_verify(client, settings, form_data):
     settings.START_BATCH_TRANSFER_UNVERIFIED = False
 
     user = UserFactory.create()
-    add_user_to_group(user, batch_transfer_group)
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
 
     source_server = DicomServer.objects.get(pk=form_data["source"])
     destination_server = DicomServer.objects.get(pk=form_data["destination"])
-    grant_access(batch_transfer_group, source_server, source=True)
-    grant_access(batch_transfer_group, destination_server, destination=True)
+    grant_access(group, source_server, source=True)
+    grant_access(group, destination_server, destination=True)
 
     client.force_login(user)
     client.post(reverse("batch_transfer_job_create"), form_data)
@@ -115,9 +115,10 @@ def test_batch_job_created_and_not_enqueued_without_auto_verify(
     assert ProcrastinateJob.objects.count() == 0
 
 
-def test_job_cant_be_created_with_missing_fields(client, batch_transfer_group, form_data):
+def test_job_cant_be_created_with_missing_fields(client, form_data):
     user = UserFactory.create()
-    add_user_to_group(user, batch_transfer_group)
+    group = create_batch_transfer_group()
+    add_user_to_group(user, group)
     client.force_login(user)
     for key_to_exclude in form_data:
         invalid_form_data = form_data.copy()

adit/batch_transfer/tests/test_views.py

@@ -0,0 +1,60 @@
+from io import BytesIO
+from unittest.mock import create_autospec
+
+import pandas as pd
+from adit_radis_shared.accounts.factories import GroupFactory
+from adit_radis_shared.common.utils.auth_utils import add_permission
+from django.core.files import File
+from django.core.files.uploadedfile import SimpleUploadedFile
+
+from adit.core.factories import DicomServerFactory
+
+
+def create_batch_transfer_group():
+    group = GroupFactory.create(name="Radiologists")
+    add_permission(group, "batch_transfer", "add_batchtransferjob")
+    add_permission(group, "batch_transfer", "view_batchtransferjob")
+    return group
+
+
+def create_data_dict():
+    return {
+        "source": DicomServerFactory(),
+        "destination": DicomServerFactory(),
+        "project_name": "Apollo project",
+        "project_description": "Fly to the moon",
+        "ethics_application_id": "12345",
+    }
+
+
+def create_file_dict():
+    file = create_autospec(File, size=5242880)
+    file.name = "sample_sheet.xlsx"
+    file.read.return_value.decode.return_value = ""
+    return {"batch_file": file}
+
+
+def create_form_data():
+    buffer = BytesIO()
+    data = pd.DataFrame(
+        [
+            ["1001", "1.2.840.113845.11.1000000001951524609.20200705182951.2689481", "WSOHMP4N"],
+            ["1002", "1.2.840.113845.11.1000000001951524609.20200705170836.2689469", "C2XJQ2AR"],
+            ["1003", "1.2.840.113845.11.1000000001951524609.20200705172608.2689471", "KRS8CZ3S"],
+        ],
+        columns=["PatientID", "StudyInstanceUID", "Pseudonym"],  # type: ignore
+    )
+    data.to_excel(buffer, index=False)  # type: ignore
+
+    return {
+        "source": DicomServerFactory.create().pk,
+        "destination": DicomServerFactory.create().pk,
+        "project_name": "Apollo project",
+        "project_description": "Fly to the moon",
+        "ethics_application_id": "12345",
+        "batch_file": SimpleUploadedFile(
+            name="sample_sheet.xlsx",
+            content=buffer.getvalue(),
+            content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        ),
+    }