# Async Testing Patterns

Testing asyncio code with pytest-asyncio.

## Setup

```bash
uv add --dev pytest-asyncio
```

```ini
# pytest.ini or pyproject.toml
[pytest]
asyncio_mode = auto  # Recommended for pytest-asyncio 0.21+
```

## Basic Async Tests

```python
import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await async_fetch_data()
    assert result["status"] == "ok"

@pytest.mark.asyncio
async def test_async_context_manager():
    async with AsyncResource() as resource:
        result = await resource.get()
        assert result is not None
```

## Async Fixtures

```python
import pytest
import aiohttp

@pytest.fixture
async def async_client():
    """Async fixture with cleanup."""
    async with aiohttp.ClientSession() as session:
        yield session
    # Session closed automatically

@pytest.fixture
async def database():
    """Async database fixture."""
    conn = await create_async_connection()
    await conn.execute("BEGIN")
    yield conn
    await conn.execute("ROLLBACK")
    await conn.close()

@pytest.mark.asyncio
async def test_with_async_fixture(async_client):
    async with async_client.get("https://httpbin.org/json") as resp:
        data = await resp.json()
        assert "slideshow" in data
```

## Fixture Scopes

```python
@pytest.fixture(scope="session")
async def app():
    """Session-scoped async fixture."""
    app = await create_app()
    yield app
    await app.shutdown()

@pytest.fixture(scope="module")
async def db_pool():
    """Module-scoped connection pool."""
    pool = await asyncpg.create_pool(DATABASE_URL)
    yield pool
    await pool.close()
```

## Testing Timeouts

```python
import asyncio

@pytest.mark.asyncio
async def test_timeout():
    with pytest.raises(asyncio.TimeoutError):
        async with asyncio.timeout(0.1):
            await asyncio.sleep(1.0)

@pytest.mark.asyncio
async def test_wait_for():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(), timeout=0.1)
```

## Testing Cancellation

```python
@pytest.mark.asyncio
async def test_task_cancellation():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(0.01)
    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task

@pytest.mark.asyncio
async def test_graceful_cancellation():
    """Test that cleanup runs on cancellation."""
    cleanup_ran = False

    async def task_with_cleanup():
        nonlocal cleanup_ran
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            cleanup_ran = True
            raise

    task = asyncio.create_task(task_with_cleanup())
    await asyncio.sleep(0.01)
    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task

    assert cleanup_ran
```

## Testing gather

```python
@pytest.mark.asyncio
async def test_gather_success():
    results = await asyncio.gather(
        async_op_1(),
        async_op_2(),
        async_op_3(),
    )
    assert len(results) == 3

@pytest.mark.asyncio
async def test_gather_with_exceptions():
    results = await asyncio.gather(
        async_op_1(),
        async_op_that_fails(),
        async_op_3(),
        return_exceptions=True
    )
    assert isinstance(results[1], Exception)
```

## Testing TaskGroup (Python 3.11+)

```python
@pytest.mark.asyncio
async def test_task_group():
    results = []

    async with asyncio.TaskGroup() as tg:
        tg.create_task(append_result(results, 1))
        tg.create_task(append_result(results, 2))
        tg.create_task(append_result(results, 3))

    assert sorted(results) == [1, 2, 3]

@pytest.mark.asyncio
async def test_task_group_exception():
    with pytest.raises(ExceptionGroup):
        async with asyncio.TaskGroup() as tg:
            tg.create_task(successful_task())
            tg.create_task(failing_task())
```

## Mocking Async Functions

```python
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_mock_async_function(mocker):
    mock = mocker.patch("mymodule.async_api_call", new_callable=AsyncMock)
    mock.return_value = {"data": "mocked"}

    result = await mymodule.fetch_data()

    assert result == {"data": "mocked"}
    mock.assert_awaited_once()

@pytest.mark.asyncio
async def test_async_side_effect(mocker):
    mock = AsyncMock()
    mock.side_effect = [
        {"page": 1},
        {"page": 2},
        ValueError("No more pages"),
    ]

    assert await mock() == {"page": 1}
    assert await mock() == {"page": 2}
    with pytest.raises(ValueError):
        await mock()
```

## Testing aiohttp

```python
import aiohttp
from aiohttp import web
import pytest

@pytest.fixture
async def app():
    """Create aiohttp app."""
    app = web.Application()
    app.router.add_get("/", home_handler)
    return app

@pytest.fixture
async def client(aiohttp_client, app):
    """Create test client."""
    return await aiohttp_client(app)

@pytest.mark.asyncio
async def test_endpoint(client):
    resp = await client.get("/")
    assert resp.status == 200
    data = await resp.json()
    assert "message" in data
```

## Testing WebSockets

```python
@pytest.mark.asyncio
async def test_websocket(aiohttp_client, app):
    client = await aiohttp_client(app)

    async with client.ws_connect("/ws") as ws:
        await ws.send_str("Hello")
        msg = await ws.receive()
        assert msg.type == aiohttp.WSMsgType.TEXT
        assert msg.data == "Hello back"
```

## Event Loop Fixtures

```python
import pytest

@pytest.fixture(scope="session")
def event_loop_policy():
    """Custom event loop policy."""
    return asyncio.DefaultEventLoopPolicy()

# For uvloop
@pytest.fixture(scope="session")
def event_loop_policy():
    import uvloop
    return uvloop.EventLoopPolicy()
```

## Testing Queues

```python
@pytest.mark.asyncio
async def test_queue_producer_consumer():
    queue = asyncio.Queue()
    results = []

    async def producer():
        for i in range(3):
            await queue.put(i)
        await queue.put(None)  # Sentinel

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:
                break
            results.append(item)

    await asyncio.gather(producer(), consumer())
    assert results == [0, 1, 2]
```

## Best Practices

1. **Use `asyncio_mode = auto`** - Simplifies test marking
2. **Scope fixtures appropriately** - Session for expensive resources
3. **Use AsyncMock** - For mocking coroutines
4. **Test cancellation** - Ensure cleanup happens
5. **Test timeouts** - Verify timeout behavior
6. **Avoid blocking calls** - Use `run_in_executor` if needed
7. **Close resources** - Use async context managers

# Coverage Strategies

Comprehensive code coverage with pytest-cov.

## Setup

```bash
uv add --dev pytest-cov
```

## Basic Usage

```bash
# Run with coverage
pytest --cov=src

# With terminal report
pytest --cov=src --cov-report=term

# With HTML report
pytest --cov=src --cov-report=html
open htmlcov/index.html

# Multiple formats
pytest --cov=src --cov-report=term --cov-report=html --cov-report=xml
```

## Coverage Configuration

### pyproject.toml

```toml
[tool.coverage.run]
source = ["src"]
branch = true
omit = [
    "*/tests/*",
    "*/__init__.py",
    "*/migrations/*",
]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise NotImplementedError",
    "if TYPE_CHECKING:",
    "if __name__ == .__main__.:",
]
fail_under = 80
show_missing = true

[tool.coverage.html]
directory = "htmlcov"
```

### .coveragerc (Alternative)

```ini
[run]
source = src
branch = true
omit =
    */tests/*
    */__init__.py

[report]
exclude_lines =
    pragma: no cover
    raise NotImplementedError
fail_under = 80

[html]
directory = htmlcov
```

## Branch Coverage

```python
# branch=true catches this
def process(value):
    if value > 0:
        return "positive"
    # Missing else branch without branch coverage
    return "non-positive"

# Test both branches
def test_positive():
    assert process(5) == "positive"

def test_non_positive():
    assert process(-1) == "non-positive"
```

## Excluding Code

```python
def debug_only():  # pragma: no cover
    """Never executed in production."""
    print("Debug info")

if TYPE_CHECKING:  # Excluded by default config
    from typing import Optional

def platform_specific():
    if sys.platform == "win32":  # pragma: no cover
        return windows_implementation()
    return unix_implementation()
```

## Coverage in CI

### GitHub Actions

```yaml
name: Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Install uv
        uses: astral-sh/setup-uv@v5

      - name: Install dependencies
        run: uv sync

      - name: Run tests with coverage
        run: uv run pytest --cov=src --cov-report=xml

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v4
        with:
          files: ./coverage.xml
          fail_ci_if_error: true
```

### Fail on Low Coverage

```bash
# Fail if coverage below 80%
pytest --cov=src --cov-fail-under=80
```

## Measuring Coverage of Specific Tests

```bash
# Coverage for specific test file
pytest tests/test_api.py --cov=src/api

# Coverage for marked tests only
pytest -m "unit" --cov=src

# Coverage for specific module
pytest --cov=src/module_name
```

## Combining Coverage

```bash
# Run tests in parallel, combine coverage
pytest -n auto --cov=src --cov-append

# Or manually combine
coverage combine
coverage report
```

## Coverage Diff (Incremental)

```bash
# Show coverage for changed lines only (with diff-cover)
uv add --dev diff-cover

uv run pytest --cov=src --cov-report=xml
uv run diff-cover coverage.xml --compare-branch=origin/main
```

## Mutation Testing

```bash
# Beyond coverage: test quality with mutmut
uv add --dev mutmut

# Run mutation testing
uv run mutmut run --paths-to-mutate=src/

# View results
uv run mutmut results
uv run mutmut html
```

## Coverage Reports

### Terminal Report

```bash
pytest --cov=src --cov-report=term-missing
```

Output:
```
Name                      Stmts   Miss Branch BrPart  Cover   Missing
---------------------------------------------------------------------
src/api.py                   50      5     12      2    88%   45-49, 67
src/utils.py                 30      0      8      0   100%
---------------------------------------------------------------------
TOTAL                        80      5     20      2    92%
```

### HTML Report

```bash
pytest --cov=src --cov-report=html
# Creates htmlcov/index.html with line-by-line highlighting
```

### XML Report (CI)

```bash
pytest --cov=src --cov-report=xml
# Creates coverage.xml for CI tools
```

### JSON Report

```bash
pytest --cov=src --cov-report=json
# Creates coverage.json for programmatic access
```

## Coverage Best Practices

### 1. Aim for Meaningful Coverage

```python
# BAD: 100% coverage but no assertions
def test_function():
    result = my_function()  # Just call it

# GOOD: Meaningful assertions
def test_function():
    result = my_function()
    assert result.status == "success"
    assert len(result.items) > 0
```

### 2. Don't Chase 100%

```python
# Some code genuinely shouldn't be tested
def __repr__(self):  # pragma: no cover
    return f"<User {self.name}>"

if __name__ == "__main__":  # pragma: no cover
    main()
```

### 3. Focus on Critical Paths

```python
# Prioritize coverage for:
# - Business logic
# - Error handling
# - Edge cases
# - Security-sensitive code
```

### 4. Use Branch Coverage

```toml
[tool.coverage.run]
branch = true
```

### 5. Track Coverage Trends

```yaml
# In CI: fail on coverage decrease
- name: Check coverage
  run: |
    pytest --cov=src --cov-report=xml
    diff-cover coverage.xml --compare-branch=origin/main --fail-under=90
```

## Quick Reference

| Command | Description |
|---------|-------------|
| `--cov=src` | Enable coverage for src/ |
| `--cov-report=term` | Terminal report |
| `--cov-report=html` | HTML report |
| `--cov-report=xml` | XML report (CI) |
| `--cov-fail-under=80` | Fail if under 80% |
| `--cov-branch` | Enable branch coverage |
| `--cov-append` | Append to existing data |
| `--no-cov` | Disable coverage |

# Advanced Fixture Patterns

Deep dive into pytest fixtures for complex testing scenarios.

## Factory Fixtures

```python
import pytest
from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str
    email: str

@pytest.fixture
def user_factory():
    """Factory to create users with custom attributes."""
    def _create_user(
        id: int = 1,
        name: str = "Test User",
        email: str = "test@example.com"
    ) -> User:
        return User(id=id, name=name, email=email)
    return _create_user

def test_user_factory(user_factory):
    user1 = user_factory()
    user2 = user_factory(id=2, name="Another User")
    assert user1.id != user2.id
```

## Fixture Dependencies

```python
@pytest.fixture
def database():
    """Base database fixture."""
    db = connect_to_test_db()
    yield db
    db.close()

@pytest.fixture
def clean_database(database):
    """Depends on database, adds cleanup."""
    database.clear_all()
    yield database
    database.clear_all()

@pytest.fixture
def seeded_database(clean_database):
    """Depends on clean_database, adds seed data."""
    clean_database.insert(SEED_DATA)
    return clean_database
```

## Autouse Fixtures

```python
@pytest.fixture(autouse=True)
def reset_environment():
    """Runs automatically before each test."""
    os.environ.clear()
    os.environ.update(TEST_ENV)
    yield
    os.environ.clear()

@pytest.fixture(autouse=True, scope="module")
def setup_logging():
    """Module-level autouse fixture."""
    logging.disable(logging.CRITICAL)
    yield
    logging.disable(logging.NOTSET)
```

## Request Fixture

```python
@pytest.fixture
def temp_file(request, tmp_path):
    """Fixture that adapts based on test parameters."""
    # Access test-specific data
    filename = getattr(request, "param", "default.txt")
    file_path = tmp_path / filename
    file_path.write_text("test content")
    return file_path

@pytest.mark.parametrize("temp_file", ["custom.txt"], indirect=True)
def test_with_custom_filename(temp_file):
    assert temp_file.name == "custom.txt"
```

## Fixture Finalization

```python
@pytest.fixture
def resource_with_finalizer(request):
    """Using request.addfinalizer for cleanup."""
    resource = allocate_resource()

    def cleanup():
        resource.release()

    request.addfinalizer(cleanup)
    return resource

# Prefer yield-based cleanup when possible
@pytest.fixture
def resource_with_yield():
    """Preferred: yield-based cleanup."""
    resource = allocate_resource()
    yield resource
    resource.release()
```

## Fixture Caching

```python
@pytest.fixture(scope="session")
def expensive_computation():
    """Computed once, cached for entire session."""
    return perform_expensive_setup()

@pytest.fixture(scope="module")
def module_cache():
    """Cached per test module."""
    return load_module_data()
```

## Parametrized Fixtures

```python
@pytest.fixture(params=["sqlite", "postgres", "mysql"])
def database_backend(request):
    """Test runs 3 times, once per backend."""
    backend = request.param
    db = create_database(backend)
    yield db
    db.close()

def test_database_operations(database_backend):
    """This test runs against all 3 databases."""
    database_backend.insert({"key": "value"})
    assert database_backend.get("key") == "value"
```

## Fixture with IDs

```python
@pytest.fixture(
    params=[
        pytest.param({"user": "admin"}, id="admin-user"),
        pytest.param({"user": "guest"}, id="guest-user"),
    ]
)
def user_context(request):
    return request.param
```

## conftest.py Organization

```
tests/
├── conftest.py              # Session/package-wide fixtures
├── unit/
│   ├── conftest.py          # Unit test fixtures
│   └── test_module.py
├── integration/
│   ├── conftest.py          # Integration fixtures
│   └── test_api.py
└── e2e/
    ├── conftest.py          # E2E fixtures
    └── test_flows.py
```

### conftest.py Example

```python
# tests/conftest.py
import pytest

def pytest_configure(config):
    """Called after command line parsing."""
    config.addinivalue_line("markers", "slow: marks slow tests")

def pytest_collection_modifyitems(config, items):
    """Modify collected tests."""
    if config.getoption("--quick"):
        skip_slow = pytest.mark.skip(reason="--quick mode")
        for item in items:
            if "slow" in item.keywords:
                item.add_marker(skip_slow)

@pytest.fixture(scope="session")
def app():
    """Application for all tests."""
    from myapp import create_app
    return create_app(testing=True)

@pytest.fixture
def client(app):
    """Test client per test."""
    return app.test_client()

@pytest.fixture
def authenticated_client(client):
    """Client with auth token."""
    client.post("/login", json={"user": "test", "pass": "test"})
    return client
```

## Fixture Best Practices

1. **Single responsibility** - Each fixture does one thing
2. **Use factory fixtures** - When tests need variations
3. **Scope appropriately** - Don't over-cache or under-cache
4. **Prefer yield** - Over request.addfinalizer
5. **Name clearly** - `db_connection` not `fixture1`
6. **Document** - Explain what fixture provides and when to use
7. **Minimize side effects** - Clean up after yourself

# Integration Testing Patterns

Patterns for testing real systems, databases, and APIs.

## Database Testing with Transactions

```python
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="session")
def engine():
    """Create test database engine."""
    engine = create_engine("postgresql://test:test@localhost/testdb")
    return engine

@pytest.fixture(scope="session")
def tables(engine):
    """Create all tables once per session."""
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db_session(engine, tables):
    """
    Transaction rollback fixture.
    Each test runs in a transaction that's rolled back.
    """
    connection = engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()

    yield session

    session.close()
    transaction.rollback()
    connection.close()


def test_user_creation(db_session):
    """Test runs in rolled-back transaction."""
    user = User(name="Test")
    db_session.add(user)
    db_session.commit()  # Committed to transaction, not DB
    assert db_session.query(User).count() == 1
    # Rolled back after test - no cleanup needed
```

## Async Database Testing

```python
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

@pytest_asyncio.fixture(scope="session")
async def async_engine():
    engine = create_async_engine("postgresql+asyncpg://test:test@localhost/testdb")
    yield engine
    await engine.dispose()

@pytest_asyncio.fixture
async def async_session(async_engine):
    """Async session with rollback."""
    async with async_engine.connect() as conn:
        await conn.begin()
        async_session = AsyncSession(bind=conn)

        yield async_session

        await async_session.close()
        await conn.rollback()


@pytest.mark.asyncio
async def test_async_query(async_session):
    result = await async_session.execute(select(User))
    users = result.scalars().all()
    assert len(users) == 0
```

## TestContainers

```python
# uv add --dev testcontainers

import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def postgres():
    """Spin up PostgreSQL container for tests."""
    with PostgresContainer("postgres:15") as postgres:
        yield postgres

@pytest.fixture(scope="session")
def postgres_url(postgres):
    """Get connection URL for containerized PostgreSQL."""
    return postgres.get_connection_url()

@pytest.fixture(scope="session")
def redis():
    """Spin up Redis container for tests."""
    with RedisContainer("redis:7") as redis:
        yield redis

@pytest.fixture
def redis_client(redis):
    """Get Redis client for container."""
    import redis as redis_lib
    client = redis_lib.from_url(redis.get_container_host_ip())
    yield client
    client.flushdb()


def test_with_real_postgres(postgres_url):
    """Test against real PostgreSQL container."""
    engine = create_engine(postgres_url)
    # Use real database
```

## FastAPI / Starlette Testing

```python
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from httpx import AsyncClient

# Synchronous testing
@pytest.fixture
def app():
    return create_app()

@pytest.fixture
def client(app):
    """Sync test client."""
    return TestClient(app)

def test_endpoint(client):
    response = client.get("/api/users")
    assert response.status_code == 200
    assert "users" in response.json()


# Async testing with httpx
@pytest.fixture
async def async_client(app):
    """Async test client for async endpoints."""
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

@pytest.mark.asyncio
async def test_async_endpoint(async_client):
    response = await async_client.get("/api/users")
    assert response.status_code == 200


# With database override
@pytest.fixture
def app_with_db(db_session):
    """Override database dependency."""
    app = create_app()

    def get_test_db():
        yield db_session

    app.dependency_overrides[get_db] = get_test_db
    yield app
    app.dependency_overrides.clear()
```

## API Testing Patterns

```python
import pytest
from dataclasses import dataclass

@dataclass
class APITestCase:
    """Structured API test case."""
    method: str
    path: str
    json: dict | None = None
    expected_status: int = 200
    expected_json: dict | None = None
    headers: dict | None = None

@pytest.mark.parametrize("test_case", [
    APITestCase("GET", "/api/users", expected_status=200),
    APITestCase("POST", "/api/users", json={"name": "Test"}, expected_status=201),
    APITestCase("GET", "/api/users/999", expected_status=404),
])
def test_api_endpoints(client, test_case):
    """Parametrized API testing."""
    response = client.request(
        method=test_case.method,
        url=test_case.path,
        json=test_case.json,
        headers=test_case.headers,
    )
    assert response.status_code == test_case.expected_status

    if test_case.expected_json:
        assert response.json() == test_case.expected_json


# Request/Response validation
def test_user_creation_flow(client):
    """Test complete user flow."""
    # Create
    response = client.post("/api/users", json={"name": "Test User"})
    assert response.status_code == 201
    user_id = response.json()["id"]

    # Read
    response = client.get(f"/api/users/{user_id}")
    assert response.status_code == 200
    assert response.json()["name"] == "Test User"

    # Update
    response = client.patch(f"/api/users/{user_id}", json={"name": "Updated"})
    assert response.status_code == 200

    # Delete
    response = client.delete(f"/api/users/{user_id}")
    assert response.status_code == 204
```

## Snapshot Testing

```python
# uv add --dev syrupy

import pytest
from syrupy.assertion import SnapshotAssertion

def test_api_response_snapshot(client, snapshot: SnapshotAssertion):
    """Compare response against stored snapshot."""
    response = client.get("/api/config")
    assert response.json() == snapshot


def test_user_serialization(snapshot):
    """Snapshot complex objects."""
    user = User(id=1, name="Test", email="test@example.com")
    assert user.dict() == snapshot


# Update snapshots: pytest --snapshot-update
```

## External Service Mocking

```python
import pytest
import responses
import respx

# responses (requests library)
@responses.activate
def test_external_api():
    responses.add(
        responses.GET,
        "https://api.example.com/data",
        json={"result": "mocked"},
        status=200
    )

    result = fetch_from_external_api()
    assert result["result"] == "mocked"


# respx (httpx library)
@pytest.fixture
def mock_api():
    with respx.mock:
        yield respx

def test_httpx_external(mock_api):
    mock_api.get("https://api.example.com/data").respond(
        json={"result": "mocked"}
    )

    result = fetch_with_httpx()
    assert result["result"] == "mocked"
```

## Factory Fixtures for Integration Tests

```python
import pytest
from faker import Faker

fake = Faker()

@pytest.fixture
def user_factory(db_session):
    """Factory for creating test users."""
    created_users = []

    def _create_user(**kwargs):
        user = User(
            name=kwargs.get("name", fake.name()),
            email=kwargs.get("email", fake.email()),
            **kwargs
        )
        db_session.add(user)
        db_session.commit()
        created_users.append(user)
        return user

    yield _create_user

    # Cleanup handled by transaction rollback


def test_user_permissions(user_factory):
    admin = user_factory(role="admin")
    user = user_factory(role="user")

    assert admin.can_delete(user)
    assert not user.can_delete(admin)
```

## Quick Reference

| Pattern | Use Case | Key Benefit |
|---------|----------|-------------|
| Transaction rollback | DB tests | Zero cleanup needed |
| TestContainers | Real services | Production-like testing |
| TestClient | API testing | Full HTTP stack |
| Snapshot testing | Complex responses | Easy regression detection |
| Factory fixtures | Data creation | Flexible test data |
| respx/responses | External APIs | Isolated testing |

# Mocking Patterns

Comprehensive guide to mocking in pytest.

## unittest.mock Basics

### Mock Object

```python
from unittest.mock import Mock

def test_mock_basics():
    mock = Mock()

    # Access any attribute (auto-created)
    mock.some_attribute
    mock.method()
    mock.nested.deeply.value

    # Configure return values
    mock.get_data.return_value = {"key": "value"}
    assert mock.get_data() == {"key": "value"}

    # Check calls
    mock.get_data.assert_called_once()
    mock.get_data.assert_called_with()  # No args
```

### MagicMock

```python
from unittest.mock import MagicMock

def test_magic_mock():
    mock = MagicMock()

    # Supports magic methods
    mock.__len__.return_value = 5
    assert len(mock) == 5

    # Iteration
    mock.__iter__.return_value = iter([1, 2, 3])
    assert list(mock) == [1, 2, 3]

    # Context manager
    mock.__enter__.return_value = "entered"
    with mock as m:
        assert m == "entered"
```

## patch Decorator

```python
from unittest.mock import patch

# Patch where used, not where defined
@patch("mymodule.requests.get")
def test_api_call(mock_get):
    mock_get.return_value.json.return_value = {"status": "ok"}

    result = mymodule.fetch_data()

    assert result["status"] == "ok"
    mock_get.assert_called_once_with("https://api.example.com/data")

# Multiple patches (applied bottom-up)
@patch("mymodule.save_to_db")
@patch("mymodule.fetch_from_api")
def test_multiple_patches(mock_fetch, mock_save):  # Note: reverse order
    mock_fetch.return_value = {"data": []}
    process_and_save()
    mock_save.assert_called_once()
```

## patch Context Manager

```python
from unittest.mock import patch

def test_with_context_manager():
    with patch("mymodule.external_service") as mock_service:
        mock_service.call.return_value = "mocked"
        result = mymodule.do_work()
        assert result == "mocked"

    # After context, original is restored
```

## patch.object

```python
from unittest.mock import patch

class MyClass:
    def method(self):
        return "real"

def test_patch_object():
    obj = MyClass()

    with patch.object(obj, "method", return_value="mocked"):
        assert obj.method() == "mocked"

    assert obj.method() == "real"  # Restored
```

## patch.dict

```python
from unittest.mock import patch
import os

def test_patch_dict():
    with patch.dict(os.environ, {"API_KEY": "test-key"}):
        assert os.environ["API_KEY"] == "test-key"

    # Clear and add
    with patch.dict(os.environ, {"NEW_VAR": "value"}, clear=True):
        assert "PATH" not in os.environ
        assert os.environ["NEW_VAR"] == "value"
```

## side_effect

```python
from unittest.mock import Mock

def test_side_effect_function():
    mock = Mock()
    mock.side_effect = lambda x: x * 2
    assert mock(5) == 10

def test_side_effect_exception():
    mock = Mock()
    mock.side_effect = ValueError("Invalid input")

    with pytest.raises(ValueError):
        mock()

def test_side_effect_list():
    mock = Mock()
    mock.side_effect = [1, 2, ValueError("Done")]

    assert mock() == 1
    assert mock() == 2
    with pytest.raises(ValueError):
        mock()
```

## spec and autospec

```python
from unittest.mock import Mock, create_autospec

class RealAPI:
    def get_user(self, user_id: int) -> dict:
        pass

    def create_user(self, name: str) -> dict:
        pass

def test_with_spec():
    # Only allows methods that exist on RealAPI
    mock = Mock(spec=RealAPI)
    mock.get_user(1)  # OK
    # mock.invalid_method()  # AttributeError

def test_with_autospec():
    # Also validates signatures
    mock = create_autospec(RealAPI)
    mock.get_user(1)  # OK
    # mock.get_user("string")  # Still OK at runtime, but IDE warns
    # mock.get_user(1, 2, 3)  # TypeError: too many args
```

## pytest-mock Plugin

```python
# uv add --dev pytest-mock

def test_with_mocker(mocker):
    # mocker is a fixture that wraps unittest.mock
    mock = mocker.patch("mymodule.external_call")
    mock.return_value = "mocked"

    result = mymodule.process()

    assert result == "mocked"
    mock.assert_called_once()

def test_spy(mocker):
    # Spy: call real method but track calls
    spy = mocker.spy(mymodule, "helper_function")

    mymodule.main_function()

    spy.assert_called()
    # Original function was actually called

def test_stub(mocker):
    # Stub: quick attribute replacement
    mocker.patch.object(MyClass, "expensive_method", return_value="cheap")
```

## Async Mocking

```python
from unittest.mock import AsyncMock

async def test_async_mock():
    mock = AsyncMock()
    mock.return_value = {"async": "result"}

    result = await mock()

    assert result == {"async": "result"}
    mock.assert_awaited_once()

@patch("mymodule.async_fetch", new_callable=AsyncMock)
async def test_patch_async(mock_fetch):
    mock_fetch.return_value = {"data": []}

    result = await mymodule.get_data()

    assert result == {"data": []}
```

## PropertyMock

```python
from unittest.mock import PropertyMock, patch

class MyClass:
    @property
    def value(self):
        return "real"

def test_property_mock():
    with patch.object(
        MyClass, "value", new_callable=PropertyMock
    ) as mock_prop:
        mock_prop.return_value = "mocked"
        obj = MyClass()
        assert obj.value == "mocked"
```

## Common Patterns

### Mock HTTP Response

```python
def test_mock_response(mocker):
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {"id": 1}
    mock_response.raise_for_status = Mock()

    mocker.patch("requests.get", return_value=mock_response)

    result = fetch_user(1)
    assert result["id"] == 1
```

### Mock File Operations

```python
from unittest.mock import mock_open, patch

def test_file_read():
    m = mock_open(read_data="file content")
    with patch("builtins.open", m):
        result = read_config("config.txt")
        assert "content" in result

def test_file_write():
    m = mock_open()
    with patch("builtins.open", m):
        write_data("output.txt", "data")
        m().write.assert_called_with("data")
```

### Mock datetime

```python
from unittest.mock import patch
from datetime import datetime

def test_mock_datetime(mocker):
    mock_dt = mocker.patch("mymodule.datetime")
    mock_dt.now.return_value = datetime(2024, 1, 15, 12, 0, 0)

    result = mymodule.get_timestamp()
    assert "2024-01-15" in result
```

## Best Practices

1. **Patch where used** - Not where defined
2. **Use autospec** - Catch API mismatches
3. **Reset mocks** - In fixtures or with `mock.reset_mock()`
4. **Don't over-mock** - Test behavior, not implementation
5. **Prefer dependency injection** - Over patching
6. **Use pytest-mock** - Cleaner syntax than unittest.mock

# Property-Based Testing with Hypothesis

Test properties of code with generated inputs, not just examples.

## Why Property Testing?

```python
# Example-based: tests specific cases
def test_sort_examples():
    assert sort([3, 1, 2]) == [1, 2, 3]
    assert sort([]) == []
    assert sort([1]) == [1]

# Property-based: tests properties for ANY input
from hypothesis import given
from hypothesis import strategies as st

@given(st.lists(st.integers()))
def test_sort_properties(lst):
    result = sort(lst)
    # Property 1: Same length
    assert len(result) == len(lst)
    # Property 2: Sorted order
    assert all(result[i] <= result[i+1] for i in range(len(result)-1))
    # Property 3: Same elements
    assert sorted(lst) == result
```

## Basic Hypothesis Usage

```python
from hypothesis import given, settings, assume
from hypothesis import strategies as st

@given(st.integers(), st.integers())
def test_addition_commutative(a, b):
    """Addition is commutative."""
    assert a + b == b + a

@given(st.text())
def test_reverse_twice(s):
    """Reversing twice returns original."""
    assert s[::-1][::-1] == s

@given(st.lists(st.integers(), min_size=1))
def test_max_in_list(lst):
    """Max is an element of the list."""
    assert max(lst) in lst
```

## Common Strategies

```python
from hypothesis import strategies as st

# Primitives
st.integers()                    # Any integer
st.integers(min_value=0)         # Non-negative
st.floats(allow_nan=False)       # Floats without NaN
st.text()                        # Unicode strings
st.text(alphabet="abc", max_size=10)
st.booleans()
st.none()
st.binary()                      # Bytes

# Collections
st.lists(st.integers())          # List of ints
st.lists(st.text(), min_size=1, max_size=10)
st.sets(st.integers())
st.frozensets(st.text())
st.dictionaries(st.text(), st.integers())

# Tuples
st.tuples(st.integers(), st.text())   # Fixed structure
st.tuples(st.integers(), st.integers(), st.integers())

# Optional / One of
st.one_of(st.integers(), st.text())   # Either type
st.none() | st.integers()              # Optional int
st.sampled_from(["red", "green", "blue"])  # Enum-like
```

## Building Custom Strategies

```python
from hypothesis import strategies as st
from dataclasses import dataclass

@dataclass
class User:
    name: str
    age: int
    email: str

# Strategy for User objects
user_strategy = st.builds(
    User,
    name=st.text(min_size=1, max_size=50),
    age=st.integers(min_value=0, max_value=150),
    email=st.emails()
)

@given(user_strategy)
def test_user_validation(user):
    assert validate_user(user)


# Composite strategies for complex logic
@st.composite
def sorted_lists(draw):
    """Generate pre-sorted lists."""
    lst = draw(st.lists(st.integers()))
    return sorted(lst)

@given(sorted_lists())
def test_binary_search(sorted_lst):
    if sorted_lst:
        target = sorted_lst[len(sorted_lst) // 2]
        assert binary_search(sorted_lst, target) != -1


# Dependent strategies
@st.composite
def list_and_index(draw):
    """Generate a list and valid index into it."""
    lst = draw(st.lists(st.integers(), min_size=1))
    index = draw(st.integers(min_value=0, max_value=len(lst)-1))
    return lst, index

@given(list_and_index())
def test_indexing(data):
    lst, index = data
    # This will never raise IndexError
    assert lst[index] is not None or lst[index] is None
```

## Filtering and Assumptions

```python
from hypothesis import given, assume
from hypothesis import strategies as st

# Filter strategy (preferred when possible)
@given(st.integers().filter(lambda x: x % 2 == 0))
def test_even_numbers(n):
    assert n % 2 == 0

# assume() for runtime filtering
@given(st.integers(), st.integers())
def test_division(a, b):
    assume(b != 0)  # Skip if b is 0
    assert (a // b) * b + (a % b) == a

# Combining filters
positive_even = st.integers(min_value=1).filter(lambda x: x % 2 == 0)
```

## Settings and Configuration

```python
from hypothesis import given, settings, Verbosity, Phase
from hypothesis import strategies as st

# Per-test settings
@settings(max_examples=500)  # More examples (default: 100)
@given(st.integers())
def test_thorough(n):
    pass

@settings(deadline=None)  # Disable timing check
@given(st.lists(st.integers()))
def test_slow_operation(lst):
    expensive_operation(lst)

@settings(
    max_examples=1000,
    verbosity=Verbosity.verbose,
    phases=[Phase.generate],  # Skip shrinking
)
@given(st.text())
def test_verbose(s):
    pass


# Profile for CI (in conftest.py)
from hypothesis import settings, Verbosity

settings.register_profile("ci", max_examples=1000)
settings.register_profile("dev", max_examples=10)
settings.register_profile("debug", max_examples=10, verbosity=Verbosity.verbose)

# Use: pytest --hypothesis-profile=ci
```

## Stateful Testing

```python
from hypothesis.stateful import RuleBasedStateMachine, rule, invariant
from hypothesis import strategies as st

class DatabaseMachine(RuleBasedStateMachine):
    """Test database operations maintain invariants."""

    def __init__(self):
        super().__init__()
        self.db = {}  # Model
        self.real_db = RealDatabase()  # System under test

    @rule(key=st.text(), value=st.integers())
    def set_value(self, key, value):
        """Set a value in both model and real DB."""
        self.db[key] = value
        self.real_db.set(key, value)

    @rule(key=st.text())
    def get_value(self, key):
        """Get value should match model."""
        expected = self.db.get(key)
        actual = self.real_db.get(key)
        assert expected == actual

    @rule(key=st.text())
    def delete_value(self, key):
        """Delete from both."""
        self.db.pop(key, None)
        self.real_db.delete(key)

    @invariant()
    def keys_match(self):
        """Keys should always match."""
        assert set(self.db.keys()) == set(self.real_db.keys())


# Run stateful tests
TestDatabase = DatabaseMachine.TestCase
```

## pytest Integration

```python
# conftest.py
from hypothesis import settings, Verbosity, Phase

# Default profile for all tests
settings.register_profile("default", max_examples=100)

# CI profile - more examples, deterministic
settings.register_profile(
    "ci",
    max_examples=500,
    derandomize=True,  # Deterministic for CI
)

# Load profile from env or default
import os
settings.load_profile(os.getenv("HYPOTHESIS_PROFILE", "default"))


# pytest.ini
# [pytest]
# addopts = --hypothesis-profile=default
```

## Shrinking Examples

```python
from hypothesis import given, settings
from hypothesis import strategies as st

@given(st.lists(st.integers()))
def test_shrinking_demo(lst):
    """Hypothesis shrinks failing inputs to minimal examples."""
    # This will fail, but Hypothesis finds minimal case
    assert sum(lst) < 100

# Hypothesis will shrink to something like:
# Falsifying example: test_shrinking_demo(lst=[100])
# Not: test_shrinking_demo(lst=[3847, -293, 10293, ...])
```

## Common Patterns

```python
# Roundtrip / Encode-Decode
@given(st.binary())
def test_compression_roundtrip(data):
    assert decompress(compress(data)) == data

@given(st.dictionaries(st.text(), st.integers()))
def test_json_roundtrip(d):
    assert json.loads(json.dumps(d)) == d


# Oracle testing (compare implementations)
@given(st.lists(st.integers()))
def test_sort_vs_stdlib(lst):
    assert my_sort(lst) == sorted(lst)


# Metamorphic relations
@given(st.lists(st.integers()))
def test_sort_idempotent(lst):
    """Sorting twice equals sorting once."""
    assert sort(sort(lst)) == sort(lst)

@given(st.lists(st.integers()), st.integers())
def test_sort_append(lst, x):
    """Appending and sorting vs inserting sorted."""
    assert sort(lst + [x]) == sort(sorted(lst) + [x])
```

## Quick Reference

| Strategy | Description |
|----------|-------------|
| `st.integers()` | Any integer |
| `st.floats()` | Floats (configure nan, inf) |
| `st.text()` | Unicode strings |
| `st.binary()` | Byte strings |
| `st.lists(st.X())` | Lists of X |
| `st.dictionaries(k, v)` | Dict with key/value strategies |
| `st.builds(Class, ...)` | Build objects |
| `st.one_of(a, b)` | Either a or b |
| `st.sampled_from([...])` | Pick from list |
| `@st.composite` | Custom strategy |

| Setting | Purpose |
|---------|---------|
| `max_examples=N` | Number of test cases |
| `deadline=None` | Disable timing |
| `derandomize=True` | Reproducible runs |
| `verbosity=Verbosity.verbose` | Debug output |

# Test Architecture Patterns

Organize tests for maintainability, speed, and confidence.

## Test Pyramid

```
                 ┌─────────────┐
                 │     E2E     │  Few, slow, high confidence
                 │   Browser   │
                 ├─────────────┤
                 │ Integration │  Moderate, real services
                 │   API/DB    │
                 ├─────────────┤
                 │    Unit     │  Many, fast, isolated
                 │  Functions  │
                 └─────────────┘
```

| Layer | Count | Speed | Scope | Tools |
|-------|-------|-------|-------|-------|
| Unit | 70% | <1ms | Single function | pytest, mock |
| Integration | 20% | <1s | Multiple components | testcontainers, FastAPI TestClient |
| E2E | 10% | <30s | Full system | Playwright, Selenium |

## Directory Structure

```
project/
├── src/
│   └── myapp/
│       ├── models/
│       ├── services/
│       └── api/
├── tests/
│   ├── conftest.py           # Shared fixtures
│   ├── unit/                  # Fast, isolated tests
│   │   ├── conftest.py
│   │   ├── test_models.py
│   │   └── test_services.py
│   ├── integration/           # Real services tests
│   │   ├── conftest.py        # DB, Redis fixtures
│   │   ├── test_api.py
│   │   └── test_repositories.py
│   ├── e2e/                   # End-to-end tests
│   │   ├── conftest.py
│   │   └── test_user_flows.py
│   └── fixtures/              # Shared test data
│       └── users.json
└── pytest.ini
```

## pytest Configuration

```ini
# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
python_classes = Test*

# Markers for test categories
markers =
    unit: Unit tests (fast, isolated)
    integration: Integration tests (requires services)
    e2e: End-to-end tests (full system)
    slow: Slow tests (>1s)

# Default options
addopts =
    -ra                 # Show summary of all except passed
    --strict-markers    # Error on unknown markers
    -q                  # Quiet mode
```

## Test Isolation Strategies

### 1. Database Isolation with Transactions

```python
@pytest.fixture
def db_session(engine):
    """Each test runs in a rolled-back transaction."""
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()
```

### 2. Schema Isolation (Parallel Safe)

```python
import uuid

@pytest.fixture(scope="session")
def test_schema(engine):
    """Create isolated schema for test session."""
    schema_name = f"test_{uuid.uuid4().hex[:8]}"

    with engine.connect() as conn:
        conn.execute(f"CREATE SCHEMA {schema_name}")
        conn.execute(f"SET search_path TO {schema_name}")

    yield schema_name

    with engine.connect() as conn:
        conn.execute(f"DROP SCHEMA {schema_name} CASCADE")
```

### 3. Container Isolation

```python
@pytest.fixture(scope="session")
def isolated_postgres():
    """Each test session gets fresh PostgreSQL."""
    with PostgresContainer("postgres:15") as pg:
        yield pg.get_connection_url()
```

## conftest.py Patterns

### Root conftest.py

```python
# tests/conftest.py
import pytest
from typing import Generator

# Session-scoped fixtures
@pytest.fixture(scope="session")
def app():
    """Create application once per session."""
    from myapp import create_app
    return create_app(testing=True)

@pytest.fixture(scope="session")
def engine(app):
    """Database engine for session."""
    return app.extensions["db"].engine

# Function-scoped (per-test)
@pytest.fixture
def client(app) -> Generator:
    """Test client per test."""
    with app.test_client() as client:
        yield client
```

### Unit Test conftest.py

```python
# tests/unit/conftest.py
import pytest
from unittest.mock import Mock

@pytest.fixture
def mock_db():
    """Mock database for unit tests."""
    return Mock()

@pytest.fixture
def mock_redis():
    """Mock Redis for unit tests."""
    return Mock()

@pytest.fixture(autouse=True)
def no_network(monkeypatch):
    """Prevent network calls in unit tests."""
    import socket
    monkeypatch.setattr(socket, "socket", Mock(side_effect=Exception("No network in unit tests!")))
```

### Integration Test conftest.py

```python
# tests/integration/conftest.py
import pytest

@pytest.fixture(scope="session")
def postgres_container():
    """PostgreSQL container for integration tests."""
    from testcontainers.postgres import PostgresContainer
    with PostgresContainer("postgres:15") as pg:
        yield pg

@pytest.fixture
def db_session(postgres_container):
    """Database session with rollback."""
    # Transaction rollback pattern
    ...
```

## Test Markers and Selection

```python
import pytest

# Mark tests by category
@pytest.mark.unit
def test_calculate_total():
    assert calculate_total([1, 2, 3]) == 6

@pytest.mark.integration
def test_save_to_database(db_session):
    user = User(name="Test")
    db_session.add(user)
    db_session.commit()
    assert user.id is not None

@pytest.mark.e2e
def test_user_signup_flow(browser):
    browser.goto("/signup")
    browser.fill("email", "test@example.com")
    browser.click("Submit")
    assert browser.url == "/dashboard"

@pytest.mark.slow
def test_data_migration():
    migrate_all_records()  # Takes 30 seconds
```

```bash
# Run specific categories
pytest -m unit            # Only unit tests
pytest -m integration     # Only integration tests
pytest -m "not slow"      # Exclude slow tests
pytest -m "unit or integration"  # Both
```

## Parallel Testing

```python
# pytest.ini
[pytest]
# Safe for parallel execution
addopts = -n auto  # Use pytest-xdist

# conftest.py - ensure isolation
@pytest.fixture(scope="session")
def worker_id(request):
    """Get unique worker ID for parallel runs."""
    if hasattr(request.config, "workerinput"):
        return request.config.workerinput["workerid"]
    return "master"

@pytest.fixture(scope="session")
def db_name(worker_id):
    """Unique database per worker."""
    return f"testdb_{worker_id}"
```

## Test Naming Conventions

```python
# Pattern: test_<unit>_<condition>_<expected>

def test_user_creation_with_valid_data_succeeds():
    pass

def test_user_creation_with_missing_email_raises_validation_error():
    pass

def test_calculate_total_with_empty_list_returns_zero():
    pass

def test_api_users_get_without_auth_returns_401():
    pass


# Or use classes for grouping
class TestUserCreation:
    def test_with_valid_data_succeeds(self):
        pass

    def test_with_missing_email_raises_validation_error(self):
        pass

    def test_with_duplicate_email_raises_conflict_error(self):
        pass
```

## Fixture Organization

```python
# tests/fixtures/factories.py
import factory
from faker import Faker

fake = Faker()

class UserFactory(factory.Factory):
    class Meta:
        model = User

    name = factory.LazyAttribute(lambda _: fake.name())
    email = factory.LazyAttribute(lambda _: fake.email())

class OrderFactory(factory.Factory):
    class Meta:
        model = Order

    user = factory.SubFactory(UserFactory)
    total = factory.LazyAttribute(lambda _: fake.pydecimal(min_value=1, max_value=1000))


# tests/conftest.py
from tests.fixtures.factories import UserFactory, OrderFactory

@pytest.fixture
def user():
    return UserFactory()

@pytest.fixture
def order(user):
    return OrderFactory(user=user)
```

## Performance Testing

```python
# uv add --dev pytest-benchmark

def test_sort_performance(benchmark):
    """Benchmark sorting algorithm."""
    data = list(range(10000, 0, -1))
    result = benchmark(sort, data)
    assert result == sorted(data)


# uv add --dev pytest-timeout
@pytest.mark.timeout(5)  # Fail if takes >5 seconds
def test_with_timeout():
    slow_operation()


# Track memory
# uv add --dev pytest-memray
@pytest.mark.limit_memory("100 MB")
def test_memory_usage():
    large_operation()
```

## Quick Reference

| Pattern | When to Use |
|---------|-------------|
| Transaction rollback | Database tests, fast isolation |
| TestContainers | Real service behavior needed |
| Schema isolation | Parallel test execution |
| Factory fixtures | Complex test data |
| Markers | Categorize and filter tests |
| conftest layers | Scope fixtures appropriately |

| Command | Purpose |
|---------|---------|
| `pytest -m unit` | Run unit tests only |
| `pytest -n auto` | Parallel execution |
| `pytest --lf` | Last failed only |
| `pytest -x` | Stop on first failure |
| `pytest --cov=src` | Coverage report |

#!/bin/bash
# Generate a pytest conftest.py with optional async/db/api fixture blocks.
#
# Usage:   generate-conftest.sh [--async] [--db] [--api]
# Input:   optional flags selecting fixture blocks; writes ./tests/conftest.py
# Output:  the generated conftest.py file plus a one-line summary on stdout
# Stderr:  none (the overwrite confirmation prompt is read from the tty)
# Exit:    0 ok (also --help, and when an existing file is kept at the prompt)
#
# Examples:
#   generate-conftest.sh
#   generate-conftest.sh --async --db
#   generate-conftest.sh --api

set -e

# --help / -h: print usage to stdout and exit 0 before any side effects (so the
# prompt never fires and no file is touched). Behavior-preserving for all other
# invocations — only -h/--help short-circuits here.
for _arg in "$@"; do
  case "$_arg" in
    -h|--help)
      cat <<'EOF'
Usage: generate-conftest.sh [--async] [--db] [--api]

Generate ./tests/conftest.py with optional fixture blocks. If a conftest.py
already exists, it prompts for confirmation (default: keep the existing file).

Options:
  --async   include async fixtures (event_loop, async_client)
  --db      include database fixtures (db_engine, db_session)
  --api     include API test fixtures (app, client, authenticated_client)

Examples:
  generate-conftest.sh
  generate-conftest.sh --async --db
  generate-conftest.sh --api
EOF
      exit 0
      ;;
  esac
done

OUTPUT="tests/conftest.py"

# Check if tests directory exists
if [[ ! -d "tests" ]]; then
    echo "Creating tests directory..."
    mkdir -p tests
fi

# Check if conftest.py already exists
if [[ -f "$OUTPUT" ]]; then
    read -p "conftest.py already exists. Overwrite? [y/N] " -n 1 -r
    echo
    if [[ ! $REPLY =~ ^[Yy]$ ]]; then
        exit 0
    fi
fi

# Parse arguments
ASYNC=""
DB=""
API=""

while [[ $# -gt 0 ]]; do
    case $1 in
        --async)
            ASYNC=1
            shift
            ;;
        --db)
            DB=1
            shift
            ;;
        --api)
            API=1
            shift
            ;;
        *)
            shift
            ;;
    esac
done

# Generate conftest.py
cat > "$OUTPUT" << 'HEADER'
"""
Pytest configuration and fixtures.
Generated by generate-conftest.sh
"""
import pytest
HEADER

# Add async imports if needed
if [[ -n "$ASYNC" ]]; then
    cat >> "$OUTPUT" << 'ASYNC_IMPORTS'
import asyncio
ASYNC_IMPORTS
fi

# Add database imports if needed
if [[ -n "$DB" ]]; then
    cat >> "$OUTPUT" << 'DB_IMPORTS'
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DB_IMPORTS
fi

# Add API imports if needed
if [[ -n "$API" ]]; then
    cat >> "$OUTPUT" << 'API_IMPORTS'
from fastapi.testclient import TestClient
# or: from flask.testing import FlaskClient
API_IMPORTS
fi

# Add base fixtures
cat >> "$OUTPUT" << 'BASE_FIXTURES'


# ============================================================
# Test Configuration
# ============================================================

def pytest_configure(config):
    """Register custom markers."""
    config.addinivalue_line("markers", "slow: marks tests as slow")
    config.addinivalue_line("markers", "integration: marks integration tests")
    config.addinivalue_line("markers", "e2e: marks end-to-end tests")


def pytest_collection_modifyitems(config, items):
    """Skip slow tests unless --slow flag is provided."""
    if not config.getoption("--slow", default=False):
        skip_slow = pytest.mark.skip(reason="use --slow to run")
        for item in items:
            if "slow" in item.keywords:
                item.add_marker(skip_slow)


def pytest_addoption(parser):
    """Add custom CLI options."""
    parser.addoption(
        "--slow",
        action="store_true",
        default=False,
        help="run slow tests"
    )


# ============================================================
# Common Fixtures
# ============================================================

@pytest.fixture
def sample_data():
    """Provide sample test data."""
    return {
        "id": 1,
        "name": "Test",
        "active": True,
    }


@pytest.fixture
def temp_file(tmp_path):
    """Create a temporary file for testing."""
    file_path = tmp_path / "test_file.txt"
    file_path.write_text("test content")
    return file_path
BASE_FIXTURES

# Add async fixtures if requested
if [[ -n "$ASYNC" ]]; then
    cat >> "$OUTPUT" << 'ASYNC_FIXTURES'


# ============================================================
# Async Fixtures
# ============================================================

@pytest.fixture(scope="session")
def event_loop():
    """Create event loop for async tests."""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest.fixture
async def async_client():
    """Async HTTP client fixture."""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        yield session
ASYNC_FIXTURES
fi

# Add database fixtures if requested
if [[ -n "$DB" ]]; then
    cat >> "$OUTPUT" << 'DB_FIXTURES'


# ============================================================
# Database Fixtures
# ============================================================

@pytest.fixture(scope="session")
def db_engine():
    """Create test database engine."""
    engine = create_engine("sqlite:///:memory:")
    # Create tables here
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(db_engine):
    """Create database session with transaction rollback."""
    Session = sessionmaker(bind=db_engine)
    session = Session()
    yield session
    session.rollback()
    session.close()
DB_FIXTURES
fi

# Add API fixtures if requested
if [[ -n "$API" ]]; then
    cat >> "$OUTPUT" << 'API_FIXTURES'


# ============================================================
# API Fixtures
# ============================================================

@pytest.fixture
def app():
    """Create test application."""
    from myapp import create_app
    app = create_app(testing=True)
    return app


@pytest.fixture
def client(app):
    """Create test client."""
    return TestClient(app)


@pytest.fixture
def authenticated_client(client):
    """Create authenticated test client."""
    # Add authentication logic here
    client.headers["Authorization"] = "Bearer test-token"
    return client
API_FIXTURES
fi

echo "Generated $OUTPUT"
echo ""
echo "Options used:"
[[ -n "$ASYNC" ]] && echo "  --async: Async fixtures included"
[[ -n "$DB" ]] && echo "  --db: Database fixtures included"
[[ -n "$API" ]] && echo "  --api: API fixtures included"

exit 0

#!/bin/bash
# Run pytest with recommended options
# Usage: ./run-tests.sh [options]
#
# Options:
#   --quick     Skip slow tests, minimal output
#   --coverage  Run with coverage report
#   --watch     Watch mode with pytest-watch
#   --failed    Re-run only failed tests
#   --debug     Enable debug output

set -e

# Colors
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

# Default options
PYTEST_ARGS="-v"
COVERAGE=""
WATCH=""

# Parse arguments
while [[ $# -gt 0 ]]; do
    case $1 in
        --quick)
            PYTEST_ARGS="-q -x --tb=short"
            shift
            ;;
        --coverage)
            COVERAGE="--cov=src --cov-report=term-missing --cov-report=html"
            shift
            ;;
        --watch)
            WATCH=1
            shift
            ;;
        --failed)
            PYTEST_ARGS="$PYTEST_ARGS --lf"
            shift
            ;;
        --debug)
            PYTEST_ARGS="$PYTEST_ARGS -s --tb=long"
            shift
            ;;
        *)
            PYTEST_ARGS="$PYTEST_ARGS $1"
            shift
            ;;
    esac
done

# Check if pytest is installed
if ! command -v pytest &> /dev/null; then
    echo -e "${RED}pytest not found. Install with: uv add --dev pytest${NC}"
    exit 1
fi

# Watch mode
if [[ -n "$WATCH" ]]; then
    if ! command -v ptw &> /dev/null; then
        echo -e "${YELLOW}pytest-watch not found. Install with: uv add --dev pytest-watch${NC}"
        exit 1
    fi
    echo -e "${GREEN}Starting watch mode...${NC}"
    ptw -- $PYTEST_ARGS $COVERAGE
    exit 0
fi

# Run tests
echo -e "${GREEN}Running pytest...${NC}"
echo "pytest $PYTEST_ARGS $COVERAGE"
echo ""

pytest $PYTEST_ARGS $COVERAGE

# Open coverage report if generated
if [[ -n "$COVERAGE" ]] && [[ -f "htmlcov/index.html" ]]; then
    echo ""
    echo -e "${GREEN}Coverage report: htmlcov/index.html${NC}"
    if command -v open &> /dev/null; then
        read -p "Open coverage report? [y/N] " -n 1 -r
        echo
        if [[ $REPLY =~ ^[Yy]$ ]]; then
            open htmlcov/index.html
        fi
    fi
fi

#!/usr/bin/env bash
# Self-test for python-pytest-ops — fully offline, deterministic, Linux-safe.
#
# generate-conftest.sh writes ./tests/conftest.py relative to the cwd and, when
# one already exists, prompts interactively (defaulting to KEEP the file). Every
# run executes inside a mktemp -d sandbox so nothing is ever written into the
# repo or a real project. Asserts the --help contract, the generated conftest's
# fixture blocks (base + --async/--db/--api), and the overwrite-safety guarantee:
# an existing user conftest.py is NOT clobbered unless the user answers the
# prompt with "y" (the documented mechanism — fed "n" here keeps the file).
#
# Usage:   bash tests/run.sh
# Exit:    0 all pass, 1 one or more failures

set -uo pipefail

HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKILL="$(dirname "$HERE")"
V="$SKILL/scripts/generate-conftest.sh"

SB="$(mktemp -d)"; trap 'rm -rf "$SB"' EXIT
PASS=0; FAIL=0
ok() { PASS=$((PASS+1)); printf '  PASS  %s\n' "$1"; }
no() { FAIL=$((FAIL+1)); printf '  FAIL  %s\n' "$1"; }
expect_exit() { [[ "$2" == "$3" ]] && ok "$1 (exit $3)" || no "$1 (want $2 got $3)"; }
expect_has()  { case "$3" in *"$2"*) ok "$1";; *) no "$1 (missing '$2')";; esac; }

echo "=== python-pytest-ops self-test ==="

# ── contract ──────────────────────────────────────────────────────────────────
echo "-- contract --"
bash -n "$V" 2>/dev/null && ok "bash -n generate-conftest.sh" || no "bash -n generate-conftest.sh"
bash "$V" --help >/dev/null 2>&1; expect_exit "--help exits 0" 0 $?
bash "$V" -h     >/dev/null 2>&1; expect_exit "-h exits 0" 0 $?
out="$(bash "$V" --help 2>/dev/null)"
expect_has "--help has Examples" "xamples" "$out"
expect_has "--help lists --async" "--async" "$out"
expect_has "--help lists --db"    "--db" "$out"
expect_has "--help lists --api"   "--api" "$out"
# --help must short-circuit before any side effect (no file created even though
# the cwd has no tests/ dir)
mkdir -p "$SB/help-cwd"
( cd "$SB/help-cwd" && bash "$V" --help >/dev/null 2>&1 )
[[ ! -e "$SB/help-cwd/tests" ]] && ok "--help creates no files" || no "--help created files"

# ── fresh generation ──────────────────────────────────────────────────────────
echo "-- fresh generation --"
mkdir -p "$SB/fresh"
( cd "$SB/fresh" && bash "$V" </dev/null >"$SB/fresh.out" 2>"$SB/fresh.err" ); expect_exit "fresh generate -> 0" 0 $?
[[ -f "$SB/fresh/tests/conftest.py" ]] && ok "creates tests/conftest.py" || no "conftest.py not created"
out="$(cat "$SB/fresh/tests/conftest.py")"
expect_has "has module docstring"      "Pytest configuration and fixtures" "$out"
expect_has "imports pytest"            "import pytest" "$out"
expect_has "registers slow marker"     '"slow: marks tests as slow"' "$out"
expect_has "has sample_data fixture"   "def sample_data():" "$out"
expect_has "has temp_file fixture"     "def temp_file(tmp_path):" "$out"
expect_has "has pytest_addoption"      "def pytest_addoption(parser):" "$out"
expect_has "summary on stdout"         "Generated tests/conftest.py" "$(cat "$SB/fresh.out")"

# ── fixture-block flags ───────────────────────────────────────────────────────
echo "-- fixture blocks --"
mkdir -p "$SB/all"
( cd "$SB/all" && bash "$V" --async --db --api </dev/null >/dev/null 2>&1 ); expect_exit "all flags -> 0" 0 $?
out="$(cat "$SB/all/tests/conftest.py")"
expect_has "--async adds event_loop"    "def event_loop():" "$out"
expect_has "--async adds async_client"  "async def async_client():" "$out"
expect_has "--db adds db_engine"        "def db_engine():" "$out"
expect_has "--db adds db_session"       "def db_session(db_engine):" "$out"
expect_has "--api adds client fixture"  "def client(app):" "$out"
expect_has "--api adds authed client"   "def authenticated_client(client):" "$out"

# base generation must NOT include the optional blocks
! grep -q "def event_loop" "$SB/fresh/tests/conftest.py" && ok "base omits async blocks" || no "base included async blocks"
! grep -q "def db_engine"  "$SB/fresh/tests/conftest.py" && ok "base omits db blocks"    || no "base included db blocks"
! grep -q "def client"     "$SB/fresh/tests/conftest.py" && ok "base omits api blocks"   || no "base included api blocks"

# ── overwrite safety ─────────────────────────────────────────────────────────
echo "-- overwrite safety --"
# Decline ("n", the documented default) MUST preserve an existing user conftest.
mkdir -p "$SB/ow/tests"
printf '# USER-SENTINEL: do not overwrite me\nimport user_thing\n' >"$SB/ow/tests/conftest.py"
( cd "$SB/ow" && printf 'n\n' | bash "$V" >/dev/null 2>&1 ); expect_exit "decline overwrite -> 0" 0 $?
out="$(cat "$SB/ow/tests/conftest.py")"
expect_has "user conftest preserved (decline)" "USER-SENTINEL" "$out"
expect_has "user import preserved (decline)"   "import user_thing" "$out"

# Confirming ("y") DOES replace the file — that is the documented overwrite path.
mkdir -p "$SB/ow2/tests"
printf '# USER-SENTINEL-2\n' >"$SB/ow2/tests/conftest.py"
( cd "$SB/ow2" && printf 'y\n' | bash "$V" >/dev/null 2>&1 ); expect_exit "confirm overwrite -> 0" 0 $?
out="$(cat "$SB/ow2/tests/conftest.py")"
! grep -q "USER-SENTINEL-2" "$SB/ow2/tests/conftest.py" && ok "confirm 'y' replaces user conftest" || no "confirm 'y' did not replace"
expect_has "new conftest has pytest import" "import pytest" "$out"

# Non-interactive (closed stdin, no answer) also keeps the existing file — the
# safe default when there is no TTY to answer the prompt.
mkdir -p "$SB/ow3/tests"
printf '# USER-SENTINEL-3\n' >"$SB/ow3/tests/conftest.py"
( cd "$SB/ow3" && bash "$V" </dev/null >/dev/null 2>&1 )
expect_has "closed-stdin keeps user conftest" "USER-SENTINEL-3" "$(cat "$SB/ow3/tests/conftest.py")"

echo ""
echo "=== $PASS passed, $FAIL failed ==="
[[ "$FAIL" -eq 0 ]] || exit 1
exit 0