Error Handling

class CannotOverwriteExistingCassetteException(Exception):
    """
    Raised when VCR cannot overwrite an existing cassette.
    
    This typically occurs in 'once' record mode when a test tries to make
    HTTP requests that aren't already recorded in the cassette.
    
    Attributes:
        cassette: The cassette that couldn't be overwritten
        failed_request: The request that couldn't be matched/recorded
    """
    
    def __init__(self, cassette, failed_request):
        """
        Initialize exception with cassette and request details.
        
        Args:
            cassette: Cassette object that couldn't be overwritten
            failed_request: Request object that caused the failure
        """
    
    cassette: Cassette
    failed_request: Request

class UnhandledHTTPRequestError(KeyError):
    """
    Raised when a cassette does not contain the request we want.
    
    This occurs when VCR intercepts an HTTP request but cannot find
    a matching recorded interaction in the current cassette.
    """

import vcr
import requests
from vcr.errors import CannotOverwriteExistingCassetteException

@vcr.use_cassette('existing.yaml', record_mode='once')
def test_with_error_handling():
    """Test that handles cassette overwrite errors."""
    try:
        # This request must already exist in the cassette
        response = requests.get('https://api.example.com/existing-endpoint')
        assert response.status_code == 200
        
        # This new request will cause an error in 'once' mode
        response = requests.get('https://api.example.com/new-endpoint')
        
    except CannotOverwriteExistingCassetteException as e:
        print(f"Cannot add new request to cassette: {e.failed_request.uri}")
        print(f"Cassette path: {e.cassette._path}")
        print(f"Record mode: {e.cassette.record_mode}")
        
        # Handle the error - perhaps switch to NEW_EPISODES mode
        # or update the test to use existing endpoints
        pass

import vcr
import requests
from vcr.errors import UnhandledHTTPRequestError

@vcr.use_cassette('partial.yaml', record_mode='none')
def test_missing_interaction():
    """Test that handles missing recorded interactions."""
    try:
        # This request must exist in the cassette
        response = requests.get('https://api.example.com/recorded-endpoint')
        assert response.status_code == 200
        
        # This request is not in the cassette
        response = requests.get('https://api.example.com/missing-endpoint')
        
    except UnhandledHTTPRequestError as e:
        print(f"No recorded interaction for request: {e}")
        
        # Handle missing interaction
        # - Add the interaction to the cassette
        # - Mock the response 
        # - Skip the test
        # - Use a different cassette
        pass

import vcr
import requests
from vcr.errors import CannotOverwriteExistingCassetteException, UnhandledHTTPRequestError

class APITestCase:
    """Test case with comprehensive VCR error handling."""
    
    def make_api_request(self, url, method='GET', **kwargs):
        """Make API request with VCR error handling."""
        try:
            response = requests.request(method, url, **kwargs)
            return response
            
        except CannotOverwriteExistingCassetteException as e:
            # Handle cassette overwrite errors
            self.handle_cassette_overwrite_error(e)
            
        except UnhandledHTTPRequestError as e:
            # Handle missing interaction errors
            self.handle_missing_interaction_error(e)
            
        except Exception as e:
            # Handle other potential errors
            self.handle_general_error(e)
    
    def handle_cassette_overwrite_error(self, error):
        """Handle cassette overwrite errors."""
        print(f"Cassette overwrite error:")
        print(f"  Request: {error.failed_request.method} {error.failed_request.uri}")
        print(f"  Cassette: {error.cassette._path}")
        print(f"  Record mode: {error.cassette.record_mode}")
        
        # Suggest solutions
        if error.cassette.record_mode == 'once':
            print("  Solution: Change record_mode to 'new_episodes' or add request to cassette")
        
        # Find similar requests for debugging
        if hasattr(error.cassette, 'find_requests_with_most_matches'):
            matches = error.cassette.find_requests_with_most_matches(error.failed_request)
            if matches:
                print(f"  Similar requests found: {len(matches)}")
                for i, (request, succeeded, failed) in enumerate(matches[:3]):
                    print(f"    {i+1}. {request.method} {request.uri}")
                    print(f"       Matched: {succeeded}")
                    print(f"       Failed: {[f[0] for f in failed]}")
    
    def handle_missing_interaction_error(self, error):
        """Handle missing interaction errors."""
        print(f"Missing interaction error: {error}")
        print("  Possible solutions:")
        print("    - Record the interaction by changing record_mode")
        print("    - Add the interaction manually to the cassette")
        print("    - Use a different cassette that contains the interaction")
        print("    - Mock the response instead of using VCR")
    
    def handle_general_error(self, error):
        """Handle other VCR-related errors."""
        print(f"VCR error: {type(error).__name__}: {error}")

import logging
import vcr
from vcr.errors import CannotOverwriteExistingCassetteException, UnhandledHTTPRequestError

# Enable VCR debug logging
logging.basicConfig(level=logging.DEBUG)
vcr_log = logging.getLogger('vcr')
vcr_log.setLevel(logging.DEBUG)

def debug_vcr_errors(test_function):
    """Decorator to add debug information to VCR errors."""
    def wrapper(*args, **kwargs):
        try:
            return test_function(*args, **kwargs)
        except CannotOverwriteExistingCassetteException as e:
            print("\n=== VCR CASSETTE OVERWRITE ERROR DEBUG ===")
            print(f"Test function: {test_function.__name__}")
            print(f"Failed request: {e.failed_request.method} {e.failed_request.uri}")
            print(f"Request headers: {dict(e.failed_request.headers)}")
            print(f"Request body: {e.failed_request.body}")
            print(f"Cassette path: {e.cassette._path}")
            print(f"Cassette record mode: {e.cassette.record_mode}")
            print(f"Cassette interactions count: {len(e.cassette.responses)}")
            raise
        except UnhandledHTTPRequestError as e:
            print("\n=== VCR UNHANDLED REQUEST ERROR DEBUG ===")
            print(f"Test function: {test_function.__name__}")
            print(f"Error details: {e}")
            raise
    return wrapper

@debug_vcr_errors
@vcr.use_cassette('debug.yaml')
def test_with_debug_info():
    """Test with enhanced error debugging."""
    response = requests.get('https://api.example.com/data')

import vcr
import requests
from vcr.errors import CannotOverwriteExistingCassetteException, UnhandledHTTPRequestError

class RecoveringVCRTest:
    """Test class that attempts to recover from VCR errors."""
    
    def __init__(self, cassette_path):
        self.cassette_path = cassette_path
        self.backup_responses = {}
    
    def make_request_with_recovery(self, url, method='GET', **kwargs):
        """Make request with automatic error recovery."""
        
        # First attempt: Use strict mode
        try:
            with vcr.use_cassette(self.cassette_path, record_mode='once'):
                return requests.request(method, url, **kwargs)
                
        except (CannotOverwriteExistingCassetteException, UnhandledHTTPRequestError):
            print(f"VCR error for {method} {url}, attempting recovery...")
            
        # Second attempt: Allow new episodes
        try:
            with vcr.use_cassette(self.cassette_path, record_mode='new_episodes'):
                return requests.request(method, url, **kwargs)
                
        except Exception as e:
            print(f"VCR recovery failed: {e}")
            
        # Final attempt: Use backup response or real request
        return self.get_backup_response(url, method) or requests.request(method, url, **kwargs)
    
    def get_backup_response(self, url, method):
        """Get backup response for URL if available."""
        key = f"{method}:{url}"
        return self.backup_responses.get(key)
    
    def set_backup_response(self, url, method, response):
        """Set backup response for URL."""
        key = f"{method}:{url}"
        self.backup_responses[key] = response

from collections import defaultdict
import vcr
from vcr.errors import CannotOverwriteExistingCassetteException, UnhandledHTTPRequestError

class VCRErrorReporter:
    """Collect and analyze VCR errors across test runs."""
    
    def __init__(self):
        self.errors = defaultdict(list)
    
    def record_error(self, error, context=None):
        """Record an error for later analysis."""
        error_type = type(error).__name__
        error_info = {
            'error': str(error),
            'context': context,
        }
        
        if isinstance(error, CannotOverwriteExistingCassetteException):
            error_info.update({
                'cassette_path': error.cassette._path,
                'record_mode': error.cassette.record_mode,
                'failed_request_uri': error.failed_request.uri,
                'failed_request_method': error.failed_request.method,
            })
        elif isinstance(error, UnhandledHTTPRequestError):
            error_info.update({
                'missing_request': str(error),
            })
        
        self.errors[error_type].append(error_info)
    
    def generate_report(self):
        """Generate error analysis report."""
        print("=== VCR ERROR ANALYSIS REPORT ===")
        
        for error_type, error_list in self.errors.items():
            print(f"\n{error_type}: {len(error_list)} occurrences")
            
            if error_type == 'CannotOverwriteExistingCassetteException':
                # Group by cassette path
                by_cassette = defaultdict(list)
                for error in error_list:
                    by_cassette[error['cassette_path']].append(error)
                
                for cassette_path, cassette_errors in by_cassette.items():
                    print(f"  Cassette: {cassette_path}")
                    print(f"    Errors: {len(cassette_errors)}")
                    for error in cassette_errors[:3]:  # Show first 3
                        print(f"      {error['failed_request_method']} {error['failed_request_uri']}")
            
            elif error_type == 'UnhandledHTTPRequestError':
                # Group by missing request pattern
                for error in error_list[:3]:  # Show first 3
                    print(f"  Missing: {error['missing_request']}")

# Usage
reporter = VCRErrorReporter()

def test_with_error_reporting():
    try:
        # Your VCR test code here
        pass
    except (CannotOverwriteExistingCassetteException, UnhandledHTTPRequestError) as e:
        reporter.record_error(e, context={'test': 'test_with_error_reporting'})
        raise

# At end of test run
reporter.generate_report()