5. Creating Custom Formatters๏
Purpose: Build specialized output formats for your tools and workflows
This tutorial teaches you how to create custom formatters for FailExtract, enabling you to generate output in any format your tools and processes require. Perfect for integrating with monitoring systems, chat platforms, and custom reporting tools.
5.1. What Youโll Learn๏
How to implement the OutputFormatter interface
How to create formatters for popular platforms (Slack, JUnit, etc.)
How to handle different data structures and edge cases
How to register and use custom formatters
How to build production-ready formatters with error handling
5.2. Prerequisites๏
Completed all previous tutorials
Understanding of Python classes and inheritance
Familiarity with output formats you want to support
25 minutes of time
5.3. Custom Formatter Architecture๏
FailExtract uses a simple interface for custom formatters:
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
class OutputFormatter(ABC):
"""Base class for all output formatters."""
@abstractmethod
def format(self,
failures: List[Dict[str, Any]],
passed: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None) -> str:
"""Format failure data into output string."""
pass
Key Design Principles:
Simple interface - Just implement one method
Rich input data - Access to failures, passed tests, and metadata
String output - Return formatted string ready for file writing
Flexible data handling - Handle missing or optional data gracefully
5.4. Your First Custom Formatter๏
Letโs create a simple formatter for team notifications:
from failextract import OutputFormatter
from typing import List, Dict, Any, Optional
import json
class TeamNotificationFormatter(OutputFormatter):
"""Simple formatter for team notifications."""
def format(self,
failures: List[Dict[str, Any]],
passed: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None) -> str:
"""Format as simple team notification."""
if not failures:
return "๐ All tests passed! Great work team!"
# Build notification message
total_tests = len(failures) + (len(passed) if passed else 0)
success_rate = ((len(passed) if passed else 0) / total_tests * 100) if total_tests > 0 else 0
message_parts = [
f"๐จ Test Results Summary",
f"",
f"โ Failures: {len(failures)}",
f"โ
Passed: {len(passed) if passed else 0}",
f"๐ Success Rate: {success_rate:.1f}%",
f"",
f"Failed Tests:"
]
# Add failure details (limit to avoid spam)
for i, failure in enumerate(failures[:5]):
test_name = failure.get('test_name', 'Unknown test')
error_msg = failure.get('exception_message', 'No error message')
message_parts.append(f" {i+1}. {test_name}: {error_msg[:80]}{'...' if len(error_msg) > 80 else ''}")
if len(failures) > 5:
message_parts.append(f" ... and {len(failures) - 5} more failures")
return "\\n".join(message_parts)
Test your formatter:
# test_custom_formatter.py
from failextract import extract_on_failure, FailureExtractor
@extract_on_failure
def test_example_failure():
assert False, "This is a test failure for our custom formatter"
if __name__ == "__main__":
# Run test to generate failure
try:
test_example_failure()
except AssertionError:
pass
# Use custom formatter
extractor = FailureExtractor()
formatter = TeamNotificationFormatter()
output = formatter.format(extractor.failures)
print(output)
# Save to file
with open("team_notification.txt", "w") as f:
f.write(output)
5.5. Real-World Example: Slack Formatter๏
Hereโs a production-ready formatter for Slack notifications:
import json
from datetime import datetime
from failextract import OutputFormatter
class SlackFormatter(OutputFormatter):
"""Professional Slack notification formatter."""
def format(self,
failures: List[Dict[str, Any]],
passed: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None) -> str:
"""Format failures as Slack message blocks."""
if not failures:
return self._create_success_message(passed, metadata)
blocks = []
# Header block
total_tests = len(failures) + (len(passed) if passed else 0)
success_rate = ((len(passed) if passed else 0) / total_tests * 100) if total_tests > 0 else 0
header_text = f"๐จ Test Failures Report"
if metadata and metadata.get('generated_at'):
header_text += f" - {metadata['generated_at']}"
blocks.append({
"type": "header",
"text": {
"type": "plain_text",
"text": header_text
}
})
# Summary block with rich formatting
summary_text = f"*{len(failures)} failures* out of {total_tests} tests "
summary_text += f"({success_rate:.1f}% success rate)"
# Add color coding based on success rate
if success_rate >= 95:
emoji = "๐ก"
elif success_rate >= 80:
emoji = "๐ "
else:
emoji = "๐ด"
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"{emoji} {summary_text}"
}
})
# Divider
blocks.append({"type": "divider"})
# Failure details (limit to avoid Slack message size limits)
for i, failure in enumerate(failures[:5]):
test_name = failure.get('test_name', 'Unknown test')
module = failure.get('test_module', 'Unknown module')
error_msg = failure.get('exception_message', 'No error message')
error_type = failure.get('exception_type', 'Unknown error')
# Truncate long error messages
if len(error_msg) > 100:
error_msg = error_msg[:100] + "..."
failure_text = f"*{test_name}*\\n"
failure_text += f"Module: `{module}`\\n"
failure_text += f"Error: {error_type}: {error_msg}"
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": failure_text
}
})
# Show remaining count if truncated
if len(failures) > 5:
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"_... and {len(failures) - 5} more failures_"
}
})
# Add action buttons for common responses
blocks.append({
"type": "actions",
"elements": [
{
"type": "button",
"text": {
"type": "plain_text",
"text": "View Full Report"
},
"style": "primary",
"url": metadata.get('report_url', '#') if metadata else '#'
},
{
"type": "button",
"text": {
"type": "plain_text",
"text": "Acknowledge"
},
"style": "danger"
}
]
})
return json.dumps({"blocks": blocks}, indent=2)
def _create_success_message(self, passed, metadata):
"""Create a success message when no failures."""
blocks = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "โ
All Tests Passed!"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*{len(passed) if passed else 0} tests* completed successfully"
}
}
]
return json.dumps({"blocks": blocks}, indent=2)
Using the Slack formatter:
# Generate Slack notification
formatter = SlackFormatter()
slack_message = formatter.format(
failures=extractor.failures,
passed=extractor.passed,
metadata={
'generated_at': datetime.now().isoformat(),
'report_url': 'https://your-ci.com/reports/123'
}
)
# Save for Slack webhook
with open("slack_notification.json", "w") as f:
f.write(slack_message)
5.6. Advanced Example: JUnit XML Formatter๏
For CI/CD integration, hereโs a JUnit XML formatter:
class JunitXMLFormatter(OutputFormatter):
"""JUnit XML formatter for CI/CD integration."""
def format(self,
failures: List[Dict[str, Any]],
passed: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None) -> str:
"""Format failures as JUnit XML."""
total_tests = len(failures) + (len(passed) if passed else 0)
total_failures = len(failures)
# Calculate timing if available
total_time = 0
if metadata and 'total_duration' in metadata:
total_time = metadata['total_duration']
xml_lines = [
'<?xml version="1.0" encoding="UTF-8"?>',
f'<testsuite name="FailExtract" tests="{total_tests}" '
f'failures="{total_failures}" errors="0" time="{total_time}">'
]
# Add failure test cases
for failure in failures:
test_name = failure.get('test_name', 'unknown_test')
classname = failure.get('test_module', 'unknown_module')
error_type = failure.get('exception_type', 'AssertionError')
error_msg = failure.get('exception_message', 'No message')
traceback = failure.get('exception_traceback', 'No traceback')
test_time = failure.get('duration', 0)
xml_lines.extend([
f' <testcase classname="{classname}" name="{test_name}" time="{test_time}">',
f' <failure type="{error_type}" message="{self._escape_xml(error_msg)}">',
f' <![CDATA[{traceback}]]>',
' </failure>',
' </testcase>'
])
# Add passed test cases
if passed:
for test in passed:
test_name = test.get('test_name', 'unknown_test')
classname = test.get('test_module', 'unknown_module')
test_time = test.get('duration', 0)
xml_lines.append(
f' <testcase classname="{classname}" name="{test_name}" time="{test_time}"/>'
)
xml_lines.append('</testsuite>')
return '\\n'.join(xml_lines)
def _escape_xml(self, text: str) -> str:
"""Escape XML special characters."""
if not text:
return ""
return (str(text).replace('&', '&')
.replace('<', '<')
.replace('>', '>')
.replace('"', '"')
.replace("'", '''))
5.7. Handling Edge Cases and Errors๏
Production formatters must handle edge cases gracefully:
class RobustFormatter(OutputFormatter):
"""Example showing robust error handling."""
def format(self,
failures: List[Dict[str, Any]],
passed: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None) -> str:
"""Robust formatter with comprehensive error handling."""
try:
return self._format_internal(failures, passed, metadata)
except Exception as e:
# Fallback to minimal format if main formatting fails
return self._create_fallback_format(failures, passed, str(e))
def _format_internal(self, failures, passed, metadata):
"""Main formatting logic."""
# Handle None inputs
failures = failures or []
passed = passed or []
metadata = metadata or {}
# Validate data structure
for failure in failures:
if not isinstance(failure, dict):
raise ValueError(f"Invalid failure data type: {type(failure)}")
# Handle empty data
if not failures and not passed:
return "No test data available"
# Main formatting logic here
result_parts = [
f"Test Results ({metadata.get('generated_at', 'unknown time')})",
f"Failures: {len(failures)}",
f"Passed: {len(passed)}"
]
# Format failures with safe access to nested data
for failure in failures:
test_name = self._safe_get(failure, 'test_name', 'Unknown test')
error_msg = self._safe_get(failure, 'exception_message', 'No error message')
# Sanitize strings to prevent injection
test_name = self._sanitize_string(test_name)
error_msg = self._sanitize_string(error_msg)
result_parts.append(f" - {test_name}: {error_msg}")
return "\\n".join(result_parts)
def _safe_get(self, data: Dict, key: str, default: str = "") -> str:
"""Safely get string value from dict."""
value = data.get(key, default)
return str(value) if value is not None else default
def _sanitize_string(self, text: str, max_length: int = 200) -> str:
"""Sanitize and truncate string."""
if not text:
return ""
# Remove control characters
sanitized = ''.join(char for char in text if ord(char) >= 32 or char in '\\n\\t')
# Truncate if too long
if len(sanitized) > max_length:
sanitized = sanitized[:max_length] + "..."
return sanitized
def _create_fallback_format(self, failures, passed, error_msg):
"""Create minimal format when main formatting fails."""
return f"Formatting error: {error_msg}\\nFailures: {len(failures or [])}\\nPassed: {len(passed or [])}"
5.8. Using Custom Formatters๏
Direct Usage
from failextract import FailureExtractor
# Create formatter and use directly
formatter = SlackFormatter()
extractor = FailureExtractor()
# Generate output
output = formatter.format(
failures=extractor.failures,
passed=extractor.passed,
metadata={'generated_at': datetime.now().isoformat()}
)
# Save or send
with open("custom_report.json", "w") as f:
f.write(output)
Integration with OutputConfig (Advanced)
For seamless integration, you can extend FailExtractโs format system:
from failextract.core.formatters.registry import FormatterRegistry
# Register custom formatter (in a real implementation)
def register_custom_formatters():
"""Register custom formatters with FailExtract."""
registry = FormatterRegistry()
# This would require extending the OutputFormat enum
# and modifying the registry system
# registry.register('slack', SlackFormatter())
# registry.register('junit', JunitXMLFormatter())
# For now, use formatters directly as shown above
5.9. Complete Custom Formatter Example๏
Hereโs a complete example showing a formatter for GitHub Issues:
#!/usr/bin/env python3
"""Complete custom formatter example for GitHub Issues"""
from failextract import OutputFormatter, extract_on_failure, FailureExtractor
from typing import List, Dict, Any, Optional
from datetime import datetime
class GitHubIssueFormatter(OutputFormatter):
"""Format failures as GitHub issue markdown."""
def format(self,
failures: List[Dict[str, Any]],
passed: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None) -> str:
"""Format as GitHub issue markdown."""
if not failures:
return self._create_success_summary(passed, metadata)
issue_parts = []
# Issue title and summary
total_tests = len(failures) + (len(passed) if passed else 0)
success_rate = ((len(passed) if passed else 0) / total_tests * 100) if total_tests > 0 else 0
issue_parts.extend([
f"# ๐จ Test Failures Report",
f"",
f"**Summary:** {len(failures)} test(s) failed out of {total_tests} total tests ({success_rate:.1f}% success rate)",
f"",
f"**Generated:** {metadata.get('generated_at', datetime.now().isoformat()) if metadata else datetime.now().isoformat()}",
f""
])
# Add environment information if available
if metadata:
issue_parts.extend([
f"## Environment",
f"",
f"- **Branch:** {metadata.get('branch', 'unknown')}",
f"- **Commit:** {metadata.get('commit', 'unknown')}",
f"- **Environment:** {metadata.get('environment', 'unknown')}",
f""
])
# Failure details
issue_parts.extend([
f"## Failed Tests",
f""
])
for i, failure in enumerate(failures, 1):
test_name = failure.get('test_name', 'Unknown test')
module = failure.get('test_module', 'Unknown module')
file_path = failure.get('test_file', 'Unknown file')
error_type = failure.get('exception_type', 'Unknown error')
error_msg = failure.get('exception_message', 'No error message')
issue_parts.extend([
f"### {i}. `{test_name}`",
f"",
f"**File:** `{file_path}` ",
f"**Module:** `{module}` ",
f"**Error Type:** `{error_type}` ",
f"",
f"**Error Message:**",
f"```",
f"{error_msg}",
f"```",
f""
])
# Add local variables if available
if 'local_variables' in failure and failure['local_variables']:
issue_parts.extend([
f"**Local Variables:**",
f"```python"
])
for var_name, var_value in failure['local_variables'].items():
# Limit variable output to prevent huge issues
value_str = str(var_value)
if len(value_str) > 100:
value_str = value_str[:100] + "..."
issue_parts.append(f"{var_name} = {value_str}")
issue_parts.extend([f"```", f""])
# Add success summary if there are passed tests
if passed:
issue_parts.extend([
f"## Passed Tests",
f"",
f"โ
{len(passed)} tests passed successfully",
f""
])
# Add troubleshooting section
issue_parts.extend([
f"## Troubleshooting",
f"",
f"- [ ] Review failed test cases above",
f"- [ ] Check environment configuration",
f"- [ ] Verify test data and fixtures",
f"- [ ] Review recent code changes",
f"- [ ] Run tests locally to reproduce",
f"",
f"## Labels",
f"",
f"`bug` `test-failure` `needs-investigation`"
])
return "\\n".join(issue_parts)
def _create_success_summary(self, passed, metadata):
"""Create success summary for all-passing runs."""
return f"""# โ
All Tests Passed!
**Summary:** {len(passed) if passed else 0} tests completed successfully
**Generated:** {metadata.get('generated_at', datetime.now().isoformat()) if metadata else datetime.now().isoformat()}
Great work! ๐"""
# Example usage
@extract_on_failure
def test_user_registration():
user_data = {"email": "invalid-email", "age": -5}
assert "@" in user_data["email"], "Invalid email format"
@extract_on_failure
def test_data_processing():
data = []
assert len(data) > 0, "No data to process"
if __name__ == "__main__":
# Run tests to generate failures
for test_func in [test_user_registration, test_data_processing]:
try:
test_func()
except AssertionError:
pass
# Generate GitHub issue
formatter = GitHubIssueFormatter()
extractor = FailureExtractor()
github_issue = formatter.format(
failures=extractor.failures,
metadata={
'generated_at': datetime.now().isoformat(),
'branch': 'feature/user-registration',
'commit': 'abc123def',
'environment': 'CI/CD'
}
)
# Save as markdown file
with open("github_issue.md", "w") as f:
f.write(github_issue)
print("Generated github_issue.md")
print(f"Found {len(extractor.failures)} failures to report")
5.10. Formatter Best Practices๏
Design Principles
Fail Gracefully - Handle missing data without crashing
Limit Output Size - Prevent overwhelming users or hitting API limits
Sanitize Input - Clean data to prevent injection or display issues
Provide Context - Include enough information for actionable responses
Consider Your Audience - Format for your specific users and tools
Performance Considerations
class PerformantFormatter(OutputFormatter):
"""Example showing performance optimizations."""
def __init__(self, max_failures: int = 50, max_message_length: int = 1000):
self.max_failures = max_failures
self.max_message_length = max_message_length
def format(self, failures, passed=None, metadata=None):
# Limit data size early to improve performance
limited_failures = failures[:self.max_failures] if failures else []
# Use list comprehension for efficiency
formatted_failures = [
self._format_single_failure(f)
for f in limited_failures
]
result = "\\n".join(formatted_failures)
# Truncate final result if too long
if len(result) > self.max_message_length:
result = result[:self.max_message_length] + "\\n... (truncated)"
return result
5.11. Next Steps๏
Now that you can create custom formatters:
Deploy in Production - Integrate formatters with your monitoring and notification systems
Share with Team - Create formatters that match your teamโs workflow tools
Contribute Back - Consider contributing useful formatters to the FailExtract project
Automate Integration - Set up automatic formatter usage in CI/CD pipelines
5.12. Key Custom Formatter Takeaways๏
โ
Simple interface - Just implement the
format() methodโ
Rich input data - Access failures, passed tests, and metadata
โ
Production-ready patterns - Error handling, sanitization, limits
โ
Platform-specific examples - Slack, JUnit, GitHub formats
โ
Performance optimization - Limit data size and processing time
โ
Flexible integration - Use directly or integrate with existing systems
You can now create formatters for any tool or platform!