fix: implement mutation testing with mutmut v3 (#280) (#724)

* fix: implement mutation testing with mutmut v3 (#280)

- Migrate mutmut configuration from setup.cfg to pyproject.toml
- Add workaround script for mutmut v3 stats collection issues
- Implement sample-based mutation testing (20 mutants by default)
- Add full mutation testing mode for comprehensive coverage
- Handle Python 3.11+ except* syntax with pragma comments
- Include plugins directory in mutation test environment
- Update Makefile with mutmut-run and mutmut-run-full targets
- Remove deprecated setup.cfg file

The mutation testing now actually runs tests against code mutations
to verify test effectiveness, providing a mutation score metric.

Signed-off-by: Mihai Criveti <[email protected]>

* Plugins now support resources pre/post

Signed-off-by: Mihai Criveti <[email protected]>

---------

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

.gitignore

@@ -1,3 +1,5 @@
+mutants
+.mutmut-cache
 CLAUDE.local.md
 .aider*
 .scannerwork

Makefile

@@ -29,7 +29,7 @@ TEST_DOCS_DIR ?= $(DOCS_DIR)/docs/test
 DIRS_TO_CLEAN := __pycache__ .pytest_cache .tox .ruff_cache .pyre .mypy_cache .pytype \
 	dist build site .eggs *.egg-info .cache htmlcov certs \
 	$(VENV_DIR) $(VENV_DIR).sbom $(COVERAGE_DIR) \
-	node_modules
+	node_modules .mutmut-cache html
 
 FILES_TO_CLEAN := .coverage coverage.xml mcp.prof mcp.pstats \
 	$(PROJECT_NAME).sbom.json \
@@ -310,6 +310,81 @@ doctest-check:
 		python3 -m pytest --doctest-modules mcpgateway/ --tb=no -q && \
 		echo '✅ All doctests passing' || (echo '❌ Doctest failures detected' && exit 1)"
 
+# =============================================================================
+# 🧬 MUTATION TESTING
+# =============================================================================
+# help: 🧬 MUTATION TESTING
+# help: mutmut-install       - Install mutmut in development virtualenv
+# help: mutmut-run           - Run mutation testing (sample of 20 mutants for quick results)
+# help: mutmut-run-full      - Run FULL mutation testing (all 11,000+ mutants - takes hours!)
+# help: mutmut-results       - Display mutation testing summary and surviving mutants
+# help: mutmut-html          - Generate browsable HTML report of mutation results
+# help: mutmut-ci            - CI-friendly mutation testing with score threshold enforcement
+# help: mutmut-clean         - Clean mutmut cache and results
+
+.PHONY: mutmut-install mutmut-run mutmut-results mutmut-html mutmut-ci mutmut-clean
+
+mutmut-install:
+	@echo "📥 Installing mutmut..."
+	@test -d "$(VENV_DIR)" || $(MAKE) venv
+	@/bin/bash -c "source $(VENV_DIR)/bin/activate && \
+		python3 -m pip install -q mutmut==3.3.1"
+
+mutmut-run: mutmut-install
+	@echo "🧬 Running mutation testing (sample mode - 20 mutants)..."
+	@echo "⏳ This should take about 2-3 minutes..."
+	@echo "📝 Target: mcpgateway/ directory"
+	@/bin/bash -c "source $(VENV_DIR)/bin/activate && \
+		cd $(PWD) && \
+		PYTHONPATH=$(PWD) python run_mutmut.py --sample"
+
+mutmut-run-full: mutmut-install
+	@echo "🧬 Running FULL mutation testing (all mutants)..."
+	@echo "⏰ WARNING: This will take a VERY long time (hours)!"
+	@echo "📝 Target: mcpgateway/ directory (11,000+ mutants)"
+	@/bin/bash -c "source $(VENV_DIR)/bin/activate && \
+		cd $(PWD) && \
+		PYTHONPATH=$(PWD) python run_mutmut.py --full"
+
+mutmut-results:
+	@echo "📊 Mutation testing results:"
+	@test -d "$(VENV_DIR)" || $(MAKE) venv
+	@/bin/bash -c "source $(VENV_DIR)/bin/activate && \
+		mutmut results || echo '⚠️  No mutation results found. Run make mutmut-run first.'"
+
+mutmut-html:
+	@echo "📄 Generating HTML mutation report..."
+	@test -d "$(VENV_DIR)" || $(MAKE) venv
+	@/bin/bash -c "source $(VENV_DIR)/bin/activate && \
+		mutmut html || echo '⚠️  No mutation results found. Run make mutmut-run first.'"
+	@[ -f html/index.html ] && echo "✅ Report available at: file://$$(pwd)/html/index.html" || true
+
+mutmut-ci: mutmut-install
+	@echo "🔍 CI mutation testing with threshold check..."
+	@echo "⚠️  Excluding gateway_service.py (uses Python 3.11+ except* syntax)"
+	@/bin/bash -c "source $(VENV_DIR)/bin/activate && \
+		cd $(PWD) && \
+		PYTHONPATH=$(PWD) mutmut run && \
+		python3 -c 'import subprocess, sys; \
+			result = subprocess.run([\"mutmut\", \"results\"], capture_output=True, text=True); \
+			import re; \
+			match = re.search(r\"killed: (\\d+) out of (\\d+)\", result.stdout); \
+			if match: \
+				killed, total = int(match.group(1)), int(match.group(2)); \
+				score = (killed / total * 100) if total > 0 else 0; \
+				print(f\"Mutation score: {score:.1f}% ({killed}/{total} killed)\"); \
+				sys.exit(0 if score >= 75 else 1); \
+			else: \
+				print(\"Could not parse mutation results\"); \
+				sys.exit(1)' || \
+		{ echo '❌ Mutation score below 75% threshold'; exit 1; }"
+
+mutmut-clean:
+	@echo "🧹 Cleaning mutmut cache..."
+	@rm -rf .mutmut-cache
+	@rm -rf html
+	@echo "✅ Mutmut cache cleaned."
+
 # =============================================================================
 # 📊 METRICS
 # =============================================================================

mcpgateway/services/gateway_service.py

@@ -507,32 +507,32 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
             await self._notify_gateway_added(db_gateway)
 
             return GatewayRead.model_validate(db_gateway).masked()
-        except* GatewayConnectionError as ge:
+        except* GatewayConnectionError as ge:  # pragma: no mutate
             if TYPE_CHECKING:
                 ge: ExceptionGroup[GatewayConnectionError]
             logger.error(f"GatewayConnectionError in group: {ge.exceptions}")
             raise ge.exceptions[0]
-        except* GatewayNameConflictError as gnce:
+        except* GatewayNameConflictError as gnce:  # pragma: no mutate
             if TYPE_CHECKING:
                 gnce: ExceptionGroup[GatewayNameConflictError]
             logger.error(f"GatewayNameConflictError in group: {gnce.exceptions}")
             raise gnce.exceptions[0]
-        except* ValueError as ve:
+        except* ValueError as ve:  # pragma: no mutate
             if TYPE_CHECKING:
                 ve: ExceptionGroup[ValueError]
             logger.error(f"ValueErrors in group: {ve.exceptions}")
             raise ve.exceptions[0]
-        except* RuntimeError as re:
+        except* RuntimeError as re:  # pragma: no mutate
             if TYPE_CHECKING:
                 re: ExceptionGroup[RuntimeError]
             logger.error(f"RuntimeErrors in group: {re.exceptions}")
             raise re.exceptions[0]
-        except* IntegrityError as ie:
+        except* IntegrityError as ie:  # pragma: no mutate
             if TYPE_CHECKING:
                 ie: ExceptionGroup[IntegrityError]
             logger.error(f"IntegrityErrors in group: {ie.exceptions}")
             raise ie.exceptions[0]
-        except* BaseException as other:  # catches every other sub-exception
+        except* BaseException as other:  # catches every other sub-exception  # pragma: no mutate
             if TYPE_CHECKING:
                 other: ExceptionGroup[BaseException]
             logger.error(f"Other grouped errors: {other.exceptions}")

mcpgateway/version.py

@@ -340,7 +340,7 @@ def _database_version() -> tuple[str, bool]:
         "postgresql": "SELECT current_setting('server_version');",
         "mysql": "SELECT version();",
     }
-    stmt = stmts.get(dialect, "SELECT version();")
+    stmt = stmts.get(dialect, "XXSELECT version();XX")
     try:
         with engine.connect() as conn:
             ver = conn.execute(text(stmt)).scalar()

mutmut_config.py

@@ -0,0 +1,19 @@
+"""Configuration for mutmut mutation testing framework."""
+
+
+def pre_mutation(context):
+    """Skip files that shouldn't be mutated."""
+    if context.filename.endswith("__init__.py"):
+        context.skip = True
+    elif "alembic" in context.filename:
+        context.skip = True
+    elif "migrations" in context.filename:
+        context.skip = True
+    # Skip gateway_service.py due to Python 3.11+ except* syntax not supported by mutmut parser
+    elif "gateway_service.py" in context.filename:
+        context.skip = True
+
+
+def post_mutation(context):
+    """Any post-mutation processing."""
+    pass

pyproject.toml

@@ -457,3 +457,25 @@ project-excludes = [
     '**/\.mypy_cache/',
 ]
 python-version = "3.11.0"
+
+# --------------------------------------------------------------------
+#  🧬 mutmut - Mutation testing configuration
+# --------------------------------------------------------------------
+[tool.mutmut]
+paths_to_mutate = ["mcpgateway/"]
+tests_dir = ["tests/"]
+do_not_mutate = ["mcpgateway/services/gateway_service.py"]
+also_copy = ["plugins/", "pyproject.toml", "mutmut_config.py"]
+
+# --------------------------------------------------------------------
+#  📊 coverage - Code coverage configuration
+# --------------------------------------------------------------------
+[tool.coverage.run]
+source = ["mcpgateway"]
+omit = [
+    "*/tests/*",
+    "*/test_*.py",
+    "*/__init__.py",
+    "*/alembic/*",
+    "*/version.py"
+]

run_mutmut.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""
+Workaround script for mutmut v3 stats collection failure.
+Generates mutants and then runs them despite stats failure.
+"""
+
+import subprocess
+import sys
+import os
+import json
+from pathlib import Path
+
+def run_command(cmd):
+    """Run a shell command and return output."""
+    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
+    return result.stdout, result.stderr, result.returncode
+
+def main():
+    # Check for command line arguments
+    sample_mode = "--sample" in sys.argv or len(sys.argv) == 1  # Default to sample mode
+    sample_size = 20  # Default sample size
+    
+    if "--full" in sys.argv:
+        sample_mode = False
+        print("🧬 Starting FULL mutation testing (this will take a long time)...")
+    else:
+        print("🧬 Starting mutation testing (sample mode)...")
+        print("💡 Tip: Use 'python run_mutmut.py --full' for complete testing")
+    
+    # Clean previous runs
+    print("🧹 Cleaning previous mutants...")
+    os.system("rm -rf mutants .mutmut-cache")
+    
+    # Generate mutants (will fail at stats but mutants are created)
+    print("📝 Generating mutants (this may take a minute)...")
+    stdout, stderr, _ = run_command("mutmut run --max-children 2 2>&1 || true")
+    
+    # Show some output to indicate progress
+    if "done in" in stdout:
+        import re
+        match = re.search(r'done in (\d+)ms', stdout)
+        if match:
+            print(f"  Generated in {int(match.group(1))/1000:.1f} seconds")
+    
+    # Check if mutants were generated
+    if not Path("mutants").exists():
+        print("❌ Failed to generate mutants")
+        return 1
+    
+    print("✅ Mutants generated successfully")
+    
+    # Get list of mutants
+    print("📊 Getting list of mutants...")
+    stdout, stderr, _ = run_command("mutmut results 2>&1 | grep -E 'mutmut_[0-9]+:' | cut -d: -f1")
+    all_mutants = [m.strip() for m in stdout.strip().split('\n') if m.strip()]
+    
+    if not all_mutants:
+        print("❌ No mutants found")
+        return 1
+    
+    # Sample mutants for quicker testing
+    import random
+    
+    print(f"🔍 Found {len(all_mutants)} total mutants")
+    
+    if sample_mode:
+        actual_sample_size = min(sample_size, len(all_mutants))
+        mutants = random.sample(all_mutants, actual_sample_size)
+        print(f"📈 Testing a sample of {len(mutants)} mutants for quick results")
+    else:
+        mutants = all_mutants
+        print(f"🚀 Testing ALL {len(mutants)} mutants (this will take a while)...")
+    
+    # Run each mutant
+    results = {"killed": 0, "survived": 0, "timeout": 0, "error": 0}
+    survived_mutants = []
+    
+    for i, mutant in enumerate(mutants, 1):
+        print(f"  [{i}/{len(mutants)}] Testing {mutant}...", end=" ", flush=True)
+        
+        # Run the mutant with a shorter timeout and minimal tests
+        cmd = f"timeout 10 bash -c 'cd mutants && MUTANT_UNDER_TEST={mutant} python -m pytest tests/unit/mcpgateway/utils/ -x --tb=no -q 2>&1'"
+        stdout, stderr, returncode = run_command(cmd)
+        
+        if returncode == 124:  # timeout
+            print("⏰ TIMEOUT")
+            results["timeout"] += 1
+        elif returncode == 0:  # tests passed = mutant survived
+            print("🙁 SURVIVED")
+            results["survived"] += 1
+            survived_mutants.append(mutant)
+        elif "FAILED" in stdout or returncode != 0:  # tests failed = mutant killed
+            print("🎉 KILLED")
+            results["killed"] += 1
+        else:
+            print("❓ ERROR")
+            results["error"] += 1
+    
+    # Print summary
+    print("\n" + "="*50)
+    print("📊 MUTATION TESTING RESULTS:")
+    print("="*50)
+    print(f"🎉 Killed:    {results['killed']} mutants")
+    print(f"🙁 Survived:  {results['survived']} mutants")
+    print(f"⏰ Timeout:   {results['timeout']} mutants")
+    print(f"❓ Error:     {results['error']} mutants")
+    
+    total = sum(results.values())
+    if total > 0:
+        score = (results['killed'] / total) * 100
+        print(f"\n📈 Mutation Score: {score:.1f}%")
+        
+        if sample_mode and len(all_mutants) > len(mutants):
+            estimated_total_score = score  # Rough estimate
+            print(f"📊 Estimated overall score: ~{estimated_total_score:.1f}% (based on sample)")
+    
+    # Show surviving mutants if any
+    if survived_mutants and len(survived_mutants) <= 5:
+        print("\n⚠️  Surviving mutants (need better tests):")
+        for mutant in survived_mutants[:5]:
+            print(f"  - {mutant}")
+            print(f"    View with: mutmut show {mutant}")
+    
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main())