Testing Framework

class QuartClient:
    """
    HTTP test client for testing Quart applications.
    
    Provides methods for making HTTP requests to test endpoints
    with full async support and response inspection.
    """
    
    async def get(
        self, 
        path: str, 
        query_string: dict | str | None = None,
        headers: dict | None = None,
        **kwargs
    ):
        """
        Make GET request to test endpoint.
        
        Args:
            path: Request path
            query_string: Query parameters as dict or string
            headers: Request headers
            **kwargs: Additional request arguments
            
        Returns:
            Response object for inspection
        """
    
    async def post(
        self,
        path: str,
        data: bytes | str | dict | None = None,
        json: dict | None = None,
        form: dict | None = None,
        files: dict | None = None,
        headers: dict | None = None,
        **kwargs
    ):
        """
        Make POST request to test endpoint.
        
        Args:
            path: Request path
            data: Raw request body data
            json: JSON data (sets appropriate content-type)
            form: Form data
            files: File uploads
            headers: Request headers
            **kwargs: Additional request arguments
            
        Returns:
            Response object for inspection
        """
    
    async def put(self, path: str, **kwargs):
        """Make PUT request to test endpoint."""
    
    async def delete(self, path: str, **kwargs):
        """Make DELETE request to test endpoint."""
    
    async def patch(self, path: str, **kwargs):
        """Make PATCH request to test endpoint."""
    
    async def head(self, path: str, **kwargs):
        """Make HEAD request to test endpoint."""
    
    async def options(self, path: str, **kwargs):
        """Make OPTIONS request to test endpoint."""
    
    async def websocket(self, path: str, **kwargs):
        """
        Create WebSocket test connection.
        
        Args:
            path: WebSocket path
            **kwargs: Additional WebSocket arguments
            
        Returns:
            WebSocket test connection (async context manager)
        """

class QuartCliRunner:
    """
    CLI test runner for testing Quart CLI commands.
    
    Provides interface for invoking CLI commands in test environment
    with output capture and result inspection.
    """
    
    def invoke(
        self, 
        cli, 
        args: list[str] | None = None, 
        input: str | bytes | None = None,
        env: dict | None = None,
        catch_exceptions: bool = True,
        **kwargs
    ):
        """
        Invoke CLI command in test environment.
        
        Args:
            cli: Click command or group to invoke
            args: Command line arguments
            input: Standard input data
            env: Environment variables
            catch_exceptions: Whether to catch exceptions
            **kwargs: Additional invoke arguments
            
        Returns:
            Click Result object with exit_code, output, and exception
        """

class TestApp(Quart):
    """
    Test application wrapper with additional testing utilities.
    
    Extends Quart application with test-specific methods and
    simplified test client/runner creation.
    """
    
    def test_client(self, **kwargs) -> QuartClient:
        """
        Create HTTP test client for this application.
        
        Args:
            **kwargs: Additional client configuration
            
        Returns:
            QuartClient instance for testing HTTP endpoints
        """
    
    def test_cli_runner(self, **kwargs) -> QuartCliRunner:
        """
        Create CLI test runner for this application.
        
        Args:
            **kwargs: Additional runner configuration
            
        Returns:
            QuartCliRunner instance for testing CLI commands
        """

def make_test_body_with_headers(
    data: bytes | str | None = None,
    form: dict | None = None,
    files: dict | None = None
) -> tuple[bytes, dict]:
    """
    Create test request body with appropriate headers.
    
    Args:
        data: Raw body data
        form: Form data
        files: File uploads
        
    Returns:
        Tuple of (body_bytes, headers_dict)
    """

def make_test_headers_path_and_query_string(
    path: str,
    headers: dict | None = None,
    query_string: dict | str | None = None
) -> tuple[dict, str, bytes]:
    """
    Create test headers, path, and query string.
    
    Args:
        path: Request path
        headers: Request headers
        query_string: Query parameters
        
    Returns:
        Tuple of (headers_dict, path_str, query_bytes)
    """

def make_test_scope(
    method: str = "GET",
    path: str = "/",
    query_string: bytes = b"",
    headers: list | None = None,
    **kwargs
) -> dict:
    """
    Create ASGI test scope dictionary.
    
    Args:
        method: HTTP method
        path: Request path
        query_string: Query string bytes
        headers: Request headers as list of tuples
        **kwargs: Additional ASGI scope fields
        
    Returns:
        ASGI scope dictionary
    """

async def no_op_push() -> None:
    """No-operation push promise function for testing."""

# Testing constants
sentinel: object
    """Sentinel object for testing placeholder values."""

class WebsocketResponseError(Exception):
    """
    Exception raised when WebSocket test encounters response error.
    
    Used to indicate protocol violations or unexpected responses
    during WebSocket testing.
    """

import pytest
from quart import Quart, jsonify, request

# Create test application
@pytest.fixture
async def app():
    app = Quart(__name__)
    
    @app.route('/')
    async def index():
        return 'Hello, World!'
    
    @app.route('/json')
    async def json_endpoint():
        return jsonify({'message': 'Hello, JSON!'})
    
    @app.route('/user/<int:user_id>')
    async def get_user(user_id):
        return jsonify({'id': user_id, 'name': f'User {user_id}'})
    
    @app.route('/create', methods=['POST'])
    async def create_item():
        data = await request.get_json()
        return jsonify({'created': data, 'id': 123}), 201
    
    return app

@pytest.fixture
async def client(app):
    return app.test_client()

# Test basic GET request
async def test_index(client):
    response = await client.get('/')
    assert response.status_code == 200
    assert await response.get_data() == b'Hello, World!'

# Test JSON response
async def test_json_endpoint(client):
    response = await client.get('/json')
    assert response.status_code == 200
    
    json_data = await response.get_json()
    assert json_data == {'message': 'Hello, JSON!'}

# Test URL parameters
async def test_user_endpoint(client):
    response = await client.get('/user/42')
    assert response.status_code == 200
    
    json_data = await response.get_json()
    assert json_data['id'] == 42
    assert json_data['name'] == 'User 42'

# Test POST with JSON data
async def test_create_item(client):
    test_data = {'name': 'Test Item', 'value': 100}
    
    response = await client.post('/create', json=test_data)
    assert response.status_code == 201
    
    json_data = await response.get_json()
    assert json_data['created'] == test_data
    assert json_data['id'] == 123

import pytest
from quart import Quart, request, session, jsonify

@pytest.fixture
async def app():
    app = Quart(__name__)
    app.secret_key = 'test-secret-key'
    
    @app.route('/upload', methods=['POST'])
    async def upload_file():
        files = await request.files
        uploaded_file = files.get('document')
        
        if uploaded_file and uploaded_file.filename:
            content = await uploaded_file.read()
            return jsonify({
                'filename': uploaded_file.filename,
                'size': len(content),
                'content_type': uploaded_file.content_type
            })
        
        return jsonify({'error': 'No file uploaded'}), 400
    
    @app.route('/form', methods=['POST'])
    async def process_form():
        form_data = await request.form
        return jsonify(dict(form_data))
    
    @app.route('/session/set/<key>/<value>')
    async def set_session(key, value):
        session[key] = value
        return jsonify({'set': {key: value}})
    
    @app.route('/session/get/<key>')
    async def get_session(key):
        value = session.get(key)
        return jsonify({key: value})
    
    @app.route('/headers')
    async def check_headers():
        return jsonify({
            'user_agent': request.headers.get('User-Agent'),
            'custom_header': request.headers.get('X-Custom-Header')
        })
    
    return app

# Test file upload
async def test_file_upload(client):
    file_data = b'This is test file content'
    
    response = await client.post('/upload', files={
        'document': (io.BytesIO(file_data), 'test.txt', 'text/plain')
    })
    
    assert response.status_code == 200
    json_data = await response.get_json()
    assert json_data['filename'] == 'test.txt'
    assert json_data['size'] == len(file_data)
    assert json_data['content_type'] == 'text/plain'

# Test form data
async def test_form_submission(client):
    form_data = {'username': 'testuser', 'email': 'test@example.com'}
    
    response = await client.post('/form', form=form_data)
    assert response.status_code == 200
    
    json_data = await response.get_json()
    assert json_data == form_data

# Test session handling
async def test_session(client):
    # Set session value
    response = await client.get('/session/set/user_id/123')
    assert response.status_code == 200
    
    # Get session value
    response = await client.get('/session/get/user_id')
    assert response.status_code == 200
    
    json_data = await response.get_json()
    assert json_data['user_id'] == '123'

# Test custom headers
async def test_custom_headers(client):
    response = await client.get('/headers', headers={
        'User-Agent': 'Test Client 1.0',
        'X-Custom-Header': 'CustomValue'
    })
    
    assert response.status_code == 200
    json_data = await response.get_json()
    assert json_data['user_agent'] == 'Test Client 1.0'
    assert json_data['custom_header'] == 'CustomValue'

# Test query parameters
async def test_query_parameters(client):
    response = await client.get('/search', query_string={
        'q': 'test query',
        'page': 2,
        'limit': 50
    })
    
    # Or using string format
    response = await client.get('/search?q=test+query&page=2&limit=50')

import pytest
from quart import Quart, websocket
import json

@pytest.fixture
async def app():
    app = Quart(__name__)
    
    @app.websocket('/ws/echo')
    async def echo_websocket():
        await websocket.accept()
        
        while True:
            try:
                message = await websocket.receive()
                await websocket.send(f'Echo: {message}')
            except ConnectionClosed:
                break
    
    @app.websocket('/ws/json')
    async def json_websocket():
        await websocket.accept()
        
        while True:
            try:
                data = await websocket.receive_json()
                response = {
                    'type': 'response',
                    'original': data,
                    'timestamp': time.time()
                }
                await websocket.send_json(response)
            except ConnectionClosed:
                break
    
    @app.websocket('/ws/auth')
    async def auth_websocket():
        # Check authentication
        token = websocket.args.get('token')
        if token != 'valid_token':
            await websocket.close(code=1008, reason='Authentication failed')
            return
        
        await websocket.accept()
        await websocket.send('Authenticated successfully')
        
        while True:
            try:
                message = await websocket.receive()
                await websocket.send(f'Authenticated echo: {message}')
            except ConnectionClosed:
                break
    
    return app

# Test basic WebSocket echo
async def test_websocket_echo(client):
    async with client.websocket('/ws/echo') as ws:
        await ws.send('Hello, WebSocket!')
        response = await ws.receive()
        assert response == 'Echo: Hello, WebSocket!'

# Test JSON WebSocket communication
async def test_websocket_json(client):
    async with client.websocket('/ws/json') as ws:
        test_data = {'message': 'Hello', 'value': 42}
        await ws.send_json(test_data)
        
        response = await ws.receive_json()
        assert response['type'] == 'response'
        assert response['original'] == test_data
        assert 'timestamp' in response

# Test WebSocket authentication
async def test_websocket_auth_success(client):
    async with client.websocket('/ws/auth', query_string={'token': 'valid_token'}) as ws:
        welcome = await ws.receive()
        assert welcome == 'Authenticated successfully'
        
        await ws.send('Test message')
        response = await ws.receive()
        assert response == 'Authenticated echo: Test message'

async def test_websocket_auth_failure(client):
    # Test authentication failure
    with pytest.raises(WebsocketResponseError):
        async with client.websocket('/ws/auth', query_string={'token': 'invalid_token'}) as ws:
            # Should raise exception due to authentication failure
            pass

# Test WebSocket connection handling
async def test_websocket_multiple_messages(client):
    async with client.websocket('/ws/echo') as ws:
        messages = ['Message 1', 'Message 2', 'Message 3']
        
        for message in messages:
            await ws.send(message)
            response = await ws.receive()
            assert response == f'Echo: {message}'

import pytest
import click
from quart import Quart

@pytest.fixture
async def app():
    app = Quart(__name__)
    
    @app.cli.command()
    @click.argument('name')
    def greet(name):
        """Greet someone."""
        click.echo(f'Hello, {name}!')
    
    @app.cli.command()
    @click.option('--count', default=1, help='Number of greetings')
    @click.argument('name')
    def multi_greet(count, name):
        """Greet someone multiple times."""
        for i in range(count):
            click.echo(f'{i+1}. Hello, {name}!')
    
    @app.cli.command()
    def database_init():
        """Initialize database."""
        click.echo('Initializing database...')
        # Simulate database initialization
        click.echo('Database initialized successfully!')
    
    return app

# Test basic CLI command
async def test_greet_command(app):
    runner = app.test_cli_runner()
    
    result = runner.invoke(app.cli, ['greet', 'World'])
    assert result.exit_code == 0
    assert 'Hello, World!' in result.output

# Test CLI command with options
async def test_multi_greet_command(app):
    runner = app.test_cli_runner()
    
    result = runner.invoke(app.cli, ['multi-greet', '--count', '3', 'Alice'])
    assert result.exit_code == 0
    assert '1. Hello, Alice!' in result.output
    assert '2. Hello, Alice!' in result.output
    assert '3. Hello, Alice!' in result.output

# Test CLI command without arguments
async def test_database_init_command(app):
    runner = app.test_cli_runner()
    
    result = runner.invoke(app.cli, ['database-init'])
    assert result.exit_code == 0
    assert 'Initializing database...' in result.output
    assert 'Database initialized successfully!' in result.output

# Test CLI command error handling
async def test_cli_error_handling(app):
    runner = app.test_cli_runner()
    
    # Test missing required argument
    result = runner.invoke(app.cli, ['greet'])
    assert result.exit_code != 0
    assert 'Error' in result.output

import pytest
from quart import Quart, render_template, request, jsonify

@pytest.fixture
async def app():
    app = Quart(__name__)
    app.config['TESTING'] = True
    
    # Mock database
    users_db = {
        1: {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'},
        2: {'id': 2, 'name': 'Bob', 'email': 'bob@example.com'}
    }
    
    @app.route('/api/users')
    async def list_users():
        return jsonify(list(users_db.values()))
    
    @app.route('/api/users/<int:user_id>')
    async def get_user(user_id):
        user = users_db.get(user_id)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        return jsonify(user)
    
    @app.route('/api/users', methods=['POST'])
    async def create_user():
        data = await request.get_json()
        
        # Validate required fields
        if not data or 'name' not in data or 'email' not in data:
            return jsonify({'error': 'Name and email required'}), 400
        
        # Create new user
        new_id = max(users_db.keys()) + 1 if users_db else 1
        user = {
            'id': new_id,
            'name': data['name'],
            'email': data['email']
        }
        users_db[new_id] = user
        
        return jsonify(user), 201
    
    @app.route('/api/users/<int:user_id>', methods=['DELETE'])
    async def delete_user(user_id):
        if user_id not in users_db:
            return jsonify({'error': 'User not found'}), 404
        
        del users_db[user_id]
        return '', 204
    
    return app

# Test complete CRUD operations
async def test_user_crud_operations(client):
    # List users (should have initial data)
    response = await client.get('/api/users')
    assert response.status_code == 200
    users = await response.get_json()
    assert len(users) == 2
    
    # Get specific user
    response = await client.get('/api/users/1')
    assert response.status_code == 200
    user = await response.get_json()
    assert user['name'] == 'Alice'
    
    # Create new user
    new_user = {'name': 'Charlie', 'email': 'charlie@example.com'}
    response = await client.post('/api/users', json=new_user)
    assert response.status_code == 201
    created_user = await response.get_json()
    assert created_user['name'] == 'Charlie'
    assert created_user['id'] == 3
    
    # Verify user was created
    response = await client.get('/api/users')
    users = await response.get_json()
    assert len(users) == 3
    
    # Delete user
    response = await client.delete('/api/users/3')
    assert response.status_code == 204
    
    # Verify user was deleted
    response = await client.get('/api/users/3')
    assert response.status_code == 404

# Test error conditions
async def test_error_conditions(client):
    # Test getting non-existent user
    response = await client.get('/api/users/999')
    assert response.status_code == 404
    error = await response.get_json()
    assert 'error' in error
    
    # Test creating user with missing data
    response = await client.post('/api/users', json={'name': 'Incomplete'})
    assert response.status_code == 400
    
    # Test deleting non-existent user
    response = await client.delete('/api/users/999')
    assert response.status_code == 404