pinellolab
diff --git a/‎CRISPResso2/CRISPRessoCORE.py‎
Lines changed: 1 addition & 1 deletion b/‎CRISPResso2/CRISPRessoCORE.py‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎CRISPResso2/CRISPRessoMultiProcessing.py‎
Lines changed: 0 additions & 3 deletions b/‎CRISPResso2/CRISPRessoMultiProcessing.py‎
Lines changed: 0 additions & 3 deletions
diff --git a/‎CRISPResso2/CRISPRessoPlot.py‎
Lines changed: 1 addition & 1 deletion b/‎CRISPResso2/CRISPRessoPlot.py‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎CRISPResso2/filterFastqs.py‎
Lines changed: 14 additions & 14 deletions b/‎CRISPResso2/filterFastqs.py‎
Lines changed: 14 additions & 14 deletions
diff --git a/‎pytest.ini‎
Lines changed: 5 additions & 0 deletions b/‎pytest.ini‎
Lines changed: 5 additions & 0 deletions
diff --git a/‎tests/df_alleles.txt‎
Lines changed: 2 additions & 0 deletions b/‎tests/df_alleles.txt‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎tests/unit_tests/conftest.py‎
Lines changed: 82 additions & 0 deletions b/‎tests/unit_tests/conftest.py‎
Lines changed: 82 additions & 0 deletions
@@ -2562,7 +2562,7 @@ def normalize_name(name, fastq_r1, fastq_r2, bam_input):
             return '%s_%s' % (get_name_from_fasta(fastq_r1), get_name_from_fasta(fastq_r2))
         elif fastq_r1:
             return '%s' % get_name_from_fasta(fastq_r1)
-        elif bam_input != '':
+        elif bam_input is not None and bam_input != '':
             return '%s' % get_name_from_bam(bam_input)
     else:
         clean_name=CRISPRessoShared.slugify(name)
 
@@ -141,7 +141,6 @@ def run_crispresso_cmds(crispresso_cmds, n_processes="1", descriptor = 'region',
         pool.terminate()
         logger.warn('Caught SIGINT. Program Terminated')
         raise Exception('CRISPResso2 Terminated')
-        exit (0)
     except Exception as e:
         print('CRISPResso2 failed')
         raise e
@@ -196,7 +195,6 @@ def input_function_chunk(df):
         pool.terminate()
         logging.warn('Caught SIGINT. Program Terminated')
         raise Exception('CRISPResso2 Terminated')
-        exit (0)
     except Exception as e:
         print('CRISPResso2 failed')
         raise e
@@ -278,7 +276,6 @@ def run_parallel_commands(commands_arr, n_processes=1, descriptor='CRISPResso2',
         pool.terminate()
         logging.warn('Caught SIGINT. Program Terminated')
         raise Exception('CRISPResso2 Terminated')
-        exit (0)
     except Exception as e:
         print('CRISPResso2 failed')
         raise e
 
@@ -71,7 +71,7 @@ def get_nuc_color(nuc, alpha):
             charSum += thisval
         charSum = (charSum/len(nuc))/90.0
 
-        return (charSum, (1-charSum), (2*charSum*(1-charSum)))
+        return (charSum, (1-charSum), (2*charSum*(1-charSum)), alpha)
 
 def get_color_lookup(nucs, alpha, custom_colors=None):
     if custom_colors is None:
 
@@ -39,10 +39,10 @@ def filterFastqs(fastq_r1=None,fastq_r2=None,fastq_r1_out=None,fastq_r2_out=None
     startTime = datetime.datetime.now()
 
     if not os.path.exists(fastq_r1):
-        raise Exception("fastq_r1 file '"+fastq_r1+"' does not exit.")
+        raise Exception("fastq_r1 file '"+fastq_r1+"' does not exist.")
 
     if fastq_r2 is not None and not os.path.exists(fastq_r2):
-        raise Exception("fastq_r2 file '"+fastq_r2+"' does not exit.")
+        raise Exception("fastq_r2 file '"+fastq_r2+"' does not exist.")
 
     ##CREATION OF FILEHANDLES##
     if fastq_r1.endswith('.gz'):
@@ -143,7 +143,7 @@ def run_mBPN(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_bp_qual_o
         npQualLine = numpy.frombuffer(qualLine, dtype=numpy.uint8)-33 #assume illumina 1.7
         npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
         npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-        f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+        f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
 
 def run_mRQ(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_bp_qual_or_N):
@@ -195,7 +195,7 @@ def run_mBP_mBPN(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_bp_qu
         if min >= min_bp_qual_in_read:
             npSeqLine = numpy.frombuffer(seqLine, 'c')
             npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
 
 def run_mRQ_mBPN(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_bp_qual_or_N):
@@ -209,7 +209,7 @@ def run_mRQ_mBPN(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_bp_qu
         if mean >= min_av_read_qual:
             npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
             npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
 
 def run_mBP_mRQ_mBPN(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_bp_qual_or_N):
@@ -225,7 +225,7 @@ def run_mBP_mRQ_mBPN(f1_in, f1_out, min_bp_qual_in_read, min_av_read_qual, min_b
             if mean >= min_av_read_qual:
                 npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
                 npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-                f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+                f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
 
 
@@ -245,10 +245,10 @@ def run_mBPN_pair(f1_in, f1_out, f2_in, f2_out, min_bp_qual_in_read, min_av_read
         npQualLine2 = numpy.frombuffer(qualLine2, dtype=numpy.uint8)-33 #assume illumina 1.7
         npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
         npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-        f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+        f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
         npSeqLine2 = numpy.frombuffer(seqLine2, 'c').copy()
         npSeqLine2[npQualLine2 < min_bp_qual_or_N] = 'N'
-        f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tostring().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
+        f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tobytes().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
 
         idLine = f1_in.readline().rstrip().decode('utf-8')
         idLine2 = f2_in.readline().rstrip().decode('utf-8')
@@ -338,10 +338,10 @@ def run_mBP_mBPN_pair(f1_in, f1_out, f2_in, f2_out, min_bp_qual_in_read, min_av_
         if min >= min_bp_qual_in_read and min2 >= min_bp_qual_in_read:
             npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
             npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
             npSeqLine2 = numpy.frombuffer(seqLine2, 'c').copy()
             npSeqLine2[npQualLine2 < min_bp_qual_or_N] = 'N'
-            f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tostring().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
+            f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tobytes().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
         idLine2 = f2_in.readline().rstrip().decode('utf-8')
 
@@ -363,10 +363,10 @@ def run_mRQ_mBPN_pair(f1_in, f1_out, f2_in, f2_out, min_bp_qual_in_read, min_av_
         if mean >= min_av_read_qual and mean2 >= min_av_read_qual:
             npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
             npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+            f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
             npSeqLine2 = numpy.frombuffer(seqLine2, 'c').copy()
             npSeqLine2[npQualLine2 < min_bp_qual_or_N] = 'N'
-            f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tostring().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
+            f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tobytes().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
         idLine2 = f2_in.readline().rstrip().decode('utf-8')
 
@@ -391,10 +391,10 @@ def run_mBP_mRQ_mBPN_pair(f1_in, f1_out, f2_in, f2_out, min_bp_qual_in_read, min
             if mean >= min_av_read_qual and mean2 >= min_av_read_qual:
                 npSeqLine = numpy.frombuffer(seqLine, 'c').copy()
                 npSeqLine[npQualLine < min_bp_qual_or_N] = 'N'
-                f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tostring().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
+                f1_out.write("%s\n%s\n%s\n%s\n"%(idLine, npSeqLine.tobytes().decode('utf-8'), plusLine.decode('utf-8'), qualLine.decode('utf-8')))
                 npSeqLine2 = numpy.frombuffer(seqLine2, 'c').copy()
                 npSeqLine2[npQualLine2 < min_bp_qual_or_N] = 'N'
-                f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tostring().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
+                f2_out.write("%s\n%s\n%s\n%s\n"%(idLine2, npSeqLine2.tobytes().decode('utf-8'), plusLine2.decode('utf-8'), qualLine2.decode('utf-8')))
         idLine = f1_in.readline().rstrip().decode('utf-8')
         idLine2 = f2_in.readline().rstrip().decode('utf-8')
 
 
@@ -0,0 +1,5 @@
+[pytest]
+testpaths = tests/unit_tests
+python_files = test_*.py
+python_functions = test_*
+addopts = -v --tb=short
@@ -0,0 +1,2 @@
+,#Reads,Aligned_Sequence,Reference_Sequence,n_inserted,n_deleted,n_mutated,Reference_Name,Read_Status,Aligned_Reference_Names,Aligned_Reference_Scores,ref_positions,%Reads
+1,100,AATACGGATGTTCCAATCAGTACGCAGAGAGTCGCCGTCTCCAAGGTGAAAGCGGAAGTAGGGCCTTCGCGCACCTCATGGAATCCCTTCTGCAGCCGCTTTTCCGAGCTTCTGGCGGTCTCAAGCACTACCTACGTCAGCACCTGGGACCCCGCCACCGTGCGCCGGGCCTTGCCGTGGGCGCGCTACCTGCGCCACATCCATCGGCGCTTTGGTCGGCATGGCCCCATTCGCACGGCTCTGGAGCGGC,CGGCCGGATGTTCCAATCAGTACGCAGAGAGTCGCCGTCTCCAAGGTGAAAGCTGAAGTAGGGCCTTCGCGCACCTCATGGAATCCCTTCTGCAGCTTTTCCGAGCTTCTGGCGGTCTCAAGCACTACCTACGTCAGCACCTGGGACCCCGCCACCGTGCGCCGGGCCTTGCAGTGGGCGCGCTACCTGCGCCACATCCATCGGCGCTTTGGTCGG,0,0,0,TEST,MODIFIED,TEST,100&100,"[0,1,2,3]",100.0
@@ -0,0 +1,82 @@
+"""Shared pytest fixtures for CRISPResso2 unit tests."""
+
+import os
+import tempfile
+
+import pytest
+
+
+@pytest.fixture
+def temp_dir():
+    """Provide a temporary directory that's cleaned up after tests."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield tmpdir
+
+
+@pytest.fixture
+def sample_fastq(temp_dir):
+    """Create a sample FASTQ file for testing."""
+    filepath = os.path.join(temp_dir, "sample.fastq")
+    with open(filepath, "w") as f:
+        f.write("@read1\nATCGATCG\n+\nIIIIIIII\n")
+        f.write("@read2\nGCTAGCTA\n+\nIIIIIIII\n")
+    return filepath
+
+
+@pytest.fixture
+def sample_fastq_low_quality(temp_dir):
+    """Create a sample FASTQ file with low quality scores."""
+    filepath = os.path.join(temp_dir, "low_quality.fastq")
+    with open(filepath, "w") as f:
+        f.write("@read1\nATCGATCG\n+\n!!!!!!!!\n")  # Quality 0
+        f.write("@read2\nGCTAGCTA\n+\n########\n")  # Quality 2
+    return filepath
+
+
+@pytest.fixture
+def sample_fastq_mixed_quality(temp_dir):
+    """Create a sample FASTQ file with mixed quality scores."""
+    filepath = os.path.join(temp_dir, "mixed_quality.fastq")
+    with open(filepath, "w") as f:
+        f.write("@read1\nATCGATCG\n+\nIIIIIIII\n")  # High quality
+        f.write("@read2\nGCTAGCTA\n+\n!!!!!!!!\n")  # Low quality
+        f.write("@read3\nAAAAAAAA\n+\nIIIIIIII\n")  # High quality
+    return filepath
+
+
+@pytest.fixture
+def empty_fastq(temp_dir):
+    """Create an empty FASTQ file."""
+    filepath = os.path.join(temp_dir, "empty.fastq")
+    with open(filepath, "w") as f:
+        f.close()
+    return filepath
+
+
+@pytest.fixture
+def aln_matrix():
+    """Load the EDNAFULL alignment matrix."""
+    from CRISPResso2 import CRISPResso2Align
+
+    return CRISPResso2Align.read_matrix("./CRISPResso2/EDNAFULL")
+
+
+@pytest.fixture
+def blosum62_matrix():
+    """Load the BLOSUM62 alignment matrix."""
+    from CRISPResso2 import CRISPResso2Align
+
+    return CRISPResso2Align.read_matrix("./CRISPResso2/BLOSUM62")
+
+
+def create_test_fastq(filepath, records):
+    """Helper function to create test FASTQ files.
+
+    Args:
+        filepath: Path to create the file at
+        records: List of tuples (name, sequence, quality)
+    """
+    with open(filepath, "w") as f:
+        for name, seq, qual in records:
+            f.write(f"@{name}\n{seq}\n+\n{qual}\n")
+    return filepath
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+,#Reads,Aligned_Sequence,Reference_Sequence,n_inserted,n_deleted,n_mutated,Reference_Name,Read_Status,Aligned_Reference_Names,Aligned_Reference_Scores,ref_positions,%Reads`
	`2`	+1,100,AATACGGATGTTCCAATCAGTACGCAGAGAGTCGCCGTCTCCAAGGTGAAAGCGGAAGTAGGGCCTTCGCGCACCTCATGGAATCCCTTCTGCAGCCGCTTTTCCGAGCTTCTGGCGGTCTCAAGCACTACCTACGTCAGCACCTGGGACCCCGCCACCGTGCGCCGGGCCTTGCCGTGGGCGCGCTACCTGCGCCACATCCATCGGCGCTTTGGTCGGCATGGCCCCATTCGCACGGCTCTGGAGCGGC,CGGCCGGATGTTCCAATCAGTACGCAGAGAGTCGCCGTCTCCAAGGTGAAAGCTGAAGTAGGGCCTTCGCGCACCTCATGGAATCCCTTCTGCAGCTTTTCCGAGCTTCTGGCGGTCTCAAGCACTACCTACGTCAGCACCTGGGACCCCGCCACCGTGCGCCGGGCCTTGCAGTGGGCGCGCTACCTGCGCCACATCCATCGGCGCTTTGGTCGG,0,0,0,TEST,MODIFIED,TEST,100&100,"[0,1,2,3]",100.0