1+ import pytest
2+ import subprocess
3+ import os
4+ import shutil
5+ import pandas as pd
6+ import numpy as np
7+ import bz2
8+ import sys
9+
10+ @pytest .fixture (scope = "module" )
11+ def demo_env ():
12+ """
13+ Sets up the test environment
14+ """
15+ project_root = os .path .dirname (os .path .dirname (os .path .abspath (__file__ )))
16+ demo_dir = os .path .join (project_root , "demo_data" )
17+ out_dir = os .path .join (project_root , "tests/test_out_pytest" )
18+
19+ if os .path .exists (out_dir ):
20+ shutil .rmtree (out_dir )
21+ os .makedirs (out_dir )
22+
23+ files_to_decompress = [
24+ "SRR12996245.1pct.kraken" ,
25+ "SRR12996245.1pct_1.fastq" ,
26+ "SRR12996245.1pct_2.fastq"
27+ ]
28+
29+ print (f"\n [Setup] Preparing test data in { out_dir } ..." )
30+
31+ for filename in files_to_decompress :
32+ source_bz2 = os .path .join (demo_dir , filename + ".bz2" )
33+ dest_file = os .path .join (out_dir , filename )
34+
35+ if os .path .exists (source_bz2 ):
36+ with bz2 .open (source_bz2 , "rb" ) as f_in , open (dest_file , "wb" ) as f_out :
37+ shutil .copyfileobj (f_in , f_out )
38+ else :
39+ pytest .fail (f"Required source file missing: { source_bz2 } " )
40+
41+ # Yield paths so the test function knows where things are
42+ yield {'out' : out_dir , 'demo' : demo_dir }
43+
44+ def test_guacamole_metrics (demo_env ):
45+ """
46+ Runs guacamole and asserts efficiency and abundance ranges.
47+ """
48+ out_dir = demo_env ['out' ]
49+ demo_dir = demo_env ['demo' ]
50+ output_file_name = "SRR12996245.1pct.guaca"
51+
52+ # Construct command
53+ cmd = [
54+ sys .executable , "-m" , "guacamole.guacamole" ,
55+ "--output" , output_file_name ,
56+ "--kraken_report" , os .path .join (demo_dir , "SRR12996245.1pct_report.txt" ),
57+ "--kraken_file" , os .path .join (out_dir , "SRR12996245.1pct.kraken" ),
58+ "--read_files" , os .path .join (out_dir , "SRR12996245.1pct_1.fastq" ), os .path .join (out_dir , "SRR12996245.1pct_2.fastq" ),
59+ "--kraken_db" , os .path .join (demo_dir , "demo_db" ),
60+ "--read_len" , "150" ,
61+ "--fragment_len" , "400" ,
62+ "--length_correction" , "True" ,
63+ "--threshold" , "5" ,
64+ "--plot" , "False"
65+ ]
66+
67+ # Run Guacamole
68+ result = subprocess .run (
69+ cmd , cwd = out_dir , stdout = subprocess .PIPE , stderr = subprocess .PIPE , text = True
70+ )
71+
72+ # Assert successful run
73+ assert result .returncode == 0 , f"Command failed with stderr: { result .stderr } "
74+
75+ # 1. Check Efficiencies (50% - 70%)
76+ eff_path = os .path .join (out_dir , "efficiencies.txt" )
77+ assert os .path .exists (eff_path ), "efficiencies.txt was not created"
78+
79+ efficiencies = np .loadtxt (eff_path )
80+ max_eff = np .argmax (efficiencies )
81+
82+ assert 20 <= max_eff <= 30 , f"Max efficiency { max_eff :.2f} is out of range [20, 30]"
83+
84+ # 2. Check Abundances (4% - 12%)
85+ guaca_path = os .path .join (out_dir , output_file_name )
86+ assert os .path .exists (guaca_path ), f"{ output_file_name } was not created"
87+
88+ df = pd .read_csv (guaca_path , sep = "\t " )
89+ abundances = df ['GuaCAMOLE_estimate' ]
90+
91+ # Vectorized check (faster and cleaner than a loop)
92+ assert (np .sum (abundances ) == pytest .approx (1.0 , abs = 0.01 )), f"Sum of abundances is not 1: { np .sum (abundances )} "
93+ assert (abundances >= 0.03 ).all (), f"Some abundances are < 4%:\n { df [abundances < 0.04 ]} "
94+ assert (abundances <= 0.12 ).all (), f"Some abundances are > 12%:\n { df [abundances > 0.12 ]} "
0 commit comments