workflow: fix package version extraction using pacman -Qp
This commit is contained in:
@@ -0,0 +1,185 @@
|
|||||||
|
---
|
||||||
|
name: python-executor
|
||||||
|
description: "Execute Python code in a safe sandboxed environment via [inference.sh](https://inference.sh). Pre-installed: NumPy, Pandas, Matplotlib, requests, BeautifulSoup, Selenium, Playwright, MoviePy, Pillow, OpenCV, trimesh, and 100+ more libraries. Use for: data processing, web scraping, image manipulation, video creation, 3D model processing, PDF generation, API calls, automation scripts. Triggers: python, execute code, run script, web scraping, data analysis, image processing, video editing, 3D models, automation, pandas, matplotlib"
|
||||||
|
allowed-tools: Bash(belt *)
|
||||||
|
---
|
||||||
|
|
||||||
|
# Python Code Executor
|
||||||
|
|
||||||
|
Execute Python code in a safe, sandboxed environment with 100+ pre-installed libraries.
|
||||||
|
|
||||||
|

|
||||||
|
|
||||||
|
## Quick Start
|
||||||
|
|
||||||
|
> Requires inference.sh CLI (`belt`). [Install instructions](https://raw.githubusercontent.com/inference-sh/skills/refs/heads/main/cli-install.md)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt login
|
||||||
|
|
||||||
|
# Run Python code
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "import pandas as pd\nprint(pd.__version__)"
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
|
||||||
|
## App Details
|
||||||
|
|
||||||
|
| Property | Value |
|
||||||
|
|----------|-------|
|
||||||
|
| App ID | `infsh/python-executor` |
|
||||||
|
| Environment | Python 3.10, CPU-only |
|
||||||
|
| RAM | 8GB (default) / 16GB (high_memory) |
|
||||||
|
| Timeout | 1-300 seconds (default: 30) |
|
||||||
|
|
||||||
|
## Input Schema
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"code": "print('Hello World!')",
|
||||||
|
"timeout": 30,
|
||||||
|
"capture_output": true,
|
||||||
|
"working_dir": null
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Pre-installed Libraries
|
||||||
|
|
||||||
|
### Web Scraping & HTTP
|
||||||
|
- `requests`, `httpx`, `aiohttp` - HTTP clients
|
||||||
|
- `beautifulsoup4`, `lxml` - HTML/XML parsing
|
||||||
|
- `selenium`, `playwright` - Browser automation
|
||||||
|
- `scrapy` - Web scraping framework
|
||||||
|
|
||||||
|
### Data Processing
|
||||||
|
- `numpy`, `pandas`, `scipy` - Numerical computing
|
||||||
|
- `matplotlib`, `seaborn`, `plotly` - Visualization
|
||||||
|
|
||||||
|
### Image Processing
|
||||||
|
- `pillow`, `opencv-python-headless` - Image manipulation
|
||||||
|
- `scikit-image`, `imageio` - Image algorithms
|
||||||
|
|
||||||
|
### Video & Audio
|
||||||
|
- `moviepy` - Video editing
|
||||||
|
- `av` (PyAV), `ffmpeg-python` - Video processing
|
||||||
|
- `pydub` - Audio manipulation
|
||||||
|
|
||||||
|
### 3D Processing
|
||||||
|
- `trimesh`, `open3d` - 3D mesh processing
|
||||||
|
- `numpy-stl`, `meshio`, `pyvista` - 3D file formats
|
||||||
|
|
||||||
|
### Documents & Graphics
|
||||||
|
- `svgwrite`, `cairosvg` - SVG creation
|
||||||
|
- `reportlab`, `pypdf2` - PDF generation
|
||||||
|
|
||||||
|
## Examples
|
||||||
|
|
||||||
|
### Web Scraping
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "import requests\nfrom bs4 import BeautifulSoup\n\nresponse = requests.get(\"https://example.com\")\nsoup = BeautifulSoup(response.content, \"html.parser\")\nprint(soup.find(\"title\").text)"
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
### Data Analysis with Visualization
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "import pandas as pd\nimport matplotlib.pyplot as plt\n\ndata = {\"name\": [\"Alice\", \"Bob\"], \"sales\": [100, 150]}\ndf = pd.DataFrame(data)\n\nplt.bar(df[\"name\"], df[\"sales\"])\nplt.savefig(\"outputs/chart.png\")\nprint(\"Chart saved!\")"
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
### Image Processing
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "from PIL import Image\nimport numpy as np\n\n# Create gradient image\narr = np.linspace(0, 255, 256*256, dtype=np.uint8).reshape(256, 256)\nimg = Image.fromarray(arr, mode=\"L\")\nimg.save(\"outputs/gradient.png\")\nprint(\"Image created!\")"
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
### Video Creation
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "from moviepy.editor import ColorClip, TextClip, CompositeVideoClip\n\nclip = ColorClip(size=(640, 480), color=(0, 100, 200), duration=3)\ntxt = TextClip(\"Hello!\", fontsize=70, color=\"white\").set_position(\"center\").set_duration(3)\nvideo = CompositeVideoClip([clip, txt])\nvideo.write_videofile(\"outputs/hello.mp4\", fps=24)\nprint(\"Video created!\")",
|
||||||
|
"timeout": 120
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3D Model Processing
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "import trimesh\n\nsphere = trimesh.creation.icosphere(subdivisions=3, radius=1.0)\nsphere.export(\"outputs/sphere.stl\")\nprint(f\"Created sphere with {len(sphere.vertices)} vertices\")"
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
### API Calls
|
||||||
|
|
||||||
|
```bash
|
||||||
|
belt app run infsh/python-executor --input '{
|
||||||
|
"code": "import requests\nimport json\n\nresponse = requests.get(\"https://api.github.com/users/octocat\")\ndata = response.json()\nprint(json.dumps(data, indent=2))"
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
## File Output
|
||||||
|
|
||||||
|
Files saved to `outputs/` are automatically returned:
|
||||||
|
|
||||||
|
```python
|
||||||
|
# These files will be in the response
|
||||||
|
plt.savefig('outputs/chart.png')
|
||||||
|
df.to_csv('outputs/data.csv')
|
||||||
|
video.write_videofile('outputs/video.mp4')
|
||||||
|
mesh.export('outputs/model.stl')
|
||||||
|
```
|
||||||
|
|
||||||
|
## Variants
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Default (8GB RAM)
|
||||||
|
belt app run infsh/python-executor --input input.json
|
||||||
|
|
||||||
|
# High memory (16GB RAM) for large datasets
|
||||||
|
belt app run infsh/python-executor@high_memory --input input.json
|
||||||
|
```
|
||||||
|
|
||||||
|
## Use Cases
|
||||||
|
|
||||||
|
- **Web scraping** - Extract data from websites
|
||||||
|
- **Data analysis** - Process and visualize datasets
|
||||||
|
- **Image manipulation** - Resize, crop, composite images
|
||||||
|
- **Video creation** - Generate videos with text overlays
|
||||||
|
- **3D processing** - Load, transform, export 3D models
|
||||||
|
- **API integration** - Call external APIs
|
||||||
|
- **PDF generation** - Create reports and documents
|
||||||
|
- **Automation** - Run any Python script
|
||||||
|
|
||||||
|
## Important Notes
|
||||||
|
|
||||||
|
- **CPU-only** - No GPU/ML libraries (use dedicated AI apps for that)
|
||||||
|
- **Safe execution** - Runs in isolated subprocess
|
||||||
|
- **Non-interactive** - Use `plt.savefig()` not `plt.show()`
|
||||||
|
- **File detection** - Output files are auto-detected and returned
|
||||||
|
|
||||||
|
## Related Skills
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# AI image generation (for ML-based images)
|
||||||
|
npx skills add inference-sh/skills@ai-image-generation
|
||||||
|
|
||||||
|
# AI video generation (for ML-based videos)
|
||||||
|
npx skills add inference-sh/skills@ai-video-generation
|
||||||
|
|
||||||
|
# LLM models (for text generation)
|
||||||
|
npx skills add inference-sh/skills@llm-models
|
||||||
|
```
|
||||||
|
|
||||||
|
## Documentation
|
||||||
|
|
||||||
|
- [Running Apps](https://inference.sh/docs/apps/running) - How to run apps via CLI
|
||||||
|
- [App Code](https://inference.sh/docs/extend/app-code) - Understanding app execution
|
||||||
|
- [Sandboxed Code Execution](https://inference.sh/blog/tools/sandboxed-execution) - Safe code execution for agents
|
||||||
|
|
||||||
@@ -0,0 +1,750 @@
|
|||||||
|
---
|
||||||
|
name: python-patterns
|
||||||
|
description: Pythonic idioms, PEP 8 standards, type hints, and best practices for building robust, efficient, and maintainable Python applications.
|
||||||
|
origin: ECC
|
||||||
|
---
|
||||||
|
|
||||||
|
# Python Development Patterns
|
||||||
|
|
||||||
|
Idiomatic Python patterns and best practices for building robust, efficient, and maintainable applications.
|
||||||
|
|
||||||
|
## When to Activate
|
||||||
|
|
||||||
|
- Writing new Python code
|
||||||
|
- Reviewing Python code
|
||||||
|
- Refactoring existing Python code
|
||||||
|
- Designing Python packages/modules
|
||||||
|
|
||||||
|
## Core Principles
|
||||||
|
|
||||||
|
### 1. Readability Counts
|
||||||
|
|
||||||
|
Python prioritizes readability. Code should be obvious and easy to understand.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: Clear and readable
|
||||||
|
def get_active_users(users: list[User]) -> list[User]:
|
||||||
|
"""Return only active users from the provided list."""
|
||||||
|
return [user for user in users if user.is_active]
|
||||||
|
|
||||||
|
|
||||||
|
# Bad: Clever but confusing
|
||||||
|
def get_active_users(u):
|
||||||
|
return [x for x in u if x.a]
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2. Explicit is Better Than Implicit
|
||||||
|
|
||||||
|
Avoid magic; be clear about what your code does.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: Explicit configuration
|
||||||
|
import logging
|
||||||
|
|
||||||
|
logging.basicConfig(
|
||||||
|
level=logging.INFO,
|
||||||
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||||
|
)
|
||||||
|
|
||||||
|
# Bad: Hidden side effects
|
||||||
|
import some_module
|
||||||
|
some_module.setup() # What does this do?
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3. EAFP - Easier to Ask Forgiveness Than Permission
|
||||||
|
|
||||||
|
Python prefers exception handling over checking conditions.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: EAFP style
|
||||||
|
def get_value(dictionary: dict, key: str) -> Any:
|
||||||
|
try:
|
||||||
|
return dictionary[key]
|
||||||
|
except KeyError:
|
||||||
|
return default_value
|
||||||
|
|
||||||
|
# Bad: LBYL (Look Before You Leap) style
|
||||||
|
def get_value(dictionary: dict, key: str) -> Any:
|
||||||
|
if key in dictionary:
|
||||||
|
return dictionary[key]
|
||||||
|
else:
|
||||||
|
return default_value
|
||||||
|
```
|
||||||
|
|
||||||
|
## Type Hints
|
||||||
|
|
||||||
|
### Basic Type Annotations
|
||||||
|
|
||||||
|
```python
|
||||||
|
from typing import Optional, List, Dict, Any
|
||||||
|
|
||||||
|
def process_user(
|
||||||
|
user_id: str,
|
||||||
|
data: Dict[str, Any],
|
||||||
|
active: bool = True
|
||||||
|
) -> Optional[User]:
|
||||||
|
"""Process a user and return the updated User or None."""
|
||||||
|
if not active:
|
||||||
|
return None
|
||||||
|
return User(user_id, data)
|
||||||
|
```
|
||||||
|
|
||||||
|
### Modern Type Hints (Python 3.9+)
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Python 3.9+ - Use built-in types
|
||||||
|
def process_items(items: list[str]) -> dict[str, int]:
|
||||||
|
return {item: len(item) for item in items}
|
||||||
|
|
||||||
|
# Python 3.8 and earlier - Use typing module
|
||||||
|
from typing import List, Dict
|
||||||
|
|
||||||
|
def process_items(items: List[str]) -> Dict[str, int]:
|
||||||
|
return {item: len(item) for item in items}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Type Aliases and TypeVar
|
||||||
|
|
||||||
|
```python
|
||||||
|
from typing import TypeVar, Union
|
||||||
|
|
||||||
|
# Type alias for complex types
|
||||||
|
JSON = Union[dict[str, Any], list[Any], str, int, float, bool, None]
|
||||||
|
|
||||||
|
def parse_json(data: str) -> JSON:
|
||||||
|
return json.loads(data)
|
||||||
|
|
||||||
|
# Generic types
|
||||||
|
T = TypeVar('T')
|
||||||
|
|
||||||
|
def first(items: list[T]) -> T | None:
|
||||||
|
"""Return the first item or None if list is empty."""
|
||||||
|
return items[0] if items else None
|
||||||
|
```
|
||||||
|
|
||||||
|
### Protocol-Based Duck Typing
|
||||||
|
|
||||||
|
```python
|
||||||
|
from typing import Protocol
|
||||||
|
|
||||||
|
class Renderable(Protocol):
|
||||||
|
def render(self) -> str:
|
||||||
|
"""Render the object to a string."""
|
||||||
|
|
||||||
|
def render_all(items: list[Renderable]) -> str:
|
||||||
|
"""Render all items that implement the Renderable protocol."""
|
||||||
|
return "\n".join(item.render() for item in items)
|
||||||
|
```
|
||||||
|
|
||||||
|
## Error Handling Patterns
|
||||||
|
|
||||||
|
### Specific Exception Handling
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: Catch specific exceptions
|
||||||
|
def load_config(path: str) -> Config:
|
||||||
|
try:
|
||||||
|
with open(path) as f:
|
||||||
|
return Config.from_json(f.read())
|
||||||
|
except FileNotFoundError as e:
|
||||||
|
raise ConfigError(f"Config file not found: {path}") from e
|
||||||
|
except json.JSONDecodeError as e:
|
||||||
|
raise ConfigError(f"Invalid JSON in config: {path}") from e
|
||||||
|
|
||||||
|
# Bad: Bare except
|
||||||
|
def load_config(path: str) -> Config:
|
||||||
|
try:
|
||||||
|
with open(path) as f:
|
||||||
|
return Config.from_json(f.read())
|
||||||
|
except:
|
||||||
|
return None # Silent failure!
|
||||||
|
```
|
||||||
|
|
||||||
|
### Exception Chaining
|
||||||
|
|
||||||
|
```python
|
||||||
|
def process_data(data: str) -> Result:
|
||||||
|
try:
|
||||||
|
parsed = json.loads(data)
|
||||||
|
except json.JSONDecodeError as e:
|
||||||
|
# Chain exceptions to preserve the traceback
|
||||||
|
raise ValueError(f"Failed to parse data: {data}") from e
|
||||||
|
```
|
||||||
|
|
||||||
|
### Custom Exception Hierarchy
|
||||||
|
|
||||||
|
```python
|
||||||
|
class AppError(Exception):
|
||||||
|
"""Base exception for all application errors."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
class ValidationError(AppError):
|
||||||
|
"""Raised when input validation fails."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
class NotFoundError(AppError):
|
||||||
|
"""Raised when a requested resource is not found."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
def get_user(user_id: str) -> User:
|
||||||
|
user = db.find_user(user_id)
|
||||||
|
if not user:
|
||||||
|
raise NotFoundError(f"User not found: {user_id}")
|
||||||
|
return user
|
||||||
|
```
|
||||||
|
|
||||||
|
## Context Managers
|
||||||
|
|
||||||
|
### Resource Management
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: Using context managers
|
||||||
|
def process_file(path: str) -> str:
|
||||||
|
with open(path, 'r') as f:
|
||||||
|
return f.read()
|
||||||
|
|
||||||
|
# Bad: Manual resource management
|
||||||
|
def process_file(path: str) -> str:
|
||||||
|
f = open(path, 'r')
|
||||||
|
try:
|
||||||
|
return f.read()
|
||||||
|
finally:
|
||||||
|
f.close()
|
||||||
|
```
|
||||||
|
|
||||||
|
### Custom Context Managers
|
||||||
|
|
||||||
|
```python
|
||||||
|
from contextlib import contextmanager
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def timer(name: str):
|
||||||
|
"""Context manager to time a block of code."""
|
||||||
|
start = time.perf_counter()
|
||||||
|
yield
|
||||||
|
elapsed = time.perf_counter() - start
|
||||||
|
print(f"{name} took {elapsed:.4f} seconds")
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
with timer("data processing"):
|
||||||
|
process_large_dataset()
|
||||||
|
```
|
||||||
|
|
||||||
|
### Context Manager Classes
|
||||||
|
|
||||||
|
```python
|
||||||
|
class DatabaseTransaction:
|
||||||
|
def __init__(self, connection):
|
||||||
|
self.connection = connection
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
self.connection.begin_transaction()
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||||
|
if exc_type is None:
|
||||||
|
self.connection.commit()
|
||||||
|
else:
|
||||||
|
self.connection.rollback()
|
||||||
|
return False # Don't suppress exceptions
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
with DatabaseTransaction(conn):
|
||||||
|
user = conn.create_user(user_data)
|
||||||
|
conn.create_profile(user.id, profile_data)
|
||||||
|
```
|
||||||
|
|
||||||
|
## Comprehensions and Generators
|
||||||
|
|
||||||
|
### List Comprehensions
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: List comprehension for simple transformations
|
||||||
|
names = [user.name for user in users if user.is_active]
|
||||||
|
|
||||||
|
# Bad: Manual loop
|
||||||
|
names = []
|
||||||
|
for user in users:
|
||||||
|
if user.is_active:
|
||||||
|
names.append(user.name)
|
||||||
|
|
||||||
|
# Complex comprehensions should be expanded
|
||||||
|
# Bad: Too complex
|
||||||
|
result = [x * 2 for x in items if x > 0 if x % 2 == 0]
|
||||||
|
|
||||||
|
# Good: Use a generator function
|
||||||
|
def filter_and_transform(items: Iterable[int]) -> list[int]:
|
||||||
|
result = []
|
||||||
|
for x in items:
|
||||||
|
if x > 0 and x % 2 == 0:
|
||||||
|
result.append(x * 2)
|
||||||
|
return result
|
||||||
|
```
|
||||||
|
|
||||||
|
### Generator Expressions
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: Generator for lazy evaluation
|
||||||
|
total = sum(x * x for x in range(1_000_000))
|
||||||
|
|
||||||
|
# Bad: Creates large intermediate list
|
||||||
|
total = sum([x * x for x in range(1_000_000)])
|
||||||
|
```
|
||||||
|
|
||||||
|
### Generator Functions
|
||||||
|
|
||||||
|
```python
|
||||||
|
def read_large_file(path: str) -> Iterator[str]:
|
||||||
|
"""Read a large file line by line."""
|
||||||
|
with open(path) as f:
|
||||||
|
for line in f:
|
||||||
|
yield line.strip()
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
for line in read_large_file("huge.txt"):
|
||||||
|
process(line)
|
||||||
|
```
|
||||||
|
|
||||||
|
## Data Classes and Named Tuples
|
||||||
|
|
||||||
|
### Data Classes
|
||||||
|
|
||||||
|
```python
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class User:
|
||||||
|
"""User entity with automatic __init__, __repr__, and __eq__."""
|
||||||
|
id: str
|
||||||
|
name: str
|
||||||
|
email: str
|
||||||
|
created_at: datetime = field(default_factory=datetime.now)
|
||||||
|
is_active: bool = True
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
user = User(
|
||||||
|
id="123",
|
||||||
|
name="Alice",
|
||||||
|
email="alice@example.com"
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
|
### Data Classes with Validation
|
||||||
|
|
||||||
|
```python
|
||||||
|
@dataclass
|
||||||
|
class User:
|
||||||
|
email: str
|
||||||
|
age: int
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
# Validate email format
|
||||||
|
if "@" not in self.email:
|
||||||
|
raise ValueError(f"Invalid email: {self.email}")
|
||||||
|
# Validate age range
|
||||||
|
if self.age < 0 or self.age > 150:
|
||||||
|
raise ValueError(f"Invalid age: {self.age}")
|
||||||
|
```
|
||||||
|
|
||||||
|
### Named Tuples
|
||||||
|
|
||||||
|
```python
|
||||||
|
from typing import NamedTuple
|
||||||
|
|
||||||
|
class Point(NamedTuple):
|
||||||
|
"""Immutable 2D point."""
|
||||||
|
x: float
|
||||||
|
y: float
|
||||||
|
|
||||||
|
def distance(self, other: 'Point') -> float:
|
||||||
|
return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5
|
||||||
|
|
||||||
|
# Usage
|
||||||
|
p1 = Point(0, 0)
|
||||||
|
p2 = Point(3, 4)
|
||||||
|
print(p1.distance(p2)) # 5.0
|
||||||
|
```
|
||||||
|
|
||||||
|
## Decorators
|
||||||
|
|
||||||
|
### Function Decorators
|
||||||
|
|
||||||
|
```python
|
||||||
|
import functools
|
||||||
|
import time
|
||||||
|
|
||||||
|
def timer(func: Callable) -> Callable:
|
||||||
|
"""Decorator to time function execution."""
|
||||||
|
@functools.wraps(func)
|
||||||
|
def wrapper(*args, **kwargs):
|
||||||
|
start = time.perf_counter()
|
||||||
|
result = func(*args, **kwargs)
|
||||||
|
elapsed = time.perf_counter() - start
|
||||||
|
print(f"{func.__name__} took {elapsed:.4f}s")
|
||||||
|
return result
|
||||||
|
return wrapper
|
||||||
|
|
||||||
|
@timer
|
||||||
|
def slow_function():
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
# slow_function() prints: slow_function took 1.0012s
|
||||||
|
```
|
||||||
|
|
||||||
|
### Parameterized Decorators
|
||||||
|
|
||||||
|
```python
|
||||||
|
def repeat(times: int):
|
||||||
|
"""Decorator to repeat a function multiple times."""
|
||||||
|
def decorator(func: Callable) -> Callable:
|
||||||
|
@functools.wraps(func)
|
||||||
|
def wrapper(*args, **kwargs):
|
||||||
|
results = []
|
||||||
|
for _ in range(times):
|
||||||
|
results.append(func(*args, **kwargs))
|
||||||
|
return results
|
||||||
|
return wrapper
|
||||||
|
return decorator
|
||||||
|
|
||||||
|
@repeat(times=3)
|
||||||
|
def greet(name: str) -> str:
|
||||||
|
return f"Hello, {name}!"
|
||||||
|
|
||||||
|
# greet("Alice") returns ["Hello, Alice!", "Hello, Alice!", "Hello, Alice!"]
|
||||||
|
```
|
||||||
|
|
||||||
|
### Class-Based Decorators
|
||||||
|
|
||||||
|
```python
|
||||||
|
class CountCalls:
|
||||||
|
"""Decorator that counts how many times a function is called."""
|
||||||
|
def __init__(self, func: Callable):
|
||||||
|
functools.update_wrapper(self, func)
|
||||||
|
self.func = func
|
||||||
|
self.count = 0
|
||||||
|
|
||||||
|
def __call__(self, *args, **kwargs):
|
||||||
|
self.count += 1
|
||||||
|
print(f"{self.func.__name__} has been called {self.count} times")
|
||||||
|
return self.func(*args, **kwargs)
|
||||||
|
|
||||||
|
@CountCalls
|
||||||
|
def process():
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Each call to process() prints the call count
|
||||||
|
```
|
||||||
|
|
||||||
|
## Concurrency Patterns
|
||||||
|
|
||||||
|
### Threading for I/O-Bound Tasks
|
||||||
|
|
||||||
|
```python
|
||||||
|
import concurrent.futures
|
||||||
|
import threading
|
||||||
|
|
||||||
|
def fetch_url(url: str) -> str:
|
||||||
|
"""Fetch a URL (I/O-bound operation)."""
|
||||||
|
import urllib.request
|
||||||
|
with urllib.request.urlopen(url) as response:
|
||||||
|
return response.read().decode()
|
||||||
|
|
||||||
|
def fetch_all_urls(urls: list[str]) -> dict[str, str]:
|
||||||
|
"""Fetch multiple URLs concurrently using threads."""
|
||||||
|
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
|
||||||
|
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
|
||||||
|
results = {}
|
||||||
|
for future in concurrent.futures.as_completed(future_to_url):
|
||||||
|
url = future_to_url[future]
|
||||||
|
try:
|
||||||
|
results[url] = future.result()
|
||||||
|
except Exception as e:
|
||||||
|
results[url] = f"Error: {e}"
|
||||||
|
return results
|
||||||
|
```
|
||||||
|
|
||||||
|
### Multiprocessing for CPU-Bound Tasks
|
||||||
|
|
||||||
|
```python
|
||||||
|
def process_data(data: list[int]) -> int:
|
||||||
|
"""CPU-intensive computation."""
|
||||||
|
return sum(x ** 2 for x in data)
|
||||||
|
|
||||||
|
def process_all(datasets: list[list[int]]) -> list[int]:
|
||||||
|
"""Process multiple datasets using multiple processes."""
|
||||||
|
with concurrent.futures.ProcessPoolExecutor() as executor:
|
||||||
|
results = list(executor.map(process_data, datasets))
|
||||||
|
return results
|
||||||
|
```
|
||||||
|
|
||||||
|
### Async/Await for Concurrent I/O
|
||||||
|
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
async def fetch_async(url: str) -> str:
|
||||||
|
"""Fetch a URL asynchronously."""
|
||||||
|
import aiohttp
|
||||||
|
async with aiohttp.ClientSession() as session:
|
||||||
|
async with session.get(url) as response:
|
||||||
|
return await response.text()
|
||||||
|
|
||||||
|
async def fetch_all(urls: list[str]) -> dict[str, str]:
|
||||||
|
"""Fetch multiple URLs concurrently."""
|
||||||
|
tasks = [fetch_async(url) for url in urls]
|
||||||
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||||
|
return dict(zip(urls, results))
|
||||||
|
```
|
||||||
|
|
||||||
|
## Package Organization
|
||||||
|
|
||||||
|
### Standard Project Layout
|
||||||
|
|
||||||
|
```
|
||||||
|
myproject/
|
||||||
|
├── src/
|
||||||
|
│ └── mypackage/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ ├── main.py
|
||||||
|
│ ├── api/
|
||||||
|
│ │ ├── __init__.py
|
||||||
|
│ │ └── routes.py
|
||||||
|
│ ├── models/
|
||||||
|
│ │ ├── __init__.py
|
||||||
|
│ │ └── user.py
|
||||||
|
│ └── utils/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ └── helpers.py
|
||||||
|
├── tests/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ ├── conftest.py
|
||||||
|
│ ├── test_api.py
|
||||||
|
│ └── test_models.py
|
||||||
|
├── pyproject.toml
|
||||||
|
├── README.md
|
||||||
|
└── .gitignore
|
||||||
|
```
|
||||||
|
|
||||||
|
### Import Conventions
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Good: Import order - stdlib, third-party, local
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import requests
|
||||||
|
from fastapi import FastAPI
|
||||||
|
|
||||||
|
from mypackage.models import User
|
||||||
|
from mypackage.utils import format_name
|
||||||
|
|
||||||
|
# Good: Use isort for automatic import sorting
|
||||||
|
# pip install isort
|
||||||
|
```
|
||||||
|
|
||||||
|
### __init__.py for Package Exports
|
||||||
|
|
||||||
|
```python
|
||||||
|
# mypackage/__init__.py
|
||||||
|
"""mypackage - A sample Python package."""
|
||||||
|
|
||||||
|
__version__ = "1.0.0"
|
||||||
|
|
||||||
|
# Export main classes/functions at package level
|
||||||
|
from mypackage.models import User, Post
|
||||||
|
from mypackage.utils import format_name
|
||||||
|
|
||||||
|
__all__ = ["User", "Post", "format_name"]
|
||||||
|
```
|
||||||
|
|
||||||
|
## Memory and Performance
|
||||||
|
|
||||||
|
### Using __slots__ for Memory Efficiency
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Bad: Regular class uses __dict__ (more memory)
|
||||||
|
class Point:
|
||||||
|
def __init__(self, x: float, y: float):
|
||||||
|
self.x = x
|
||||||
|
self.y = y
|
||||||
|
|
||||||
|
# Good: __slots__ reduces memory usage
|
||||||
|
class Point:
|
||||||
|
__slots__ = ['x', 'y']
|
||||||
|
|
||||||
|
def __init__(self, x: float, y: float):
|
||||||
|
self.x = x
|
||||||
|
self.y = y
|
||||||
|
```
|
||||||
|
|
||||||
|
### Generator for Large Data
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Bad: Returns full list in memory
|
||||||
|
def read_lines(path: str) -> list[str]:
|
||||||
|
with open(path) as f:
|
||||||
|
return [line.strip() for line in f]
|
||||||
|
|
||||||
|
# Good: Yields lines one at a time
|
||||||
|
def read_lines(path: str) -> Iterator[str]:
|
||||||
|
with open(path) as f:
|
||||||
|
for line in f:
|
||||||
|
yield line.strip()
|
||||||
|
```
|
||||||
|
|
||||||
|
### Avoid String Concatenation in Loops
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Bad: O(n²) due to string immutability
|
||||||
|
result = ""
|
||||||
|
for item in items:
|
||||||
|
result += str(item)
|
||||||
|
|
||||||
|
# Good: O(n) using join
|
||||||
|
result = "".join(str(item) for item in items)
|
||||||
|
|
||||||
|
# Good: Using StringIO for building
|
||||||
|
from io import StringIO
|
||||||
|
|
||||||
|
buffer = StringIO()
|
||||||
|
for item in items:
|
||||||
|
buffer.write(str(item))
|
||||||
|
result = buffer.getvalue()
|
||||||
|
```
|
||||||
|
|
||||||
|
## Python Tooling Integration
|
||||||
|
|
||||||
|
### Essential Commands
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Code formatting
|
||||||
|
black .
|
||||||
|
isort .
|
||||||
|
|
||||||
|
# Linting
|
||||||
|
ruff check .
|
||||||
|
pylint mypackage/
|
||||||
|
|
||||||
|
# Type checking
|
||||||
|
mypy .
|
||||||
|
|
||||||
|
# Testing
|
||||||
|
pytest --cov=mypackage --cov-report=html
|
||||||
|
|
||||||
|
# Security scanning
|
||||||
|
bandit -r .
|
||||||
|
|
||||||
|
# Dependency management
|
||||||
|
pip-audit
|
||||||
|
safety check
|
||||||
|
```
|
||||||
|
|
||||||
|
### pyproject.toml Configuration
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[project]
|
||||||
|
name = "mypackage"
|
||||||
|
version = "1.0.0"
|
||||||
|
requires-python = ">=3.9"
|
||||||
|
dependencies = [
|
||||||
|
"requests>=2.31.0",
|
||||||
|
"pydantic>=2.0.0",
|
||||||
|
]
|
||||||
|
|
||||||
|
[project.optional-dependencies]
|
||||||
|
dev = [
|
||||||
|
"pytest>=7.4.0",
|
||||||
|
"pytest-cov>=4.1.0",
|
||||||
|
"black>=23.0.0",
|
||||||
|
"ruff>=0.1.0",
|
||||||
|
"mypy>=1.5.0",
|
||||||
|
]
|
||||||
|
|
||||||
|
[tool.black]
|
||||||
|
line-length = 88
|
||||||
|
target-version = ['py39']
|
||||||
|
|
||||||
|
[tool.ruff]
|
||||||
|
line-length = 88
|
||||||
|
select = ["E", "F", "I", "N", "W"]
|
||||||
|
|
||||||
|
[tool.mypy]
|
||||||
|
python_version = "3.9"
|
||||||
|
warn_return_any = true
|
||||||
|
warn_unused_configs = true
|
||||||
|
disallow_untyped_defs = true
|
||||||
|
|
||||||
|
[tool.pytest.ini_options]
|
||||||
|
testpaths = ["tests"]
|
||||||
|
addopts = "--cov=mypackage --cov-report=term-missing"
|
||||||
|
```
|
||||||
|
|
||||||
|
## Quick Reference: Python Idioms
|
||||||
|
|
||||||
|
| Idiom | Description |
|
||||||
|
|-------|-------------|
|
||||||
|
| EAFP | Easier to Ask Forgiveness than Permission |
|
||||||
|
| Context managers | Use `with` for resource management |
|
||||||
|
| List comprehensions | For simple transformations |
|
||||||
|
| Generators | For lazy evaluation and large datasets |
|
||||||
|
| Type hints | Annotate function signatures |
|
||||||
|
| Dataclasses | For data containers with auto-generated methods |
|
||||||
|
| `__slots__` | For memory optimization |
|
||||||
|
| f-strings | For string formatting (Python 3.6+) |
|
||||||
|
| `pathlib.Path` | For path operations (Python 3.4+) |
|
||||||
|
| `enumerate` | For index-element pairs in loops |
|
||||||
|
|
||||||
|
## Anti-Patterns to Avoid
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Bad: Mutable default arguments
|
||||||
|
def append_to(item, items=[]):
|
||||||
|
items.append(item)
|
||||||
|
return items
|
||||||
|
|
||||||
|
# Good: Use None and create new list
|
||||||
|
def append_to(item, items=None):
|
||||||
|
if items is None:
|
||||||
|
items = []
|
||||||
|
items.append(item)
|
||||||
|
return items
|
||||||
|
|
||||||
|
# Bad: Checking type with type()
|
||||||
|
if type(obj) == list:
|
||||||
|
process(obj)
|
||||||
|
|
||||||
|
# Good: Use isinstance
|
||||||
|
if isinstance(obj, list):
|
||||||
|
process(obj)
|
||||||
|
|
||||||
|
# Bad: Comparing to None with ==
|
||||||
|
if value == None:
|
||||||
|
process()
|
||||||
|
|
||||||
|
# Good: Use is
|
||||||
|
if value is None:
|
||||||
|
process()
|
||||||
|
|
||||||
|
# Bad: from module import *
|
||||||
|
from os.path import *
|
||||||
|
|
||||||
|
# Good: Explicit imports
|
||||||
|
from os.path import join, exists
|
||||||
|
|
||||||
|
# Bad: Bare except
|
||||||
|
try:
|
||||||
|
risky_operation()
|
||||||
|
except:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Good: Specific exception
|
||||||
|
try:
|
||||||
|
risky_operation()
|
||||||
|
except SpecificError as e:
|
||||||
|
logger.error(f"Operation failed: {e}")
|
||||||
|
```
|
||||||
|
|
||||||
|
__Remember__: Python code should be readable, explicit, and follow the principle of least surprise. When in doubt, prioritize clarity over cleverness.
|
||||||
@@ -0,0 +1,622 @@
|
|||||||
|
---
|
||||||
|
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.
|
||||||
|
---
|
||||||
|
|
||||||
|
# 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)
|
||||||
|
```
|
||||||
|
|
||||||
|
For advanced patterns including async testing, monkeypatching, temporary files, conftest setup, property-based testing, database testing, CI/CD integration, and configuration files, see [references/advanced-patterns.md](references/advanced-patterns.md)
|
||||||
|
|
||||||
|
## Test Design Principles
|
||||||
|
|
||||||
|
### One Behavior Per Test
|
||||||
|
|
||||||
|
Each test should verify exactly one behavior. This makes failures easy to diagnose and tests easy to maintain.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# BAD - testing multiple behaviors
|
||||||
|
def test_user_service():
|
||||||
|
user = service.create_user(data)
|
||||||
|
assert user.id is not None
|
||||||
|
assert user.email == data["email"]
|
||||||
|
updated = service.update_user(user.id, {"name": "New"})
|
||||||
|
assert updated.name == "New"
|
||||||
|
|
||||||
|
# GOOD - focused tests
|
||||||
|
def test_create_user_assigns_id():
|
||||||
|
user = service.create_user(data)
|
||||||
|
assert user.id is not None
|
||||||
|
|
||||||
|
def test_create_user_stores_email():
|
||||||
|
user = service.create_user(data)
|
||||||
|
assert user.email == data["email"]
|
||||||
|
|
||||||
|
def test_update_user_changes_name():
|
||||||
|
user = service.create_user(data)
|
||||||
|
updated = service.update_user(user.id, {"name": "New"})
|
||||||
|
assert updated.name == "New"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Test Error Paths
|
||||||
|
|
||||||
|
Always test failure cases, not just happy paths.
|
||||||
|
|
||||||
|
```python
|
||||||
|
def test_get_user_raises_not_found():
|
||||||
|
with pytest.raises(UserNotFoundError) as exc_info:
|
||||||
|
service.get_user("nonexistent-id")
|
||||||
|
|
||||||
|
assert "nonexistent-id" in str(exc_info.value)
|
||||||
|
|
||||||
|
def test_create_user_rejects_invalid_email():
|
||||||
|
with pytest.raises(ValueError, match="Invalid email format"):
|
||||||
|
service.create_user({"email": "not-an-email"})
|
||||||
|
```
|
||||||
|
|
||||||
|
## 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 Convention
|
||||||
|
|
||||||
|
A common pattern: `test_<unit>_<scenario>_<expected_outcome>`. Adapt to your team's preferences.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Pattern: test_<unit>_<scenario>_<expected>
|
||||||
|
def test_create_user_with_valid_data_returns_user():
|
||||||
|
...
|
||||||
|
|
||||||
|
def test_create_user_with_duplicate_email_raises_conflict():
|
||||||
|
...
|
||||||
|
|
||||||
|
def test_get_user_with_unknown_id_returns_none():
|
||||||
|
...
|
||||||
|
|
||||||
|
# Good test names - clear and descriptive
|
||||||
|
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 - avoid these
|
||||||
|
def test_1(): # Not descriptive
|
||||||
|
pass
|
||||||
|
|
||||||
|
def test_user(): # Too vague
|
||||||
|
pass
|
||||||
|
|
||||||
|
def test_function(): # Doesn't explain what's tested
|
||||||
|
pass
|
||||||
|
```
|
||||||
|
|
||||||
|
### Testing Retry Behavior
|
||||||
|
|
||||||
|
Verify that retry logic works correctly using mock side effects.
|
||||||
|
|
||||||
|
```python
|
||||||
|
from unittest.mock import Mock
|
||||||
|
|
||||||
|
def test_retries_on_transient_error():
|
||||||
|
"""Test that service retries on transient failures."""
|
||||||
|
client = Mock()
|
||||||
|
# Fail twice, then succeed
|
||||||
|
client.request.side_effect = [
|
||||||
|
ConnectionError("Failed"),
|
||||||
|
ConnectionError("Failed"),
|
||||||
|
{"status": "ok"},
|
||||||
|
]
|
||||||
|
|
||||||
|
service = ServiceWithRetry(client, max_retries=3)
|
||||||
|
result = service.fetch()
|
||||||
|
|
||||||
|
assert result == {"status": "ok"}
|
||||||
|
assert client.request.call_count == 3
|
||||||
|
|
||||||
|
def test_gives_up_after_max_retries():
|
||||||
|
"""Test that service stops retrying after max attempts."""
|
||||||
|
client = Mock()
|
||||||
|
client.request.side_effect = ConnectionError("Failed")
|
||||||
|
|
||||||
|
service = ServiceWithRetry(client, max_retries=3)
|
||||||
|
|
||||||
|
with pytest.raises(ConnectionError):
|
||||||
|
service.fetch()
|
||||||
|
|
||||||
|
assert client.request.call_count == 3
|
||||||
|
|
||||||
|
def test_does_not_retry_on_permanent_error():
|
||||||
|
"""Test that permanent errors are not retried."""
|
||||||
|
client = Mock()
|
||||||
|
client.request.side_effect = ValueError("Invalid input")
|
||||||
|
|
||||||
|
service = ServiceWithRetry(client, max_retries=3)
|
||||||
|
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
service.fetch()
|
||||||
|
|
||||||
|
# Only called once - no retry for ValueError
|
||||||
|
assert client.request.call_count == 1
|
||||||
|
```
|
||||||
|
|
||||||
|
### Mocking Time with Freezegun
|
||||||
|
|
||||||
|
Use freezegun to control time in tests for predictable time-dependent behavior.
|
||||||
|
|
||||||
|
```python
|
||||||
|
from freezegun import freeze_time
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
@freeze_time("2026-01-15 10:00:00")
|
||||||
|
def test_token_expiry():
|
||||||
|
"""Test token expires at correct time."""
|
||||||
|
token = create_token(expires_in_seconds=3600)
|
||||||
|
assert token.expires_at == datetime(2026, 1, 15, 11, 0, 0)
|
||||||
|
|
||||||
|
@freeze_time("2026-01-15 10:00:00")
|
||||||
|
def test_is_expired_returns_false_before_expiry():
|
||||||
|
"""Test token is not expired when within validity period."""
|
||||||
|
token = create_token(expires_in_seconds=3600)
|
||||||
|
assert not token.is_expired()
|
||||||
|
|
||||||
|
@freeze_time("2026-01-15 12:00:00")
|
||||||
|
def test_is_expired_returns_true_after_expiry():
|
||||||
|
"""Test token is expired after validity period."""
|
||||||
|
token = Token(expires_at=datetime(2026, 1, 15, 11, 30, 0))
|
||||||
|
assert token.is_expired()
|
||||||
|
|
||||||
|
def test_with_time_travel():
|
||||||
|
"""Test behavior across time using freeze_time context."""
|
||||||
|
with freeze_time("2026-01-01") as frozen_time:
|
||||||
|
item = create_item()
|
||||||
|
assert item.created_at == datetime(2026, 1, 1)
|
||||||
|
|
||||||
|
# Move forward in time
|
||||||
|
frozen_time.move_to("2026-01-15")
|
||||||
|
assert item.age_days == 14
|
||||||
|
```
|
||||||
|
|
||||||
|
### 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/
|
||||||
|
```
|
||||||
|
|
||||||
|
For advanced patterns (async testing, monkeypatching, property-based testing, database testing, CI/CD integration, and configuration), see [references/advanced-patterns.md](references/advanced-patterns.md)
|
||||||
@@ -0,0 +1,411 @@
|
|||||||
|
# Python Testing Patterns — Advanced Reference
|
||||||
|
|
||||||
|
Advanced testing patterns including async code, monkeypatching, temporary files, conftest setup, property-based testing, database testing, CI/CD integration, and configuration.
|
||||||
|
|
||||||
|
## 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 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",
|
||||||
|
]
|
||||||
|
```
|
||||||
@@ -22,8 +22,7 @@ jobs:
|
|||||||
- name: Extract package version
|
- name: Extract package version
|
||||||
id: pkg_version
|
id: pkg_version
|
||||||
run: |
|
run: |
|
||||||
PKGFILE=$(basename external/*.pkg.tar.zst)
|
PKGVERSION=$(pacman -Qp external/*.pkg.tar.zst | awk '{print $2}')
|
||||||
PKGVERSION=$(echo "$PKGFILE" | sed 's/^rscli-git-\([^-]*-[^-]*\)-any\.pkg\.tar\.zst$/\1/')
|
|
||||||
echo "PKGVERSION=$PKGVERSION" >> $GITHUB_ENV
|
echo "PKGVERSION=$PKGVERSION" >> $GITHUB_ENV
|
||||||
echo "version=$PKGVERSION" >> $GITHUB_OUTPUT
|
echo "version=$PKGVERSION" >> $GITHUB_OUTPUT
|
||||||
- name: Upload Artifact
|
- name: Upload Artifact
|
||||||
@@ -45,7 +44,7 @@ jobs:
|
|||||||
path: ./dist
|
path: ./dist
|
||||||
- name: Create Git Tag
|
- name: Create Git Tag
|
||||||
run: |
|
run: |
|
||||||
VERSION=$(ls dist/*.pkg.tar.zst | sed 's/^rscli-git-\(r[^-]*-[^-]*\)-.*$/\1/')
|
VERSION="${{ needs.build_arch.outputs.version }}"
|
||||||
git tag "$VERSION"
|
git tag "$VERSION"
|
||||||
git push origin "$VERSION"
|
git push origin "$VERSION"
|
||||||
echo "VERSION=$VERSION" >> $GITHUB_ENV
|
echo "VERSION=$VERSION" >> $GITHUB_ENV
|
||||||
|
|||||||
@@ -0,0 +1,37 @@
|
|||||||
|
# AGENTS.md
|
||||||
|
|
||||||
|
## Project Overview
|
||||||
|
- **Project**: rscli - Python CLI for a social network API
|
||||||
|
- **Entry point**: `rcli` command (defined in pyproject.toml as `rscli.main:app`)
|
||||||
|
- **Python version**: >= 3.14
|
||||||
|
- **Dependencies**: prettytable, requests, typer
|
||||||
|
|
||||||
|
## Running the CLI
|
||||||
|
```bash
|
||||||
|
rcli --help # Show commands
|
||||||
|
rcli version # Show version
|
||||||
|
rcli login # Login to API
|
||||||
|
rcli api_uri <url> # Set API URI
|
||||||
|
```
|
||||||
|
|
||||||
|
## Configuration
|
||||||
|
- Config stored at: `~/.config/rs-cli/config.json`
|
||||||
|
- Contains: `token`, `API_URI`
|
||||||
|
|
||||||
|
## Development
|
||||||
|
- Install locally: `pip install -e .` or use `uv pip install -e .`
|
||||||
|
- Project uses **uv** for dependency management (see uv.lock)
|
||||||
|
|
||||||
|
## CI/Build
|
||||||
|
- Gitea workflows in `.gitea/workflows/build.yml`
|
||||||
|
- Builds Arch Linux package (.pkg.tar.zst)
|
||||||
|
- Publishes releases to Gitea
|
||||||
|
|
||||||
|
## Available Skills
|
||||||
|
- python-patterns
|
||||||
|
- python-testing-patterns
|
||||||
|
- python-executor
|
||||||
|
|
||||||
|
## Notes
|
||||||
|
- No test suite currently exists in the repo
|
||||||
|
- Config class supports dot-notation paths (e.g., `config.get("token")`)
|
||||||
@@ -0,0 +1,20 @@
|
|||||||
|
{
|
||||||
|
"version": 1,
|
||||||
|
"skills": {
|
||||||
|
"python-executor": {
|
||||||
|
"source": "inferen-sh/skills",
|
||||||
|
"sourceType": "autoskills-registry",
|
||||||
|
"computedHash": "bd2d874c27788964fadf03c0840049de0409407f06f883ee90886164cb22ef69"
|
||||||
|
},
|
||||||
|
"python-patterns": {
|
||||||
|
"source": "affaan-m/everything-claude-code",
|
||||||
|
"sourceType": "autoskills-registry",
|
||||||
|
"computedHash": "5c344cc64b19a9a7aecb9b5fa977175acd8e55a169f86901d28eb57c637869b4"
|
||||||
|
},
|
||||||
|
"python-testing-patterns": {
|
||||||
|
"source": "wshobson/agents",
|
||||||
|
"sourceType": "autoskills-registry",
|
||||||
|
"computedHash": "07b87d62993c0b6159a91d18fc8723b7f4c13d5000c0984266c207493ce641ff"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user