Add integration tests for mass_nifti_pic.py

python/lib/db/queries/file.py

@@ -26,22 +26,6 @@ def try_get_file_with_unique_combination(
     ).scalar_one_or_none()
 
 
-def try_get_parameter_value_with_file_id_parameter_name(
-    db: Database,
-    file_id: int,
-    parameter_name: str
-) -> DbFileParameter | None:
-    """
-    Get parameter value from file ID and parameter name, or return `None` if no entry was found
-    """
-
-    return db.execute(select(DbFileParameter)
-        .join(DbFileParameter.type)
-        .where(DbParameterType.name == parameter_name)
-        .where(DbFileParameter.file_id == file_id)
-    ).scalar_one_or_none()
-
-
 def try_get_file_with_hash(db: Database, file_hash: str) -> DbFile | None:
     """
     Get an imaging file from the database using its BLAKE2b or MD5 hash, or return `None` if no

python/lib/db/queries/file_parameter.py

@@ -0,0 +1,30 @@
+from sqlalchemy import delete, select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.file_parameter import DbFileParameter
+from lib.db.models.parameter_type import DbParameterType
+
+
+def delete_file_parameter(db: Database, file_parameter_id: int):
+    """
+    Delete from the database a parameter value based on a file ID and parameter ID.
+    """
+
+    db.execute(delete(DbFileParameter)
+        .where(DbFileParameter.id == file_parameter_id))
+
+
+def try_get_parameter_value_with_file_id_parameter_name(
+    db: Database,
+    file_id: int,
+    parameter_name: str
+) -> DbFileParameter | None:
+    """
+    Get parameter value from file ID and parameter name, or return `None` if no entry was found
+    """
+
+    return db.execute(select(DbFileParameter)
+        .join(DbFileParameter.type)
+        .where(DbParameterType.name == parameter_name)
+        .where(DbFileParameter.file_id == file_id)
+    ).scalar_one_or_none()

python/tests/integration/scripts/test_mass_nifti_pic.py

@@ -3,8 +3,8 @@
 from datetime import datetime
 
 from lib.db.models.file import DbFile
-from lib.db.queries.file import delete_file, try_get_parameter_value_with_file_id_parameter_name
-from lib.db.queries.parameter_file import delete_file_parameter
+from lib.db.queries.file import delete_file
+from lib.db.queries.file_parameter import delete_file_parameter, try_get_parameter_value_with_file_id_parameter_name
 from lib.exitcode import INVALID_ARG, INVALID_PATH, MISSING_ARG, SUCCESS
 from tests.util.database import get_integration_database_session
 from tests.util.run_integration_script import run_integration_script
@@ -145,12 +145,13 @@ def test_force_option():
     # database connection
     db = get_integration_database_session()
     file_pic_data = try_get_parameter_value_with_file_id_parameter_name(db, 2, 'check_pic_filename')
-    if file_pic_data:
-        # remove file from the file system before recreating it
-        # otherwise get Operation Not Permitted (specific to the test environment)
-        pic_to_remove = os.path.join('/data/loris/pic/', str(file_pic_data.value))
-        if os.path.exists(pic_to_remove):
-            os.remove(pic_to_remove)
+    assert file_pic_data is not None
+    # remove file from the file system before recreating it
+    # otherwise get Operation Not Permitted (specific to the test environment)
+    # TODO: Investigate
+    pic_to_remove = os.path.join('/data/loris/pic/', str(file_pic_data.value))
+    if os.path.exists(pic_to_remove):
+        os.remove(pic_to_remove)
 
     process = run_integration_script([
         'mass_nifti_pic.py',
@@ -185,14 +186,14 @@ def test_running_on_a_text_file():
     db = get_integration_database_session()
 
     # insert fake text file with file ID 1
-    file = DbFile()
-    file.id                  = 1
-    file.rel_path            = 'test.txt'
-    file.file_type           = 'txt'
-    file.session_id          = 564
-    file.output_type         = 'native'
-    file.insert_time         = int(datetime.now().timestamp())
-    file.inserted_by_user_id = 'test'
+    file = DbFile(
+        rel_path            = 'test.txt',
+        file_type           = 'txt',
+        session_id          = 564,
+        output_type         = 'native',
+        insert_time         = int(datetime.now().timestamp()),
+        inserted_by_user_id = 'test'
+    )
     db.add(file)
     db.commit()
 
@@ -211,7 +212,7 @@ def test_running_on_a_text_file():
     assert process.stderr == ""
 
     # Clean up the file that was inserted before this test
-    delete_file(db, 1)
+    delete_file(db, file.id)
     db.commit()
 
 
@@ -225,14 +226,16 @@ def test_successful_run():
 
     # Remove file ID 2 pic from the database and filesystem
     file_pic_data = try_get_parameter_value_with_file_id_parameter_name(db, 2, 'check_pic_filename')
-    if file_pic_data:
-        # remove file from the file system before recreating it
-        pic_to_remove = os.path.join('/data/loris/pic/', str(file_pic_data.value))
-        if os.path.exists(pic_to_remove):
-            os.remove(pic_to_remove)
-        # delete pic entry based on its parameter file ID
-        delete_file_parameter(db, file_pic_data.id)
-        db.commit()
+    assert file_pic_data is not None
+    # remove file from the file system before recreating it
+    # otherwise get Operation Not Permitted (specific to the test environment)
+    # TODO: Investigate
+    pic_to_remove = os.path.join('/data/loris/pic/', str(file_pic_data.value))
+    if os.path.exists(pic_to_remove):
+        os.remove(pic_to_remove)
+    # delete pic entry based on its parameter file ID
+    delete_file_parameter(db, file_pic_data.id)
+    db.commit()
 
     file_pic_data = try_get_parameter_value_with_file_id_parameter_name(db, 2, 'check_pic_filename')
     assert file_pic_data is None

python/tests/integration/scripts/test_run_nifti_insertion.py

@@ -2,10 +2,8 @@
 import shutil
 from os.path import basename
 
-from lib.db.queries.file import (
-    try_get_file_with_unique_combination,
-    try_get_parameter_value_with_file_id_parameter_name,
-)
+from lib.db.queries.file import try_get_file_with_unique_combination
+from lib.db.queries.file_parameter import try_get_parameter_value_with_file_id_parameter_name
 from lib.db.queries.mri_protocol_violated_scan import get_protocol_violated_scans_with_unique_series_combination
 from lib.db.queries.mri_upload import get_mri_upload_with_patient_name
 from lib.db.queries.mri_violation_log import get_violations_log_with_unique_series_combination