Edit the unit test for notebookC

good_standard_metagv0.csv

@@ -0,0 +1,38 @@
+[Header],,,,,,,,
+IEMFileVersion,4,,,,,,,
+SheetType,tellseq_metag,,,,,,,
+SheetVersion,10,,,,,,,
+Investigator Name,Knight,,,,,,,
+Experiment Name,RKL_experiment,,,,,,,
+Date,2025-08-14,,,,,,,
+Workflow,GenerateFASTQ,,,,,,,
+Application,FASTQ Only,,,,,,,
+Assay,Metagenomic,,,,,,,
+Description,,,,,,,,
+Chemistry,Default,,,,,,,
+,,,,,,,,
+[Reads],,,,,,,,
+151,,,,,,,,
+151,,,,,,,,
+,,,,,,,,
+[Settings],,,,,,,,
+ReverseComplement,0,,,,,,,
+,,,,,,,,
+[Data],,,,,,,,
+Sample_ID,Sample_Name,Sample_Plate,well_id_384,barcode_id,Sample_Project,Well_description,Lane,
+sample_1,sample.1,A1,False,C501,MyProject_99999,A1.sample.1.False,1,
+sample_2,sample.2,A2,False,C509,MyProject_99999,A2.sample.2.False,1,
+sample_3,sample.3,A3,False,C520,MyProject_99999,A3.sample.3.False,1,
+,,,,,,,,
+[Bioinformatics],,,,,,,,
+Sample_Project,QiitaID,BarcodesAreRC,ForwardAdapter,ReverseAdapter,HumanFiltering,library_construction_protocol,experiment_design_description,contains_replicates
+MyProject_99999,99999,True,AACC,GGTT,False,some protocol,some description,False
+,,,,,,,,
+[Contact],,,,,,,,
+Sample_Project,Email,,,,,,,
+MyProject_99999,foo@bar.org,,,,,,,
+,,,,,,,,
+[SampleContext],,,,,,,,
+sample_name,sample_type,primary_qiita_study,secondary_qiita_studies,,,,,
+sample.3,control blank,99999,,,,,,
+,,,,,,,,

notebooks/test_output/SampleSheets/Tellseq_samplesheet_instrument_iseq_set_col19to24.csv

@@ -2,7 +2,7 @@
 Experiment Name,RKLtest_col19to24,,,,,,,,,,
 Investigator Name,Enter the investigator name (optional),,,,,,,,,,
 Project Name,TestProjA_10002_1_2_3_4_10to1dilution,,,,,,,,,,
-Date,2025-04-01,,,,,,,,,,
+Date,2026-02-02,,,,,,,,,,
 Workflow,GenerateFASTQ,,,,,,,,,,
 Library Prep Kit,TELLSEQ,,,,,,,,,,
 [Manifest],,,,,,,,,,,

notebooks/tests/test2_tellseq_C_equal_volume_pooling.py

@@ -0,0 +1,52 @@
+import unittest
+import papermill as pm
+import tempfile
+from pathlib import Path
+import os
+
+# ==========================================
+# 1. Configuration & Parameters
+# ==========================================
+NOTEBOOK = "tellseq_D_variable_volume_pooling.ipynb"
+CURRENT_SET_ID = "col19to24"  # Change this ID for different test sets
+EXPECTED_PICKLIST = f"Tellseq_iSeqnormpool_set_{CURRENT_SET_ID}.txt"
+
+class TestTellseqD_Simple(unittest.TestCase):
+    def setUp(self):
+        """Set up standard paths"""
+        self.base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        self.test_output_dir = os.path.join(self.base_dir, 'test_output')
+        self.test_data_dir = os.path.join(self.base_dir, 'test_data')
+
+    def test_variable_volume_pooling(self):
+        """Execute notebook and verify picklist generation"""
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_root = Path(tmp_dir)
+
+            # 2. Define Parameters for the Notebook
+
+            test_values = {
+                "plate_df_set_fp": os.path.join(self.test_output_dir, "QC", f"Tellseq_plate_df_C_set_{CURRENT_SET_ID}.txt"),
+                "read_counts_fps": [os.path.join(self.test_data_dir, "Demux", "Tellseq_fastqc_sequence_counts.tsv")],
+                "dynamic_range": 5,
+                "iseqnormed_picklist_fbase": str(tmp_root / "Tellseq_iSeqnormpool")
+            }
+
+            # 3. Run Notebook
+            pm.execute_notebook(
+                input_path=os.path.join(self.base_dir, NOTEBOOK),
+                output_path=str(tmp_root / "executed_test.ipynb"),
+                parameters={"test_dict": test_values},
+                log_output=True,
+            )
+
+            # 4. Verify Output File
+            produced_file = tmp_root / EXPECTED_PICKLIST
+
+            # Check if file exists
+            self.assertTrue(produced_file.exists(), f"Missing output file: {EXPECTED_PICKLIST}")
+
+
+if __name__ == "__main__":
+    unittest.main()

notebooks/tests/test2_tellseq_D_variable_volume_pooling.py

@@ -0,0 +1,86 @@
+import unittest
+import papermill as pm
+import tempfile
+from pathlib import Path
+import os
+import glob
+
+NOTEBOOK_NAME = "tellseq_D_variable_volume_pooling.ipynb"
+
+class TestTellseqD(unittest.TestCase):
+
+    def setUp(self):
+        """Set up paths before starting tests"""
+        self.test_file_path = os.path.abspath(__file__)
+        self.notebooks_dir = os.path.dirname(os.path.dirname(self.test_file_path)) 
+        self.test_output_dir = os.path.join(self.notebooks_dir, 'test_output')
+        self.test_data_dir = os.path.join(self.notebooks_dir, 'test_data')
+
+    def test_iseqnorm_picklist_dynamic(self):
+        """Test Notebook D dynamically by extracting the ID from the input filename"""
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_path = Path(tmp_dir)
+
+
+            search_pattern = os.path.join(self.test_output_dir, "QC", "Tellseq_plate_df_C_set_*.txt")
+            matching_files = glob.glob(search_pattern)
+
+            if not matching_files:
+                self.fail(f"Could not find input file for testing: {search_pattern}")
+
+
+            input_plate_c_fp = matching_files[0]
+
+
+            filename = os.path.basename(input_plate_c_fp)
+            extracted_id = filename.split('_set_')[-1].replace('.txt', '')
+
+
+            test_values = {
+                'plate_df_set_fp': input_plate_c_fp,
+                'read_counts_fps': [
+                    os.path.join(self.test_data_dir, "Demux", "Tellseq_fastqc_sequence_counts.tsv")
+                ],
+                'dynamic_range': 5,
+                'iseqnormed_picklist_fbase': str(tmp_path / "Tellseq_iSeqnormpool")
+            }
+
+            # 5. Execute Notebook (Overwrites the top-level 'test_dict = None')
+            pm.execute_notebook(
+                input_path=os.path.join(self.notebooks_dir, NOTEBOOK_NAME),
+                output_path=str(tmp_path / "executed_test_D.ipynb"),
+                parameters={'test_dict': test_values},
+                log_output=True,
+            )
+
+            # 6. Dynamically verify the output filename
+            # Based on notebook logic: {{fbase}}_set_{{extracted_id}}.txt is generated
+            expected_filename = f"Tellseq_iSeqnormpool_set_{extracted_id}.txt"
+            produced_picklist_fp = tmp_path / expected_filename
+
+            # Check if the file exists
+            self.assertTrue(produced_picklist_fp.exists(),
+                            msg=f"Notebook failed to produce the output file: {expected_filename}")
+
+            # 7. Compare the actual result with the Golden File (expected value)
+            golden_picklist_fp = os.path.join(self.test_output_dir, "Pooling", expected_filename)
+
+            if os.path.exists(golden_picklist_fp):
+                with open(produced_picklist_fp, 'r') as out, open(golden_picklist_fp, 'r') as gold:
+                    out_lines = out.readlines()
+                    gold_lines = gold.readlines()
+
+                    # Compare line count and content
+                    self.assertEqual(len(out_lines), len(gold_lines), 
+                                     msg=f"Line count mismatch: {expected_filename}")
+
+                    for i, (o_line, g_line) in enumerate(zip(out_lines, gold_lines), 1):
+                        self.assertEqual(o_line.strip(), g_line.strip(),
+                                         msg=f"Content mismatch in {expected_filename} at line {i}")
+                print(f"Test Passed: {expected_filename} verified successfully (Extracted ID: {extracted_id})")
+            else:
+                print(f"Warning: Golden file for comparison is missing: {golden_picklist_fp}")
+
+if __name__ == "__main__":
+    unittest.main()

notebooks/tests/test_tellseq_C_equal_volume_pooling_1.py

@@ -0,0 +1,70 @@
+import unittest
+import papermill as pm
+import tempfile
+from pathlib import Path
+import os
+
+NOTEBOOK = "tellseq_C_equal_volume_pooling.ipynb"
+PICKLIST_FNAME = "Tellseq_iSeqnormpool_set_col19to24.txt"
+
+
+class TestTellseqD(unittest.TestCase):
+    def setUp(self):
+        self.notebooks_dir = os.path.dirname(os.path.dirname(__file__))
+        self.test_output_dir = os.path.join(self.notebooks_dir, 'test_output')
+
+    def test_iseqnorm_picklist(self):
+        """Verify notebook produces expected output for iSeqnormed picklist."""
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_path = Path(tmp_dir)
+
+            run_params = {
+                    """
+                    'plate_df_set_fp': f"{self.test_output_dir}/QC/"
+                                       f"Tellseq_plate_df_C_set_col19to24.txt",
+                    'read_counts_fps': [
+                        f"{self.notebooks_dir}/test_data/Demux/"
+                        f"Tellseq_fastqc_sequence_counts.tsv"],
+                    'dynamic_range': 5,
+                    'iseqnormed_picklist_fbase':
+                        f"{tmp_path}/Tellseq_iSeqnormpool"
+                    """
+                    'full_plate_fp' : f"{self.test_output_dir}/QC/Tellseq_plate_df_B.txt",
+                    'expt_config_fp' : f"{self.test_output_dir}/QC/Tellseq_expt_info.yml",
+                    'current_set_id' : "col19to24",
+                    'total_vol' : 190,
+                    'evp_picklist_fbase' : f"{self.test_output_dir}/Indices/Tellseq_evp",
+                    'machine_samplesheet_fbase' : f"{self.test_output_dir}/SampleSheets/Tellseq_samplesheet_instrument_iseq",
+                    'spp_samplesheet_fbase' : f"{self.test_output_dir}/SampleSheets/Tellseq_samplesheet_spp",
+                    'iseq_sequencer' : 'iSeq',
+                    'novaseq_sequencer' : 'NovaSeqXPlus'
+
+            }
+
+            pm.execute_notebook(
+                input_path=f"{self.notebooks_dir}/{NOTEBOOK}",
+                output_path=f"{tmp_path}/test_iseqnorm_picklist.ipynb",
+                parameters=run_params,
+                log_output=True,
+            )
+
+            out_iseqnormed_picklist_fp = f"{tmp_path}/{PICKLIST_FNAME}"
+            self.assertTrue(os.path.exists(out_iseqnormed_picklist_fp),
+                            msg="Notebook did not produce desired file.")
+
+            exp_iseqnormed_fp = \
+                f"{self.test_output_dir}/Pooling/{PICKLIST_FNAME}"
+            with open(out_iseqnormed_picklist_fp, 'r') as out:
+                with open(exp_iseqnormed_fp, 'r') as test:
+                    out_lines = out.readlines()
+                    test_lines = test.readlines()
+                    for out_line, test_line in zip(out_lines,
+                                                   test_lines):
+                        self.assertEqual(out_line, test_line,
+                                         msg=("Lines of output" +
+                                              "and test don't match"))
+
+
+if __name__ == "__main__":
+    unittest.main()

notebooks/tests/test_tellseq_D_variable_volume_pooling.py

@@ -7,20 +7,17 @@
 NOTEBOOK = "tellseq_D_variable_volume_pooling.ipynb"
 PICKLIST_FNAME = "Tellseq_iSeqnormpool_set_col19to24.txt"
 
-
 class TestTellseqD(unittest.TestCase):
+
     def setUp(self):
-        self.notebooks_dir = os.path.dirname(os.path.dirname(__file__))
+        self.notebooks_dir = os.path.dirname(os.path.dirname(__file__)) 
         self.test_output_dir = os.path.join(self.notebooks_dir, 'test_output')
 
-    def test_iseqnorm_picklist(self):
-        """Verify notebook produces expected output for iSeqnormed picklist."""
 
+    def test_iseqnorm_picklist(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
             tmp_path = Path(tmp_dir)
-
             run_params = {
-                'test_dict': {
                     'plate_df_set_fp': f"{self.test_output_dir}/QC/"
                                        f"Tellseq_plate_df_C_set_col19to24.txt",
                     'read_counts_fps': [
@@ -29,7 +26,6 @@ def test_iseqnorm_picklist(self):
                     'dynamic_range': 5,
                     'iseqnormed_picklist_fbase':
                         f"{tmp_path}/Tellseq_iSeqnormpool"
-                }
             }
 
             pm.execute_notebook(
@@ -55,6 +51,5 @@ def test_iseqnorm_picklist(self):
                                          msg=("Lines of output" +
                                               "and test don't match"))
 
-
 if __name__ == "__main__":
     unittest.main()