一键导入
pytest
Write, structure, and maintain high-quality pytest test suites following official pytest documentation best practices.
用 Codex 或 Claude 帮你安装 复制这段 Prompt,粘贴到 Codex、Claude 或其他助手里,让它检查 Skill 页面并帮你完成安装。
菜单
Write, structure, and maintain high-quality pytest test suites following official pytest documentation best practices.
用 Codex 或 Claude 帮你安装 复制这段 Prompt,粘贴到 Codex、Claude 或其他助手里,让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
Creates, configures, and updates Databricks Lakeflow Spark Declarative Pipelines (SDP/LDP) using serverless compute. Handles data ingestion with streaming tables, materialized views, CDC, SCD Type 2, and Auto Loader ingestion patterns. Use when building data pipelines, working with Delta Live Tables, ingesting streaming data, implementing change data capture, or when the user mentions SDP, LDP, DLT, Lakeflow pipelines, streaming tables, or bronze/silver/gold medallion architectures.
Write clean, testable, and maintainable Python code using functional programming principles — pure functions, immutability, higher-order functions, and composable pipelines.
Write idiomatic, readable Python code following PEP 8 – the official Python style guide.
Write idiomatic, performant PySpark code following the Palantir PySpark Style Guide.
| name | pytest |
| description | Write, structure, and maintain high-quality pytest test suites following official pytest documentation best practices. |
When writing or reviewing pytest tests, follow the rules and patterns below, derived from the pytest documentation.
Every test should be decomposed into four clear phases:
assert statements.yield fixtures so cleanup is automatic).def test_email_received(sending_user, receiving_user):
# Act
sending_user.send_email(Email("Hi", "Body"), receiving_user)
# Assert
assert Email("Hi", "Body") in receiving_user.inbox
@pytest.fixtureimport pytest
@pytest.fixture
def user():
return User(name="Alice")
yield for teardown (preferred over addfinalizer)@pytest.fixture
def db_connection():
conn = create_connection()
yield conn # test runs here
conn.close() # teardown always executes
| Scope | Destroyed after |
|---|---|
function (default) | each test function |
class | last test in the class |
module | last test in the module |
package | last test in the package |
session | end of the entire test session |
Use scope="session" or scope="module" for expensive resources (SparkSession, DB connections):
@pytest.fixture(scope="session")
def spark():
session = SparkSession.builder.master("local[1]").getOrCreate()
yield session
session.stop()
conftest.pyPlace fixtures used across multiple test files in conftest.py. Pytest discovers them automatically — no import needed.
tests/
conftest.py ← shared fixtures live here
test_users.py
test_orders.py
autouse=True for setup that applies to every test in scope@pytest.fixture(autouse=True)
def clean_tmp_dir(tmp_path):
yield
shutil.rmtree(tmp_path, ignore_errors=True)
tmp_path over manual temp directoriesdef test_writes_file(tmp_path):
output = tmp_path / "result.csv"
write_data(output)
assert output.exists()
monkeypatch for environment and attribute patchingdef test_reads_env_var(monkeypatch):
monkeypatch.setenv("API_KEY", "test-key")
assert get_api_key() == "test-key"
caplog for log assertionsdef test_logs_warning(caplog):
import logging
with caplog.at_level(logging.WARNING):
process_data([])
assert "no rows" in caplog.text
capsys / capfd for stdout/stderr assertionsdef test_prints_header(capsys):
print_report_header()
captured = capsys.readouterr()
assert "Report" in captured.out
@pytest.mark.parametrize to eliminate duplicate test functionsWhen multiple test functions differ only in their inputs and expected outputs, collapse them into one parametrized function:
# bad – three functions testing the same logic
def test_add_two_and_three():
assert add(2, 3) == 5
def test_add_zero_and_five():
assert add(0, 5) == 5
def test_add_negative():
assert add(-1, 1) == 0
# good – one parametrized function
@pytest.mark.parametrize(("a", "b", "expected"), [
(2, 3, 5),
(0, 5, 5),
(-1, 1, 0),
])
def test_add(a, b, expected):
assert add(a, b) == expected
pytest.param for readable IDs and per-case marks@pytest.mark.parametrize(("value", "expected"), [
pytest.param("hello", True, id="valid-string"),
pytest.param("", False, id="empty-string"),
pytest.param(None, False, id="none-value", marks=pytest.mark.xfail),
])
def test_is_valid(value, expected):
assert is_valid(value) == expected
@pytest.mark.parametrize("fmt", ["csv", "parquet", "json"])
@pytest.mark.parametrize("compression", [None, "gzip", "snappy"])
def test_write_format(fmt, compression):
... # runs 9 combinations
indirect=True to pass parameters through a fixture@pytest.fixture
def db(request):
return connect(request.param)
@pytest.mark.parametrize("db", ["sqlite", "postgres"], indirect=True)
def test_insert(db):
db.execute("INSERT INTO t VALUES (1)")
pyproject.toml to avoid warnings[tool.pytest.ini_options]
markers = [
"spark: tests requiring a live SparkSession",
"integration: slow end-to-end tests",
]
pytestmarkpytestmark = pytest.mark.spark
@pytest.mark.skip(reason="upstream API not available in CI")
def test_external_call(): ...
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX paths only")
def test_posix_paths(): ...
@pytest.mark.xfail(reason="known bug #123", strict=True)
def test_known_failure(): ...
assert — pytest rewrites it for rich failure messagesassert result == expected # pytest shows both values on failure
assert "error" in message
assert df.count() > 0
pytest.approx for floating-point comparisonsassert 0.1 + 0.2 == pytest.approx(0.3)
assert computed_revenue == pytest.approx(expected_revenue, rel=1e-3)
pytest.raises as a context manager for exception testingwith pytest.raises(ValueError, match="must be positive"):
process(value=-1)
Always include a match= pattern — it guards against catching the wrong ValueError.
ExceptionInfo when you need the exception objectwith pytest.raises(RuntimeError) as exc_info:
risky_operation()
assert "timeout" in str(exc_info.value)
assert exc_info.type is RuntimeError
src layout)pyproject.toml
src/
mypackage/
module.py
tests/
conftest.py
test_module.py
Configure pythonpath in pyproject.toml when using src layout:
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
test_*.py or *_test.pytest_*Test* (no __init__ method)pyproject.toml)[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
addopts = ["-ra", "--strict-markers", "--strict-config"]
markers = [
"spark: tests requiring a live SparkSession",
"slow: long-running integration tests",
]
Key options:
--strict-markers: turns undeclared marker usage into an error.--strict-config: turns config warnings into errors.-ra: show short summary of all non-passing tests at the end.pytest-covInstall pytest-cov:
pip install pytest-cov
Run with an annotated coverage report:
pytest --cov --cov-report=annotate:cov_annotate
For a specific module:
pytest --cov=mypackage --cov-report=annotate:cov_annotate tests/
Inspect cov_annotate/: lines starting with ! are not covered.
Add tests until all ! lines are covered.
Enforce a minimum threshold in CI:
pytest --cov=mypackage --cov-fail-under=90
pytest.skip()def test_gpu_feature():
if not torch.cuda.is_available():
pytest.skip("no GPU")
...
pytest.importorskip for optional dependenciesnp = pytest.importorskip("numpy")
def test_array_sum():
assert np.sum([1, 2, 3]) == 6
Prefer unittest.mock (stdlib) or pytest-mock (mocker fixture).
from unittest.mock import MagicMock, patch
def test_calls_api(monkeypatch):
mock_client = MagicMock()
monkeypatch.setattr("mymodule.api_client", mock_client)
call_service()
mock_client.post.assert_called_once()
Use patch.object for patching attributes on live modules:
with patch.object(module, "expensive_func", return_value=42):
assert compute() == 42
Pure functions (no side effects, deterministic output) are the easiest to test.
Always test them with @pytest.mark.parametrize:
@pytest.mark.parametrize(("row_type", "num_rows", "expected_len"), [
("users", 5, 5),
("products", 0, 0),
("orders", 10, 10),
])
def test_generate_rows_count(row_type, num_rows, expected_len):
kwargs = {"num_users": 20, "num_products": 20} if row_type == "orders" else {}
rows = generate_list_of_rows(row_type, num_rows=num_rows, **kwargs)
assert len(rows) == expected_len
| Anti-pattern | What to do instead |
|---|---|
| Multiple test functions that differ only in data | Use @pytest.mark.parametrize |
setup/teardown methods (xUnit style) | Use yield fixtures |
import inside test bodies | Import at module level or in conftest.py |
print() for debugging | Use capfd/capsys, or -s flag |
| Hardcoded absolute paths | Use tmp_path fixture or pathlib.Path |
Bare pytest.raises(Exception) | Always include match= pattern |
Session-level fixtures with scope="function" for heavy setup | Use scope="session" or scope="module" |
Asserting True / False directly on method calls | Assert the actual value: assert compute() == expected |
| Tests that depend on execution order | Make each test independent; use fixtures for shared state |
Catching broad exceptions in pytest.raises | Specify the exact exception type and a match= regex |
| Flag | Purpose |
|---|---|
-v | verbose: show each test name |
-ra | show summary of all non-passing results |
-k "expr" | run tests whose name matches expression |
-m "marker" | run tests with a specific marker |
--co / --collect-only | list collected tests without running |
--fixtures | list available fixtures |
-x / --exitfirst | stop after first failure |
--lf / --last-failed | re-run only previously failed tests |
-s | disable output capture (see print output) |
--tb=short | shorter traceback format |
--strict-markers | error on undeclared markers |