#!/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