"""Tests for testing_engine.pipeline module."""
import pytest
from pathlib import Path
from unittest.mock import Mock, patch, MagicMock
from testing_engine.pipeline import full_test_pipeline, _extract_coverage
class TestFullTestPipeline:
"""Test suite for full_test_pipeline function."""
def test_full_test_pipeline_creates_test_folders(self, tmp_path):
"""Test that the pipeline creates required test folder structure."""
# Create a simple module
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
# Mock the subprocess to return success
mock_process = Mock()
mock_process.communicate.return_value = ("All tests passed\n", None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify test folders were created
assert (tmp_path / "tests" / "unit").exists()
assert (tmp_path / "tests" / "integration").exists()
assert (tmp_path / "tests" / "e2e").exists()
assert (tmp_path / "tests" / "snapshots").exists()
def test_full_test_pipeline_generates_tests(self, tmp_path):
"""Test that the pipeline generates unit tests."""
# Create a simple module
module = tmp_path / "math_utils.py"
module.write_text("def multiply(x, y): return x * y")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
mock_process.communicate.return_value = ("1 passed in 0.01s\n", None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify test was generated
assert result["tests_generated"] == 1
# Verify test file exists
test_file = tmp_path / "tests" / "unit" / "test_math_utils.py"
assert test_file.exists()
def test_full_test_pipeline_runs_pytest(self, tmp_path):
"""Test that the pipeline executes pytest."""
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
mock_output = "===== test session starts =====\n5 passed in 0.50s\n"
mock_process.communicate.return_value = (mock_output, None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify pytest was called
mock_popen.assert_called_once()
call_args = mock_popen.call_args
assert "pytest" in call_args.kwargs.get("args", call_args.args[0])
def test_full_test_pipeline_custom_test_command(self, tmp_path):
"""Test that the pipeline accepts a custom test command."""
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
custom_command = "pytest --verbose --maxfail=5"
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
mock_process.communicate.return_value = ("Tests completed\n", None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(
str(tmp_path),
language="python",
test_command=custom_command,
)
# Verify custom command was used
call_args = mock_popen.call_args
assert custom_command in str(call_args)
def test_full_test_pipeline_detects_failures(self, tmp_path):
"""Test that the pipeline detects test failures."""
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
mock_output = "===== FAILED test_something =====\n1 failed, 2 passed\n"
mock_process.communicate.return_value = (mock_output, None)
mock_process.returncode = 1
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify failure was detected
assert result["failures"] == 1
assert result["exit_code"] == 1
def test_full_test_pipeline_extracts_coverage(self, tmp_path):
"""Test that the pipeline extracts coverage percentage."""
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
mock_output = """
===== test session starts =====
Coverage: 87.5%
3 passed in 0.25s
"""
mock_process.communicate.return_value = (mock_output, None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify coverage was extracted
assert result["coverage_percent"] == 87.5
def test_full_test_pipeline_writes_report(self, tmp_path):
"""Test that the pipeline writes test output to a report file."""
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
test_output = "Test output content\n5 passed in 0.10s\n"
mock_process.communicate.return_value = (test_output, None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify report file was created
report_file = tmp_path / ".mcp" / "test_output.log"
assert report_file.exists()
# Verify content was written
content = report_file.read_text()
assert test_output in content
# Verify report path is in result
assert str(report_file) in result["report_files"]
def test_full_test_pipeline_returns_result_dict(self, tmp_path):
"""Test that the pipeline returns a properly structured result dictionary."""
module = tmp_path / "calculator.py"
module.write_text("def add(a, b): return a + b")
with patch('testing_engine.pipeline.subprocess.Popen') as mock_popen:
mock_process = Mock()
mock_process.communicate.return_value = ("All tests passed\n", None)
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = full_test_pipeline(str(tmp_path), language="python")
# Verify result structure
assert "tests_generated" in result
assert "tests_run" in result
assert "failures" in result
assert "coverage_percent" in result
assert "report_files" in result
assert "exit_code" in result
assert isinstance(result["tests_generated"], int)
assert isinstance(result["exit_code"], int)
assert isinstance(result["report_files"], list)
class TestExtractCoverage:
"""Test suite for _extract_coverage helper function."""
def test_extract_coverage_from_pytest_output(self):
"""Test extracting coverage percentage from pytest output."""
text = """
===== test session starts =====
collected 10 items
tests/test_foo.py .......... [100%]
Coverage: 85.7%
===== 10 passed in 0.50s =====
"""
coverage = _extract_coverage(text)
assert coverage == 85.7
def test_extract_coverage_with_multiple_percentages(self):
"""Test that the first percentage is extracted."""
text = "Coverage: 92.3% of code covered. Total: 45.6% executed."
coverage = _extract_coverage(text)
assert coverage == 92.3
def test_extract_coverage_no_match(self):
"""Test that -1.0 is returned when no coverage found."""
text = "All tests passed without coverage information"
coverage = _extract_coverage(text)
assert coverage == -1.0
def test_extract_coverage_zero_percent(self):
"""Test extracting 0% coverage."""
text = "Coverage: 0.0% - no code covered"
coverage = _extract_coverage(text)
assert coverage == 0.0
def test_extract_coverage_hundred_percent(self):
"""Test extracting 100% coverage."""
text = "Coverage: 100.0% - all code covered"
coverage = _extract_coverage(text)
assert coverage == 100.0
def test_extract_coverage_integer_format(self):
"""Test extracting coverage in integer format."""
text = "Coverage report: 75.0%"
coverage = _extract_coverage(text)
assert coverage == 75.0
