Generates mutations in Kubernetes operator controllers using controller-runtime for mutation testing, targeting reconciliation logic, error handling, requeue behavior, status updates, and API interactions.
From testingnpx claudepluginhub openshift-eng/ai-helpers --plugin testingThis skill uses the workspace's default tool permissions.
apply_mutation.pygenerate_mutations.pygenerate_mutations_efficient.pyDesigns and optimizes AI agent action spaces, tool definitions, observation formats, error recovery, and context for higher task completion rates.
Reorganizes X and LinkedIn networks: review-first pruning of low-value follows, priority-based add/follow recommendations, and drafts warm outreach in user's voice.
Compares coding agents like Claude Code and Aider on custom YAML-defined codebase tasks using git worktrees, measuring pass rate, cost, time, and consistency.
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.
Use this skill when:
Kubernetes operators have specific patterns that should be tested:
1. Reconciliation Logic:
2. Error Handling:
3. Requeue Behavior:
ctrl.Result{Requeue: true}ctrl.Result{RequeueAfter: duration}4. Status Updates:
5. API Interactions:
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
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
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
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
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
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
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!
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).
# 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")