tessl/pypi-pook

HTTP traffic mocking and expectations made easy for Python testing and development

—

Pending

Overview

Eval results

Files

Engine Management

Name: tessl/pypi-pook
Author: tessl

Engine control and state management functions for controlling pook's HTTP interception capabilities, networking modes, and mock lifecycle. These functions provide fine-grained control over when and how pook intercepts HTTP traffic.

Capabilities

Engine Activation and Deactivation

Controls when pook starts and stops intercepting HTTP traffic, with support for both programmatic control and decorator patterns.

def activate(fn=None):
    """
    Enables HTTP traffic interceptors. Can be used as a decorator or called directly.
    
    Parameters:
    - fn (function|coroutinefunction, optional): Function to decorate when used as decorator
    
    Returns:
    function: Decorator wrapper when used as decorator, None otherwise
    """

def on(fn=None):
    """
    Alias to activate(). Enables HTTP traffic interceptors.
    
    Parameters:
    - fn (function|coroutinefunction, optional): Function to decorate when used as decorator
    
    Returns:
    function: Decorator wrapper when used as decorator, None otherwise
    """

def disable():
    """
    Disables HTTP traffic interceptors without flushing mocks.
    
    The mocks remain registered and can be reactivated later.
    """

def off():
    """
    Disables mock engine, HTTP interceptors, and flushes all registered mocks.
    
    This is equivalent to calling both disable() and reset().
    """

Usage examples:

import pook
import requests

# Direct activation/deactivation
pook.activate()
pook.get('https://api.example.com/data').reply(200).json({'data': []})

response = requests.get('https://api.example.com/data')  # Intercepted
print(response.json())

pook.disable()  # Stop intercepting
pook.activate() # Resume with same mocks

pook.off()      # Stop and clear all mocks

# Using as decorator
@pook.activate  
def test_with_mocks():
    pook.get('https://api.example.com/test').reply(200)
    response = requests.get('https://api.example.com/test')
    assert response.status_code == 200
    # Pook automatically deactivated after function

@pook.on  # Alias to activate
def another_test():
    pook.post('https://api.example.com/create').reply(201)
    # Test implementation

# Async function support
@pook.activate
async def async_test():
    pook.get('https://api.example.com/async').reply(200)
    # Async HTTP client calls will be intercepted

Engine State Management

Functions for resetting and managing the internal state of pook's mock engine.

def reset():
    """
    Resets current mock engine state, flushing all registered mocks 
    without disabling the engine.
    
    The engine remains active if it was previously activated.
    """

def engine():
    """
    Returns the current running mock engine instance.
    
    Returns:
    Engine: Current engine instance for direct manipulation
    """

def set_mock_engine(engine):
    """
    Sets a custom mock engine, replacing the built-in one.
    
    Parameters:
    - engine (MockEngine): Custom mock engine to use
    """

Usage examples:

import pook

# Engine state management
pook.activate()
pook.get('https://api.example.com/test1').reply(200)
pook.get('https://api.example.com/test2').reply(200)

print(f"Pending mocks: {pook.pending()}")  # 2

pook.reset()  # Clear mocks but keep engine active
print(f"Pending mocks: {pook.pending()}")  # 0
print(f"Engine active: {pook.isactive()}")  # True

# Direct engine access
current_engine = pook.engine()
print(f"Engine networking mode: {current_engine.networking}")
print(f"Number of mocks: {len(current_engine.mocks)}")

# Custom engine (advanced usage)
from pook import Engine, MockEngine
custom_engine = MockEngine(Engine(network=True))
pook.set_mock_engine(custom_engine)

Context Managers

Context manager functions for creating isolated mock scopes with automatic cleanup and optional networking support.

def use(network=False):
    """
    Creates a new isolated mock engine to be used via context manager.
    
    Parameters:
    - network (bool, optional): Enable networking mode. Defaults to False
    
    Returns:
    Context manager yielding Engine instance
    """

def use_network():
    """
    Creates a new mock engine with networking enabled as context manager.
    
    Equivalent to use(network=True).
    
    Returns:
    Context manager yielding Engine instance
    """

Usage examples:

import pook
import requests

# Isolated mock scope
with pook.use() as engine:
    # All mocks created here are isolated
    pook.get('https://api.example.com/isolated').reply(200).json({'isolated': True})
    
    response = requests.get('https://api.example.com/isolated')
    assert response.json()['isolated'] == True
    
    # Check engine state
    print(f"Mocks in this engine: {len(engine.mocks)}")

# Outside context, mocks are gone and engine is restored
print(f"Global mocks: {pook.pending()}")  # 0

# Network-enabled context
with pook.use_network() as engine:
    # Only some requests are mocked, others hit real network
    pook.get('https://httpbin.org/json').reply(200).json({'mocked': True})
    
    # This hits the mock
    mock_response = requests.get('https://httpbin.org/json')
    print(mock_response.json())  # {'mocked': True}
    
    # This hits real network (if allowed)
    real_response = requests.get('https://httpbin.org/ip')
    print(real_response.json())  # Real response from httpbin

# Nested contexts
with pook.use() as outer_engine:
    pook.get('https://api.example.com/outer').reply(200)
    
    with pook.use() as inner_engine:
        pook.get('https://api.example.com/inner').reply(200)
        # Inner context has different mocks
        
    # Back to outer context

Network Mode Control

Functions for controlling when pook allows real network traffic for unmatched requests.

def enable_network(*hostnames):
    """
    Enables real networking mode for unmatched mocks in the current mock engine.
    
    Parameters:
    - *hostnames (str): Optional hostnames to enable networking for.
                        If not provided, enables for all hosts.
    """

def disable_network():
    """
    Disables real traffic networking mode in the current mock engine.
    
    All unmatched requests will be blocked/rejected.
    """

def use_network_filter(*fn):
    """
    Adds network filters to determine if certain outgoing unmatched 
    HTTP traffic can establish real network connections.
    
    Parameters:
    - *fn (function): Filter functions that return True to allow network access
    """

Usage examples:

import pook
import requests

# Enable networking for all hosts
pook.activate()
pook.enable_network()

# Mock some requests, allow others through
pook.get('https://api.example.com/mock-me').reply(200).json({'mocked': True})

# This is mocked
mock_response = requests.get('https://api.example.com/mock-me')
print(mock_response.json())  # {'mocked': True}

# This goes to real network
real_response = requests.get('https://httpbin.org/ip')
print(real_response.json())  # Real IP response

# Enable networking for specific hosts only
pook.disable_network()
pook.enable_network('httpbin.org', 'api.github.com')

# This goes to real network
requests.get('https://httpbin.org/json')  # Allowed

# This would be blocked
# requests.get('https://api.example.com/real')  # Would raise exception

# Network filtering with custom logic
def allow_safe_hosts(request):
    """Allow requests to known safe hosts."""
    safe_hosts = ['httpbin.org', 'jsonplaceholder.typicode.com']
    return any(host in request.url for host in safe_hosts)

def allow_get_only(request):
    """Only allow GET requests through."""
    return request.method.upper() == 'GET'

pook.use_network_filter(allow_safe_hosts, allow_get_only)

# Only GET requests to safe hosts will be allowed through
requests.get('https://httpbin.org/get')      # Allowed
# requests.post('https://httpbin.org/post')  # Blocked (POST method)
# requests.get('https://evil.com/data')      # Blocked (unsafe host)

Advanced Engine Patterns

Complex engine management scenarios and patterns for sophisticated testing setups.

import pook
from pook import Engine, MockEngine

# Custom engine configuration
def create_test_engine():
    """Create a custom configured engine for testing."""
    engine = Engine(network=True)
    engine.debug = True  # Enable debug mode
    return MockEngine(engine)

# Use custom engine
original_engine = pook.engine()
test_engine = create_test_engine()
pook.set_mock_engine(test_engine)

try:
    # Test with custom engine
    pook.activate()
    pook.get('https://api.test.com/data').reply(200)
    
    # Your tests here
    
finally:
    # Restore original engine
    pook.set_mock_engine(original_engine)

# Conditional engine setup
def setup_test_environment(use_network=False, debug=False):
    """Setup test environment with specific engine configuration."""
    if use_network:
        with pook.use_network() as engine:
            if debug:
                engine.debug = True
            yield engine
    else:
        with pook.use() as engine:
            if debug:
                engine.debug = True
            yield engine

# Usage
with setup_test_environment(use_network=True, debug=True) as engine:
    # Test in network-enabled, debug mode
    pook.get('https://api.example.com/test').reply(200)
    # Some requests mocked, others hit real network

# Engine state inspection and management
@pook.activate
def complex_test_scenario():
    """Test scenario with multiple mock phases."""
    
    # Phase 1: Setup initial mocks
    pook.get('https://api.example.com/setup').reply(200).json({'phase': 1})
    
    # Verify setup
    assert pook.pending() == 1
    assert pook.isactive()
    
    # Execute phase 1
    response1 = requests.get('https://api.example.com/setup')
    assert response1.json()['phase'] == 1
    
    # Phase 2: Reset and add new mocks
    pook.reset()  # Clear mocks but keep engine active
    pook.post('https://api.example.com/execute').reply(201).json({'phase': 2})
    
    # Execute phase 2
    response2 = requests.post('https://api.example.com/execute')
    assert response2.json()['phase'] == 2
    
    # Verify all mocks were consumed
    assert pook.isdone()

Engine Class Reference

class Engine:
    """
    Engine represents the mock interceptor and matcher engine responsible 
    for triggering interceptors and matching outgoing HTTP traffic.
    """
    
    def __init__(self, network=False):
        """
        Parameters:
        - network (bool, optional): Enables/disables real networking mode
        """
    
    # Key attributes
    debug: bool                    # Enables/disables debug mode
    active: bool                   # Current engine activation status  
    networking: bool               # Current engine networking mode status
    mocks: list                    # List of registered Mock instances
    filters: list                  # Engine-level mock filter functions
    mappers: list                  # Engine-level mock mapper functions
    unmatched_reqs: list          # Unmatched outgoing HTTP requests
    network_filters: list         # Real networking mode filters
    
    def activate(self):
        """Activates registered interceptors."""
        
    def disable(self):
        """Disables interceptors."""
        
    def reset(self):
        """Resets and flushes engine state."""
        
    def mock(self, url=None, **kw):
        """
        Creates and registers new HTTP mock.
        
        Parameters:
        - url (str, optional): Request URL to mock
        - **kw: Additional Mock constructor arguments
        
        Returns:
        Mock: New mock instance
        """
        
    def enable_network(self, *hostnames):
        """
        Enables real networking mode.
        
        Parameters:
        - *hostnames (str): Optional hostnames to enable networking for
        """
        
    def disable_network(self):
        """Disables real networking mode."""

class MockEngine:
    """
    MockEngine represents the low-level mocking engine abstraction layer 
    between pook and the underlying mocking mechanism.
    """
    
    def __init__(self, engine):
        """
        Parameters:
        - engine (Engine): Injected pook engine to be used
        """
    
    engine: Engine                        # Stores pook engine to be used
    interceptors: list                   # HTTP traffic interceptors
    
    def activate(self):
        """Activates registered interceptors."""
        
    def disable(self):
        """Disables interceptors."""

Install with Tessl CLI