Testing

uv add --dev pytest pytest-cov pytest-mock
uv run pytest                  # all tests
uv run pytest tests/test_cart.py::TestAdd -q
uvx --with pytest-cov pytest   # no project; throwaway env

import pytest
 
from shop.cart import Cart
 
 
def test_total_sums_lines() -> None:
    cart = Cart()
    cart.add("tea", price=4.5, qty=2)
    assert cart.total() == 9.0
 
 
class TestAdd:  # no __init__, so pytest collects it
    def test_rejects_zero_qty(self) -> None:
        with pytest.raises(ValueError, match="positive"):
            Cart().add("tea", 4.5, qty=0)

>       assert got == {"name": "ada", "roles": ["dev"]}
E       AssertionError: assert {'name': 'ada...s': ['admin']} == {'name': 'ada...les': ['dev']}
E         Omitting 1 identical items, use -vv to show
E         Differing items:
E         {'roles': ['admin']} != {'roles': ['dev']}

import warnings
 
import pytest
 
from shop.net import parse_port
 
 
def test_raises_with_message() -> None:
    with pytest.raises(ValueError, match=r"range: \d+"):
        parse_port("70000")
 
 
def test_inspect_exception() -> None:
    with pytest.raises(ValueError) as exc_info:
        parse_port("http")
    assert "invalid literal" in str(exc_info.value)
    assert exc_info.type is ValueError
 
 
def test_floats() -> None:
    assert 0.1 + 0.2 == pytest.approx(0.3)
    assert [0.1 + 0.2, 1.0] == pytest.approx([0.3, 1.0])
    assert 103 == pytest.approx(100, rel=0.05)  # within 5 %
 
 
def test_warning() -> None:
    with pytest.warns(DeprecationWarning, match="v1"):
        warnings.warn("v1 is old", DeprecationWarning, 2)
 
 
def test_exception_group() -> None:
    with pytest.RaisesGroup(ValueError, TypeError):
        errors = [ValueError(), TypeError()]
        raise ExceptionGroup("batch", errors)

import sqlite3
from collections.abc import Iterator
from pathlib import Path
 
import pytest
 
 
@pytest.fixture(scope="session")
def db_path(
    tmp_path_factory: pytest.TempPathFactory,
) -> Path:
    return tmp_path_factory.mktemp("db") / "test.sqlite3"
 
 
@pytest.fixture
def conn(db_path: Path) -> Iterator[sqlite3.Connection]:
    conn = sqlite3.connect(db_path)
    yield conn  # the test runs here
    conn.close()  # teardown runs even if the test failed
 
 
@pytest.fixture(autouse=True)
def _utc(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("TZ", "UTC")  # every test, no request

from dataclasses import dataclass
from typing import Protocol
 
import pytest
 
 
@dataclass(frozen=True)
class User:
    name: str
    admin: bool = False
 
 
class MakeUser(Protocol):
    def __call__(
        self, name: str = ..., *, admin: bool = ...
    ) -> User: ...
 
 
@pytest.fixture
def make_user() -> MakeUser:
    def make(
        name: str = "ada", *, admin: bool = False
    ) -> User:
        return User(name, admin)
 
    return make
 
 
def test_factory(make_user: MakeUser) -> None:
    admin = make_user("root", admin=True)
    assert admin.admin and not make_user().admin

import logging
import os
 
import pytest
 
log = logging.getLogger("shop")
 
 
def test_env(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("API_URL", "http://test")
    monkeypatch.delenv("HOME", raising=False)
    assert os.environ["API_URL"] == "http://test"
 
 
def test_stdout(capsys: pytest.CaptureFixture[str]) -> None:
    print("hello")
    out, err = capsys.readouterr()
    assert out == "hello\n" and err == ""
 
 
def test_logs(caplog: pytest.LogCaptureFixture) -> None:
    with caplog.at_level(logging.WARNING, logger="shop"):
        log.warning("low stock: %s", "tea")
    assert caplog.messages == ["low stock: tea"]
 
 
def test_subtests(subtests: pytest.Subtests) -> None:
    for n in (2, 4, 6):
        with subtests.test("even", n=n):
            assert n % 2 == 0  # each failure reported

from dataclasses import dataclass
 
import pytest
 
from shop.net import parse_port
 
 
@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("80", 80),
        ("65535", 65_535),
        pytest.param(" 443 ", 443, id="whitespace"),
        pytest.param(
            "0x50",
            80,
            marks=pytest.mark.xfail(raises=ValueError),
        ),
    ],
)
def test_parse(raw: str, expected: int) -> None:
    assert parse_port(raw) == expected
 
 
@pytest.mark.parametrize("x", [1, 2])
@pytest.mark.parametrize("y", [10, 20])
def test_grid(x: int, y: int) -> None:  # 4 cases
    assert x < y
 
 
@dataclass(frozen=True)
class Conn:
    driver: str
 
 
@pytest.fixture
def conn(request: pytest.FixtureRequest) -> Conn:
    return Conn(driver=request.param)  # from parametrize
 
 
@pytest.mark.parametrize(
    "conn", ["sqlite", "postgres"], indirect=True
)
def test_indirect(conn: Conn) -> None:
    assert conn.driver in {"sqlite", "postgres"}
 
 
@pytest.fixture(params=["utf-8", "latin-1"], ids=str.upper)
def encoding(request: pytest.FixtureRequest) -> str:
    return str(request.param)  # users run once per param
 
 
def test_encode(encoding: str) -> None:
    assert "cafe".encode(encoding) == b"cafe"

test_params.py::test_parse[80-80]
test_params.py::test_parse[whitespace]
test_params.py::test_grid[10-1]
test_params.py::test_indirect[sqlite]
test_params.py::test_encode[UTF-8]

import sys
 
import pytest
 
 
@pytest.mark.skip(reason="flaky upstream")
def test_skipped() -> None: ...
 
 
@pytest.mark.skipif(sys.platform == "win32", reason="posix")
def test_posix_only() -> None: ...
 
 
@pytest.mark.xfail(reason="bug #42", strict=True)
def test_known_bug() -> None:
    assert int("1e3") == 1000
 
 
@pytest.mark.slow  # custom: register it in config
def test_big_import() -> None:
    np = pytest.importorskip("numpy")  # skip if missing
    assert np.zeros(3).sum() == 0
 
 
pytestmark = pytest.mark.integration  # whole module

uv run pytest -m "not slow"
uv run pytest -m "integration and not slow"
uv run pytest -k "parse and not whitespace"  # name match

[tool.pytest]
minversion = "9.0"
testpaths = ["tests"]
addopts = ["-ra", "--import-mode=importlib"]
strict = true  # strict config, markers, xfail, param ids
markers = [
  "slow: takes more than a second",
  "integration: needs real services",
]
filterwarnings = [
  "error",  # warnings fail tests
  "ignore::DeprecationWarning:somelib.*",
]
log_level = "INFO"  # what caplog captures
asyncio_mode = "auto"  # pytest-asyncio, see below

from datetime import UTC, datetime
from unittest.mock import AsyncMock, create_autospec, patch
 
import pytest
from pytest_mock import MockerFixture
 
from shop import report
from shop.notify import Mailer, welcome
 
FIXED = datetime(2026, 9, 25, tzinfo=UTC)
 
 
def test_patch_where_used() -> None:
    with patch("shop.report.now", return_value=FIXED):
        assert report.stamp() == "2026-09-25"
 
 
def test_autospec_checks_signature() -> None:
    with patch("shop.report.now", autospec=True) as now:
        now.return_value = FIXED
        report.stamp()
        with pytest.raises(TypeError):
            now("unexpected arg")  # real now() takes none
 
 
async def test_async_method() -> None:
    mailer = create_autospec(Mailer, instance=True)
    mailer.send.return_value = True  # async def → AsyncMock
    assert await welcome(mailer, "a@x.dev")
    mailer.send.assert_awaited_once_with(
        "a@x.dev", "Welcome!"
    )
 
 
async def test_side_effects() -> None:
    send = AsyncMock(side_effect=[True, OSError("down")])
    assert await send()
    with pytest.raises(OSError):
        await send()
 
 
def test_mocker(mocker: MockerFixture) -> None:
    now = mocker.patch("shop.report.now", return_value=FIXED)
    spy = mocker.spy(report, "stamp")  # real call, recorded
    report.stamp()
    now.assert_called_once_with()
    assert spy.spy_return == "2026-09-25"

import asyncio
from collections.abc import AsyncIterator
 
import pytest
import pytest_asyncio
 
 
@pytest_asyncio.fixture
async def queue() -> AsyncIterator[asyncio.Queue[int]]:
    q: asyncio.Queue[int] = asyncio.Queue()
    yield q
    q.shutdown()
 
 
@pytest.mark.asyncio
async def test_queue(queue: asyncio.Queue[int]) -> None:
    await queue.put(1)
    assert await queue.get() == 1

from collections.abc import AsyncIterator
 
import anyio
import pytest
 
pytestmark = pytest.mark.anyio
 
 
@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"  # params=["asyncio", "trio"] for both
 
 
@pytest.fixture
async def answer() -> AsyncIterator[int]:
    yield 42
 
 
async def test_sleep(answer: int) -> None:
    with anyio.fail_after(1):  # timeout
        await anyio.sleep(0)
    assert answer == 42

uv add --dev pytest-cov
uv run pytest --cov --cov-report=term-missing
uv run pytest --cov --cov-report=html  # htmlcov/index.html
uv run pytest --cov --cov-report=xml   # CI upload

[tool.coverage.run]
branch = true  # count both sides of each if
source = ["shop"]
patch = ["subprocess"]  # measure child processes too
 
[tool.coverage.report]
show_missing = true
skip_covered = true
fail_under = 90
exclude_also = [
  "if TYPE_CHECKING:",
  "raise NotImplementedError",
  "@overload",
]

import json
 
from hypothesis import example, given, settings
from hypothesis import strategies as st
 
from shop.cart import Cart
 
json_values = st.recursive(
    st.none() | st.booleans() | st.integers() | st.text(),
    lambda kids: st.lists(kids)
    | st.dictionaries(st.text(), kids),
)
 
 
@given(json_values)
def test_json_roundtrip(value: object) -> None:
    assert json.loads(json.dumps(value)) == value
 
 
@given(st.lists(st.integers()))
@example([])  # always tried, on top of random cases
def test_sort_idempotent(xs: list[int]) -> None:
    assert sorted(sorted(xs)) == sorted(xs)
 
 
prices = st.floats(0, 1e6, allow_nan=False)
qtys = st.integers(min_value=1, max_value=99)
 
 
@settings(max_examples=500)
@given(st.lists(st.tuples(prices, qtys), max_size=20))
def test_total_never_negative(
    lines: list[tuple[float, int]],
) -> None:
    cart = Cart()
    for price, qty in lines:
        cart.add("sku", price, qty)
    assert cart.total() >= 0

import pytest
from syrupy.assertion import SnapshotAssertion
from syrupy.extensions.json import JSONSnapshotExtension
 
from shop.cart import Cart
from shop.invoice import render
 
 
@pytest.fixture
def snapshot_json(
    snapshot: SnapshotAssertion,
) -> SnapshotAssertion:
    return snapshot.use_extension(JSONSnapshotExtension)
 
 
def test_invoice(snapshot: SnapshotAssertion) -> None:
    cart = Cart()
    cart.add("tea", 4.5, 2)
    assert render(cart) == snapshot  # __snapshots__/*.ambr
 
 
def test_invoice_json(
    snapshot_json: SnapshotAssertion,
) -> None:
    assert render(Cart()) == snapshot_json  # one .json each

uv add --dev syrupy
uv run pytest --snapshot-update  # write or refresh
uv run pytest                    # compare; unused ones fail

from collections.abc import Iterator
 
import httpx2
import pytest
from fastapi.testclient import TestClient
 
from shop.api import app, get_store
 
 
def fake_store() -> dict[int, str]:
    return {1: "tea"}
 
 
@pytest.fixture
def client() -> Iterator[TestClient]:
    app.dependency_overrides[get_store] = fake_store
    with TestClient(app) as c:  # runs lifespan events
        yield c
    app.dependency_overrides.clear()
 
 
def test_read_item(client: TestClient) -> None:
    r = client.get("/items/1")
    assert r.status_code == 200
    assert r.json() == {"name": "tea"}
 
 
async def test_async_client() -> None:
    app.dependency_overrides[get_store] = fake_store
    transport = httpx2.ASGITransport(app=app)
    async with httpx2.AsyncClient(
        transport=transport, base_url="http://test"
    ) as ac:
        r = await ac.get("/items/1")
    app.dependency_overrides.clear()
    assert r.json() == {"name": "tea"}

import json
from pathlib import Path
 
import pytest
 
 
@pytest.fixture
def project(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> Path:
    cfg = {"port": 8080}
    (tmp_path / "config.json").write_text(json.dumps(cfg))
    monkeypatch.chdir(tmp_path)  # restored after the test
    return tmp_path
 
 
def test_reads_cwd_config(project: Path) -> None:
    cfg = json.loads(Path("config.json").read_text())
    assert cfg["port"] == 8080

from datetime import UTC, datetime, timedelta
 
import time_machine
from time_machine import TimeMachineFixture
 
from shop.report import stamp
 
 
def test_frozen(time_machine: TimeMachineFixture) -> None:
    time_machine.move_to(
        datetime(2026, 9, 25, 9, tzinfo=UTC), tick=False
    )
    assert stamp() == "2026-09-25"
    time_machine.shift(timedelta(days=1))
    assert stamp() == "2026-09-26"
 
 
@time_machine.travel(datetime(2000, 1, 1, tzinfo=UTC))
def test_decorator() -> None:
    assert datetime.now(UTC).year == 2000

import pytest
 
from shop.net import parse_port
 
type Row = tuple[str, int | None]  # None means "raises"
 
CASES: dict[str, Row] = {
    "http": ("80", 80),
    "max": ("65535", 65_535),
    "too-big": ("65536", None),
    "not-a-number": ("http", None),
}
 
 
@pytest.mark.parametrize(
    ("raw", "want"), CASES.values(), ids=CASES.keys()
)
def test_parse_port(raw: str, want: int | None) -> None:
    if want is None:
        with pytest.raises(ValueError):
            parse_port(raw)
    else:
        assert parse_port(raw) == want

import httpx
import pytest
import respx
 
from shop.weather import temperature, temperature_sync
 
URL = "https://api.example.com/w/oslo"
 
 
async def test_ok(respx_mock: respx.MockRouter) -> None:
    body = {"temp": 4.5}
    route = respx_mock.get(URL).respond(200, json=body)
    async with httpx.AsyncClient() as client:
        assert await temperature(client, "oslo") == 4.5
    assert route.call_count == 1
 
 
def test_errors(respx_mock: respx.MockRouter) -> None:
    respx_mock.get(host="api.example.com").mock(
        side_effect=[httpx.Response(503), httpx.ConnectError]
    )
    with pytest.raises(httpx.HTTPStatusError):
        temperature_sync("oslo")
    with pytest.raises(httpx.ConnectError):
        temperature_sync("oslo")

from collections.abc import Iterator
 
import pytest
import sqlalchemy as sa
from sqlalchemy.orm import Session
 
from shop.models import Base
 
 
@pytest.fixture(scope="session")
def engine() -> sa.Engine:
    engine = sa.create_engine("postgresql+psycopg://…/test")
    Base.metadata.create_all(engine)
    return engine
 
 
@pytest.fixture
def session(engine: sa.Engine) -> Iterator[Session]:
    with engine.connect() as conn, conn.begin() as tx:
        with Session(
            bind=conn,
            join_transaction_mode="create_savepoint",
        ) as s:
            yield s  # commit() only releases a SAVEPOINT
        tx.rollback()

uv run pytest --lf           # last failed only
uv run pytest --ff -x        # failed first, stop at one
uv run pytest --sw           # fix one at a time, resume
uv run pytest -n auto        # pytest-xdist, all cores
# only tests whose code changed (pytest-testmon)
uv run --with pytest-testmon pytest --testmon
# re-run on save (pytest-watcher)
uv run --with pytest-watcher ptw . --now

import logging
import warnings
 
import pytest
 
 
def old_api() -> int:
    warnings.warn("use new_api()", DeprecationWarning, 2)
    logging.getLogger("shop").error("old_api called")
    return 1
 
 
def test_warns_and_logs(
    caplog: pytest.LogCaptureFixture,
) -> None:
    with pytest.warns(DeprecationWarning, match="new_api"):
        assert old_api() == 1
    record = ("shop", logging.ERROR, "old_api called")
    assert record in caplog.record_tuples
 
 
def test_no_warnings() -> None:
    with warnings.catch_warnings():
        warnings.simplefilter("error")  # any warning fails
        int("80")