diff --git a/.agents/skills/python-error-handling/SKILL.md b/.agents/skills/python-error-handling/SKILL.md new file mode 100644 index 000000000..41214dab8 --- /dev/null +++ b/.agents/skills/python-error-handling/SKILL.md @@ -0,0 +1,193 @@ +--- +name: python-error-handling +description: Python error handling patterns including input validation, exception hierarchies, and partial failure handling. Use when implementing validation logic, designing exception strategies, handling batch processing failures, or building robust APIs. +--- + +# Python Error Handling + +Build robust Python applications with proper input validation, meaningful exceptions, and graceful failure handling. Good error handling makes debugging easier and systems more reliable. + +## When to Use This Skill + +- Validating user input and API parameters +- Designing exception hierarchies for applications +- Handling partial failures in batch operations +- Converting external data to domain types +- Building user-friendly error messages +- Implementing fail-fast validation patterns + +## Core Concepts + +### 1. Fail Fast + +Validate inputs early, before expensive operations. Report all validation errors at once when possible. + +### 2. Meaningful Exceptions + +Use appropriate exception types with context. Messages should explain what failed, why, and how to fix it. + +### 3. Partial Failures + +In batch operations, don't let one failure abort everything. Track successes and failures separately. + +### 4. Preserve Context + +Chain exceptions to maintain the full error trail for debugging. + +## Quick Start + +```python +def fetch_page(url: str, page_size: int) -> Page: + if not url: + raise ValueError("'url' is required") + if not 1 <= page_size <= 100: + raise ValueError(f"'page_size' must be 1-100, got {page_size}") + # Now safe to proceed... +``` + +## Fundamental Patterns + +### Pattern 1: Early Input Validation + +Validate all inputs at API boundaries before any processing begins. + +```python +def process_order( + order_id: str, + quantity: int, + discount_percent: float, +) -> OrderResult: + """Process an order with validation.""" + # Validate required fields + if not order_id: + raise ValueError("'order_id' is required") + + # Validate ranges + if quantity <= 0: + raise ValueError(f"'quantity' must be positive, got {quantity}") + + if not 0 <= discount_percent <= 100: + raise ValueError( + f"'discount_percent' must be 0-100, got {discount_percent}" + ) + + # Validation passed, proceed with processing + return _process_validated_order(order_id, quantity, discount_percent) +``` + +### Pattern 2: Convert to Domain Types Early + +Parse strings and external data into typed domain objects at system boundaries. + +```python +from enum import Enum + +class OutputFormat(Enum): + JSON = "json" + CSV = "csv" + PARQUET = "parquet" + +def parse_output_format(value: str) -> OutputFormat: + """Parse string to OutputFormat enum. + + Args: + value: Format string from user input. + + Returns: + Validated OutputFormat enum member. + + Raises: + ValueError: If format is not recognized. + """ + try: + return OutputFormat(value.lower()) + except ValueError: + valid_formats = [f.value for f in OutputFormat] + raise ValueError( + f"Invalid format '{value}'. " + f"Valid options: {', '.join(valid_formats)}" + ) + +# Usage at API boundary +def export_data(data: list[dict], format_str: str) -> bytes: + output_format = parse_output_format(format_str) # Fail fast + # Rest of function uses typed OutputFormat + ... +``` + +### Pattern 3: Pydantic for Complex Validation + +Use Pydantic models for structured input validation with automatic error messages. + +```python +from pydantic import BaseModel, Field, field_validator + +class CreateUserInput(BaseModel): + """Input model for user creation.""" + + email: str = Field(..., min_length=5, max_length=255) + name: str = Field(..., min_length=1, max_length=100) + age: int = Field(ge=0, le=150) + + @field_validator("email") + @classmethod + def validate_email_format(cls, v: str) -> str: + if "@" not in v or "." not in v.split("@")[-1]: + raise ValueError("Invalid email format") + return v.lower() + + @field_validator("name") + @classmethod + def normalize_name(cls, v: str) -> str: + return v.strip().title() + +# Usage +try: + user_input = CreateUserInput( + email="user@example.com", + name="john doe", + age=25, + ) +except ValidationError as e: + # Pydantic provides detailed error information + print(e.errors()) +``` + +### Pattern 4: Map Errors to Standard Exceptions + +Use Python's built-in exception types appropriately, adding context as needed. + +| Failure Type | Exception | Example | +|--------------|-----------|---------| +| Invalid input | `ValueError` | Bad parameter values | +| Wrong type | `TypeError` | Expected string, got int | +| Missing item | `KeyError` | Dict key not found | +| Operational failure | `RuntimeError` | Service unavailable | +| Timeout | `TimeoutError` | Operation took too long | +| File not found | `FileNotFoundError` | Path doesn't exist | +| Permission denied | `PermissionError` | Access forbidden | + +```python +# Good: Specific exception with context +raise ValueError(f"'page_size' must be 1-100, got {page_size}") + +# Avoid: Generic exception, no context +raise Exception("Invalid parameter") +``` + +## Detailed worked examples and patterns + +Detailed sections (starting with `## Advanced Patterns`) live in `references/details.md`. Read that file when the navigation summary above is insufficient. + +## Best Practices Summary + +1. **Validate early** - Check inputs before expensive operations +2. **Use specific exceptions** - `ValueError`, `TypeError`, not generic `Exception` +3. **Include context** - Messages should explain what, why, and how to fix +4. **Convert types at boundaries** - Parse strings to enums/domain types early +5. **Chain exceptions** - Use `raise ... from e` to preserve debug info +6. **Handle partial failures** - Don't abort batches on single item errors +7. **Use Pydantic** - For complex input validation with structured errors +8. **Document failure modes** - Docstrings should list possible exceptions +9. **Log with context** - Include IDs, counts, and other debugging info +10. **Test error paths** - Verify exceptions are raised correctly diff --git a/.agents/skills/python-error-handling/references/details.md b/.agents/skills/python-error-handling/references/details.md new file mode 100644 index 000000000..64f6611d9 --- /dev/null +++ b/.agents/skills/python-error-handling/references/details.md @@ -0,0 +1,171 @@ +# python-error-handling — detailed worked examples + +## Advanced Patterns + +### Pattern 5: Custom Exceptions with Context + +Create domain-specific exceptions that carry structured information. + +```python +class ApiError(Exception): + """Base exception for API errors.""" + + def __init__( + self, + message: str, + status_code: int, + response_body: str | None = None, + ) -> None: + self.status_code = status_code + self.response_body = response_body + super().__init__(message) + +class RateLimitError(ApiError): + """Raised when rate limit is exceeded.""" + + def __init__(self, retry_after: int) -> None: + self.retry_after = retry_after + super().__init__( + f"Rate limit exceeded. Retry after {retry_after}s", + status_code=429, + ) + +# Usage +def handle_response(response: Response) -> dict: + match response.status_code: + case 200: + return response.json() + case 401: + raise ApiError("Invalid credentials", 401) + case 404: + raise ApiError(f"Resource not found: {response.url}", 404) + case 429: + retry_after = int(response.headers.get("Retry-After", 60)) + raise RateLimitError(retry_after) + case code if 400 <= code < 500: + raise ApiError(f"Client error: {response.text}", code) + case code if code >= 500: + raise ApiError(f"Server error: {response.text}", code) +``` + +### Pattern 6: Exception Chaining + +Preserve the original exception when re-raising to maintain the debug trail. + +```python +import httpx + +class ServiceError(Exception): + """High-level service operation failed.""" + pass + +def upload_file(path: str) -> str: + """Upload file and return URL.""" + try: + with open(path, "rb") as f: + response = httpx.post("https://upload.example.com", files={"file": f}) + response.raise_for_status() + return response.json()["url"] + except FileNotFoundError as e: + raise ServiceError(f"Upload failed: file not found at '{path}'") from e + except httpx.HTTPStatusError as e: + raise ServiceError( + f"Upload failed: server returned {e.response.status_code}" + ) from e + except httpx.RequestError as e: + raise ServiceError(f"Upload failed: network error") from e +``` + +### Pattern 7: Batch Processing with Partial Failures + +Never let one bad item abort an entire batch. Track results per item. + +```python +from dataclasses import dataclass + +@dataclass +class BatchResult[T]: + """Results from batch processing.""" + + succeeded: dict[int, T] # index -> result + failed: dict[int, Exception] # index -> error + + @property + def success_count(self) -> int: + return len(self.succeeded) + + @property + def failure_count(self) -> int: + return len(self.failed) + + @property + def all_succeeded(self) -> bool: + return len(self.failed) == 0 + +def process_batch(items: list[Item]) -> BatchResult[ProcessedItem]: + """Process items, capturing individual failures. + + Args: + items: Items to process. + + Returns: + BatchResult with succeeded and failed items by index. + """ + succeeded: dict[int, ProcessedItem] = {} + failed: dict[int, Exception] = {} + + for idx, item in enumerate(items): + try: + result = process_single_item(item) + succeeded[idx] = result + except Exception as e: + failed[idx] = e + + return BatchResult(succeeded=succeeded, failed=failed) + +# Caller handles partial results +result = process_batch(items) +if not result.all_succeeded: + logger.warning( + f"Batch completed with {result.failure_count} failures", + failed_indices=list(result.failed.keys()), + ) +``` + +### Pattern 8: Progress Reporting for Long Operations + +Provide visibility into batch progress without coupling business logic to UI. + +```python +from collections.abc import Callable + +ProgressCallback = Callable[[int, int, str], None] # current, total, status + +def process_large_batch( + items: list[Item], + on_progress: ProgressCallback | None = None, +) -> BatchResult: + """Process batch with optional progress reporting. + + Args: + items: Items to process. + on_progress: Optional callback receiving (current, total, status). + """ + total = len(items) + succeeded = {} + failed = {} + + for idx, item in enumerate(items): + if on_progress: + on_progress(idx, total, f"Processing {item.id}") + + try: + succeeded[idx] = process_single_item(item) + except Exception as e: + failed[idx] = e + + if on_progress: + on_progress(total, total, "Complete") + + return BatchResult(succeeded=succeeded, failed=failed) +``` diff --git a/.agents/skills/python-pro/SKILL.md b/.agents/skills/python-pro/SKILL.md new file mode 100644 index 000000000..fa5a326db --- /dev/null +++ b/.agents/skills/python-pro/SKILL.md @@ -0,0 +1,161 @@ +--- +name: python-pro +description: Master Python 3.12+ with modern features, async programming, performance optimization, and production-ready practices. Expert in the latest Python ecosystem including uv, ruff, pydantic, and FastAPI. +risk: critical +source: community +date_added: '2026-02-27' +--- +You are a Python expert specializing in modern Python 3.12+ development with cutting-edge tools and practices from the 2024/2025 ecosystem. + +## Use this skill when + +- Writing or reviewing Python 3.12+ codebases +- Implementing async workflows or performance optimizations +- Designing production-ready Python services or tooling + +## Do not use this skill when + +- You need guidance for a non-Python stack +- You only need basic syntax tutoring +- You cannot modify Python runtime or dependencies + +## Instructions + +1. Confirm runtime, dependencies, and performance targets. +2. Choose patterns (async, typing, tooling) that match requirements. +3. Implement and test with modern tooling. +4. Profile and tune for latency, memory, and correctness. + +## Purpose +Expert Python developer mastering Python 3.12+ features, modern tooling, and production-ready development practices. Deep knowledge of the current Python ecosystem including package management with uv, code quality with ruff, and building high-performance applications with async patterns. + +## Capabilities + +### Modern Python Features +- Python 3.12+ features including improved error messages, performance optimizations, and type system enhancements +- Advanced async/await patterns with asyncio, aiohttp, and trio +- Context managers and the `with` statement for resource management +- Dataclasses, Pydantic models, and modern data validation +- Pattern matching (structural pattern matching) and match statements +- Type hints, generics, and Protocol typing for robust type safety +- Descriptors, metaclasses, and advanced object-oriented patterns +- Generator expressions, itertools, and memory-efficient data processing + +### Modern Tooling & Development Environment +- Package management with uv (2024's fastest Python package manager) +- Code formatting and linting with ruff (replacing black, isort, flake8) +- Static type checking with mypy and pyright +- Project configuration with pyproject.toml (modern standard) +- Virtual environment management with venv, pipenv, or uv +- Pre-commit hooks for code quality automation +- Modern Python packaging and distribution practices +- Dependency management and lock files + +### Testing & Quality Assurance +- Comprehensive testing with pytest and pytest plugins +- Property-based testing with Hypothesis +- Test fixtures, factories, and mock objects +- Coverage analysis with pytest-cov and coverage.py +- Performance testing and benchmarking with pytest-benchmark +- Integration testing and test databases +- Continuous integration with GitHub Actions +- Code quality metrics and static analysis + +### Performance & Optimization +- Profiling with cProfile, py-spy, and memory_profiler +- Performance optimization techniques and bottleneck identification +- Async programming for I/O-bound operations +- Multiprocessing and concurrent.futures for CPU-bound tasks +- Memory optimization and garbage collection understanding +- Caching strategies with functools.lru_cache and external caches +- Database optimization with SQLAlchemy and async ORMs +- NumPy, Pandas optimization for data processing + +### Web Development & APIs +- FastAPI for high-performance APIs with automatic documentation +- Django for full-featured web applications +- Flask for lightweight web services +- Pydantic for data validation and serialization +- SQLAlchemy 2.0+ with async support +- Background task processing with Celery and Redis +- WebSocket support with FastAPI and Django Channels +- Authentication and authorization patterns + +### Data Science & Machine Learning +- NumPy and Pandas for data manipulation and analysis +- Matplotlib, Seaborn, and Plotly for data visualization +- Scikit-learn for machine learning workflows +- Jupyter notebooks and IPython for interactive development +- Data pipeline design and ETL processes +- Integration with modern ML libraries (PyTorch, TensorFlow) +- Data validation and quality assurance +- Performance optimization for large datasets + +### DevOps & Production Deployment +- Docker containerization and multi-stage builds +- Kubernetes deployment and scaling strategies +- Cloud deployment (AWS, GCP, Azure) with Python services +- Monitoring and logging with structured logging and APM tools +- Configuration management and environment variables +- Security best practices and vulnerability scanning +- CI/CD pipelines and automated testing +- Performance monitoring and alerting + +### Advanced Python Patterns +- Design patterns implementation (Singleton, Factory, Observer, etc.) +- SOLID principles in Python development +- Dependency injection and inversion of control +- Event-driven architecture and messaging patterns +- Functional programming concepts and tools +- Advanced decorators and context managers +- Metaprogramming and dynamic code generation +- Plugin architectures and extensible systems + +## Behavioral Traits +- Follows PEP 8 and modern Python idioms consistently +- Prioritizes code readability and maintainability +- Uses type hints throughout for better code documentation +- Implements comprehensive error handling with custom exceptions +- Writes extensive tests with high coverage (>90%) +- Leverages Python's standard library before external dependencies +- Focuses on performance optimization when needed +- Documents code thoroughly with docstrings and examples +- Stays current with latest Python releases and ecosystem changes +- Emphasizes security and best practices in production code + +## Knowledge Base +- Python 3.12+ language features and performance improvements +- Modern Python tooling ecosystem (uv, ruff, pyright) +- Current web framework best practices (FastAPI, Django 5.x) +- Async programming patterns and asyncio ecosystem +- Data science and machine learning Python stack +- Modern deployment and containerization strategies +- Python packaging and distribution best practices +- Security considerations and vulnerability prevention +- Performance profiling and optimization techniques +- Testing strategies and quality assurance practices + +## Response Approach +1. **Analyze requirements** for modern Python best practices +2. **Suggest current tools and patterns** from the 2024/2025 ecosystem +3. **Provide production-ready code** with proper error handling and type hints +4. **Include comprehensive tests** with pytest and appropriate fixtures +5. **Consider performance implications** and suggest optimizations +6. **Document security considerations** and best practices +7. **Recommend modern tooling** for development workflow +8. **Include deployment strategies** when applicable + +## Example Interactions +- "Help me migrate from pip to uv for package management" +- "Optimize this Python code for better async performance" +- "Design a FastAPI application with proper error handling and validation" +- "Set up a modern Python project with ruff, mypy, and pytest" +- "Implement a high-performance data processing pipeline" +- "Create a production-ready Dockerfile for a Python application" +- "Design a scalable background task system with Celery" +- "Implement modern authentication patterns in FastAPI" + +## Limitations +- Use this skill only when the task clearly matches the scope described above. +- Do not treat the output as a substitute for environment-specific validation, testing, or expert review. +- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing. diff --git a/.agents/skills/python-project-structure/SKILL.md b/.agents/skills/python-project-structure/SKILL.md new file mode 100644 index 000000000..504df638b --- /dev/null +++ b/.agents/skills/python-project-structure/SKILL.md @@ -0,0 +1,252 @@ +--- +name: python-project-structure +description: Python project organization, module architecture, and public API design. Use when setting up new projects, organizing modules, defining public interfaces with __all__, or planning directory layouts. +--- + +# Python Project Structure & Module Architecture + +Design well-organized Python projects with clear module boundaries, explicit public interfaces, and maintainable directory structures. Good organization makes code discoverable and changes predictable. + +## When to Use This Skill + +- Starting a new Python project from scratch +- Reorganizing an existing codebase for clarity +- Defining module public APIs with `__all__` +- Deciding between flat and nested directory structures +- Determining test file placement strategies +- Creating reusable library packages + +## Core Concepts + +### 1. Module Cohesion + +Group related code that changes together. A module should have a single, clear purpose. + +### 2. Explicit Interfaces + +Define what's public with `__all__`. Everything not listed is an internal implementation detail. + +### 3. Flat Hierarchies + +Prefer shallow directory structures. Add depth only for genuine sub-domains. + +### 4. Consistent Conventions + +Apply naming and organization patterns uniformly across the project. + +## Quick Start + +``` +myproject/ +├── src/ +│ └── myproject/ +│ ├── __init__.py +│ ├── services/ +│ ├── models/ +│ └── api/ +├── tests/ +├── pyproject.toml +└── README.md +``` + +## Fundamental Patterns + +### Pattern 1: One Concept Per File + +Each file should focus on a single concept or closely related set of functions. Consider splitting when a file: + +- Handles multiple unrelated responsibilities +- Grows beyond 300-500 lines (varies by complexity) +- Contains classes that change for different reasons + +```python +# Good: Focused files +# user_service.py - User business logic +# user_repository.py - User data access +# user_models.py - User data structures + +# Avoid: Kitchen sink files +# user.py - Contains service, repository, models, utilities... +``` + +### Pattern 2: Explicit Public APIs with `__all__` + +Define the public interface for every module. Unlisted members are internal implementation details. + +```python +# mypackage/services/__init__.py +from .user_service import UserService +from .order_service import OrderService +from .exceptions import ServiceError, ValidationError + +__all__ = [ + "UserService", + "OrderService", + "ServiceError", + "ValidationError", +] + +# Internal helpers remain private by omission +# from .internal_helpers import _validate_input # Not exported +``` + +### Pattern 3: Flat Directory Structure + +Prefer minimal nesting. Deep hierarchies make imports verbose and navigation difficult. + +``` +# Preferred: Flat structure +project/ +├── api/ +│ ├── routes.py +│ └── middleware.py +├── services/ +│ ├── user_service.py +│ └── order_service.py +├── models/ +│ ├── user.py +│ └── order.py +└── utils/ + └── validation.py + +# Avoid: Deep nesting +project/core/internal/services/impl/user/ +``` + +Add sub-packages only when there's a genuine sub-domain requiring isolation. + +### Pattern 4: Test File Organization + +Choose one approach and apply it consistently throughout the project. + +**Option A: Colocated Tests** + +``` +src/ +├── user_service.py +├── test_user_service.py +├── order_service.py +└── test_order_service.py +``` + +Benefits: Tests live next to the code they verify. Easy to see coverage gaps. + +**Option B: Parallel Test Directory** + +``` +src/ +├── services/ +│ ├── user_service.py +│ └── order_service.py +tests/ +├── services/ +│ ├── test_user_service.py +│ └── test_order_service.py +``` + +Benefits: Clean separation between production and test code. Standard for larger projects. + +## Advanced Patterns + +### Pattern 5: Package Initialization + +Use `__init__.py` to provide a clean public interface for package consumers. + +```python +# mypackage/__init__.py +"""MyPackage - A library for doing useful things.""" + +from .core import MainClass, HelperClass +from .exceptions import PackageError, ConfigError +from .config import Settings + +__all__ = [ + "MainClass", + "HelperClass", + "PackageError", + "ConfigError", + "Settings", +] + +__version__ = "1.0.0" +``` + +Consumers can then import directly from the package: + +```python +from mypackage import MainClass, Settings +``` + +### Pattern 6: Layered Architecture + +Organize code by architectural layer for clear separation of concerns. + +``` +myapp/ +├── api/ # HTTP handlers, request/response +│ ├── routes/ +│ └── middleware/ +├── services/ # Business logic +├── repositories/ # Data access +├── models/ # Domain entities +├── schemas/ # API schemas (Pydantic) +└── config/ # Configuration +``` + +Each layer should only depend on layers below it, never above. + +### Pattern 7: Domain-Driven Structure + +For complex applications, organize by business domain rather than technical layer. + +``` +ecommerce/ +├── users/ +│ ├── models.py +│ ├── services.py +│ ├── repository.py +│ └── api.py +├── orders/ +│ ├── models.py +│ ├── services.py +│ ├── repository.py +│ └── api.py +└── shared/ + ├── database.py + └── exceptions.py +``` + +## File and Module Naming + +### Conventions + +- Use `snake_case` for all file and module names: `user_repository.py` +- Avoid abbreviations that obscure meaning: `user_repository.py` not `usr_repo.py` +- Match class names to file names: `UserService` in `user_service.py` + +### Import Style + +Use absolute imports for clarity and reliability: + +```python +# Preferred: Absolute imports +from myproject.services import UserService +from myproject.models import User + +# Avoid: Relative imports +from ..services import UserService +from . import models +``` + +Relative imports can break when modules are moved or reorganized. + +## Best Practices Summary + +1. **Keep files focused** - One concept per file, consider splitting at 300-500 lines (varies by complexity) +2. **Define `__all__` explicitly** - Make public interfaces clear +3. **Prefer flat structures** - Add depth only for genuine sub-domains +4. **Use absolute imports** - More reliable and clearer +5. **Be consistent** - Apply patterns uniformly across the project +6. **Match names to content** - File names should describe their purpose +7. **Separate concerns** - Keep layers distinct and dependencies flowing one direction +8. **Document your structure** - Include a README explaining the organization diff --git a/.agents/skills/python-resource-management/SKILL.md b/.agents/skills/python-resource-management/SKILL.md new file mode 100644 index 000000000..553604fe4 --- /dev/null +++ b/.agents/skills/python-resource-management/SKILL.md @@ -0,0 +1,243 @@ +--- +name: python-resource-management +description: Python resource management with context managers, cleanup patterns, and streaming. Use when managing connections, file handles, implementing cleanup logic, or building streaming responses with accumulated state. +--- + +# Python Resource Management + +Manage resources deterministically using context managers. Resources like database connections, file handles, and network sockets should be released reliably, even when exceptions occur. + +## When to Use This Skill + +- Managing database connections and connection pools +- Working with file handles and I/O +- Implementing custom context managers +- Building streaming responses with state +- Handling nested resource cleanup +- Creating async context managers + +## Core Concepts + +### 1. Context Managers + +The `with` statement ensures resources are released automatically, even on exceptions. + +### 2. Protocol Methods + +`__enter__`/`__exit__` for sync, `__aenter__`/`__aexit__` for async resource management. + +### 3. Unconditional Cleanup + +`__exit__` always runs, regardless of whether an exception occurred. + +### 4. Exception Handling + +Return `True` from `__exit__` to suppress exceptions, `False` to propagate them. + +## Quick Start + +```python +from contextlib import contextmanager + +@contextmanager +def managed_resource(): + resource = acquire_resource() + try: + yield resource + finally: + resource.cleanup() + +with managed_resource() as r: + r.do_work() +``` + +## Fundamental Patterns + +### Pattern 1: Class-Based Context Manager + +Implement the context manager protocol for complex resources. + +```python +class DatabaseConnection: + """Database connection with automatic cleanup.""" + + def __init__(self, dsn: str) -> None: + self._dsn = dsn + self._conn: Connection | None = None + + def connect(self) -> None: + """Establish database connection.""" + self._conn = psycopg.connect(self._dsn) + + def close(self) -> None: + """Close connection if open.""" + if self._conn is not None: + self._conn.close() + self._conn = None + + def __enter__(self) -> "DatabaseConnection": + """Enter context: connect and return self.""" + self.connect() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + """Exit context: always close connection.""" + self.close() + +# Usage with context manager (preferred) +with DatabaseConnection(dsn) as db: + result = db.execute(query) + +# Manual management when needed +db = DatabaseConnection(dsn) +db.connect() +try: + result = db.execute(query) +finally: + db.close() +``` + +### Pattern 2: Async Context Manager + +For async resources, implement the async protocol. + +```python +class AsyncDatabasePool: + """Async database connection pool.""" + + def __init__(self, dsn: str, min_size: int = 1, max_size: int = 10) -> None: + self._dsn = dsn + self._min_size = min_size + self._max_size = max_size + self._pool: asyncpg.Pool | None = None + + async def __aenter__(self) -> "AsyncDatabasePool": + """Create connection pool.""" + self._pool = await asyncpg.create_pool( + self._dsn, + min_size=self._min_size, + max_size=self._max_size, + ) + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + """Close all connections in pool.""" + if self._pool is not None: + await self._pool.close() + + async def execute(self, query: str, *args) -> list[dict]: + """Execute query using pooled connection.""" + async with self._pool.acquire() as conn: + return await conn.fetch(query, *args) + +# Usage +async with AsyncDatabasePool(dsn) as pool: + users = await pool.execute("SELECT * FROM users WHERE active = $1", True) +``` + +### Pattern 3: Using @contextmanager Decorator + +Simplify context managers with the decorator for straightforward cases. + +```python +from contextlib import contextmanager, asynccontextmanager +import time +import structlog + +logger = structlog.get_logger() + +@contextmanager +def timed_block(name: str): + """Time a block of code.""" + start = time.perf_counter() + try: + yield + finally: + elapsed = time.perf_counter() - start + logger.info(f"{name} completed", duration_seconds=round(elapsed, 3)) + +# Usage +with timed_block("data_processing"): + process_large_dataset() + +@asynccontextmanager +async def database_transaction(conn: AsyncConnection): + """Manage database transaction.""" + await conn.execute("BEGIN") + try: + yield conn + await conn.execute("COMMIT") + except Exception: + await conn.execute("ROLLBACK") + raise + +# Usage +async with database_transaction(conn) as tx: + await tx.execute("INSERT INTO users ...") + await tx.execute("INSERT INTO audit_log ...") +``` + +### Pattern 4: Unconditional Resource Release + +Always clean up resources in `__exit__`, regardless of exceptions. + +```python +class FileProcessor: + """Process file with guaranteed cleanup.""" + + def __init__(self, path: str) -> None: + self._path = path + self._file: IO | None = None + self._temp_files: list[Path] = [] + + def __enter__(self) -> "FileProcessor": + self._file = open(self._path, "r") + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + """Clean up all resources unconditionally.""" + # Close main file + if self._file is not None: + self._file.close() + + # Clean up any temporary files + for temp_file in self._temp_files: + try: + temp_file.unlink() + except OSError: + pass # Best effort cleanup + + # Return None/False to propagate any exception +``` + +## Detailed worked examples and patterns + +Detailed sections (starting with `## Advanced Patterns`) live in `references/details.md`. Read that file when the navigation summary above is insufficient. + +## Best Practices Summary + +1. **Always use context managers** - For any resource that needs cleanup +2. **Clean up unconditionally** - `__exit__` runs even on exception +3. **Don't suppress unexpectedly** - Return `False` unless suppression is intentional +4. **Use @contextmanager** - For simple resource patterns +5. **Implement both protocols** - Support `with` and manual management +6. **Use ExitStack** - For dynamic numbers of resources +7. **Accumulate efficiently** - List + join, not string concatenation +8. **Track metrics** - Time-to-first-byte matters for streaming +9. **Document behavior** - Especially exception suppression +10. **Test cleanup paths** - Verify resources are released on errors diff --git a/.agents/skills/python-resource-management/references/details.md b/.agents/skills/python-resource-management/references/details.md new file mode 100644 index 000000000..a1f1775a2 --- /dev/null +++ b/.agents/skills/python-resource-management/references/details.md @@ -0,0 +1,183 @@ +# python-resource-management — detailed worked examples + +## Advanced Patterns + +### Pattern 5: Selective Exception Suppression + +Only suppress specific, documented exceptions. + +```python +class StreamWriter: + """Writer that handles broken pipe gracefully.""" + + def __init__(self, stream) -> None: + self._stream = stream + + def __enter__(self) -> "StreamWriter": + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> bool: + """Clean up, suppressing BrokenPipeError on shutdown.""" + self._stream.close() + + # Suppress BrokenPipeError (client disconnected) + # This is expected behavior, not an error + if exc_type is BrokenPipeError: + return True # Exception suppressed + + return False # Propagate all other exceptions +``` + +### Pattern 6: Streaming with Accumulated State + +Maintain both incremental chunks and accumulated state during streaming. + +```python +from collections.abc import Generator +from dataclasses import dataclass, field + +@dataclass +class StreamingResult: + """Accumulated streaming result.""" + + chunks: list[str] = field(default_factory=list) + _finalized: bool = False + + @property + def content(self) -> str: + """Get accumulated content.""" + return "".join(self.chunks) + + def add_chunk(self, chunk: str) -> None: + """Add chunk to accumulator.""" + if self._finalized: + raise RuntimeError("Cannot add to finalized result") + self.chunks.append(chunk) + + def finalize(self) -> str: + """Mark stream complete and return content.""" + self._finalized = True + return self.content + +def stream_with_accumulation( + response: StreamingResponse, +) -> Generator[tuple[str, str], None, str]: + """Stream response while accumulating content. + + Yields: + Tuple of (accumulated_content, new_chunk) for each chunk. + + Returns: + Final accumulated content. + """ + result = StreamingResult() + + for chunk in response.iter_content(): + result.add_chunk(chunk) + yield result.content, chunk + + return result.finalize() +``` + +### Pattern 7: Efficient String Accumulation + +Avoid O(n²) string concatenation when accumulating. + +```python +def accumulate_stream(stream) -> str: + """Efficiently accumulate stream content.""" + # BAD: O(n²) due to string immutability + # content = "" + # for chunk in stream: + # content += chunk # Creates new string each time + + # GOOD: O(n) with list and join + chunks: list[str] = [] + for chunk in stream: + chunks.append(chunk) + return "".join(chunks) # Single allocation +``` + +### Pattern 8: Tracking Stream Metrics + +Measure time-to-first-byte and total streaming time. + +```python +import time +from collections.abc import Generator + +def stream_with_metrics( + response: StreamingResponse, +) -> Generator[str, None, dict]: + """Stream response while collecting metrics. + + Yields: + Content chunks. + + Returns: + Metrics dictionary. + """ + start = time.perf_counter() + first_chunk_time: float | None = None + chunk_count = 0 + total_bytes = 0 + + for chunk in response.iter_content(): + if first_chunk_time is None: + first_chunk_time = time.perf_counter() - start + + chunk_count += 1 + total_bytes += len(chunk.encode()) + yield chunk + + total_time = time.perf_counter() - start + + return { + "time_to_first_byte_ms": round((first_chunk_time or 0) * 1000, 2), + "total_time_ms": round(total_time * 1000, 2), + "chunk_count": chunk_count, + "total_bytes": total_bytes, + } +``` + +### Pattern 9: Managing Multiple Resources with ExitStack + +Handle a dynamic number of resources cleanly. + +```python +from contextlib import ExitStack, AsyncExitStack +from pathlib import Path + +def process_files(paths: list[Path]) -> list[str]: + """Process multiple files with automatic cleanup.""" + results = [] + + with ExitStack() as stack: + # Open all files - they'll all be closed when block exits + files = [stack.enter_context(open(p)) for p in paths] + + for f in files: + results.append(f.read()) + + return results + +async def process_connections(hosts: list[str]) -> list[dict]: + """Process multiple async connections.""" + results = [] + + async with AsyncExitStack() as stack: + connections = [ + await stack.enter_async_context(connect_to_host(host)) + for host in hosts + ] + + for conn in connections: + results.append(await conn.fetch_data()) + + return results +``` diff --git a/.agents/skills/python-testing-patterns/SKILL.md b/.agents/skills/python-testing-patterns/SKILL.md new file mode 100644 index 000000000..f37c8a3c3 --- /dev/null +++ b/.agents/skills/python-testing-patterns/SKILL.md @@ -0,0 +1,45 @@ +--- +name: python-testing-patterns +description: "Implement comprehensive testing strategies with pytest, fixtures, mocking, and test-driven development. Use when writing Python tests, setting up test suites, or implementing testing best practices." +risk: safe +source: community +date_added: "2026-02-27" +--- + +# Python Testing Patterns + +Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices. + +## Use this skill when + +- Writing unit tests for Python code +- Setting up test suites and test infrastructure +- Implementing test-driven development (TDD) +- Creating integration tests for APIs and services +- Mocking external dependencies and services +- Testing async code and concurrent operations +- Setting up continuous testing in CI/CD +- Implementing property-based testing +- Testing database operations +- Debugging failing tests + +## Do not use this skill when + +- The task is unrelated to python testing patterns +- You need a different domain or tool outside this scope + +## Instructions + +- Clarify goals, constraints, and required inputs. +- Apply relevant best practices and validate outcomes. +- Provide actionable steps and verification. +- If detailed examples are required, open `resources/implementation-playbook.md`. + +## Resources + +- `resources/implementation-playbook.md` for detailed patterns and examples. + +## Limitations +- Use this skill only when the task clearly matches the scope described above. +- Do not treat the output as a substitute for environment-specific validation, testing, or expert review. +- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing. diff --git a/.agents/skills/python-testing-patterns/resources/implementation-playbook.md b/.agents/skills/python-testing-patterns/resources/implementation-playbook.md new file mode 100644 index 000000000..da3202cdb --- /dev/null +++ b/.agents/skills/python-testing-patterns/resources/implementation-playbook.md @@ -0,0 +1,906 @@ +# Python Testing Patterns Implementation Playbook + +This file contains detailed patterns, checklists, and code samples referenced by the skill. + +# Python Testing Patterns + +Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices. + +## When to Use This Skill + +- Writing unit tests for Python code +- Setting up test suites and test infrastructure +- Implementing test-driven development (TDD) +- Creating integration tests for APIs and services +- Mocking external dependencies and services +- Testing async code and concurrent operations +- Setting up continuous testing in CI/CD +- Implementing property-based testing +- Testing database operations +- Debugging failing tests + +## Core Concepts + +### 1. Test Types +- **Unit Tests**: Test individual functions/classes in isolation +- **Integration Tests**: Test interaction between components +- **Functional Tests**: Test complete features end-to-end +- **Performance Tests**: Measure speed and resource usage + +### 2. Test Structure (AAA Pattern) +- **Arrange**: Set up test data and preconditions +- **Act**: Execute the code under test +- **Assert**: Verify the results + +### 3. Test Coverage +- Measure what code is exercised by tests +- Identify untested code paths +- Aim for meaningful coverage, not just high percentages + +### 4. Test Isolation +- Tests should be independent +- No shared state between tests +- Each test should clean up after itself + +## Quick Start + +```python +# test_example.py +def add(a, b): + return a + b + +def test_add(): + """Basic test example.""" + result = add(2, 3) + assert result == 5 + +def test_add_negative(): + """Test with negative numbers.""" + assert add(-1, 1) == 0 + +# Run with: pytest test_example.py +``` + +## Fundamental Patterns + +### Pattern 1: Basic pytest Tests + +```python +# test_calculator.py +import pytest + +class Calculator: + """Simple calculator for testing.""" + + def add(self, a: float, b: float) -> float: + return a + b + + def subtract(self, a: float, b: float) -> float: + return a - b + + def multiply(self, a: float, b: float) -> float: + return a * b + + def divide(self, a: float, b: float) -> float: + if b == 0: + raise ValueError("Cannot divide by zero") + return a / b + + +def test_addition(): + """Test addition.""" + calc = Calculator() + assert calc.add(2, 3) == 5 + assert calc.add(-1, 1) == 0 + assert calc.add(0, 0) == 0 + + +def test_subtraction(): + """Test subtraction.""" + calc = Calculator() + assert calc.subtract(5, 3) == 2 + assert calc.subtract(0, 5) == -5 + + +def test_multiplication(): + """Test multiplication.""" + calc = Calculator() + assert calc.multiply(3, 4) == 12 + assert calc.multiply(0, 5) == 0 + + +def test_division(): + """Test division.""" + calc = Calculator() + assert calc.divide(6, 3) == 2 + assert calc.divide(5, 2) == 2.5 + + +def test_division_by_zero(): + """Test division by zero raises error.""" + calc = Calculator() + with pytest.raises(ValueError, match="Cannot divide by zero"): + calc.divide(5, 0) +``` + +### Pattern 2: Fixtures for Setup and Teardown + +```python +# test_database.py +import pytest +from typing import Generator + +class Database: + """Simple database class.""" + + def __init__(self, connection_string: str): + self.connection_string = connection_string + self.connected = False + + def connect(self): + """Connect to database.""" + self.connected = True + + def disconnect(self): + """Disconnect from database.""" + self.connected = False + + def query(self, sql: str) -> list: + """Execute query.""" + if not self.connected: + raise RuntimeError("Not connected") + return [{"id": 1, "name": "Test"}] + + +@pytest.fixture +def db() -> Generator[Database, None, None]: + """Fixture that provides connected database.""" + # Setup + database = Database("sqlite:///:memory:") + database.connect() + + # Provide to test + yield database + + # Teardown + database.disconnect() + + +def test_database_query(db): + """Test database query with fixture.""" + results = db.query("SELECT * FROM users") + assert len(results) == 1 + assert results[0]["name"] == "Test" + + +@pytest.fixture(scope="session") +def app_config(): + """Session-scoped fixture - created once per test session.""" + return { + "database_url": "postgresql://localhost/test", + "api_key": "test-key", + "debug": True + } + + +@pytest.fixture(scope="module") +def api_client(app_config): + """Module-scoped fixture - created once per test module.""" + # Setup expensive resource + client = {"config": app_config, "session": "active"} + yield client + # Cleanup + client["session"] = "closed" + + +def test_api_client(api_client): + """Test using api client fixture.""" + assert api_client["session"] == "active" + assert api_client["config"]["debug"] is True +``` + +### Pattern 3: Parameterized Tests + +```python +# test_validation.py +import pytest + +def is_valid_email(email: str) -> bool: + """Check if email is valid.""" + return "@" in email and "." in email.split("@")[1] + + +@pytest.mark.parametrize("email,expected", [ + ("user@example.com", True), + ("test.user@domain.co.uk", True), + ("invalid.email", False), + ("@example.com", False), + ("user@domain", False), + ("", False), +]) +def test_email_validation(email, expected): + """Test email validation with various inputs.""" + assert is_valid_email(email) == expected + + +@pytest.mark.parametrize("a,b,expected", [ + (2, 3, 5), + (0, 0, 0), + (-1, 1, 0), + (100, 200, 300), + (-5, -5, -10), +]) +def test_addition_parameterized(a, b, expected): + """Test addition with multiple parameter sets.""" + from test_calculator import Calculator + calc = Calculator() + assert calc.add(a, b) == expected + + +# Using pytest.param for special cases +@pytest.mark.parametrize("value,expected", [ + pytest.param(1, True, id="positive"), + pytest.param(0, False, id="zero"), + pytest.param(-1, False, id="negative"), +]) +def test_is_positive(value, expected): + """Test with custom test IDs.""" + assert (value > 0) == expected +``` + +### Pattern 4: Mocking with unittest.mock + +```python +# test_api_client.py +import pytest +from unittest.mock import Mock, patch, MagicMock +import requests + +class APIClient: + """Simple API client.""" + + def __init__(self, base_url: str): + self.base_url = base_url + + def get_user(self, user_id: int) -> dict: + """Fetch user from API.""" + response = requests.get(f"{self.base_url}/users/{user_id}") + response.raise_for_status() + return response.json() + + def create_user(self, data: dict) -> dict: + """Create new user.""" + response = requests.post(f"{self.base_url}/users", json=data) + response.raise_for_status() + return response.json() + + +def test_get_user_success(): + """Test successful API call with mock.""" + client = APIClient("https://api.example.com") + + mock_response = Mock() + mock_response.json.return_value = {"id": 1, "name": "John Doe"} + mock_response.raise_for_status.return_value = None + + with patch("requests.get", return_value=mock_response) as mock_get: + user = client.get_user(1) + + assert user["id"] == 1 + assert user["name"] == "John Doe" + mock_get.assert_called_once_with("https://api.example.com/users/1") + + +def test_get_user_not_found(): + """Test API call with 404 error.""" + client = APIClient("https://api.example.com") + + mock_response = Mock() + mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") + + with patch("requests.get", return_value=mock_response): + with pytest.raises(requests.HTTPError): + client.get_user(999) + + +@patch("requests.post") +def test_create_user(mock_post): + """Test user creation with decorator syntax.""" + client = APIClient("https://api.example.com") + + mock_post.return_value.json.return_value = {"id": 2, "name": "Jane Doe"} + mock_post.return_value.raise_for_status.return_value = None + + user_data = {"name": "Jane Doe", "email": "jane@example.com"} + result = client.create_user(user_data) + + assert result["id"] == 2 + mock_post.assert_called_once() + call_args = mock_post.call_args + assert call_args.kwargs["json"] == user_data +``` + +### Pattern 5: Testing Exceptions + +```python +# test_exceptions.py +import pytest + +def divide(a: float, b: float) -> float: + """Divide a by b.""" + if b == 0: + raise ZeroDivisionError("Division by zero") + if not isinstance(a, (int, float)) or not isinstance(b, (int, float)): + raise TypeError("Arguments must be numbers") + return a / b + + +def test_zero_division(): + """Test exception is raised for division by zero.""" + with pytest.raises(ZeroDivisionError): + divide(10, 0) + + +def test_zero_division_with_message(): + """Test exception message.""" + with pytest.raises(ZeroDivisionError, match="Division by zero"): + divide(5, 0) + + +def test_type_error(): + """Test type error exception.""" + with pytest.raises(TypeError, match="must be numbers"): + divide("10", 5) + + +def test_exception_info(): + """Test accessing exception info.""" + with pytest.raises(ValueError) as exc_info: + int("not a number") + + assert "invalid literal" in str(exc_info.value) +``` + +## Advanced Patterns + +### Pattern 6: Testing Async Code + +```python +# test_async.py +import pytest +import asyncio + +async def fetch_data(url: str) -> dict: + """Fetch data asynchronously.""" + await asyncio.sleep(0.1) + return {"url": url, "data": "result"} + + +@pytest.mark.asyncio +async def test_fetch_data(): + """Test async function.""" + result = await fetch_data("https://api.example.com") + assert result["url"] == "https://api.example.com" + assert "data" in result + + +@pytest.mark.asyncio +async def test_concurrent_fetches(): + """Test concurrent async operations.""" + urls = ["url1", "url2", "url3"] + tasks = [fetch_data(url) for url in urls] + results = await asyncio.gather(*tasks) + + assert len(results) == 3 + assert all("data" in r for r in results) + + +@pytest.fixture +async def async_client(): + """Async fixture.""" + client = {"connected": True} + yield client + client["connected"] = False + + +@pytest.mark.asyncio +async def test_with_async_fixture(async_client): + """Test using async fixture.""" + assert async_client["connected"] is True +``` + +### Pattern 7: Monkeypatch for Testing + +```python +# test_environment.py +import os +import pytest + +def get_database_url() -> str: + """Get database URL from environment.""" + return os.environ.get("DATABASE_URL", "sqlite:///:memory:") + + +def test_database_url_default(): + """Test default database URL.""" + # Will use actual environment variable if set + url = get_database_url() + assert url + + +def test_database_url_custom(monkeypatch): + """Test custom database URL with monkeypatch.""" + monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test") + assert get_database_url() == "postgresql://localhost/test" + + +def test_database_url_not_set(monkeypatch): + """Test when env var is not set.""" + monkeypatch.delenv("DATABASE_URL", raising=False) + assert get_database_url() == "sqlite:///:memory:" + + +class Config: + """Configuration class.""" + + def __init__(self): + self.api_key = "production-key" + + def get_api_key(self): + return self.api_key + + +def test_monkeypatch_attribute(monkeypatch): + """Test monkeypatching object attributes.""" + config = Config() + monkeypatch.setattr(config, "api_key", "test-key") + assert config.get_api_key() == "test-key" +``` + +### Pattern 8: Temporary Files and Directories + +```python +# test_file_operations.py +import pytest +from pathlib import Path + +def save_data(filepath: Path, data: str): + """Save data to file.""" + filepath.write_text(data) + + +def load_data(filepath: Path) -> str: + """Load data from file.""" + return filepath.read_text() + + +def test_file_operations(tmp_path): + """Test file operations with temporary directory.""" + # tmp_path is a pathlib.Path object + test_file = tmp_path / "test_data.txt" + + # Save data + save_data(test_file, "Hello, World!") + + # Verify file exists + assert test_file.exists() + + # Load and verify data + data = load_data(test_file) + assert data == "Hello, World!" + + +def test_multiple_files(tmp_path): + """Test with multiple temporary files.""" + files = { + "file1.txt": "Content 1", + "file2.txt": "Content 2", + "file3.txt": "Content 3" + } + + for filename, content in files.items(): + filepath = tmp_path / filename + save_data(filepath, content) + + # Verify all files created + assert len(list(tmp_path.iterdir())) == 3 + + # Verify contents + for filename, expected_content in files.items(): + filepath = tmp_path / filename + assert load_data(filepath) == expected_content +``` + +### Pattern 9: Custom Fixtures and Conftest + +```python +# conftest.py +"""Shared fixtures for all tests.""" +import pytest + +@pytest.fixture(scope="session") +def database_url(): + """Provide database URL for all tests.""" + return "postgresql://localhost/test_db" + + +@pytest.fixture(autouse=True) +def reset_database(database_url): + """Auto-use fixture that runs before each test.""" + # Setup: Clear database + print(f"Clearing database: {database_url}") + yield + # Teardown: Clean up + print("Test completed") + + +@pytest.fixture +def sample_user(): + """Provide sample user data.""" + return { + "id": 1, + "name": "Test User", + "email": "test@example.com" + } + + +@pytest.fixture +def sample_users(): + """Provide list of sample users.""" + return [ + {"id": 1, "name": "User 1"}, + {"id": 2, "name": "User 2"}, + {"id": 3, "name": "User 3"}, + ] + + +# Parametrized fixture +@pytest.fixture(params=["sqlite", "postgresql", "mysql"]) +def db_backend(request): + """Fixture that runs tests with different database backends.""" + return request.param + + +def test_with_db_backend(db_backend): + """This test will run 3 times with different backends.""" + print(f"Testing with {db_backend}") + assert db_backend in ["sqlite", "postgresql", "mysql"] +``` + +### Pattern 10: Property-Based Testing + +```python +# test_properties.py +from hypothesis import given, strategies as st +import pytest + +def reverse_string(s: str) -> str: + """Reverse a string.""" + return s[::-1] + + +@given(st.text()) +def test_reverse_twice_is_original(s): + """Property: reversing twice returns original.""" + assert reverse_string(reverse_string(s)) == s + + +@given(st.text()) +def test_reverse_length(s): + """Property: reversed string has same length.""" + assert len(reverse_string(s)) == len(s) + + +@given(st.integers(), st.integers()) +def test_addition_commutative(a, b): + """Property: addition is commutative.""" + assert a + b == b + a + + +@given(st.lists(st.integers())) +def test_sorted_list_properties(lst): + """Property: sorted list is ordered.""" + sorted_lst = sorted(lst) + + # Same length + assert len(sorted_lst) == len(lst) + + # All elements present + assert set(sorted_lst) == set(lst) + + # Is ordered + for i in range(len(sorted_lst) - 1): + assert sorted_lst[i] <= sorted_lst[i + 1] +``` + +## Testing Best Practices + +### Test Organization + +```python +# tests/ +# __init__.py +# conftest.py # Shared fixtures +# test_unit/ # Unit tests +# test_models.py +# test_utils.py +# test_integration/ # Integration tests +# test_api.py +# test_database.py +# test_e2e/ # End-to-end tests +# test_workflows.py +``` + +### Test Naming + +```python +# Good test names +def test_user_creation_with_valid_data(): + """Clear name describes what is being tested.""" + pass + + +def test_login_fails_with_invalid_password(): + """Name describes expected behavior.""" + pass + + +def test_api_returns_404_for_missing_resource(): + """Specific about inputs and expected outcomes.""" + pass + + +# Bad test names +def test_1(): # Not descriptive + pass + + +def test_user(): # Too vague + pass + + +def test_function(): # Doesn't explain what's tested + pass +``` + +### Test Markers + +```python +# test_markers.py +import pytest + +@pytest.mark.slow +def test_slow_operation(): + """Mark slow tests.""" + import time + time.sleep(2) + + +@pytest.mark.integration +def test_database_integration(): + """Mark integration tests.""" + pass + + +@pytest.mark.skip(reason="Feature not implemented yet") +def test_future_feature(): + """Skip tests temporarily.""" + pass + + +@pytest.mark.skipif(os.name == "nt", reason="Unix only test") +def test_unix_specific(): + """Conditional skip.""" + pass + + +@pytest.mark.xfail(reason="Known bug #123") +def test_known_bug(): + """Mark expected failures.""" + assert False + + +# Run with: +# pytest -m slow # Run only slow tests +# pytest -m "not slow" # Skip slow tests +# pytest -m integration # Run integration tests +``` + +### Coverage Reporting + +```bash +# Install coverage +pip install pytest-cov + +# Run tests with coverage +pytest --cov=myapp tests/ + +# Generate HTML report +pytest --cov=myapp --cov-report=html tests/ + +# Fail if coverage below threshold +pytest --cov=myapp --cov-fail-under=80 tests/ + +# Show missing lines +pytest --cov=myapp --cov-report=term-missing tests/ +``` + +## Testing Database Code + +```python +# test_database_models.py +import pytest +from sqlalchemy import create_engine, Column, Integer, String +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker, Session + +Base = declarative_base() + + +class User(Base): + """User model.""" + __tablename__ = "users" + + id = Column(Integer, primary_key=True) + name = Column(String(50)) + email = Column(String(100), unique=True) + + +@pytest.fixture(scope="function") +def db_session() -> Session: + """Create in-memory database for testing.""" + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + + SessionLocal = sessionmaker(bind=engine) + session = SessionLocal() + + yield session + + session.close() + + +def test_create_user(db_session): + """Test creating a user.""" + user = User(name="Test User", email="test@example.com") + db_session.add(user) + db_session.commit() + + assert user.id is not None + assert user.name == "Test User" + + +def test_query_user(db_session): + """Test querying users.""" + user1 = User(name="User 1", email="user1@example.com") + user2 = User(name="User 2", email="user2@example.com") + + db_session.add_all([user1, user2]) + db_session.commit() + + users = db_session.query(User).all() + assert len(users) == 2 + + +def test_unique_email_constraint(db_session): + """Test unique email constraint.""" + from sqlalchemy.exc import IntegrityError + + user1 = User(name="User 1", email="same@example.com") + user2 = User(name="User 2", email="same@example.com") + + db_session.add(user1) + db_session.commit() + + db_session.add(user2) + + with pytest.raises(IntegrityError): + db_session.commit() +``` + +## CI/CD Integration + +```yaml +# .github/workflows/test.yml +name: Tests + +on: [push, pull_request] + +jobs: + test: + runs-on: ubuntu-latest + + strategy: + matrix: + python-version: ["3.9", "3.10", "3.11", "3.12"] + + steps: + - uses: actions/checkout@v3 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + pip install -e ".[dev]" + pip install pytest pytest-cov + + - name: Run tests + run: | + pytest --cov=myapp --cov-report=xml + + - name: Upload coverage + uses: codecov/codecov-action@v3 + with: + file: ./coverage.xml +``` + +## Configuration Files + +```ini +# pytest.ini +[pytest] +testpaths = tests +python_files = test_*.py +python_classes = Test* +python_functions = test_* +addopts = + -v + --strict-markers + --tb=short + --cov=myapp + --cov-report=term-missing +markers = + slow: marks tests as slow + integration: marks integration tests + unit: marks unit tests + e2e: marks end-to-end tests +``` + +```toml +# pyproject.toml +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +addopts = [ + "-v", + "--cov=myapp", + "--cov-report=term-missing", +] + +[tool.coverage.run] +source = ["myapp"] +omit = ["*/tests/*", "*/migrations/*"] + +[tool.coverage.report] +exclude_lines = [ + "pragma: no cover", + "def __repr__", + "raise AssertionError", + "raise NotImplementedError", +] +``` + +## Resources + +- **pytest documentation**: https://docs.pytest.org/ +- **unittest.mock**: https://docs.python.org/3/library/unittest.mock.html +- **hypothesis**: Property-based testing +- **pytest-asyncio**: Testing async code +- **pytest-cov**: Coverage reporting +- **pytest-mock**: pytest wrapper for mock + +## Best Practices Summary + +1. **Write tests first** (TDD) or alongside code +2. **One assertion per test** when possible +3. **Use descriptive test names** that explain behavior +4. **Keep tests independent** and isolated +5. **Use fixtures** for setup and teardown +6. **Mock external dependencies** appropriately +7. **Parametrize tests** to reduce duplication +8. **Test edge cases** and error conditions +9. **Measure coverage** but focus on quality +10. **Run tests in CI/CD** on every commit diff --git a/.claude/commands/opsx/apply.md b/.claude/commands/opsx/apply.md new file mode 100644 index 000000000..731c78958 --- /dev/null +++ b/.claude/commands/opsx/apply.md @@ -0,0 +1,184 @@ +--- +name: "OPSX: Apply" +description: "Implement tasks from an OpenSpec change (Experimental)" +allowed-tools: Bash(openspec:*) +category: "Workflow" +tags: ["workflow", "artifacts", "experimental"] +--- + +Implement tasks from an OpenSpec change. + +**Store selection:** If the user names a store (a store is a standalone OpenSpec repo registered on this machine) or the work lives in one, run `openspec store list --json` to discover registered store ids, then pass `--store ` on the commands that read or write specs and changes (`new change`, `status`, `instructions`, `list`, `show`, `validate`, `archive`, `doctor`, `context`, `schemas`, `view`). Once selected, treat `--store ` as sticky for the rest of the workflow. Every unscoped example of those commands below is shorthand: before running it, append the flag. For example, run `openspec status --change "" --json --store ""`, not the unscoped form shown below. Other commands do not take the flag. Hints printed by commands already carry the flag; keep it on follow-ups. Without a store, commands act on the nearest local `openspec/` root. + +**Input**: Optionally specify a change name (e.g., `/opsx:apply add-auth`). If omitted, check if it can be inferred from conversation context. If vague or ambiguous you MUST prompt for available changes. + +**Steps** + +1. **Select the change** + + If a name is provided, use it. Otherwise: + - Infer from conversation context if the user mentioned a change + - Auto-select if only one active change exists + - If ambiguous, run `openspec list --json` to get available changes and ask the user to select one + + Always announce: "Using change: " and how to override (e.g., `/opsx:apply `). + +2. **Check status to understand the schema** + ```bash + openspec status --change "" --json + ``` + Parse the JSON to understand: + - `schemaName`: The workflow being used (e.g., "spec-driven") + - `planningHome`, `changeRoot`, and `actionContext`: planning scope and edit constraints + - Which artifact contains the tasks (typically "tasks" for spec-driven, check status for others) + +3. **Get apply instructions** + + ```bash + openspec instructions apply --change "" --json + ``` + + This returns: + - `contextFiles`: artifact ID -> array of concrete file paths (varies by schema - could be proposal/specs/design/tasks or spec/tests/implementation/docs) + - Progress (total, complete, remaining) + - Task list with status + - Dynamic instruction based on current state + - Optional `context`: current required project instruction input from the selected root + - Optional `operationGuidance`: current advisory guidance for apply + + **Handle states:** + - If `state: "blocked"` (missing artifacts): show message, suggest using `/opsx:continue` (if it is not installed, run `openspec status --change "" --json` to see the next artifact and `openspec instructions --change "" --json` for how to create it) + - If `state: "all_done"`: congratulate, suggest archive + - Otherwise: proceed to implementation + + Treat `context` as a required prompt-level input. Read and consider it, and + apply relevant project facts, conventions, and constraints while implementing. + Treat `operationGuidance` as optional additive advice. Read and consider every + entry, and follow entries that are applicable and compatible with the built-in + workflow. + + Keep both fields separate from CLI-returned state, missing artifacts, tasks, + progress, `contextFiles`, and the built-in `instruction`. They are not + evidence of task completion, do not replace the built-in instruction, and do + not permit bypassing a blocked state. If context conflicts with the built-in + instruction, an explicit user choice, or a CLI-controlled value, report the + conflict and preserve the controlling value. If guidance is inapplicable or + conflicts with those controlling inputs, do not follow it and explain why. + These are prompt-level behavior contracts, not enforceable checks. + +4. **Read context files** + + Read every file path listed under `contextFiles` from the apply instructions output. + The files depend on the schema being used: + - **spec-driven**: proposal, specs, design, tasks + - Other schemas: follow the contextFiles from CLI output + + Do not copy `context` or `operationGuidance` verbatim into implementation + files or planning artifacts unless the user separately asks for that content. + +5. **Show current progress** + + Display: + - Schema being used + - Progress: "N/M tasks complete" + - Remaining tasks overview + - Dynamic instruction from CLI + +6. **Implement tasks (loop until done or blocked)** + + For each pending task: + - Show which task is being worked on + - Make the code changes required + - Keep changes minimal and focused + - Mark task complete in the tasks file: `- [ ]` → `- [x]` + - Continue to next task + + **Pause if:** + - Task is unclear → ask for clarification + - Implementation reveals a design issue → suggest updating artifacts + - A task needs work beyond what the spec and tasks describe, or you are tempted to drop, narrow, defer, or accept exceptions to specified behavior to make it fit → surface the added scope and ask; do not absorb it silently + - Error or blocker encountered → report and wait for guidance + - User interrupts + +7. **On completion or pause, show status** + + Display: + - Tasks completed this session + - Overall progress: "N/M tasks complete" + - If all done: suggest archive + - If paused: explain why and wait for guidance + +**Output During Implementation** + +``` +## Implementing: (schema: ) + +Working on task 3/7: +[...implementation happening...] +✓ Task complete + +Working on task 4/7: +[...implementation happening...] +✓ Task complete +``` + +**Output On Completion** + +``` +## Implementation Complete + +**Change:** +**Schema:** +**Progress:** 7/7 tasks complete ✓ + +### Completed This Session +- [x] Task 1 +- [x] Task 2 +... + +All tasks complete! You can archive this change with `/opsx:archive`. +``` + +**Output On Pause (Issue Encountered)** + +``` +## Implementation Paused + +**Change:** +**Schema:** +**Progress:** 4/7 tasks complete + +### Issue Encountered + + +**Options:** +1.