Factor out some common setup code

tests/experiment/conftest.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Optional
+import urllib.request
+import pytest
+from nidm.experiment.tools.rest import RestParser
+
+# We will test example NIDM files downloaded from
+# the GitHub dbkeator/simple2_NIDM_examples repo
+#
+# DBK: this is a bit unsafe as the TTL files in the github repo above can change and the UUID will change since they are randomly
+# generated at this point.  It's probably more robust to explicitly create these files for the time being and explicitly set the
+# UUID in the test file:
+# For example:  kwargs={Constants.NIDM_PROJECT_NAME:"FBIRN_PhaseIII",Constants.NIDM_PROJECT_IDENTIFIER:1200,Constants.NIDM_PROJECT_DESCRIPTION:"Test investigation2"}
+#               project = Project(uuid="_654321",attributes=kwargs)
+
+
+@pytest.fixture(scope="session")
+def brain_vol_files(tmp_path_factory: pytest.TempPathFactory) -> list[str]:
+    tmp_path = tmp_path_factory.mktemp("brain_vol_files")
+    urllib.request.urlretrieve(
+        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/CMU_a/nidm.ttl",
+        tmp_path / "cmu_a.nidm.ttl",
+    )
+    urllib.request.urlretrieve(
+        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/Caltech/nidm.ttl",
+        tmp_path / "caltech.nidm.ttl",
+    )
+    return [
+        str(tmp_path / "cmu_a.nidm.ttl"),
+        str(tmp_path / "caltech.nidm.ttl"),
+    ]
+
+
+@dataclass
+class BrainVol:
+    files: list[str]
+    restParser: RestParser
+    cmu_test_project_uuid: str
+    cmu_test_subject_uuid: str
+
+
+@pytest.fixture(scope="session")
+def brain_vol(brain_vol_files: list[str]) -> BrainVol:
+    restParser = RestParser(output_format=RestParser.OBJECT_FORMAT)
+    projects = restParser.run(brain_vol_files, "/projects")
+    cmu_test_project_uuid: Optional[str] = None
+    for p in projects:
+        proj_info = restParser.run(brain_vol_files, f"/projects/{p}")
+        if (
+            isinstance(proj_info, dict)
+            and proj_info.get("dctypes:title") == "ABIDE - CMU_a"
+        ):
+            cmu_test_project_uuid = p
+            break
+    assert cmu_test_project_uuid is not None
+    subjects = restParser.run(
+        brain_vol_files, f"/projects/{cmu_test_project_uuid}/subjects"
+    )
+    cmu_test_subject_uuid = subjects["uuid"][0]
+    return BrainVol(
+        brain_vol_files,
+        restParser,
+        cmu_test_project_uuid,
+        cmu_test_subject_uuid,
+    )

tests/experiment/test_navigate.py

@@ -1,99 +1,78 @@
+from __future__ import annotations
 from dataclasses import dataclass
 import re
 import urllib.request
 import pytest
 from nidm.experiment import Navigate
 
-BRAIN_VOL_FILES = ["cmu_a.nidm.ttl", "caltech.nidm.ttl"]
-OPENNEURO_FILES = ["ds000110.nidm.ttl"]
-
 
 @dataclass
-class Setup:
-    brain_vol_files: list[str]
-    openneuro_files: list[str]
-    project_uri: str
-    openneuro_project_uri: str
+class ProjectData:
+    files: list[str]
+    uri: str
 
 
 @pytest.fixture(scope="module")
-def setup(tmp_path_factory: pytest.TempPathFactory) -> Setup:
-    tmp_path = tmp_path_factory.mktemp("setup")
-
-    urllib.request.urlretrieve(
-        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/CMU_a/nidm.ttl",
-        tmp_path / "cmu_a.nidm.ttl",
-    )
-
-    urllib.request.urlretrieve(
-        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/Caltech/nidm.ttl",
-        tmp_path / "caltech.nidm.ttl",
-    )
-
-    brain_vol_files = [str(tmp_path / fname) for fname in BRAIN_VOL_FILES]
-
+def brain_vol(brain_vol_files: list[str]) -> ProjectData:
     projects = Navigate.getProjects(brain_vol_files)
     project_uri = projects[0]
+    return ProjectData(brain_vol_files, project_uri)
+
 
+@pytest.fixture(scope="module")
+def openneuro(tmp_path_factory: pytest.TempPathFactory) -> ProjectData:
+    tmp_path = tmp_path_factory.mktemp("openneuro")
     urllib.request.urlretrieve(
         "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/openneuro/ds000110/nidm.ttl",
         tmp_path / "ds000110.nidm.ttl",
     )
-
-    openneuro_files = [str(tmp_path / fname) for fname in OPENNEURO_FILES]
-
-    projects2 = Navigate.getProjects(openneuro_files)
-    openneuro_project_uri = projects2[0]
-
-    return Setup(
-        brain_vol_files=brain_vol_files,
-        openneuro_files=openneuro_files,
-        project_uri=project_uri,
-        openneuro_project_uri=openneuro_project_uri,
-    )
+    files = [str(tmp_path / "ds000110.nidm.ttl")]
+    projects2 = Navigate.getProjects(files)
+    project_uri = projects2[0]
+    return ProjectData(files, project_uri)
 
 
-def test_navigate_get_projects(setup: Setup) -> None:
-    projects = Navigate.getProjects(setup.brain_vol_files)
+def test_navigate_get_projects(brain_vol: ProjectData) -> None:
+    projects = Navigate.getProjects(brain_vol.files)
     assert len(projects) == 2
 
 
-def test_navigate_get_sessions(setup: Setup) -> None:
-    sessions = Navigate.getSessions(setup.brain_vol_files, setup.project_uri)
+def test_navigate_get_sessions(brain_vol: ProjectData) -> None:
+    sessions = Navigate.getSessions(brain_vol.files, brain_vol.uri)
     assert len(sessions) > 0
 
 
-def test_navigate_get_acquisitions_for_session(setup: Setup) -> None:
-    sessions = Navigate.getSessions(setup.brain_vol_files, setup.project_uri)
+def test_navigate_get_acquisitions_for_session(brain_vol: ProjectData) -> None:
+    sessions = Navigate.getSessions(brain_vol.files, brain_vol.uri)
     for _ in sessions:
-        acquisitions = Navigate.getAcquisitions(setup.brain_vol_files, sessions[0])
+        acquisitions = Navigate.getAcquisitions(brain_vol.files, sessions[0])
         assert len(acquisitions) > 0
 
 
-def test_navigate_get_subjects_for_acquisition(setup: Setup) -> None:
+def test_navigate_get_subjects_for_acquisition(brain_vol: ProjectData) -> None:
     subjects = set()
-    sessions = Navigate.getSessions(setup.brain_vol_files, setup.project_uri)
+    sessions = Navigate.getSessions(brain_vol.files, brain_vol.uri)
     for s in sessions:
-        acquisitions = Navigate.getAcquisitions(setup.brain_vol_files, s)
+        acquisitions = Navigate.getAcquisitions(brain_vol.files, s)
         for acq in acquisitions:
-            sub = Navigate.getSubject(setup.brain_vol_files, acq)
+            sub = Navigate.getSubject(brain_vol.files, acq)
             assert sub is not None
             subjects.add(sub)
     assert len(subjects) > 5
 
 
-def test_navigate_get_acquisition_data_by_session(setup: Setup) -> None:
+def test_navigate_get_acquisition_data_by_session(openneuro: ProjectData) -> None:
     set_of_keys_returned = set()
     set_of_activities = set()
 
-    sessions = Navigate.getSessions(setup.openneuro_files, setup.openneuro_project_uri)
+    sessions = Navigate.getSessions(openneuro.files, openneuro.uri)
     assert len(sessions) > 0
     for s in sessions:
-        acquisitions = Navigate.getAcquisitions(setup.openneuro_files, s)
+        acquisitions = Navigate.getAcquisitions(openneuro.files, s)
         assert len(acquisitions) > 0
         for a in acquisitions:
             set_of_activities.add(str(a))
-            ad = Navigate.getActivityData(setup.openneuro_files, a)
+            ad = Navigate.getActivityData(openneuro.files, a)
             assert len(ad.data) > 5
             for vt in ad.data:
                 set_of_keys_returned.add(vt.label)
@@ -104,20 +83,20 @@ def test_navigate_get_acquisition_data_by_session(setup: Setup) -> None:
     assert "hadAcquisitionModality" in set_of_keys_returned
 
 
-def test_navigate_get_acquisition_data_by_subject(setup: Setup) -> None:
+def test_navigate_get_acquisition_data_by_subject(openneuro: ProjectData) -> None:
     set_of_keys_returned = set()
     set_of_activities = set()
 
-    subjects = Navigate.getSubjects(setup.openneuro_files, setup.openneuro_project_uri)
+    subjects = Navigate.getSubjects(openneuro.files, openneuro.uri)
     assert len(subjects) > 0
     for s in subjects:
         activities = Navigate.getActivities(
-            nidm_file_tuples=tuple(setup.openneuro_files), subject_id=s
+            nidm_file_tuples=tuple(openneuro.files), subject_id=s
         )
         assert len(activities) > 0
         for a in activities:
             set_of_activities.add(str(a))
-            ad = Navigate.getActivityData(setup.openneuro_files, a)
+            ad = Navigate.getActivityData(openneuro.files, a)
             assert len(ad.data) > 5
             for vt in ad.data:
                 set_of_keys_returned.add(vt.label)
@@ -128,9 +107,9 @@ def test_navigate_get_acquisition_data_by_subject(setup: Setup) -> None:
     assert "hadImageUsageType" in set_of_keys_returned
 
 
-def test_navigate_get_sub_uuid_from_id(setup: Setup) -> None:
+def test_navigate_get_sub_uuid_from_id(brain_vol: ProjectData) -> None:
     uuids = Navigate.getSubjectUUIDsfromID(
-        nidm_file_tuples=setup.brain_vol_files, sub_id="50653"
+        nidm_file_tuples=brain_vol.files, sub_id="50653"
     )
     assert len(uuids) == 1
     assert re.match(

tests/experiment/test_query.py

@@ -1,8 +1,8 @@
+from __future__ import annotations
 from dataclasses import dataclass
 from pathlib import Path
 import tempfile
 from typing import Optional
-import urllib.request
 import prov.model as pm
 import pytest
 from nidm.core import Constants
@@ -17,51 +17,29 @@
 from nidm.experiment.CDE import download_cde_files
 import nidm.experiment.Navigate
 
-ABIDE_FILES = ("cmu_a.nidm.ttl",)
-
-
-# We will test example NIDM files downloaded from
-# the GitHub dbkeator/simple2_NIDM_examples repo
-#
-# DBK: this is a bit unsafe as the TTL files in the github repo above can change and the UUID will change since they are randomly
-# generated at this point.  It's probably more robust to explicitly create these files for the time being and explicitly set the
-# UUID in the test file:
-# For example:  kwargs={Constants.NIDM_PROJECT_NAME:"FBIRN_PhaseIII",Constants.NIDM_PROJECT_IDENTIFIER:1200,Constants.NIDM_PROJECT_DESCRIPTION:"Test investigation2"}
-#               project = Project(uuid="_654321",attributes=kwargs)
-
 
 @dataclass
-class Setup:
-    abide_files: list[str]
+class ProjectData:
+    files: list[str]
     cmu_test_project_uuid: str
 
 
 @pytest.fixture(scope="module", autouse="True")
-def setup(tmp_path_factory: pytest.TempPathFactory) -> Setup:
-    tmp_path = tmp_path_factory.mktemp("setup")
-
-    urllib.request.urlretrieve(
-        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/CMU_a/nidm.ttl",
-        tmp_path / "cmu_a.nidm.ttl",
-    )
-
-    abide_files = [str(tmp_path / fname) for fname in ABIDE_FILES]
-
-    projects = Query.GetProjectsUUID(abide_files)
+def abide(brain_vol_files) -> ProjectData:
+    files = [f for f in brain_vol_files if Path(f).name == "cmu_a.nidm.ttl"]
+    assert files
+    projects = Query.GetProjectsUUID(files)
     cmu_test_project_uuid: Optional[str] = None
     for p in projects:
-        proj_info = nidm.experiment.Navigate.GetProjectAttributes(abide_files, p)
+        proj_info = nidm.experiment.Navigate.GetProjectAttributes(files, p)
         if (
             "dctypes:title" in proj_info.keys()
             and proj_info["dctypes:title"] == "ABIDE - CMU_a"
         ):
             cmu_test_project_uuid = p
             break
     assert cmu_test_project_uuid is not None
-    return Setup(
-        abide_files=abide_files,
-        cmu_test_project_uuid=cmu_test_project_uuid,
-    )
+    return ProjectData(files, cmu_test_project_uuid)
 
 
 def test_GetProjectMetadata(tmp_path: Path) -> None:
@@ -289,13 +267,13 @@ def makeProjectTestFile2(filename):
     return saveProject(filename, project)
 
 
-def test_GetProjectsMetadata(setup: Setup, tmp_path: Path) -> None:
+def test_GetProjectsMetadata(abide: ProjectData, tmp_path: Path) -> None:
     p1 = makeProjectTestFile(str(tmp_path / "testfile.ttl"))
     p2 = makeProjectTestFile2(str(tmp_path / "testfile2.ttl"))
     files = [
         str(tmp_path / "testfile.ttl"),
         str(tmp_path / "testfile2.ttl"),
-        *setup.abide_files,
+        *abide.files,
     ]
 
     parsed = Query.GetProjectsMetadata(files)
@@ -331,8 +309,8 @@ def test_prefix_helpers():
     assert Query.matchPrefix("http://purl.org/nidash/fsl#xyz", short=True) == "fsl"
 
 
-def test_getProjectAcquisitionObjects(setup: Setup) -> None:
-    files = setup.abide_files
+def test_getProjectAcquisitionObjects(abide: ProjectData) -> None:
+    files = abide.files
 
     project_list = Query.GetProjectsUUID(files)
     project_uuid = str(project_list[0])
@@ -341,10 +319,10 @@ def test_getProjectAcquisitionObjects(setup: Setup) -> None:
     assert isinstance(objects, list)
 
 
-def test_GetProjectAttributes(setup: Setup) -> None:
-    files = setup.abide_files
+def test_GetProjectAttributes(abide: ProjectData) -> None:
+    files = abide.files
 
-    project_uuid = setup.cmu_test_project_uuid
+    project_uuid = abide.cmu_test_project_uuid
     project_attributes = nidm.experiment.Navigate.GetProjectAttributes(
         files, project_uuid
     )

tests/experiment/tools/test_nidm_lingreg.py

@@ -1,25 +1,11 @@
+from __future__ import annotations
 from pathlib import Path
-import urllib.request
 import click
 from click.testing import CliRunner
 import pytest
 from nidm.experiment.tools.nidm_linreg import linear_regression
 
 
-@pytest.fixture(scope="module")
-def brain_vol_files(tmp_path_factory: pytest.TempPathFactory) -> list[str]:
-    tmp_path = tmp_path_factory.mktemp("brain_vol_files")
-    urllib.request.urlretrieve(
-        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/CMU_a/nidm.ttl",
-        tmp_path / "cmu_a.nidm.ttl",
-    )
-    urllib.request.urlretrieve(
-        "https://raw.githubusercontent.com/dbkeator/simple2_NIDM_examples/master/datasets.datalad.org/abide/RawDataBIDS/Caltech/nidm.ttl",
-        tmp_path / "caltech.nidm.ttl",
-    )
-    return [str(tmp_path / "cmu_a.nidm.ttl"), str(tmp_path / "caltech.nidm.ttl")]
-
-
 def call_click_command(cmd, *args, **kwargs):
     """Wrapper to call a click command