Request Recording

from responses._recorder import record

@record(file_path="responses.yaml")
def function_making_requests():
    """
    Decorator to record HTTP requests and responses.

    Parameters:
    - file_path: str or Path, where to save recorded responses (default: "response.yaml")

    During execution, all HTTP requests are passed through to real servers
    and the responses are recorded. The recorded data can later be used
    to create mock responses for testing.
    """

from responses._recorder import record
import requests

@record(file_path="api_responses.yaml")
def fetch_user_data():
    # These requests hit real servers and are recorded
    user_response = requests.get("https://api.example.com/users/1")
    posts_response = requests.get("https://api.example.com/users/1/posts")
    
    return user_response.json(), posts_response.json()

# Run the function to record responses
user_data, posts_data = fetch_user_data()

# The file api_responses.yaml now contains the recorded responses
# which can be loaded later for mocking

class Recorder:
    def __init__(
        self,
        *,
        target="requests.adapters.HTTPAdapter.send",
        registry=OrderedRegistry
    ):
        """
        Create a Recorder instance for capturing HTTP requests.

        Parameters:
        - target: str, method path to patch for interception
        - registry: Registry class, typically OrderedRegistry for recording

        The Recorder extends RequestsMock to capture real responses
        instead of returning mock responses.
        """

    def record(self, *, file_path="response.yaml"):
        """
        Decorator method for recording requests to a file.

        Parameters:
        - file_path: str or Path, output file for recorded responses

        Returns:
        Decorator function that records HTTP requests during execution
        """

    def dump_to_file(
        self,
        file_path,
        *,
        registered=None
    ):
        """
        Save recorded responses to a YAML file.

        Parameters:
        - file_path: str or Path, output file path
        - registered: List of BaseResponse objects (optional, uses current registry)

        Exports recorded responses in YAML format compatible with
        the _add_from_file method for later mocking.
        """

from responses._recorder import Recorder

def test_recording():
    recorder = Recorder()
    
    with recorder:
        # Make real requests
        response = requests.get("https://api.example.com/data")
        post_response = requests.post(
            "https://api.example.com/submit",
            json={"key": "value"}
        )
    
    # Save recorded responses
    recorder.dump_to_file("recorded_responses.yaml")
    
    # Later, load for mocking
    @responses.activate
    def test_with_recorded():
        responses._add_from_file("recorded_responses.yaml")
        
        # Now requests will use recorded responses instead of hitting real servers
        mock_response = requests.get("https://api.example.com/data")
        assert mock_response.status_code == 200

def _add_from_file(file_path):
    """
    Load response configurations from a YAML file.

    Parameters:
    - file_path: str or Path to YAML file containing response configurations

    The YAML file should follow this structure:
    responses:
      - response:
          method: HTTP_METHOD
          url: URL_PATTERN
          body: RESPONSE_BODY
          status: STATUS_CODE
          headers: HEADERS_DICT
          content_type: CONTENT_TYPE
          auto_calculate_content_length: BOOLEAN

    This function is typically used with files generated by the Recorder
    but can also load manually created response configurations.
    """

responses:
  - response:
      method: GET
      url: https://api.example.com/users/1
      body: '{"id": 1, "name": "John Doe", "email": "john@example.com"}'
      status: 200
      headers:
        Content-Type: application/json
        X-RateLimit-Remaining: "99"
      content_type: application/json
      auto_calculate_content_length: false
  - response:
      method: POST
      url: https://api.example.com/users
      body: '{"id": 2, "name": "Jane Smith", "email": "jane@example.com"}'
      status: 201
      headers:
        Content-Type: application/json
        Location: https://api.example.com/users/2
      content_type: application/json
      auto_calculate_content_length: false

import responses

@responses.activate
def test_from_file():
    # Load recorded responses
    responses._add_from_file("api_responses.yaml")
    
    # Make requests that will hit the loaded mocks
    user_response = requests.get("https://api.example.com/users/1")
    create_response = requests.post(
        "https://api.example.com/users",
        json={"name": "New User", "email": "new@example.com"}
    )
    
    assert user_response.status_code == 200
    assert create_response.status_code == 201
    assert "Location" in create_response.headers

from responses._recorder import record, _add_from_file
import responses
import requests

# Step 1: Record real API interactions
@record(file_path="integration_responses.yaml")
def record_integration_flow():
    # Authenticate
    auth_response = requests.post("https://api.example.com/auth", {
        "username": "testuser",
        "password": "testpass"
    })
    token = auth_response.json()["token"]
    
    # Make authenticated requests
    headers = {"Authorization": f"Bearer {token}"}
    user_data = requests.get("https://api.example.com/profile", headers=headers)
    
    # Update profile
    updated_profile = requests.put(
        "https://api.example.com/profile",
        headers=headers,
        json={"bio": "Updated bio"}
    )
    
    return user_data.json(), updated_profile.json()

# Run recording (hits real API)
user_profile, updated_data = record_integration_flow()

# Step 2: Use recorded responses in tests
@responses.activate
def test_integration_flow():
    # Load recorded responses
    responses._add_from_file("integration_responses.yaml")
    
    # Run same flow against mocks
    auth_response = requests.post("https://api.example.com/auth", {
        "username": "testuser", 
        "password": "testpass"
    })
    
    token = auth_response.json()["token"]
    headers = {"Authorization": f"Bearer {token}"}
    
    profile_response = requests.get("https://api.example.com/profile", headers=headers)
    update_response = requests.put(
        "https://api.example.com/profile",
        headers=headers,
        json={"bio": "Updated bio"}
    )
    
    # Assertions based on recorded data
    assert profile_response.status_code == 200
    assert update_response.status_code == 200
    assert "token" in auth_response.json()

# Run test (uses mocks, no network calls)
test_integration_flow()

# Global recorder instance and decorator
recorder: Recorder  # Pre-configured Recorder instance
record: callable    # Decorator function from global recorder instance

# File loading function
_add_from_file: callable  # Function to load responses from YAML files