hotwa
8d11216481
- Add property-based tests for PixiRunner - Add HAN055.fna test data file - Update README with pixi installation and usage guide - Update .gitignore for pixi and test artifacts - Update CLI to remove Docker-related arguments
477 lines
21 KiB
Python
477 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Property-based tests for PixiRunner.
|
|
|
|
Uses hypothesis library for property-based testing as specified in the design document.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from hypothesis import given, settings, HealthCheck, strategies as st
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
|
from pixi_runner import PixiRunner, PixiRunnerError
|
|
|
|
|
|
def create_mock_pixi_toml(tmp_path: Path) -> Path:
|
|
"""Create a mock pixi.toml file for testing."""
|
|
pixi_toml = tmp_path / "pixi.toml"
|
|
pixi_toml.write_text('[workspace]\nname = "test"\n[environments]\ndigger = ["digger"]\n')
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_pixi_toml(tmp_path: Path) -> Path:
|
|
"""Create a mock pixi.toml file for testing."""
|
|
return create_mock_pixi_toml(tmp_path)
|
|
|
|
|
|
sequence_types = st.sampled_from(["nucl", "orfs", "prot"])
|
|
file_suffixes = st.sampled_from([".fna", ".fasta", ".fa", ".ffn", ".faa"])
|
|
thread_counts = st.integers(min_value=1, max_value=64)
|
|
|
|
|
|
class TestDiggerCommandConstruction:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 1: Digger command construction correctness**
|
|
**Validates: Requirements 2.2**
|
|
"""
|
|
|
|
@given(sequence_type=sequence_types, scaf_suffix=file_suffixes, threads=thread_counts)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_command_starts_with_pixi_run(self, mock_pixi_toml, sequence_type, scaf_suffix, threads):
|
|
"""Command SHALL start with 'pixi run -e digger BtToxin_Digger'."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
cmd = runner.build_digger_command(Path("/test"), sequence_type, scaf_suffix, threads)
|
|
assert cmd[:5] == ["pixi", "run", "-e", "digger", "BtToxin_Digger"]
|
|
|
|
@given(sequence_type=sequence_types, scaf_suffix=file_suffixes, threads=thread_counts)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_command_contains_required_arguments(self, mock_pixi_toml, sequence_type, scaf_suffix, threads):
|
|
"""Command SHALL contain --SeqPath, --SequenceType, and --threads."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
cmd = runner.build_digger_command(Path("/test"), sequence_type, scaf_suffix, threads)
|
|
assert "--SeqPath" in cmd and "--SequenceType" in cmd and "--threads" in cmd
|
|
|
|
@given(sequence_type=sequence_types, scaf_suffix=file_suffixes, threads=thread_counts)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_command_contains_scaf_suffix_for_nucl(self, mock_pixi_toml, sequence_type, scaf_suffix, threads):
|
|
"""Command SHALL contain --Scaf_suffix when sequence_type is 'nucl'."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
cmd = runner.build_digger_command(Path("/test"), sequence_type, scaf_suffix, threads)
|
|
if sequence_type == "nucl":
|
|
assert "--Scaf_suffix" in cmd
|
|
idx = cmd.index("--Scaf_suffix")
|
|
assert cmd[idx + 1] == scaf_suffix
|
|
|
|
|
|
class TestResultDictionaryCompleteness:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 2: Result dictionary completeness**
|
|
**Validates: Requirements 2.3**
|
|
"""
|
|
|
|
@given(exit_code=st.integers(-128, 255), stdout=st.text(max_size=50), stderr=st.text(max_size=50))
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_result_contains_required_keys(self, mock_pixi_toml, exit_code, stdout, stderr):
|
|
"""Result SHALL contain 'success', 'exit_code', 'logs', and 'status'."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
mock_result = MagicMock(returncode=exit_code, stdout=stdout, stderr=stderr)
|
|
with patch("pixi_runner.subprocess.run", return_value=mock_result):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
inp, out, log = Path(tmpdir)/"input", Path(tmpdir)/"output", Path(tmpdir)/"logs"
|
|
inp.mkdir()
|
|
(inp / "test.fna").write_text(">s\nA\n")
|
|
result = runner.run_bttoxin_digger(inp, out, log)
|
|
assert all(k in result for k in ["success", "exit_code", "logs", "status"])
|
|
assert isinstance(result["success"], bool) and isinstance(result["exit_code"], int)
|
|
|
|
|
|
class TestFailureHandling:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 3: Failure status on non-zero exit**
|
|
**Validates: Requirements 2.4**
|
|
"""
|
|
|
|
@given(exit_code=st.integers(1, 255))
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_nonzero_exit_returns_failure(self, mock_pixi_toml, exit_code):
|
|
"""Non-zero exit SHALL return success=False and status='failed'."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
mock_result = MagicMock(returncode=exit_code, stdout="", stderr="Error")
|
|
with patch("pixi_runner.subprocess.run", return_value=mock_result):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
inp, out, log = Path(tmpdir)/"input", Path(tmpdir)/"output", Path(tmpdir)/"logs"
|
|
inp.mkdir()
|
|
(inp / "test.fna").write_text(">s\nA\n")
|
|
result = runner.run_bttoxin_digger(inp, out, log)
|
|
assert result["success"] is False and result["status"] == "failed"
|
|
|
|
|
|
class TestErrorMessageGuidance:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 7: Error message contains actionable guidance**
|
|
**Validates: Requirements 5.1, 5.2, 5.3, 5.4**
|
|
"""
|
|
|
|
def test_pixi_not_installed_error_contains_guidance(self, mock_pixi_toml):
|
|
"""Error message SHALL contain actionable instructions."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
with patch("pixi_runner.subprocess.run", side_effect=FileNotFoundError("pixi")):
|
|
result = runner.check_environment()
|
|
assert result["error"] and any(k in result["error"].lower() for k in ["install", "pixi", "run"])
|
|
|
|
@given(error_type=st.sampled_from(["pixi_missing", "env_missing"]))
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_all_errors_contain_guidance(self, mock_pixi_toml, error_type):
|
|
"""All error types SHALL contain actionable guidance."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="nonexistent")
|
|
def mock_run(cmd, **kw):
|
|
m = MagicMock(returncode=0, stdout='{"environments_info":[]}', stderr="")
|
|
if error_type == "pixi_missing":
|
|
raise FileNotFoundError()
|
|
return m
|
|
with patch("pixi_runner.subprocess.run", side_effect=mock_run):
|
|
result = runner.check_environment()
|
|
assert result["error"] and any(k in result["error"].lower() for k in ["install", "pixi", "run"])
|
|
|
|
|
|
class TestShotterCommandConstruction:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 4: Shotter command uses pipeline environment**
|
|
**Validates: Requirements 3.2**
|
|
"""
|
|
|
|
@given(
|
|
min_identity=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
min_coverage=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
allow_unknown=st.booleans(),
|
|
require_index=st.booleans(),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_shotter_command_starts_with_pixi_pipeline(
|
|
self, mock_pixi_toml, min_identity, min_coverage, allow_unknown, require_index
|
|
):
|
|
"""Shotter command SHALL start with 'pixi run -e pipeline python'."""
|
|
from pixi_runner import build_shotter_command
|
|
|
|
cmd = build_shotter_command(
|
|
pixi_project_dir=mock_pixi_toml,
|
|
script_path=Path("/scripts/bttoxin_shoter.py"),
|
|
toxicity_csv=Path("/data/toxicity.csv"),
|
|
all_toxins=Path("/output/All_Toxins.txt"),
|
|
output_dir=Path("/output"),
|
|
min_identity=min_identity,
|
|
min_coverage=min_coverage,
|
|
allow_unknown_families=allow_unknown,
|
|
require_index_hit=require_index,
|
|
)
|
|
assert cmd[:4] == ["pixi", "run", "-e", "pipeline"]
|
|
assert cmd[4] == "python"
|
|
|
|
@given(
|
|
min_identity=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
min_coverage=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_shotter_command_contains_script_path(
|
|
self, mock_pixi_toml, min_identity, min_coverage
|
|
):
|
|
"""Shotter command SHALL include the bttoxin_shoter.py script path."""
|
|
from pixi_runner import build_shotter_command
|
|
|
|
script = Path("/scripts/bttoxin_shoter.py")
|
|
cmd = build_shotter_command(
|
|
pixi_project_dir=mock_pixi_toml,
|
|
script_path=script,
|
|
toxicity_csv=Path("/data/toxicity.csv"),
|
|
all_toxins=Path("/output/All_Toxins.txt"),
|
|
output_dir=Path("/output"),
|
|
min_identity=min_identity,
|
|
min_coverage=min_coverage,
|
|
)
|
|
assert str(script) in cmd
|
|
|
|
|
|
|
|
class TestPlotCommandConstruction:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 5: Plot command uses pipeline environment**
|
|
**Validates: Requirements 3.3**
|
|
"""
|
|
|
|
@given(
|
|
merge_unresolved=st.booleans(),
|
|
report_mode=st.sampled_from(["paper", "summary"]),
|
|
lang=st.sampled_from(["zh", "en"]),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_plot_command_starts_with_pixi_pipeline(
|
|
self, mock_pixi_toml, merge_unresolved, report_mode, lang
|
|
):
|
|
"""Plot command SHALL start with 'pixi run -e pipeline python'."""
|
|
from pixi_runner import build_plot_command
|
|
|
|
cmd = build_plot_command(
|
|
pixi_project_dir=mock_pixi_toml,
|
|
script_path=Path("/scripts/plot_shotter.py"),
|
|
strain_scores=Path("/output/strain_target_scores.tsv"),
|
|
toxin_support=Path("/output/toxin_support.tsv"),
|
|
species_scores=Path("/output/strain_target_species_scores.tsv"),
|
|
out_dir=Path("/output"),
|
|
merge_unresolved=merge_unresolved,
|
|
report_mode=report_mode,
|
|
lang=lang,
|
|
)
|
|
assert cmd[:4] == ["pixi", "run", "-e", "pipeline"]
|
|
assert cmd[4] == "python"
|
|
|
|
@given(
|
|
merge_unresolved=st.booleans(),
|
|
report_mode=st.sampled_from(["paper", "summary"]),
|
|
lang=st.sampled_from(["zh", "en"]),
|
|
per_hit_strain=st.one_of(st.none(), st.text(min_size=1, max_size=20, alphabet=st.characters(whitelist_categories=('L', 'N')))),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_plot_command_contains_script_path(
|
|
self, mock_pixi_toml, merge_unresolved, report_mode, lang, per_hit_strain
|
|
):
|
|
"""Plot command SHALL include the plot_shotter.py script path."""
|
|
from pixi_runner import build_plot_command
|
|
|
|
script = Path("/scripts/plot_shotter.py")
|
|
cmd = build_plot_command(
|
|
pixi_project_dir=mock_pixi_toml,
|
|
script_path=script,
|
|
strain_scores=Path("/output/strain_target_scores.tsv"),
|
|
toxin_support=Path("/output/toxin_support.tsv"),
|
|
species_scores=Path("/output/strain_target_species_scores.tsv"),
|
|
out_dir=Path("/output"),
|
|
merge_unresolved=merge_unresolved,
|
|
report_mode=report_mode,
|
|
lang=lang,
|
|
per_hit_strain=per_hit_strain,
|
|
)
|
|
assert str(script) in cmd
|
|
|
|
|
|
|
|
class TestBundleCreation:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 6: Bundle creation correctness**
|
|
**Validates: Requirements 3.5**
|
|
"""
|
|
|
|
@given(
|
|
digger_files=st.lists(st.text(min_size=1, max_size=20, alphabet=st.characters(whitelist_categories=('L', 'N'))), min_size=0, max_size=5),
|
|
shotter_files=st.lists(st.text(min_size=1, max_size=20, alphabet=st.characters(whitelist_categories=('L', 'N'))), min_size=0, max_size=5),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
|
|
def test_bundle_contains_correct_arcnames(self, digger_files, shotter_files):
|
|
"""Bundle SHALL contain directories with correct arcnames ('digger' and 'shotter')."""
|
|
from pixi_runner import create_pipeline_bundle, verify_bundle_contents
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
tmp = Path(tmpdir)
|
|
digger_dir = tmp / "digger_output"
|
|
shotter_dir = tmp / "shotter_output"
|
|
bundle_path = tmp / "test_bundle.tar.gz"
|
|
|
|
# Create directories with some files
|
|
digger_dir.mkdir(parents=True, exist_ok=True)
|
|
shotter_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
for f in digger_files:
|
|
if f: # Skip empty strings
|
|
(digger_dir / f"{f}.txt").write_text("test")
|
|
for f in shotter_files:
|
|
if f: # Skip empty strings
|
|
(shotter_dir / f"{f}.txt").write_text("test")
|
|
|
|
# Create bundle
|
|
success = create_pipeline_bundle(bundle_path, digger_dir, shotter_dir)
|
|
assert success
|
|
|
|
# Verify contents
|
|
result = verify_bundle_contents(bundle_path)
|
|
|
|
# Check arcnames
|
|
if digger_files:
|
|
assert result["has_digger"], "Bundle should contain 'digger' directory"
|
|
assert any(m.startswith("digger/") or m == "digger" for m in result["members"])
|
|
if shotter_files:
|
|
assert result["has_shotter"], "Bundle should contain 'shotter' directory"
|
|
assert any(m.startswith("shotter/") or m == "shotter" for m in result["members"])
|
|
|
|
@given(
|
|
has_digger=st.booleans(),
|
|
has_shotter=st.booleans(),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_bundle_handles_missing_directories(self, has_digger, has_shotter):
|
|
"""Bundle creation SHALL handle missing directories gracefully."""
|
|
from pixi_runner import create_pipeline_bundle, verify_bundle_contents
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
tmp = Path(tmpdir)
|
|
digger_dir = tmp / "digger_output"
|
|
shotter_dir = tmp / "shotter_output"
|
|
bundle_path = tmp / "test_bundle.tar.gz"
|
|
|
|
# Conditionally create directories
|
|
if has_digger:
|
|
digger_dir.mkdir(parents=True, exist_ok=True)
|
|
(digger_dir / "test.txt").write_text("digger content")
|
|
if has_shotter:
|
|
shotter_dir.mkdir(parents=True, exist_ok=True)
|
|
(shotter_dir / "test.txt").write_text("shotter content")
|
|
|
|
# Create bundle
|
|
success = create_pipeline_bundle(bundle_path, digger_dir, shotter_dir)
|
|
assert success
|
|
|
|
# Verify contents match what was created
|
|
result = verify_bundle_contents(bundle_path)
|
|
assert result["has_digger"] == has_digger
|
|
assert result["has_shotter"] == has_shotter
|
|
|
|
|
|
|
|
class TestCLIArgumentPassthrough:
|
|
"""
|
|
**Feature: pixi-conda-migration, Property 8: CLI argument passthrough**
|
|
**Validates: Requirements 4.3, 6.4**
|
|
"""
|
|
|
|
@given(
|
|
min_identity=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
min_coverage=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
allow_unknown=st.booleans(),
|
|
require_index=st.booleans(),
|
|
lang=st.sampled_from(["zh", "en"]),
|
|
threads=st.integers(min_value=1, max_value=64),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_shotter_args_passthrough(
|
|
self, mock_pixi_toml, min_identity, min_coverage, allow_unknown, require_index, lang, threads
|
|
):
|
|
"""CLI arguments SHALL be correctly passed to shotter command without modification."""
|
|
from pixi_runner import build_shotter_command
|
|
|
|
cmd = build_shotter_command(
|
|
pixi_project_dir=mock_pixi_toml,
|
|
script_path=Path("/scripts/bttoxin_shoter.py"),
|
|
toxicity_csv=Path("/data/toxicity.csv"),
|
|
all_toxins=Path("/output/All_Toxins.txt"),
|
|
output_dir=Path("/output"),
|
|
min_identity=min_identity,
|
|
min_coverage=min_coverage,
|
|
allow_unknown_families=allow_unknown,
|
|
require_index_hit=require_index,
|
|
)
|
|
|
|
# Verify min_identity is passed correctly
|
|
if min_identity > 0:
|
|
assert "--min_identity" in cmd
|
|
idx = cmd.index("--min_identity")
|
|
assert float(cmd[idx + 1]) == min_identity
|
|
|
|
# Verify min_coverage is passed correctly
|
|
if min_coverage > 0:
|
|
assert "--min_coverage" in cmd
|
|
idx = cmd.index("--min_coverage")
|
|
assert float(cmd[idx + 1]) == min_coverage
|
|
|
|
# Verify allow_unknown_families flag
|
|
if not allow_unknown:
|
|
assert "--disallow_unknown_families" in cmd
|
|
else:
|
|
assert "--disallow_unknown_families" not in cmd
|
|
|
|
# Verify require_index_hit flag
|
|
if require_index:
|
|
assert "--require_index_hit" in cmd
|
|
else:
|
|
assert "--require_index_hit" not in cmd
|
|
|
|
@given(
|
|
merge_unresolved=st.booleans(),
|
|
report_mode=st.sampled_from(["paper", "summary"]),
|
|
lang=st.sampled_from(["zh", "en"]),
|
|
per_hit_strain=st.one_of(st.none(), st.text(min_size=1, max_size=20, alphabet=st.characters(whitelist_categories=('L', 'N')))),
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_plot_args_passthrough(
|
|
self, mock_pixi_toml, merge_unresolved, report_mode, lang, per_hit_strain
|
|
):
|
|
"""CLI arguments SHALL be correctly passed to plot command without modification."""
|
|
from pixi_runner import build_plot_command
|
|
|
|
cmd = build_plot_command(
|
|
pixi_project_dir=mock_pixi_toml,
|
|
script_path=Path("/scripts/plot_shotter.py"),
|
|
strain_scores=Path("/output/strain_target_scores.tsv"),
|
|
toxin_support=Path("/output/toxin_support.tsv"),
|
|
species_scores=Path("/output/strain_target_species_scores.tsv"),
|
|
out_dir=Path("/output"),
|
|
merge_unresolved=merge_unresolved,
|
|
report_mode=report_mode,
|
|
lang=lang,
|
|
per_hit_strain=per_hit_strain,
|
|
)
|
|
|
|
# Verify merge_unresolved flag
|
|
if merge_unresolved:
|
|
assert "--merge_unresolved" in cmd
|
|
else:
|
|
assert "--merge_unresolved" not in cmd
|
|
|
|
# Verify report_mode is passed correctly
|
|
assert "--report_mode" in cmd
|
|
idx = cmd.index("--report_mode")
|
|
assert cmd[idx + 1] == report_mode
|
|
|
|
# Verify lang is passed correctly
|
|
assert "--lang" in cmd
|
|
idx = cmd.index("--lang")
|
|
assert cmd[idx + 1] == lang
|
|
|
|
# Verify per_hit_strain is passed correctly when provided
|
|
if per_hit_strain:
|
|
assert "--per_hit_strain" in cmd
|
|
idx = cmd.index("--per_hit_strain")
|
|
assert cmd[idx + 1] == per_hit_strain
|
|
|
|
@given(
|
|
sequence_type=sequence_types,
|
|
scaf_suffix=file_suffixes,
|
|
threads=thread_counts,
|
|
)
|
|
@settings(max_examples=100, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
|
def test_digger_args_passthrough(
|
|
self, mock_pixi_toml, sequence_type, scaf_suffix, threads
|
|
):
|
|
"""CLI arguments SHALL be correctly passed to digger command without modification."""
|
|
runner = PixiRunner(pixi_project_dir=mock_pixi_toml, env_name="digger")
|
|
cmd = runner.build_digger_command(Path("/test"), sequence_type, scaf_suffix, threads)
|
|
|
|
# Verify sequence_type is passed correctly
|
|
assert "--SequenceType" in cmd
|
|
idx = cmd.index("--SequenceType")
|
|
assert cmd[idx + 1] == sequence_type
|
|
|
|
# Verify threads is passed correctly
|
|
assert "--threads" in cmd
|
|
idx = cmd.index("--threads")
|
|
assert int(cmd[idx + 1]) == threads
|
|
|
|
# Verify scaf_suffix is passed correctly for nucl type
|
|
if sequence_type == "nucl":
|
|
assert "--Scaf_suffix" in cmd
|
|
idx = cmd.index("--Scaf_suffix")
|
|
assert cmd[idx + 1] == scaf_suffix
|