Live Server

@pytest.fixture(scope=_determine_scope)
def live_server(request, app, pytestconfig):
    """
    Run application in a separate process.
    
    When the live_server fixture is applied, the url_for function
    works as expected with external URLs.
    
    Parameters:
        request: pytest request object
        app: Flask application instance
        pytestconfig: pytest configuration object
        
    Returns:
        LiveServer: Live server instance with start/stop control
    """

import requests
from flask import url_for

def test_live_server_basic(live_server):
    """Basic live server functionality"""
    # Server automatically starts
    response = requests.get(live_server.url('/ping'))
    assert response.status_code == 200
    assert response.json() == {'ping': 'pong'}

def test_with_external_url(live_server):
    """Using url_for with live server generates complete URLs"""
    # url_for automatically includes host and port when live_server is active
    index_url = url_for('index', _external=True)
    assert index_url.startswith('http://localhost:')
    
    response = requests.get(index_url)
    assert response.status_code == 200

def test_integration_with_external_service(live_server):
    """Testing integration with external services"""
    import subprocess
    
    # Example: test webhook endpoint with external tool
    webhook_url = live_server.url('/webhooks/github')
    
    # Simulate external service calling webhook
    payload = {'action': 'opened', 'pull_request': {'id': 123}}
    response = requests.post(webhook_url, json=payload)
    
    assert response.status_code == 200
    assert response.json()['status'] == 'processed'

class LiveServer:
    """
    The helper class used to manage a live server. Handles creation and
    stopping application in a separate process.
    
    Parameters:
        app: The Flask application to run
        host: The host where to listen (default localhost)  
        port: The port to run application
        wait: The timeout after which test case is aborted if
              application is not started
        clean_stop: Whether to attempt clean shutdown (default False)
    """
    
    def __init__(self, app, host, port, wait, clean_stop=False):
        """
        Initialize live server instance.
        
        Parameters:
            app: Flask application instance
            host: str, host address to bind to
            port: int, port number to listen on
            wait: float, timeout in seconds for server startup
            clean_stop: bool, whether to attempt graceful shutdown
        """
        
    def start(self):
        """Start application in a separate process."""
        
    def stop(self):
        """Stop application process."""
        
    def url(self, url=""):
        """
        Returns the complete url based on server options.
        
        Parameters:
            url: str, path to append to base URL (default: "")
            
        Returns:
            str: Complete URL including protocol, host, port, and path
        """

def test_live_server_methods(live_server):
    """Test LiveServer methods"""
    # Get base URL
    base_url = live_server.url()
    assert base_url.startswith('http://localhost:')
    
    # Get URL with path
    api_url = live_server.url('/api/v1/users')
    assert api_url == base_url + '/api/v1/users'
    
    # Server is automatically started
    response = requests.get(live_server.url('/health'))
    assert response.status_code == 200

def test_manual_server_control():
    """Manual server lifecycle control (advanced usage)"""
    from pytest_flask.live_server import LiveServer
    from myapp import create_app
    
    app = create_app()
    server = LiveServer(app, 'localhost', 0, 5)
    
    try:
        server.start()
        response = requests.get(server.url('/ping'))
        assert response.status_code == 200
    finally:
        server.stop()

# Start server automatically (enabled by default)
pytest --start-live-server

# Don't start server automatically
pytest --no-start-live-server

# Set server startup timeout in seconds (default: 5.0)
pytest --live-server-wait=10

# Attempt to kill the live server cleanly (default: True)
pytest --live-server-clean-stop

# Terminate the server forcefully after stop
pytest --no-live-server-clean-stop

# Set server host (default: localhost)
pytest --live-server-host=0.0.0.0

# Set fixed port for the live_server fixture (default: 0 for random port)
pytest --live-server-port=8080

[tool:pytest]
live_server_scope = session  # Default: session
# Options: function, class, module, package, session

@pytest.fixture
def app():
    app = create_app()
    
    # Set fixed port for live server
    app.config['LIVESERVER_PORT'] = 5555
    
    return app

def test_custom_port_config(live_server):
    """Test with custom port configuration"""
    assert ':5555' in live_server.url()

# pytest.ini configuration
[tool:pytest]
live_server_scope = class

# Server shared across all test methods in class
class TestIntegrationSuite:
    
    def test_first_endpoint(self, live_server):
        response = requests.get(live_server.url('/api/users'))
        assert response.status_code == 200
        
    def test_second_endpoint(self, live_server):
        # Same server instance as previous test
        response = requests.get(live_server.url('/api/posts'))
        assert response.status_code == 200

def test_webhook_processing(live_server):
    """Test webhook endpoint with actual HTTP calls"""
    import threading
    import time
    
    # Set up webhook URL
    webhook_url = live_server.url('/webhooks/payment')
    
    # Simulate external payment service calling webhook
    def send_webhook():
        time.sleep(0.1)  # Small delay
        payload = {
            'event': 'payment.completed',
            'payment_id': 'pay_123',
            'amount': 1000
        }
        requests.post(webhook_url, json=payload)
    
    # Start webhook sender in background
    thread = threading.Thread(target=send_webhook)
    thread.start()
    
    # Test the application processes webhook correctly
    # (This would typically check database or queue state)
    thread.join()
    
    # Verify webhook was processed
    response = requests.get(live_server.url('/api/payments/pay_123'))
    assert response.status_code == 200
    assert response.json()['status'] == 'completed'