Content Attachments

class Content:
    """
    A MIME-like Content object for test attachments.
    
    Content objects can be serialized to bytes and provide
    structured data attachment capabilities for test results.
    """
    
    def __init__(self, content_type, get_bytes):
        """
        Create a Content object.
        
        Args:
            content_type (ContentType): MIME content type
            get_bytes (callable): Function returning byte iterator
        """
    
    def iter_bytes(self):
        """
        Iterate over bytestrings of the serialized content.
        
        Yields:
            bytes: Chunks of content data
        """
    
    def iter_text(self):
        """
        Iterate over text of the serialized content.
        
        Only valid for text MIME types. Uses ISO-8859-1 if
        no charset parameter is present.
        
        Yields:
            str: Text chunks
            
        Raises:
            ValueError: If content type is not text/*
        """
    
    def as_text(self):
        """
        Return all content as text string.
        
        Loads all content into memory. For large content,
        use iter_text() instead.
        
        Returns:
            str: Complete text content
            
        Raises:
            ValueError: If content type is not text/*
        """
    
    def __eq__(self, other):
        """
        Compare content objects for equality.
        
        Args:
            other (Content): Content to compare with
            
        Returns:
            bool: True if content types and data are equal
        """

class TracebackContent(Content):
    """
    Content object for Python tracebacks.
    
    Specialized content type for capturing and
    attaching exception tracebacks to test results.
    """
    
    def __init__(self, err, test):
        """
        Create traceback content from exception.
        
        Args:
            err: Exception information (type, value, traceback)
            test: Test case that generated the exception
        """

class ContentType:
    """
    A content type from IANA media types registry.
    
    Represents MIME content types with parameters
    for proper content handling and display.
    """
    
    def __init__(self, primary_type, sub_type, parameters=None):
        """
        Create a ContentType.
        
        Args:
            primary_type (str): Primary type (e.g., "text", "application")
            sub_type (str): Sub type (e.g., "plain", "json")
            parameters (dict): Optional type parameters
        """
    
    @property
    def type(self):
        """
        Primary content type.
        
        Returns:
            str: Primary type (e.g., "text")
        """
    
    @property
    def subtype(self):
        """
        Content subtype.
        
        Returns:
            str: Sub type (e.g., "plain")
        """
    
    @property
    def parameters(self):
        """
        Content type parameters.
        
        Returns:
            dict: Parameters like charset, boundary, etc.
        """
    
    def __repr__(self):
        """
        String representation of content type.
        
        Returns:
            str: MIME type string (e.g., "text/plain; charset=utf8")
        """

# Predefined content types
JSON = ContentType("application", "json")
UTF8_TEXT = ContentType("text", "plain", {"charset": "utf8"})

def text_content(text):
    """
    Create text content from string.
    
    Args:
        text (str): Text content to attach
        
    Returns:
        Content: Text content object with UTF-8 encoding
    """

def json_content(json_data):
    """
    Create JSON content from Python objects.
    
    Args:
        json_data: Python object to serialize as JSON
        
    Returns:
        Content: JSON content object
    """

def content_from_file(path, content_type=None, chunk_size=DEFAULT_CHUNK_SIZE):
    """
    Create content object from file.
    
    Args:
        path (str): File path to read
        content_type (ContentType): Optional content type
        chunk_size (int): Read chunk size
        
    Returns:
        Content: File content object
    """

def content_from_stream(stream, content_type, seek_offset=None, 
                       seek_whence=0, chunk_size=DEFAULT_CHUNK_SIZE):
    """
    Create content object from stream/file-like object.
    
    Args:
        stream: File-like object to read from
        content_type (ContentType): Content type
        seek_offset (int): Optional seek position
        seek_whence (int): Seek reference point
        chunk_size (int): Read chunk size
        
    Returns:
        Content: Stream content object
    """

def attach_file(test, name, path, content_type=None):
    """
    Attach file content to test results.
    
    Convenience function for attaching files to test cases
    with automatic content type detection.
    
    Args:
        test: TestCase to attach content to
        name (str): Attachment name/identifier
        path (str): File path to attach
        content_type (ContentType): Optional content type
    """

DEFAULT_CHUNK_SIZE = 4096
"""Default chunk size for reading content streams."""

STDOUT_LINE = "\nStdout:\n%s"
"""Format string for stdout content presentation."""

STDERR_LINE = "\nStderr:\n%s"  
"""Format string for stderr content presentation."""

import testtools
from testtools.content import text_content, json_content

class MyTest(testtools.TestCase):
    
    def test_with_debug_info(self):
        # Attach text debug information
        debug_info = "Processing started at 10:30 AM\nUser: alice\nMode: batch"
        self.addDetail('debug_log', text_content(debug_info))
        
        # Attach structured data
        state = {
            'user_id': 12345,
            'session': 'abc-def-123',
            'permissions': ['read', 'write'],
            'timestamp': '2023-10-15T10:30:00Z'
        }
        self.addDetail('application_state', json_content(state))
        
        # Perform test
        result = process_user_request()
        self.assertEqual(result.status, 'success')
    
    def test_with_file_attachment(self):
        # Generate test data file
        test_data_file = '/tmp/test_data.csv'
        with open(test_data_file, 'w') as f:
            f.write('name,age,city\nAlice,30,Boston\nBob,25,Seattle\n')
        
        # Attach the file to test results
        testtools.content.attach_file(
            self, 
            'input_data', 
            test_data_file,
            testtools.content_type.ContentType('text', 'csv')
        )
        
        # Test file processing
        result = process_csv_file(test_data_file)
        self.assertEqual(len(result), 2)

import testtools
from testtools.content import content_from_file, content_from_stream
import io

class FileProcessingTest(testtools.TestCase):
    
    def test_log_file_processing(self):
        log_file = '/var/log/application.log'
        
        # Attach log file content
        if os.path.exists(log_file):
            log_content = content_from_file(
                log_file,
                testtools.content_type.UTF8_TEXT,
                chunk_size=8192
            )
            self.addDetail('application_logs', log_content)
        
        # Test log processing
        result = analyze_logs(log_file)
        self.assertGreater(result.warning_count, 0)
    
    def test_stream_processing(self):
        # Create in-memory stream
        data_stream = io.StringIO("line1\nline2\nline3\n")
        
        # Attach stream content
        stream_content = content_from_stream(
            data_stream,
            testtools.content_type.UTF8_TEXT,
            seek_offset=0
        )
        self.addDetail('input_stream', stream_content)
        
        # Test stream processing
        data_stream.seek(0)  # Reset for actual processing
        result = process_stream(data_stream)
        self.assertEqual(len(result), 3)

import testtools
from testtools.content import Content, ContentType
import gzip
import json

class AdvancedContentTest(testtools.TestCase):
    
    def test_with_compressed_content(self):
        # Create custom compressed content
        original_data = json.dumps({
            'large_dataset': list(range(10000)),
            'metadata': {'compression': 'gzip'}
        })
        
        def get_compressed_bytes():
            compressed = gzip.compress(original_data.encode('utf-8'))
            yield compressed
        
        compressed_content = Content(
            ContentType('application', 'json', {'encoding': 'gzip'}),
            get_compressed_bytes
        )
        
        self.addDetail('compressed_data', compressed_content)
        
        # Test with large dataset
        result = process_large_dataset()
        self.assertIsNotNone(result)
    
    def test_with_binary_content(self):
        # Attach binary file (e.g., screenshot)
        screenshot_path = '/tmp/test_screenshot.png'
        if os.path.exists(screenshot_path):
            binary_content = content_from_file(
                screenshot_path,
                ContentType('image', 'png')
            )
            self.addDetail('failure_screenshot', binary_content)
        
        # Test UI functionality
        result = interact_with_ui()
        self.assertTrue(result.success)
    
    def test_multiple_content_types(self):
        # Attach multiple types of debugging content
        
        # Configuration data
        config = {'debug': True, 'timeout': 30}
        self.addDetail('configuration', json_content(config))
        
        # Log excerpts
        recent_logs = get_recent_log_entries()
        self.addDetail('recent_logs', text_content('\n'.join(recent_logs)))
        
        # Performance metrics
        metrics = measure_performance()
        self.addDetail('performance_metrics', json_content(metrics))
        
        # Environment information
        env_info = {
            'python_version': sys.version,
            'platform': platform.platform(),
            'memory_usage': get_memory_usage()
        }
        self.addDetail('environment', json_content(env_info))
        
        # Run the actual test
        result = complex_operation()
        self.assertEqual(result.status, 'completed')

import testtools
from testtools.content import text_content
from fixtures import Fixture

class LoggingFixture(Fixture):
    """Fixture that captures logs and attaches them to tests."""
    
    def __init__(self):
        super().__init__()
        self.log_entries = []
    
    def _setUp(self):
        # Set up log capture
        self.original_handler = setup_log_capture(self.log_entries)
        self.addCleanup(restore_log_handler, self.original_handler)
    
    def get_log_content(self):
        """Get captured logs as content."""
        return text_content('\n'.join(self.log_entries))

class MyIntegrationTest(testtools.TestCase):
    
    def test_with_log_capture(self):
        # Use logging fixture
        log_fixture = self.useFixture(LoggingFixture())
        
        # Perform operations that generate logs
        service = ExternalService()
        result = service.complex_operation()
        
        # Attach captured logs
        self.addDetail('operation_logs', log_fixture.get_log_content())
        
        # Verify results
        self.assertEqual(result.status, 'success')
        self.assertIn('INFO', '\n'.join(log_fixture.log_entries))

import testtools
from testtools.content import json_content

class TestResultAnalyzer(testtools.TestResult):
    """Custom result class that analyzes attached content."""
    
    def __init__(self):
        super().__init__()
        self.content_analysis = {}
    
    def addSuccess(self, test, details=None):
        super().addSuccess(test, details)
        if details:
            self.analyze_content(test.id(), details)
    
    def addFailure(self, test, err, details=None):
        super().addFailure(test, err, details)
        if details:
            self.analyze_content(test.id(), details)
    
    def analyze_content(self, test_id, details):
        """Analyze attached content for insights."""
        analysis = {}
        
        for name, content in details.items():
            if content.content_type.type == 'application' and \
               content.content_type.subtype == 'json':
                # Analyze JSON content
                try:
                    data = json.loads(content.as_text())
                    analysis[name] = {
                        'type': 'json',
                        'keys': list(data.keys()) if isinstance(data, dict) else None,
                        'size': len(str(data))
                    }
                except:
                    analysis[name] = {'type': 'json', 'error': 'parse_failed'}
            
            elif content.content_type.type == 'text':
                # Analyze text content
                text = content.as_text()
                analysis[name] = {
                    'type': 'text',
                    'lines': len(text.split('\n')),
                    'chars': len(text),
                    'contains_error': 'ERROR' in text.upper()
                }
        
        self.content_analysis[test_id] = analysis
    
    def get_content_summary(self):
        """Get summary of all attached content."""
        return json_content(self.content_analysis)

# Usage
result = TestResultAnalyzer()
suite.run(result)

# Get content analysis
content_summary = result.get_content_summary()
print("Content analysis:", content_summary.as_text())