Modular Integration vs Unit — What Each Test Catches¶

def set_status(self, order_id: str, status: str, payment_id: str | None = None) -> None:
    self.conn.execute(
        "UPDATE orders.orders SET status = %s, payment_id = coalesce(%s, payment_id) WHERE id = %s",
        (status, payment_id, order_id),
    )

FAILED tests/unit/test_place_order_mocks.py::test_paid_order_is_marked_paid_and_published
FAILED tests/unit/test_place_order_mocks.py::test_out_of_stock_marks_order_failed
FAILED tests/unit/test_place_order_mocks.py::test_declined_payment_releases_stock

FAILED tests/module/test_orders_http.py::test_place_and_read_order
E   At index 0 diff: {'sku': 'book', 'qty': 25, 'price': '2.00'} != {'sku': 'book', 'qty': 2, 'price': '25.00'}
FAILED tests/module/test_place_order.py::test_paid_order_is_stored_charged_and_published
E   AssertionError: assert [('book', 25), ('lamp', 60)] == [('book', 2), ('lamp', 1)]

E   psycopg.errors.UndefinedColumn: column "paymentid" of relation "orders" does not exist

shipping = Decimal("0") if goods > FREE_SHIPPING_FROM else SHIPPING   # was >=

FAILED tests/unit/test_pricing.py::test_total[lines1-None-expected1]      # exactly 100.00

FAILED tests/module/test_orders_http.py::test_errors_map_to_status_codes[body0-headers0-409]

        assert inventory.reserve("o1", {"book": 1, "lamp": 2}) is False
>       assert (inventory.stock_of("book"), inventory.stock_of("lamp")) == (3, 1)
E       assert (2, 1) == (3, 1)
FAILED tests/contracts/test_inventory_contract.py::test_reserve_takes_all_lines_or_none[fake]

# tests/contracts/test_inventory_contract.py
"""One set of tests for the real inventory module and for the fake the orders tests use.
If the fake drifts from the real module, these tests fail for the fake."""

from decimal import Decimal

import pytest

from shop.inventory import InventoryModule
from tests.fakes import FakeInventory

pytestmark = pytest.mark.module


@pytest.fixture(params=["fake", "real"])
def inventory(request):
    if request.param == "fake":
        inventory = FakeInventory()
    else:
        inventory = InventoryModule(request.getfixturevalue("conn"))
    inventory.add_product("book", Decimal("25.00"), stock=3)
    inventory.add_product("lamp", Decimal("60.00"), stock=1)
    return inventory


def test_price_of_known_sku(inventory):
    assert inventory.price_of("book") == Decimal("25.00")


def test_price_of_unknown_sku_raises_key_error(inventory):
    with pytest.raises(KeyError):
        inventory.price_of("nope")


def test_reserve_takes_all_lines_or_none(inventory):
    assert inventory.reserve("o1", {"book": 1, "lamp": 2}) is False
    assert (inventory.stock_of("book"), inventory.stock_of("lamp")) == (3, 1)


def test_release_returns_reserved_stock(inventory):
    assert inventory.reserve("o1", {"book": 2}) is True
    inventory.release("o1")
    assert inventory.stock_of("book") == 3


def test_release_of_unknown_order_is_a_no_op(inventory):
    inventory.release("never-reserved")
    assert inventory.stock_of("book") == 3

# shop/inventory/__init__.py
"""The inventory module: owns stock and prices. Its own schema, its own public API."""

from decimal import Decimal

import psycopg

SCHEMA = """
CREATE SCHEMA IF NOT EXISTS inventory;
CREATE TABLE IF NOT EXISTS inventory.products (
    sku   text PRIMARY KEY,
    price numeric(12, 2) NOT NULL,
    stock integer NOT NULL CHECK (stock >= 0)
);
CREATE TABLE IF NOT EXISTS inventory.reservations (
    order_id text NOT NULL,
    sku      text NOT NULL REFERENCES inventory.products(sku),
    qty      integer NOT NULL,
    PRIMARY KEY (order_id, sku)
);
"""


class UnknownSku(KeyError):
    pass


class InventoryModule:
    def __init__(self, conn: psycopg.Connection) -> None:
        self.conn = conn

    @staticmethod
    def migrate(conn: psycopg.Connection) -> None:
        conn.execute(SCHEMA)

    def add_product(self, sku: str, price: Decimal, stock: int) -> None:
        with self.conn.transaction():
            self.conn.execute("INSERT INTO inventory.products VALUES (%s, %s, %s)", (sku, price, stock))

    def price_of(self, sku: str) -> Decimal:
        row = self.conn.execute("SELECT price FROM inventory.products WHERE sku = %s", (sku,)).fetchone()
        if row is None:
            raise UnknownSku(sku)
        return row[0]

    def stock_of(self, sku: str) -> int:
        return self.conn.execute("SELECT stock FROM inventory.products WHERE sku = %s", (sku,)).fetchone()[0]

    def reserve(self, order_id: str, items: dict[str, int]) -> bool:
        try:
            with self.conn.transaction():                  # all lines or none
                for sku, qty in sorted(items.items()):
                    updated = self.conn.execute(
                        "UPDATE inventory.products SET stock = stock - %s WHERE sku = %s AND stock >= %s",
                        (qty, sku, qty),
                    ).rowcount
                    if updated == 0:
                        raise _NotEnough(sku)
                    self.conn.execute("INSERT INTO inventory.reservations VALUES (%s, %s, %s)", (order_id, sku, qty))
        except _NotEnough:
            return False
        return True

    def release(self, order_id: str) -> None:
        with self.conn.transaction():
            rows = self.conn.execute(
                "DELETE FROM inventory.reservations WHERE order_id = %s RETURNING sku, qty", (order_id,)
            ).fetchall()
            for sku, qty in rows:
                self.conn.execute("UPDATE inventory.products SET stock = stock + %s WHERE sku = %s", (qty, sku))


class _NotEnough(Exception):
    pass