Mutation Generator for Operator Controllers

This skill generates mutations (deliberate bugs) in operator controller code to enable mutation testing. It focuses on patterns common in Kubernetes operators built with controller-runtime.

When to Use This Skill

Use this skill when:

• You need to generate mutants for operator mutation testing
• You want to create realistic bugs that tests should catch
• You're analyzing operator controller code for mutation points
• You need operator-specific mutations (requeue, status updates, API calls)

Prerequisites

1. Go operator code using controller-runtime framework
2. Python 3.8+ for mutation generation scripts
3. Go AST parsing capability (via go/parser or Python ast module)

Mutation Strategy for Operators

Understanding Operator Mutation Points

Kubernetes operators have specific patterns that should be tested:

1. Reconciliation Logic:

• Conditional checks on resource state
• Decision trees for state transitions
• Owner reference checks
• Finalizer logic

2. Error Handling:

• API call error handling
• Client-go errors (NotFound, AlreadyExists, Conflict)
• Wrapped errors
• Error return paths

3. Requeue Behavior:

• ctrl.Result{Requeue: true}
• ctrl.Result{RequeueAfter: duration}
• Rate limiting
• Conditional requeueing

4. Status Updates:

• Condition setting (Ready, Available, Degraded)
• ObservedGeneration tracking
• Status subresource updates
• Partial status updates

5. API Interactions:

• Get vs. List operations
• Create vs. Update logic
• Patch vs. Update
• Delete with preconditions

Implementation Steps

Step 1: Parse Controller Code

1.1 Identify Reconcile Functions

Look for the standard controller-runtime reconciliation signature:

func (r *MyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error)

1.2 Extract Abstract Syntax Tree (AST)

Use Go's parser to build AST:

import go_parser  # or use subprocess to call go/ast tools

def parse_controller(file_path):
    """Parse Go controller file into AST."""
    # Read file
    with open(file_path, 'r') as f:
        source = f.read()
    
    # Parse into AST
    ast = parse_go_file(source)
    
    # Find Reconcile function
    reconcile_func = find_reconcile_function(ast)
    
    return {
        'source': source,
        'ast': ast,
        'reconcile': reconcile_func,
        'file_path': file_path
    }

1.3 Identify Mutation Candidates

Walk the AST to find mutation points:

def find_mutation_points(ast_node):
    """Find all locations where mutations can be applied."""
    mutation_points = []
    
    # Walk AST nodes
    for node in walk_ast(ast_node):
        if is_conditional(node):
            mutation_points.append({
                'type': 'conditional',
                'location': node.position,
                'original': node.text,
                'node': node
            })
        elif is_error_check(node):
            mutation_points.append({
                'type': 'error-handling',
                'location': node.position,
                'original': node.text,
                'node': node
            })
        # ... more patterns
    
    return mutation_points

---

Step 2: Generate Conditional Mutations

2.1 Comparison Operator Mutations

Mutate comparison operators:

CONDITIONAL_MUTATIONS = {
    '==': ['!='],
    '!=': ['=='],
    '<': ['>', '<=', '>='],
    '>': ['<', '<=', '>='],
    '<=': ['<', '>='],
    '>=': ['>', '<='],
}

def mutate_conditional(node):
    """Generate mutations for conditional expressions."""
    mutations = []
    
    if node.operator in CONDITIONAL_MUTATIONS:
        for new_op in CONDITIONAL_MUTATIONS[node.operator]:
            mutations.append({
                'id': generate_mutant_id(),
                'type': 'conditional',
                'description': f'Change {node.operator} to {new_op}',
                'location': f'{node.file}:{node.line}',
                'original': node.text,
                'mutated': node.text.replace(node.operator, new_op, 1),
                'pattern': 'comparison-operator'
            })
    
    return mutations

2.2 Boolean Expression Mutations

Negate boolean expressions:

def mutate_boolean_expr(node):
    """Negate boolean expressions."""
    mutations = []
    
    # if condition → if !condition
    # if !condition → if condition
    
    if node.is_negated:
        # Remove negation
        mutated = node.text.lstrip('!')
        description = 'Remove negation'
    else:
        # Add negation
        mutated = f'!({node.text})'
        description = 'Add negation'
    
    mutations.append({
        'id': generate_mutant_id(),
        'type': 'conditional',
        'description': description,
        'location': f'{node.file}:{node.line}',
        'original': node.text,
        'mutated': mutated,
        'pattern': 'boolean-negation'
    })
    
    return mutations

2.3 Operator-Specific Conditional Mutations

Target operator patterns:

# Example: Mutate finalizer checks
# Original: if contains(obj.Finalizers, MyFinalizer)
# Mutated:  if !contains(obj.Finalizers, MyFinalizer)

# Example: Mutate generation checks
# Original: if obj.Generation != obj.Status.ObservedGeneration
# Mutated:  if obj.Generation == obj.Status.ObservedGeneration

---

Step 3: Generate Error Handling Mutations

3.1 Remove Error Checks

Most critical mutation type for operators:

def mutate_error_handling(node):
    """Mutate error handling code."""
    mutations = []
    
    # Pattern: if err != nil { return ... }
    if is_error_check_pattern(node):
        mutations.append({
            'id': generate_mutant_id(),
            'type': 'error-handling',
            'description': 'Remove error check',
            'location': f'{node.file}:{node.line}',
            'original': node.text,
            'mutated': f'// MUTANT: Removed error check\n// {node.text}',
            'pattern': 'remove-error-check'
        })
    
    return mutations

3.2 Change Error Returns

def mutate_error_return(node):
    """Mutate error return statements."""
    mutations = []
    
    # Pattern: return ctrl.Result{}, err
    # Mutate to: return ctrl.Result{}, nil
    
    if is_error_return(node):
        mutations.append({
            'id': generate_mutant_id(),
            'type': 'error-handling',
            'description': 'Return nil instead of error',
            'location': f'{node.file}:{node.line}',
            'original': node.text,
            'mutated': node.text.replace(', err', ', nil'),
            'pattern': 'error-return-nil'
        })
    
    return mutations

3.3 Ignore Specific Error Types

Kubernetes-specific error mutations:

def mutate_k8s_errors(node):
    """Mutate Kubernetes error handling."""
    mutations = []
    
    # Pattern: if errors.IsNotFound(err)
    # Mutate to: if errors.IsAlreadyExists(err)
    
    # Pattern: if errors.IsAlreadyExists(err)
    # Mutate to: if errors.IsNotFound(err)
    
    if is_k8s_error_check(node):
        error_types = ['IsNotFound', 'IsAlreadyExists', 'IsConflict', 'IsInvalid']
        current_type = extract_error_type(node)
        
        for new_type in error_types:
            if new_type != current_type:
                mutations.append({
                    'id': generate_mutant_id(),
                    'type': 'error-handling',
                    'description': f'Change {current_type} to {new_type}',
                    'location': f'{node.file}:{node.line}',
                    'original': node.text,
                    'mutated': node.text.replace(current_type, new_type),
                    'pattern': 'k8s-error-type'
                })
    
    return mutations

---

Step 4: Generate Requeue Mutations

4.1 Toggle Requeue Flag

def mutate_requeue(node):
    """Mutate requeue behavior."""
    mutations = []
    
    # Pattern: return ctrl.Result{}, nil
    # Mutate to: return ctrl.Result{Requeue: true}, nil
    
    if is_result_return(node):
        # If no requeue, add requeue
        if 'Requeue' not in node.text:
            mutations.append({
                'id': generate_mutant_id(),
                'type': 'requeue',
                'description': 'Add unnecessary requeue',
                'location': f'{node.file}:{node.line}',
                'original': node.text,
                'mutated': node.text.replace('ctrl.Result{}', 'ctrl.Result{Requeue: true}'),
                'pattern': 'add-requeue'
            })
        
        # If requeue exists, remove it
        if 'Requeue: true' in node.text:
            mutations.append({
                'id': generate_mutant_id(),
                'type': 'requeue',
                'description': 'Remove requeue flag',
                'location': f'{node.file}:{node.line}',
                'original': node.text,
                'mutated': node.text.replace('Requeue: true', 'Requeue: false'),
                'pattern': 'remove-requeue'
            })
    
    return mutations

4.2 Change Requeue Timing

def mutate_requeue_after(node):
    """Mutate RequeueAfter duration."""
    mutations = []
    
    # Pattern: RequeueAfter: 5 * time.Second
    # Mutate to: RequeueAfter: 0
    # Or:        RequeueAfter: 5 * time.Minute (change unit)
    
    if 'RequeueAfter' in node.text:
        mutations.extend([
            {
                'id': generate_mutant_id(),
                'type': 'requeue',
                'description': 'Set RequeueAfter to zero',
                'location': f'{node.file}:{node.line}',
                'original': node.text,
                'mutated': set_requeue_after_zero(node.text),
                'pattern': 'requeue-timing-zero'
            },
            {
                'id': generate_mutant_id(),
                'type': 'requeue',
                'description': 'Change RequeueAfter time unit',
                'location': f'{node.file}:{node.line}',
                'original': node.text,
                'mutated': change_time_unit(node.text),
                'pattern': 'requeue-timing-unit'
            }
        ])
    
    return mutations

---

Step 5: Generate Status Update Mutations

5.1 Skip Status Updates

def mutate_status_update(node):
    """Mutate status update calls."""
    mutations = []
    
    # Pattern: r.Status().Update(ctx, obj)
    # Mutate to: // MUTANT: Skipped status update
    
    if is_status_update_call(node):
        mutations.append({
            'id': generate_mutant_id(),
            'type': 'status',
            'description': 'Skip status update',
            'location': f'{node.file}:{node.line}',
            'original': node.text,
            'mutated': f'// MUTANT: Skipped status update\n// {node.text}',
            'pattern': 'skip-status-update'
        })
    
    return mutations

5.2 Change Condition Values

def mutate_condition(node):
    """Mutate condition setting."""
    mutations = []
    
    # Pattern: SetCondition(Ready, True, "Ready", "...")
    # Mutate to: SetCondition(Ready, False, "Ready", "...")
    
    if is_set_condition_call(node):
        # Toggle condition status
        if 'True' in node.text:
            mutated = node.text.replace('True', 'False', 1)
            desc = 'Change condition to False'
        else:
            mutated = node.text.replace('False', 'True', 1)
            desc = 'Change condition to True'
        
        mutations.append({
            'id': generate_mutant_id(),
            'type': 'status',
            'description': desc,
            'location': f'{node.file}:{node.line}',
            'original': node.text,
            'mutated': mutated,
            'pattern': 'condition-value'
        })
    
    return mutations

---

Step 6: Generate API Call Mutations

6.1 Change API Operation Type

def mutate_api_call(node):
    """Mutate Kubernetes API calls."""
    mutations = []
    
    # Pattern: r.Get(ctx, key, obj)
    # Mutate to: r.List(ctx, obj) [intentional API misuse]
    
    if is_client_call(node):
        operation = extract_operation(node)  # Get, List, Create, Update, Delete
        
        alternative_ops = {
            'Get': ['List'],  # Get → List (wrong cardinality)
            'Update': ['Patch'],  # Update → Patch (different semantics)
            'Create': ['Update'],  # Create → Update (wrong operation)
        }
        
        if operation in alternative_ops:
            for alt_op in alternative_ops[operation]:
                mutations.append({
                    'id': generate_mutant_id(),
                    'type': 'api-calls',
                    'description': f'Change {operation} to {alt_op}',
                    'location': f'{node.file}:{node.line}',
                    'original': node.text,
                    'mutated': node.text.replace(operation, alt_op, 1),
                    'pattern': 'api-operation-type'
                })
    
    return mutations

---

Step 7: Apply Mutations In-Place (No Copies!)

7.1 Generate Mutation Metadata Only

For each mutation, generate metadata WITHOUT copying files:

def generate_mutation_metadata(mutation, operator_path, output_file):
    """Generate mutation metadata for in-place application."""
    
    mutant_id = mutation['id']
    
    # Save mutation metadata only (no file copies)
    metadata = {
        'id': mutant_id,
        'type': mutation['type'],
        'description': mutation['description'],
        'file': mutation['file'],
        'line': mutation['line'],
        'original': mutation['original'],
        'mutated': mutation['mutated'],
        'pattern': mutation['pattern']
    }
    
    return metadata

def save_all_mutations(mutations, output_file):
    """Save all mutation definitions to a single JSON file."""
    
    with open(output_file, 'w') as f:
        json.dump({
            'total_mutations': len(mutations),
            'mutations': mutations
        }, f, indent=2)

Note: The mutation testing workflow will apply each mutation in-place to the original file, run tests, then immediately revert the change. This avoids creating GB of repository copies.

7.2 Apply and Revert Mutations In-Place

def apply_mutation_to_file(file_path, mutation):
    """Apply mutation in-place to the original file."""
    
    with open(file_path, 'r') as f:
        lines = f.readlines()
    
    # Find and replace the specific line
    target_line = mutation['line'] - 1  # 0-indexed
    
    if target_line < len(lines):
        # Replace exact match on that line
        lines[target_line] = lines[target_line].replace(
            mutation['original'],
            mutation['mutated'],
            1  # Replace only first occurrence
        )
    
    # Write back to the ORIGINAL file
    with open(file_path, 'w') as f:
        f.writelines(lines)

def revert_mutation(file_path, mutation):
    """Revert mutation by restoring the original code."""
    
    with open(file_path, 'r') as f:
        lines = f.readlines()
    
    target_line = mutation['line'] - 1
    
    if target_line < len(lines):
        # Restore original code
        lines[target_line] = lines[target_line].replace(
            mutation['mutated'],
            mutation['original'],
            1
        )
    
    with open(file_path, 'w') as f:
        f.writelines(lines)

Usage Pattern: Apply mutation → Run tests → Immediately revert. No file copies needed!

---

Output Format

The mutation generator produces:

1. Mutations Catalog (JSON)

{
  "total_mutations": 145,
  "mutations_by_type": {
    "conditional": 42,
    "error-handling": 38,
    "requeue": 18,
    "status": 25,
    "api-calls": 15,
    "returns": 7
  },
  "mutations": [
    {
      "id": "mutant-001",
      "type": "error-handling",
      "description": "Remove error check after API Get",
      "file": "controllers/pod_controller.go",
      "line": 87,
      "function": "Reconcile",
      "pattern": "remove-error-check",
      "original": "if err != nil { return ctrl.Result{}, err }",
      "mutated": "// MUTANT: Removed error check"
    }
  ]
}

2. Mutation Metadata Storage

.work/mutation-testing/
├── mutations.json        # All mutation definitions
├── results/
│   ├── mutant-001-result.json     # Metadata about this mutation
│   ├── mutant-001-output.txt      # Test output
│   ├── mutant-002-result.json
│   └── mutant-002-output.txt
...

Note: Mutations are applied in-place to the original files, tested, then immediately reverted. No full repository copies are created, keeping disk usage minimal (<1MB).

Error Handling

• No controllers found: Warn user and provide suggestions
• Parse errors: Skip files with syntax errors, report them
• Invalid mutations: Validate mutations don't break Go syntax
• Disk space: Mutations are applied in-place with minimal disk usage (<1MB for metadata); verify sufficient temporary storage and available inodes if needed

Best Practices

1. Focus on High-Value Mutations: Prioritize error handling and conditionals
2. Avoid Equivalent Mutants: Don't generate mutations that don't change behavior
3. Limit Mutations: For large controllers, consider sampling strategy
4. Validate Syntax: Ensure mutated code is syntactically valid before including in mutation list
5. Track Coverage: Keep mapping of which mutations test which behavior
6. In-Place Efficiency: Generate metadata only; apply/revert mutations during testing to minimize disk usage

Example Usage

# Generate mutation metadata (no file copies!)
mutations = generate_mutations(
    operator_path="/path/to/operator",
    mutation_types=["error-handling", "conditional", "requeue"],
    output_file=".work/mutation-testing/mutations.json"
)

print(f"Generated {len(mutations)} mutation definitions")
print(f"Metadata saved to: {output_file} (Total: <1MB)")
print(f"No repository copies created - mutations applied in-place during testing")

See Also

• mutation-tester skill - Tests each mutant
• go-mutesting - Existing Go mutation testing tool
• Mutation Testing Best Practices