From Test Cases to Mathematical Guarantees
Pillar: #7 of 12 - Supporting Priority
Timeline: 6 Weeks
Dependencies: #02 (Mathematical Pipeline)
Expected ROI: 90% bug detection pre-merge, automatic test generation, invariant-based quality gates
Executive Summary
Transform testing from example-based (hoping we thought of the right cases) to property-based (proving properties hold for all inputs). This pillar implements Hypothesis-style property testing, automatic invariant extraction, mutation testing for test quality, and contract testing for service boundaries.
Key Insight: One property test can be worth 1000 unit tests.
Problem Statement
Current State: Example-Based Testing
# Traditional unit test - covers ONE case
def test_sort_simple():
assert sort([3, 1, 2]) == [1, 2, 3]
# What about:
# - Empty list?
# - Single element?
# - Already sorted?
# - Reverse sorted?
# - Duplicates?
# - Negative numbers?
# - Very large lists?
# - Unicode strings?
# ...
# We're playing "guess the edge case"
Test Suite Problems
Common Testing Failures:
1. Coverage != Correctness
- 90% line coverage, but missing edge cases
- Tests pass on happy path, fail on real inputs
2. Specification Mismatch
- Tests verify implementation, not requirements
- Refactor breaks tests but code is correct
3. Combinatorial Explosion
- Too many input combinations to test manually
- Developers skip edge cases due to effort
4. Test Maintenance Burden
- Every code change requires test updates
- Tests become stale and misleading
Maestro-Specific Testing Gaps
# File: tests/test_workflow.py
# Current: Manual test cases
def test_workflow_execution():
workflow = create_simple_workflow()
result = await execute(workflow)
assert result.success
# Missing:
# - What if a phase fails mid-execution?
# - What if multiple phases run concurrently?
# - What if context exceeds token limit?
# - What if LLM returns malformed response?
# - What about race conditions in async code?
Solution Architecture
Multi-Layer Testing Pyramid
┌─────────────────────┐
│ Formal Proofs │ ← Critical paths
│ (Z3, TLA+) │ Mathematically proven
├─────────────────────┤
│ Property Tests │ ← Core logic
│ (Hypothesis) │ Thousands of inputs
├─────────────────────┤
│ Contract Tests │ ← Service boundaries
│ (Pact, Schemathesis) │ API compatibility
├─────────────────────┤
│ Mutation Tests │ ← Test quality
│ (mutmut) │ Verify tests catch bugs
├─────────────────────┤
│ Integration Tests │ ← System behavior
│ (pytest) │ E2E scenarios
├─────────────────────┤
│ Unit Tests │ ← Edge cases
│ (pytest) │ Specific examples
└─────────────────────┘
Core Components
1. Property-Based Test Generator
from hypothesis import given, strategies as st, assume
class PropertyTestGenerator:
"""
Generate property-based tests from function signatures and docstrings.
"""
def generate_tests(self, function: Callable) -> List[PropertyTest]:
# Extract type hints
hints = get_type_hints(function)
# Generate strategy for each parameter
strategies = {
name: self._type_to_strategy(type_hint)
for name, type_hint in hints.items()
if name != 'return'
}
# Extract properties from docstring
properties = self._extract_properties(function)
return [
PropertyTest(
function=function,
strategies=strategies,
property=prop,
)
for prop in properties
]
def _type_to_strategy(self, type_hint: type) -> st.SearchStrategy:
"""Map Python types to Hypothesis strategies."""
if type_hint == int:
return st.integers()
elif type_hint == str:
return st.text()
elif type_hint == float:
return st.floats(allow_nan=False)
elif get_origin(type_hint) == list:
inner = get_args(type_hint)[0]
return st.lists(self._type_to_strategy(inner))
elif get_origin(type_hint) == dict:
k, v = get_args(type_hint)
return st.dictionaries(
self._type_to_strategy(k),
self._type_to_strategy(v)
)
else:
return st.from_type(type_hint)
# Generated property test example
@given(xs=st.lists(st.integers()))
def test_sort_preserves_length(xs):
"""Length invariant for sort."""
assert len(sort(xs)) == len(xs)
@given(xs=st.lists(st.integers()))
def test_sort_preserves_elements(xs):
"""Elements invariant for sort."""
assert sorted(sort(xs)) == sorted(xs) # Same elements
@given(xs=st.lists(st.integers()))
def test_sort_is_ordered(xs):
"""Ordering property for sort."""
result = sort(xs)
for i in range(len(result) - 1):
assert result[i] <= result[i + 1]
@given(xs=st.lists(st.integers()))
def test_sort_is_idempotent(xs):
"""Idempotence property for sort."""
assert sort(sort(xs)) == sort(xs)
class InvariantExtractor:
"""
Extract invariants from code, comments, and specifications.
"""
def extract_invariants(self, codebase: Codebase) -> List[Invariant]:
invariants = []
for function in codebase.functions:
# From docstrings
invariants.extend(self._from_docstring(function))
# From type hints (e.g., NonEmpty, Positive)
invariants.extend(self._from_types(function))
# From code patterns (e.g., len(output) == len(input))
invariants.extend(self._from_code_analysis(function))
# From contracts (@requires, @ensures)
invariants.extend(self._from_contracts(function))
return invariants
def _from_code_analysis(self, function: Function) -> List[Invariant]:
"""Infer invariants from common patterns."""
invariants = []
# Pattern: output length == input length
if self._preserves_length(function):
invariants.append(Invariant(
name="length_preservation",
expression="len(output) == len(input)",
function=function,
))
# Pattern: no side effects on input
if self._is_pure(function):
invariants.append(Invariant(
name="purity",
expression="input is unchanged after call",
function=function,
))
# Pattern: idempotent
if self._is_idempotent(function):
invariants.append(Invariant(
name="idempotence",
expression="f(f(x)) == f(x)",
function=function,
))
return invariants
def generate_property_tests(self, invariants: List[Invariant]) -> str:
"""Generate Hypothesis tests from invariants."""
tests = []
for inv in invariants:
test = f'''
@given({self._generate_strategy(inv.function)})
def test_{inv.function.name}_{inv.name}({self._params(inv.function)}):
"""Property: {inv.expression}"""
result = {inv.function.name}({self._args(inv.function)})
assert {inv.to_assertion()}
'''
tests.append(test)
return "\n".join(tests)
3. Mutation Testing Engine
class MutationTestingEngine:
"""
Verify test quality by introducing bugs and checking if tests catch them.
"""
MUTATIONS = [
("==", "!="), # Equality
(">", ">="), # Comparison
("and", "or"), # Boolean
("+", "-"), # Arithmetic
("True", "False"), # Boolean literal
("append", "extend"), # Method call
("return x", "return None"), # Return value
]
async def run_mutation_testing(
self,
module: Module,
tests: List[Test]
) -> MutationReport:
killed = 0
survived = 0
mutants = []
for mutant in self._generate_mutants(module):
# Apply mutation
mutated_code = self._apply_mutation(module.source, mutant)
# Run tests against mutant
result = await self._run_tests(mutated_code, tests)
if result.any_failed:
killed += 1
mutants.append(MutantResult(mutant, "KILLED"))
else:
survived += 1
mutants.append(MutantResult(mutant, "SURVIVED"))
return MutationReport(
killed=killed,
survived=survived,
mutation_score=killed / (killed + survived),
survivors=self._analyze_survivors(mutants),
)
def _analyze_survivors(self, mutants: List[MutantResult]) -> List[TestGap]:
"""Identify missing test cases from surviving mutants."""
gaps = []
for mutant in mutants:
if mutant.status == "SURVIVED":
gaps.append(TestGap(
location=mutant.location,
mutation=mutant.mutation,
suggested_test=self._suggest_test(mutant),
))
return gaps
4. Contract Testing System
class ContractTestingSystem:
"""
API contract testing between services.
Consumer-driven contracts ensure compatibility.
"""
async def generate_contract(
self,
consumer: Service,
provider: Service
) -> Contract:
# Analyze consumer's API calls
interactions = await self._analyze_interactions(consumer, provider)
# Generate contract from interactions
contract = Contract(
consumer=consumer.name,
provider=provider.name,
interactions=[
Interaction(
description=i.description,
request=i.request,
response=i.expected_response,
)
for i in interactions
]
)
return contract
async def verify_provider(
self,
contract: Contract,
provider: Service
) -> VerificationResult:
"""Verify provider satisfies consumer's contract."""
failures = []
for interaction in contract.interactions:
# Make actual request
response = await self._call_provider(
provider,
interaction.request
)
# Compare with expected
if not self._matches(response, interaction.response):
failures.append(ContractViolation(
interaction=interaction,
expected=interaction.response,
actual=response,
))
return VerificationResult(
passed=len(failures) == 0,
failures=failures,
)
async def generate_from_openapi(self, spec: OpenAPISpec) -> List[Test]:
"""Generate tests from OpenAPI specification."""
from schemathesis import from_schema
schema = from_schema(spec)
tests = []
for endpoint in schema.endpoints():
@given(case=endpoint.as_strategy())
def test_endpoint(case):
response = case.call()
case.validate_response(response)
tests.append(test_endpoint)
return tests
5. Fuzzing Engine
class FuzzingEngine:
"""
Find edge cases through intelligent input generation.
"""
def fuzz(
self,
function: Callable,
duration: timedelta = timedelta(minutes=5)
) -> FuzzingReport:
crashes = []
edge_cases = []
coverage_data = CoverageTracker()
deadline = now() + duration
while now() < deadline:
# Generate input based on coverage feedback
input_data = self._generate_input(function, coverage_data)
try:
# Execute with coverage tracking
with coverage_data.track():
result = function(**input_data)
# Check for interesting behaviors
if self._is_edge_case(result, input_data):
edge_cases.append(EdgeCase(input_data, result))
except Exception as e:
crashes.append(Crash(input_data, e))
return FuzzingReport(
crashes=crashes,
edge_cases=edge_cases,
coverage=coverage_data.percentage,
suggested_tests=self._generate_test_cases(crashes, edge_cases),
)
def _generate_input(
self,
function: Callable,
coverage: CoverageTracker
) -> Dict[str, Any]:
"""Coverage-guided input generation."""
# Start with random input
base_input = self._random_input(function)
# Mutate to explore uncovered branches
for _ in range(10):
mutated = self._mutate_input(base_input)
if coverage.would_increase(function, mutated):
return mutated
return base_input
Integration Points
Files to Modify
CI Pipeline Integration
# .github/workflows/tests.yml
jobs:
unit-tests:
runs-on: ubuntu-latest
steps:
- run: pytest tests/unit/ -v
property-tests:
runs-on: ubuntu-latest
steps:
- run: pytest tests/properties/ -v --hypothesis-profile=ci
contract-tests:
runs-on: ubuntu-latest
steps:
- run: pytest tests/contracts/ -v
- name: Publish Contract
run: pact-broker publish contracts/
mutation-tests:
runs-on: ubuntu-latest
steps:
- run: mutmut run --paths-to-mutate=src/
- name: Check Mutation Score
run: |
score=$(mutmut results --json | jq '.mutation_score')
if (( $(echo "$score < 0.80" | bc -l) )); then
echo "Mutation score $score below 80% threshold"
exit 1
fi
fuzzing:
runs-on: ubuntu-latest
steps:
- run: python -m pytest tests/fuzz/ --fuzz-time=300
Hypothesis Configuration
# tests/conftest.py
from hypothesis import settings, Phase, Verbosity
# CI profile: more examples, stricter
settings.register_profile(
"ci",
max_examples=1000,
deadline=None,
suppress_health_check=[],
phases=[Phase.explicit, Phase.reuse, Phase.generate, Phase.shrink],
)
# Dev profile: faster feedback
settings.register_profile(
"dev",
max_examples=100,
deadline=timedelta(seconds=1),
)
# Load profile from environment
settings.load_profile(os.environ.get("HYPOTHESIS_PROFILE", "dev"))
Implementation Plan
Week 1: Property Testing Foundation
- [ ] Add Hypothesis to dependencies
- [ ] Create property test generators
- [ ] Convert critical unit tests to property tests
- [ ] Set up CI integration
- [ ] Build invariant extractor
- [ ] Analyze codebase for patterns
- [ ] Generate invariant catalog
- [ ] Auto-generate property tests
Week 3: Mutation Testing
- [ ] Integrate mutmut
- [ ] Establish baseline mutation score
- [ ] Identify test gaps from survivors
- [ ] Add missing tests
Week 4: Contract Testing
- [ ] Set up Pact broker
- [ ] Generate contracts from API specs
- [ ] Create consumer tests
- [ ] Verify provider compliance
Week 5: Fuzzing Integration
- [ ] Build fuzzing engine
- [ ] Integrate with coverage tools
- [ ] Run fuzzing on critical paths
- [ ] Convert findings to regression tests
Week 6: Production Rollout
- [ ] Enforce mutation score threshold in CI
- [ ] Add property tests for new code
- [ ] Document testing patterns
- [ ] Training for development team
Expected Results
Bug Detection
Test Quality
Maintenance Burden
Risks & Mitigations
Success Criteria
- #02 Mathematical Pipeline: Formal proofs for critical paths
- #06 Autonomous CI/CD: Tests gate deployments
- #08 Observability: Test metrics in dashboards
Status: Ready for Implementation
Key Insight: One property test can replace 1000 unit tests
Owner: Quality Engineering Team
