feat: updates integration test for bergamo

tests/integration/bergamo/__init__.py

@@ -0,0 +1 @@
+"""Package for bergamo integration tests."""

tests/integration/bergamo/session.py

@@ -0,0 +1,63 @@
+"""Integration test for bergamo session"""
+
+import argparse
+import json
+import os
+import sys
+import unittest
+from pathlib import Path
+
+from aind_metadata_mapper.bergamo.session import BergamoEtl, JobSettings
+
+EXPECTED_OUTPUT_FILE_PATH = (
+    Path(os.path.dirname(os.path.realpath(__file__)))
+    / ".."
+    / ".."
+    / "resources"
+    / "bergamo"
+    / "session.json"
+)
+
+
+class IntegrationTestBergamo(unittest.TestCase):
+    """Integration test for Bergamo"""
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        """Set up the class."""
+        with open(EXPECTED_OUTPUT_FILE_PATH, "r") as f:
+            expected_output_json = json.load(f)
+        cls.expected_output = expected_output_json
+
+    def test_run_job(self):
+        """Tests run_job on actual raw data source."""
+        input_source: str = getattr(IntegrationTestBergamo, "input_source")
+        job_settings = JobSettings(
+            input_source=Path(input_source),
+            experimenter_full_name=["Jane Doe"],
+            subject_id="706957",
+            imaging_laser_wavelength=405,
+            fov_imaging_depth=150,
+            fov_targeted_structure="M1",
+            notes=None,
+        )
+        bergamo_job = BergamoEtl(job_settings=job_settings)
+        response = bergamo_job.run_job()
+        actual_session = json.loads(response.data)
+        self.assertEqual(self.expected_output, actual_session)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--input_source",
+        type=str,
+        required=True,
+        help="The input source for the ETL job.",
+    )
+    parser.add_argument("unittest_args", nargs="*")
+
+    args = parser.parse_args()
+    setattr(IntegrationTestBergamo, "input_source", args.input_source)
+    sys.argv[1:] = args.unittest_args
+    unittest.main()

tests/test_dynamic_routing/test_neuropixels_rig.py

@@ -24,7 +24,7 @@ def test_update_modification_date(self):
         """Test ETL workflow with inferred probe mapping."""
         etl = NeuropixelsRigEtl(
             RESOURCES_DIR / "base_rig.json",
-            Path("/"),
+            Path("abc"),
         )
         extracted = etl._extract()
         transformed = etl._transform(extracted)

tests/test_dynamic_routing/test_utils.py

@@ -40,6 +40,6 @@ def setup_neuropixels_etl_resources(
     """
     return (
         RESOURCES_DIR / "base_rig.json",
-        Path("/"),  # hopefully file writes are mocked
+        Path("abc"),  # hopefully file writes are mocked
         Rig.model_validate_json(expected_json.read_text()),
     )

tests/test_mesoscope/test_session.py

@@ -148,7 +148,7 @@ def test_extract_no_input_source(
         """Tests that _extract raises a ValueError"""
         mock_is_dir.return_value = True
         mock_path_exists.return_value = False
-        mock_path_glob.return_value = iter([Path("/somedir/a")])
+        mock_path_glob.return_value = iter([Path("somedir/a")])
         etl1 = MesoscopeEtl(
             job_settings=self.example_job_settings,
         )

tests/test_open_ephys/test_rig.py

@@ -19,7 +19,7 @@
 BASE_RIG_MISSING_PROBE_PATH = (
     RESOURCES_DIR / "dynamic_routing" / "base-missing-probe_rig.json"
 )
-OUTPUT_DIR = Path("")  # File writes will be mocked
+OUTPUT_DIR = Path("abc")  # File writes will be mocked
 
 
 class TestOpenEphysRigEtl(unittest.TestCase):

tests/test_open_ephys/test_utils/__init__.py

@@ -0,0 +1 @@
+"""Test modules in utils package"""

tests/test_open_ephys/test_utils/test_stim_utils.py

@@ -116,7 +116,7 @@ def test_enforce_df_column_order(self):
             }
         )
         result_df = stim.enforce_df_column_order(df, column_order)
-        pd.testing.assert_frame_equal(result_df, expected_df)
+        pd.testing.assert_frame_equal(result_df, expected_df, check_like=True)
 
         # Test case: Specified column order with all columns
         column_order = ["C", "A", "D", "B"]