Merge pull request #4 from codefitz/codex/add-tests-for-vcf-header-and-duplicates

Improve VCF header validation

Author: codefitz

tests/test_vcf_validation.py

@@ -0,0 +1,64 @@
+import tempfile
+import os
+import unittest
+from vcf_validation import validate_vcf
+
+class TestVCFValidation(unittest.TestCase):
+    def _write_temp_vcf(self, content):
+        tmp = tempfile.NamedTemporaryFile(delete=False, mode='w', suffix='.vcf')
+        tmp.write(content)
+        tmp.flush()
+        tmp.close()
+        self.addCleanup(lambda: os.remove(tmp.name))
+        return tmp.name
+
+    def test_valid_file(self):
+        content = """##fileformat=VCFv4.2
+##contig=<ID=1>
+#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tS1\tS2
+1\t1\tGAIN1\tA\t<CNV>\t30\tPASS\tSVTYPE=CNV\tCN:GT\t0/1\t1/1
+"""
+        path = self._write_temp_vcf(content)
+        # Should not raise SystemExit
+        validate_vcf(path)
+
+    def test_missing_fileformat(self):
+        content = """##contig=<ID=1>
+#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tS1
+1\t1\tGAIN1\tA\t<CNV>\t30\tPASS\tSVTYPE=CNV\tCN:GT\t0/1
+"""
+        path = self._write_temp_vcf(content)
+        with self.assertRaises(SystemExit):
+            validate_vcf(path)
+
+    def test_missing_chrom_header(self):
+        content = """##fileformat=VCFv4.2
+##contig=<ID=1>
+1\t1\tGAIN1\tA\t<CNV>\t30\tPASS\tSVTYPE=CNV\tCN:GT\t0/1
+"""
+        path = self._write_temp_vcf(content)
+        with self.assertRaises(SystemExit):
+            validate_vcf(path)
+
+    def test_missing_format_with_genotypes(self):
+        content = """##fileformat=VCFv4.2
+##contig=<ID=1>
+#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tS1
+1\t1\tGAIN1\tA\t<CNV>\t30\tPASS\tSVTYPE=CNV\t0/1
+"""
+        path = self._write_temp_vcf(content)
+        with self.assertRaises(SystemExit):
+            validate_vcf(path)
+
+    def test_duplicate_sample_names(self):
+        content = """##fileformat=VCFv4.2
+##contig=<ID=1>
+#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tS1\tS1
+1\t1\tGAIN1\tA\t<CNV>\t30\tPASS\tSVTYPE=CNV\tCN:GT\t0/1\t0/1
+"""
+        path = self._write_temp_vcf(content)
+        with self.assertRaises(SystemExit):
+            validate_vcf(path)
+
+if __name__ == '__main__':
+    unittest.main()

vcf_validation.py

@@ -19,6 +19,29 @@
 # 1.0.0 : WMF : Created.
 # 1.0.1 : WMF : Updated with Congenica rules.
 # 1.0.2 : WMF : Added support for bgzipped files. Updated error message for Alternate Alleles.
+# 1.0.3 : WMF : Added header validations and duplicate sample detection.
+#
+# Header line syntax
+# ------------------
+# The header line names the 8 fixed, mandatory columns. These columns are as follows:
+#   1. #CHROM
+#   2. POS
+#   3. ID
+#   4. REF
+#   5. ALT
+#   6. QUAL
+#   7. FILTER
+#   8. INFO
+# If genotype data is present in the file, these are followed by a FORMAT column header, then an arbitrary number
+# of sample IDs. Duplicate sample IDs are not allowed. The header line is tab-delimited.
+#
+# Congenica Strict Rules
+# contig = ID=1-22, ID=chr* - this fails
+# Must have a FORMAT field (9 columns)
+# INFO Must contain SVTYPE=CNV
+# ALT must be <CNV> for CNV types
+# ID must contain "LOSS" or "GAIN"
+# FORMAT field must have "CN"
 
 import sys
 import re
@@ -119,20 +142,33 @@ def validate_vcf(vcf_file, strict=False, report=False):
     with open_func(vcf_file, 'rt') as file:
 
         line_number = 0
+        fileformat_found = False
+        header_found = False
         for line in file:
             line_number += 1
             if line.startswith("##"):
-                if strict and line.startswith("##contig"):
+                if line.startswith("##fileformat"):
+                    fileformat_found = True
+                if line.startswith("##contig"):
                     contig_info = line.split('<',1)[1].split('>')[0]
                     id_info = [x for x in contig_info.split(',') if x.startswith('ID=')]
                     if id_info:
                         contig_id = id_info[0].split('=')[1]
                         if contig_id.startswith("chr"):
                             print(f"Error: Contig ID starts with 'chr' on line {line_number}: {line.strip()}")
                             sys.exit(1)
-                            
                 continue
             elif line.startswith("#CHROM"):
+                header_found = True
+                header_fields = line.strip().split('\t')
+                if len(header_fields) > 8 and header_fields[8] != "FORMAT":
+                    print("Error: FORMAT column missing from header line")
+                    sys.exit(1)
+                if len(header_fields) > 9:
+                    sample_names = header_fields[9:]
+                    if len(sample_names) != len(set(sample_names)):
+                        print("Error: Duplicate sample names in header line")
+                        sys.exit(1)
                 continue
             else:
                 fields = line.strip().split('\t')
@@ -150,6 +186,13 @@ def validate_vcf(vcf_file, strict=False, report=False):
                 validate_info(fields[7], line_number, line)
                 validate_format(fields[8], line_number, line)
 
+        if not fileformat_found:
+            print("Error: Missing ##fileformat header")
+            sys.exit(1)
+        if not header_found:
+            print("Error: Missing #CHROM header line")
+            sys.exit(1)
+
     if report:
         print("VCF file validation completed. No structural errors found.")