Pytest Тестовый Фреймворк

# Pytest Configuration Examples

# 1. pytest.ini - Basic configuration
[tool:pytest]
# Test discovery patterns
python_files = test_*.py *_test.py
python_classes = Test*
python_functions = test_*

# Test directories
testpaths = tests

# Output options
addopts =
    -v
    --tb=short
    --strict-markers
    --disable-warnings
    --color=yes

# Minimum version required
minversion = 6.0

# Markers
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: marks tests as integration tests
    unit: marks tests as unit tests
    smoke: marks tests as smoke tests
    regression: marks tests as regression tests

# 2. pyproject.toml - Modern configuration
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-app"
version = "1.0.0"
description = "A sample Python application"
dependencies = [
    "fastapi",
    "sqlalchemy",
    "pydantic"
]

[project.optional-dependencies]
test = [
    "pytest>=7.0.0",
    "pytest-cov>=4.0.0",
    "pytest-mock>=3.10.0",
    "pytest-asyncio>=0.21.0",
    "pytest-xdist>=3.0.0",
    "httpx",
    "factory-boy"
]

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
    "-v",
    "--tb=short",
    "--strict-markers",
    "--strict-config",
    "--cov=src",
    "--cov-report=term-missing",
    "--cov-report=html",
    "--cov-report=xml",
]
markers = [
    "slow: marks tests as slow (deselect with '-m "not slow"')",
    "integration: marks tests as integration tests",
    "unit: marks tests as unit tests",
    "smoke: marks tests as smoke tests",
]

# 3. conftest.py - Shared fixtures and configuration
import pytest
import tempfile
import shutil
from pathlib import Path
from typing import Generator, Any
import asyncio
from unittest.mock import Mock, AsyncMock

# Test database setup
@pytest.fixture(scope="session")
def test_db_url() -> str:
    """Provide test database URL."""
    return "sqlite:///:memory:"

@pytest.fixture(scope="function")
def test_db(test_db_url: str) -> Generator[Any, None, None]:
    """Create a test database session."""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker, Session

    engine = create_engine(test_db_url)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    # Create tables
    from src.models import Base
    Base.metadata.create_all(bind=engine)

    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()
        engine.dispose()

# Temporary directory fixture
@pytest.fixture
def temp_dir() -> Generator[Path, None, None]:
    """Create a temporary directory for tests."""
    temp_path = Path(tempfile.mkdtemp())
    try:
        yield temp_path
    finally:
        shutil.rmtree(temp_path)

# Mock external API
@pytest.fixture
def mock_api_client():
    """Mock API client for testing."""
    client = Mock()
    client.get.return_value = {"status": "success", "data": {}}
    client.post.return_value = {"status": "created", "data": {}}
    client.put.return_value = {"status": "updated", "data": {}}
    client.delete.return_value = {"status": "deleted", "data": {}}
    return client

# Async mock fixture
@pytest.fixture
def async_mock():
    """Create an async mock object."""
    return AsyncMock()

# Test user fixture
@pytest.fixture
def test_user():
    """Create a test user."""
    return {
        "id": 1,
        "username": "testuser",
        "email": "[email protected]",
        "is_active": True
    }

# Event loop for async tests
@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

# Environment setup
@pytest.fixture(autouse=True)
def setup_test_environment(monkeypatch):
    """Setup test environment variables."""
    monkeypatch.setenv("TESTING", "true")
    monkeypatch.setenv("DATABASE_URL", "sqlite:///:memory:")
    monkeypatch.setenv("SECRET_KEY", "test-secret-key")

# Logging configuration
@pytest.fixture(autouse=True)
def configure_logging():
    """Configure logging for tests."""
    import logging
    logging.getLogger().setLevel(logging.WARNING)

# 4. tests/__init__.py
# Empty file to make tests directory a Python package

# 5. tests/test_basic.py - Basic test examples
import pytest
from src.calculator import Calculator

class TestCalculatorBasics:
    """Basic calculator tests."""

    def test_calculator_addition(self):
        """Test basic addition."""
        calc = Calculator()
        assert calc.add(2, 3) == 5
        assert calc.add(-1, 1) == 0
        assert calc.add(0, 0) == 0

    def test_calculator_subtraction(self):
        """Test basic subtraction."""
        calc = Calculator()
        assert calc.subtract(5, 3) == 2
        assert calc.subtract(1, 1) == 0
        assert calc.subtract(0, 5) == -5

    def test_calculator_multiplication(self):
        """Test basic multiplication."""
        calc = Calculator()
        assert calc.multiply(3, 4) == 12
        assert calc.multiply(-2, 3) == -6
        assert calc.multiply(0, 5) == 0

    def test_calculator_division(self):
        """Test basic division."""
        calc = Calculator()
        assert calc.divide(10, 2) == 5
        assert calc.divide(-6, 3) == -2
        assert calc.divide(0, 5) == 0

    def test_division_by_zero(self):
        """Test division by zero raises exception."""
        calc = Calculator()
        with pytest.raises(ZeroDivisionError, match="Cannot divide by zero"):
            calc.divide(5, 0)

# Pytest Fixtures and Parametrization Examples

# 1. conftest.py - Advanced fixtures
import pytest
import tempfile
import json
from pathlib import Path
from typing import List, Dict, Any, Generator
import factory
from factory import fuzzy
from datetime import datetime, timedelta
import random

# Factory Boy setup for test data
class UserFactory(factory.Factory):
    class Meta:
        model = dict

    id = factory.Sequence(lambda n: n + 1)
    username = factory.Faker('user_name')
    email = factory.LazyAttribute(lambda obj: f'{obj.username}@example.com')
    first_name = factory.Faker('first_name')
    last_name = factory.Faker('last_name')
    is_active = True
    created_at = factory.LazyFunction(datetime.now)

class ProductFactory(factory.Factory):
    class Meta:
        model = dict

    id = factory.Sequence(lambda n: n + 1)
    name = factory.Faker('word')
    price = fuzzy.FuzzyDecimal(1.0, 1000.0, 2)
    description = factory.Faker('text', max_nb_chars=200)
    in_stock = fuzzy.FuzzyChoice([True, False])
    created_at = factory.LazyFunction(datetime.now)

# Database fixtures with different scopes
@pytest.fixture(scope="function")
def clean_db():
    """Create a clean database for each test."""
    # In a real app, this would setup a fresh database
    return {
        'users': [],
        'products': [],
        'orders': []
    }

@pytest.fixture(scope="session")
def shared_db():
    """Create a shared database for the entire session."""
    # Setup once for all tests
    db = {'users': [], 'products': [], 'orders': []}
    yield db
    # Cleanup after all tests

# Parametrized test data fixtures
@pytest.fixture(params=[
    {"username": "user1", "email": "[email protected]", "role": "admin"},
    {"username": "user2", "email": "[email protected]", "role": "user"},
    {"username": "user3", "email": "[email protected]", "role": "moderator"}
])
def user_data(request):
    """Provide different user data for tests."""
    return request.param

@pytest.fixture(params=[
    ("valid_token", "active"),
    ("expired_token", "inactive"),
    ("invalid_token", "inactive")
])
def auth_scenarios(request):
    """Provide different authentication scenarios."""
    return request.param

# File fixtures
@pytest.fixture
def sample_json_file(temp_dir):
    """Create a sample JSON file for testing."""
    data = {
        "users": [
            {"id": 1, "name": "John Doe", "email": "[email protected]"},
            {"id": 2, "name": "Jane Smith", "email": "[email protected]"}
        ],
        "settings": {
            "theme": "dark",
            "notifications": True
        }
    }

    file_path = temp_dir / "sample.json"
    with open(file_path, 'w') as f:
        json.dump(data, f, indent=2)

    return file_path

# API fixtures
@pytest.fixture
def api_client():
    """Mock API client."""
    class MockAPIClient:
        def __init__(self):
            self.responses = {}
            self.requests = []

        def set_response(self, endpoint, response):
            self.responses[endpoint] = response

        def get(self, endpoint):
            self.requests.append(('GET', endpoint))
            return self.responses.get(endpoint, {})

        def post(self, endpoint, data):
            self.requests.append(('POST', endpoint, data))
            return self.responses.get(endpoint, {})

    return MockAPIClient()

# 2. tests/test_fixtures.py - Using fixtures
import pytest
from src.user_service import UserService
from src.models import User

class TestUserServiceWithFixtures:
    """Test UserService using fixtures."""

    def test_create_user_with_factory(self, test_db):
        """Test creating a user with factory data."""
        user_data = UserFactory()

        user_service = UserService(test_db)
        user = user_service.create_user(user_data)

        assert user['id'] is not None
        assert user['username'] == user_data['username']
        assert user['email'] == user_data['email']
        assert len(test_db['users']) == 1

    def test_user_with_different_roles(self, user_data, test_db):
        """Test user creation with different roles."""
        user_service = UserService(test_db)
        user = user_service.create_user(user_data)

        assert user['username'] == user_data['username']
        assert 'role' in user_data  # Parametrized data

    def test_authentication_scenarios(self, auth_scenarios):
        """Test different authentication scenarios."""
        token, expected_status = auth_scenarios

        # Simulate authentication
        def authenticate(token):
            if token == "valid_token":
                return "active"
            return "inactive"

        status = authenticate(token)
        assert status == expected_status

    def test_file_operations(self, sample_json_file):
        """Test file operations with sample file."""
        with open(sample_json_file, 'r') as f:
            data = json.load(f)

        assert 'users' in data
        assert len(data['users']) == 2
        assert data['settings']['theme'] == 'dark'

    def test_api_client_usage(self, api_client):
        """Test API client with fixture."""
        # Setup mock response
        api_client.set_response('/users', {'users': []})

        # Make API call
        response = api_client.get('/users')

        # Verify request was made
        assert ('GET', '/users') in api_client.requests
        assert response == {'users': []}

# 3. tests/test_parametrization.py - Parametrization examples
import pytest

class TestParametrization:
    """Examples of test parametrization."""

    # Simple parametrization
    @pytest.mark.parametrize("input_num,expected", [
        (2, 4),
        (3, 9),
        (4, 16),
        (5, 25),
        (0, 0),
        (-2, 4)
    ])
    def test_square(self, input_num, expected):
        """Test square function with various inputs."""
        def square(x):
            return x * x

        assert square(input_num) == expected

    # Parametrization with ids
    @pytest.mark.parametrize(
        "email,valid",
        [
            ("[email protected]", True, "valid_email"),
            ("[email protected]", True, "valid_email_with_dots"),
            ("[email protected]", True, "valid_email_with_plus"),
            ("invalid-email", False, "invalid_no_at"),
            ("@example.com", False, "invalid_no_user"),
            ("test@", False, "invalid_no_domain")
        ],
        ids=[
            "valid_basic",
            "valid_with_dots",
            "valid_with_plus",
            "invalid_no_at",
            "invalid_no_user",
            "invalid_no_domain"
        ]
    )
    def test_email_validation(self, email, valid):
        """Test email validation with various formats."""
        import re

        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$'
        is_valid = bool(re.match(email_regex, email))

        assert is_valid == valid

    # Multiple parameters parametrization
    @pytest.mark.parametrize(
        "a,b,operation,expected",
        [
            (2, 3, "add", 5),
            (10, 5, "subtract", 5),
            (4, 3, "multiply", 12),
            (15, 3, "divide", 5),
            (0, 5, "add", 5),
            (10, 0, "multiply", 0)
        ]
    )
    def test_calculator_operations(self, a, b, operation, expected):
        """Test calculator with different operations."""
        def calculate(x, y, op):
            if op == "add":
                return x + y
            elif op == "subtract":
                return x - y
            elif op == "multiply":
                return x * y
            elif op == "divide":
                return x // y
            else:
                raise ValueError(f"Unknown operation: {op}")

        result = calculate(a, b, operation)
        assert result == expected

    # Using pytest.param for test identification
    @pytest.mark.parametrize(
        "user_input,expected_status",
        [
            pytest.param({"username": "valid_user", "password": "valid_pass"}, 200,
                        id="valid_credentials"),
            pytest.param({"username": "valid_user", "password": "wrong_pass"}, 401,
                        id="wrong_password"),
            pytest.param({"username": "nonexistent_user", "password": "any_pass"}, 401,
                        id="nonexistent_user"),
            pytest.param({"username": "", "password": "valid_pass"}, 400,
                        id="empty_username"),
            pytest.param({"username": "valid_user", "password": ""}, 400,
                        id="empty_password")
        ]
    )
    def test_user_login(self, user_input, expected_status):
        """Test user login with various credentials."""
        # Mock authentication service
        users = {"valid_user": "valid_pass"}

        def authenticate(username, password):
            if not username or not password:
                return 400
            if username in users and users[username] == password:
                return 200
            return 401

        status = authenticate(
            user_input.get("username", ""),
            user_input.get("password", "")
        )

        assert status == expected_status

    # Parametrization with fixtures
    @pytest.mark.parametrize("product_data", [
        {"name": "Laptop", "price": 999.99, "category": "electronics"},
        {"name": "Book", "price": 19.99, "category": "books"},
        {"name": "Shirt", "price": 29.99, "category": "clothing"}
    ])
    def test_product_creation(self, product_data, test_db):
        """Test product creation with different data."""
        from src.product_service import ProductService

        service = ProductService(test_db)
        product = service.create_product(product_data)

        assert product['name'] == product_data['name']
        assert product['price'] == product_data['price']
        assert product['category'] == product_data['category']
        assert len(test_db['products']) == 1

# 4. Dynamic parametrization
def pytest_generate_tests(metafunc):
    """Generate tests dynamically based on function parameters."""

    if "currency_pair" in metafunc.fixturenames:
        # Define currency pairs for testing
        currency_pairs = [
            ("USD", "EUR"),
            ("USD", "GBP"),
            ("EUR", "GBP"),
            ("USD", "JPY"),
            ("EUR", "JPY")
        ]

        metafunc.parametrize("currency_pair", currency_pairs)

    if "test_environment" in metafunc.fixturenames:
        # Define test environments
        environments = [
            "development",
            "staging",
            "production"
        ]

        metafunc.parametrize("test_environment", environments)

class TestDynamicParametrization:
    """Tests using dynamic parametrization."""

    def test_currency_conversion(self, currency_pair):
        """Test currency conversion between pairs."""
        from_currency, to_currency = currency_pair

        # Mock exchange rates
        rates = {
            ("USD", "EUR"): 0.85,
            ("USD", "GBP"): 0.73,
            ("EUR", "GBP"): 0.86,
            ("USD", "JPY"): 110.0,
            ("EUR", "JPY"): 129.0
        }

        def convert(amount, from_curr, to_curr):
            rate = rates.get((from_curr, to_curr))
            if rate:
                return amount * rate
            # Reverse the pair
            reverse_rate = rates.get((to_curr, from_curr))
            return amount / reverse_rate if reverse_rate else None

        result = convert(100, from_currency, to_currency)
        assert result is not None
        assert result > 0

# Pytest Async Testing Examples

# 1. pytest.ini or pyproject.toml - Async configuration
# Add to pytest.ini:
# [tool:pytest]
# asyncio_mode = auto

# Or to pyproject.toml:
# [tool.pytest.ini_options]
# asyncio_mode = "auto"
# async_test_mode = "auto"

# 2. conftest.py - Async fixtures
import pytest
import asyncio
import aiohttp
from typing import AsyncGenerator, Any
from unittest.mock import AsyncMock, MagicMock

# Async database fixture
@pytest.fixture(scope="function")
async def async_db():
    """Create an async database connection."""
    # Mock async database
    class AsyncDatabase:
        def __init__(self):
            self.data = {}
            self.connected = False

        async def connect(self):
            await asyncio.sleep(0.1)  # Simulate connection
            self.connected = True

        async def disconnect(self):
            await asyncio.sleep(0.1)  # Simulate disconnection
            self.connected = False

        async def get(self, key):
            await asyncio.sleep(0.01)  # Simulate I/O
            return self.data.get(key)

        async def set(self, key, value):
            await asyncio.sleep(0.01)  # Simulate I/O
            self.data[key] = value

        async def delete(self, key):
            await asyncio.sleep(0.01)  # Simulate I/O
            return self.data.pop(key, None)

    db = AsyncDatabase()
    await db.connect()

    try:
        yield db
    finally:
        await db.disconnect()

# Async HTTP client fixture
@pytest.fixture(scope="function")
async def http_client():
    """Create an async HTTP client."""
    class MockAsyncClient:
        def __init__(self):
            self.responses = {}
            self.requests = []

        def set_response(self, method, url, response):
            key = (method.upper(), url)
            self.responses[key] = response

        async def get(self, url, headers=None):
            self.requests.append(('GET', url, headers))
            return self.responses.get(('GET', url), MagicMock(status=404))

        async def post(self, url, data=None, json=None, headers=None):
            self.requests.append(('POST', url, data, json, headers))
            return self.responses.get(('POST', url), MagicMock(status=404))

        async def put(self, url, data=None, json=None, headers=None):
            self.requests.append(('PUT', url, data, json, headers))
            return self.responses.get(('PUT', url), MagicMock(status=404))

        async def delete(self, url, headers=None):
            self.requests.append(('DELETE', url, headers))
            return self.responses.get(('DELETE', url), MagicMock(status=404))

    return MockAsyncClient()

# Async service fixture
@pytest.fixture(scope="function")
async def async_service(async_db):
    """Create an async service with database."""
    class AsyncUserService:
        def __init__(self, db):
            self.db = db

        async def get_user(self, user_id):
            user_data = await self.db.get(f"user:{user_id}")
            if user_data:
                return User(**user_data)
            return None

        async def create_user(self, user_data):
            user_id = len([k for k in self.db.data.keys() if k.startswith("user:")]) + 1
            user_data['id'] = user_id
            await self.db.set(f"user:{user_id}", user_data)
            return User(**user_data)

        async def update_user(self, user_id, updates):
            user_data = await self.db.get(f"user:{user_id}")
            if user_data:
                user_data.update(updates)
                await self.db.set(f"user:{user_id}", user_data)
                return User(**user_data)
            return None

        async def delete_user(self, user_id):
            return await self.db.delete(f"user:{user_id}")

    class User:
        def __init__(self, id, username, email, **kwargs):
            self.id = id
            self.username = username
            self.email = email
            for key, value in kwargs.items():
                setattr(self, key, value)

    return AsyncUserService(async_db)

# Async event loop fixture
@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

# Async mock fixture
@pytest.fixture
def async_mock():
    """Create an async mock object."""
    return AsyncMock()

# 3. tests/test_async_basic.py - Basic async tests
import pytest
import asyncio

class TestAsyncBasics:
    """Basic async testing examples."""

    @pytest.mark.asyncio
    async def test_simple_async_function(self):
        """Test a simple async function."""
        async def compute_square(x):
            await asyncio.sleep(0.1)  # Simulate async work
            return x * x

        result = await compute_square(5)
        assert result == 25

    @pytest.mark.asyncio
    async def test_async_with_timeout(self):
        """Test async function with timeout."""
        async def slow_function():
            await asyncio.sleep(2)
            return "completed"

        # Should complete within timeout
        result = await asyncio.wait_for(slow_function(), timeout=3.0)
        assert result == "completed"

    @pytest.mark.asyncio
    async def test_async_exception_handling(self):
        """Test async exception handling."""
        async def failing_function():
            await asyncio.sleep(0.1)
            raise ValueError("Async error occurred")

        with pytest.raises(ValueError, match="Async error occurred"):
            await failing_function()

    @pytest.mark.asyncio
    async def test_async_context_manager(self):
        """Test async context manager."""
        class AsyncContextManager:
            async def __aenter__(self):
                await asyncio.sleep(0.1)
                return {"initialized": True}

            async def __aexit__(self, exc_type, exc_val, exc_tb):
                await asyncio.sleep(0.1)
                return False

        async with AsyncContextManager() as context:
            assert context["initialized"] is True

# 4. tests/test_async_database.py - Database testing
import pytest
from datetime import datetime

class TestAsyncDatabase:
    """Test async database operations."""

    @pytest.mark.asyncio
    async def test_async_crud_operations(self, async_service):
        """Test Create, Read, Update, Delete operations."""
        # Create user
        user_data = {
            "username": "testuser",
            "email": "[email protected]",
            "created_at": datetime.now().isoformat()
        }

        created_user = await async_service.create_user(user_data)
        assert created_user.id is not None
        assert created_user.username == "testuser"
        assert created_user.email == "[email protected]"

        # Read user
        retrieved_user = await async_service.get_user(created_user.id)
        assert retrieved_user is not None
        assert retrieved_user.id == created_user.id
        assert retrieved_user.username == "testuser"

        # Update user
        updates = {"email": "[email protected]"}
        updated_user = await async_service.update_user(created_user.id, updates)
        assert updated_user is not None
        assert updated_user.email == "[email protected]"

        # Delete user
        deleted = await async_service.delete_user(created_user.id)
        assert deleted is not None

        # Verify deletion
        final_user = await async_service.get_user(created_user.id)
        assert final_user is None

    @pytest.mark.asyncio
    async def test_async_concurrent_operations(self, async_service):
        """Test concurrent async operations."""
        # Create multiple users concurrently
        user_creation_tasks = []
        for i in range(5):
            user_data = {
                "username": f"user{i}",
                "email": f"user{i}@example.com"
            }
            task = async_service.create_user(user_data)
            user_creation_tasks.append(task)

        created_users = await asyncio.gather(*user_creation_tasks)
        assert len(created_users) == 5

        # Verify all users were created with different IDs
        user_ids = [user.id for user in created_users]
        assert len(set(user_ids)) == 5  # All IDs should be unique

# 5. tests/test_async_http.py - HTTP client testing
import pytest
from unittest.mock import MagicMock

class TestAsyncHTTP:
    """Test async HTTP operations."""

    @pytest.mark.asyncio
    async def test_async_get_request(self, http_client):
        """Test async GET request."""
        # Setup mock response
        mock_response = MagicMock()
        mock_response.status = 200
        mock_response.json = AsyncMock(return_value={"message": "success"})

        http_client.set_response("GET", "/api/users", mock_response)

        # Make request
        response = await http_client.get("/api/users")

        assert response.status == 200
        assert ("GET", "/api/users", None) in http_client.requests

        data = await response.json()
        assert data["message"] == "success"

    @pytest.mark.asyncio
    async def test_async_post_request(self, http_client):
        """Test async POST request."""
        # Setup mock response
        mock_response = MagicMock()
        mock_response.status = 201
        mock_response.json = AsyncMock(return_value={"id": 1, "name": "John"})

        http_client.set_response("POST", "/api/users", mock_response)

        # Make request
        response = await http_client.post(
            "/api/users",
            json={"name": "John", "email": "[email protected]"}
        )

        assert response.status == 201
        assert ("POST", "/api/users", None, {"name": "John", "email": "[email protected]"}, None) in http_client.requests

    @pytest.mark.asyncio
    async def test_async_request_with_headers(self, http_client):
        """Test async request with custom headers."""
        # Setup mock response
        mock_response = MagicMock()
        mock_response.status = 200
        mock_response.json = AsyncMock(return_value={"authenticated": True})

        http_client.set_response("GET", "/api/protected", mock_response)

        # Make request with headers
        headers = {"Authorization": "Bearer token123"}
        response = await http_client.get("/api/protected", headers=headers)

        assert response.status == 200
        assert ("GET", "/api/protected", headers) in http_client.requests

# 6. tests/test_async_websockets.py - WebSocket testing
import pytest
import json
from unittest.mock import AsyncMock, MagicMock

class TestAsyncWebSockets:
    """Test async WebSocket operations."""

    @pytest.mark.asyncio
    async def test_websocket_connection(self, async_mock):
        """Test WebSocket connection and messaging."""
        # Mock WebSocket connection
        websocket = AsyncMock()
        websocket.recv = AsyncMock(side_effect=[
            json.dumps({"type": "message", "content": "Hello"}),
            json.dumps({"type": "disconnect"}),
        ])
        websocket.send = AsyncMock()

        # Simulate WebSocket handler
        async def handle_websocket(websocket):
            try:
                while True:
                    message = await websocket.recv()
                    data = json.loads(message)

                    if data["type"] == "message":
                        response = {"type": "echo", "content": data["content"]}
                        await websocket.send(json.dumps(response))
                    elif data["type"] == "disconnect":
                        break
            except Exception as e:
                print(f"WebSocket error: {e}")

        # Run handler
        await handle_websocket(websocket)

        # Verify messages were received and responses sent
        assert websocket.recv.call_count == 2
        assert websocket.send.call_count == 1

        # Check echo response
        sent_response = json.loads(websocket.send.call_args[0][0])
        assert sent_response["type"] == "echo"
        assert sent_response["content"] == "Hello"

    @pytest.mark.asyncio
    async def test_websocket_broadcast(self):
        """Test WebSocket broadcast to multiple clients."""
        # Mock multiple WebSocket clients
        clients = [AsyncMock() for _ in range(3)]

        async def broadcast_message(websockets, message):
            """Broadcast message to all connected websockets."""
            tasks = [ws.send(message) for ws in websockets]
            await asyncio.gather(*tasks)

        # Broadcast message
        message = json.dumps({"type": "broadcast", "content": "Hello everyone!"})
        await broadcast_message(clients, message)

        # Verify all clients received the message
        for client in clients:
            client.send.assert_called_once_with(message)

# 7. Performance testing with async
class TestAsyncPerformance:
    """Test async performance patterns."""

    @pytest.mark.asyncio
    async def test_async_vs_sync_performance(self):
        """Compare async vs sync performance."""
        import time

        # Synchronous approach
        def sync_approach():
            time.sleep(0.1)  # Simulate blocking I/O
            return "sync_result"

        # Asynchronous approach
        async def async_approach():
            await asyncio.sleep(0.1)  # Simulate non-blocking I/O
            return "async_result"

        # Test synchronous execution
        start_time = time.time()
        results = []
        for _ in range(3):
            result = sync_approach()
            results.append(result)
        sync_time = time.time() - start_time

        # Test asynchronous execution
        start_time = time.time()
        async_tasks = [async_approach() for _ in range(3)]
        results = await asyncio.gather(*async_tasks)
        async_time = time.time() - start_time

        # Async should be faster for concurrent I/O operations
        assert async_time < sync_time
        assert len(results) == 3
        assert all(result == "async_result" for result in results)