Backend Live Coding

# ======================================================================
# Q: How do you test a FastAPI endpoint without a real database and without real
# auth?
#
# app.dependency_overrides replaces any Depends, so I do not need monkeypatch.
# For the database I prefer real PostgreSQL (Docker or testcontainers) with a
# transaction that rolls back at the end of each test, because SQLite does not
# have the same constraints, JSONB and locking. In Django the equivalent is
# pytest-django with @pytest.mark.django_db.
# ======================================================================

from fastapi.testclient import TestClient
from app.main import app
from app.deps import get_session, current_user, get_erp

@pytest.fixture
def client(db_session):                       # db_session: a transaction with rollback
    fake_erp = FakeErpClient()
    app.dependency_overrides[get_session] = lambda: db_session
    app.dependency_overrides[current_user] = lambda: User(id=1, tenant_id=7,
                                                          perms={'orders:write'})
    app.dependency_overrides[get_erp] = lambda: fake_erp
    with TestClient(app) as c:
        c.fake_erp = fake_erp
        yield c
    app.dependency_overrides.clear()

def test_create_order_pushes_to_erp(client):
    r = client.post('/v1/orders', json={'lines': [{'sku': 'A1', 'qty': 2}]})
    assert r.status_code == 201
    assert client.fake_erp.sent[0][0] == r.json()['id']

def test_validation_error(client):
    r = client.post('/v1/orders', json={'lines': [{'sku': 'A1', 'qty': 0}]})
    assert r.status_code == 422

# ======================================================================
# Q: Show me pytest with fixtures and parametrize for a pricing rule.
#
# The fixture provides shared data, and parametrize turns the business table
# into tests. I often take the table of examples from the product owner as is:
# it also becomes documentation of the rule.
# ======================================================================

import pytest
from decimal import Decimal as D
from pricing import Line, order_total

@pytest.fixture
def lines():
    return [Line(D('10.00'), 3), Line(D('2.50'), 4)]       # 30 + 10 = 40

@pytest.mark.parametrize('discount, expected', [
    (D('0'),     D('40.00')),
    (D('0.10'),  D('36.00')),
    (D('0.333'), D('26.68')),      # 26.68 after ROUND_HALF_UP
    (D('1'),     D('0.00')),
], ids=['none', '10pct', 'odd', 'full'])
def test_order_total(lines, discount, expected):
    assert order_total(lines, discount) == expected

def test_empty_cart_is_zero():
    assert order_total([], D('0.1')) == D('0.00')

# ======================================================================
# Q: Repository pattern and dependency injection: do you need them in Django?
#
# In Django the ORM is already a repository, so I do not wrap every query. I do
# keep ports for external dependencies, because that is where the real value is.
# In FastAPI/SQLAlchemy the repository is more natural.
# Protocol gives structural typing: the fake does not need to inherit from
# anything, and mypy checks that it matches.
# ======================================================================

from typing import Protocol

class ErpClient(Protocol):
    def push_order(self, order_id: int, payload: dict) -> str: ...

class HttpErpClient:
    def __init__(self, session: httpx.Client, base_url: str):
        self._s, self._url = session, base_url
    def push_order(self, order_id, payload):
        r = self._s.post(f'{self._url}/orders', json=payload,
                         headers={'Idempotency-Key': f'order-{order_id}'}, timeout=10)
        r.raise_for_status()
        return r.json()['erp_ref']

class FakeErpClient:                          # for tests
    def __init__(self): self.sent = []
    def push_order(self, order_id, payload):
        self.sent.append((order_id, payload)); return 'ERP-1'

def sync_order(order: Order, erp: ErpClient) -> None:    # injected
    order.external_ref = erp.push_order(order.id, serialize(order))
    order.save(update_fields=['external_ref'])

# ======================================================================
# Q: Take this "fat" Django view and make it testable.
#
# What I gained: the pricing rule is tested with plain pytest in milliseconds.
# The service has a transaction and clean domain exceptions. The ERP and the
# email go out via Celery with on_commit, so we do not send an order to the ERP
# if it was rolled back, and the user does not wait for the ERP. I also fixed
# the N+1 and the float. The same service is called by the view, a management
# command or a Celery task.
# ======================================================================

# BEFORE: everything inside the view, impossible to test without HTTP, DB and ERP
@api_view(['POST'])
def place_order(request):
    cart = Cart.objects.get(user=request.user)
    total = 0
    for item in cart.items.all():                      # N+1
        price = item.product.price
        if request.user.customer.is_b2b:
            price = price * 0.9
        total += price * item.qty                      # float
    if total > request.user.customer.credit_limit:
        return Response({'error': 'credit'}, status=400)
    order = Order.objects.create(customer=request.user.customer, total=total)
    requests.post(ERP_URL, json={'id': order.id, 'total': total})   # inside the request
    send_mail('Order', f'#{order.id}', None, [request.user.email])
    return Response({'id': order.id})



# AFTER: pure rule + service + thin view + async side-effects
# pricing.py (pure, no Django)
def order_total(lines: list[Line], discount: Decimal) -> Decimal:
    gross = sum((l.unit_price * l.qty for l in lines), Decimal('0'))
    return (gross * (1 - discount)).quantize(Decimal('0.01'), ROUND_HALF_UP)

# services.py
class CreditLimitExceeded(Exception): ...

@transaction.atomic
def place_order(customer: Customer, cart: Cart) -> Order:
    lines = [Line(i.product.price, i.qty)
             for i in cart.items.select_related('product')]
    total = order_total(lines, customer.discount_rate)
    if total > customer.credit_limit:
        raise CreditLimitExceeded(total)
    order = Order.objects.create(customer=customer, total=total)
    transaction.on_commit(lambda: push_order_to_erp.delay(order.id))
    transaction.on_commit(lambda: send_order_email.delay(order.id))
    return order

# views.py
@api_view(['POST'])
def place_order_view(request):
    try:
        order = services.place_order(request.user.customer,
                                     Cart.objects.get(user=request.user))
    except CreditLimitExceeded:
        return Response({'code': 'credit_limit_exceeded'}, status=422)
    return Response({'id': order.id}, status=201)

# ======================================================================
# Q: A bug appeared sometime in the last 200 commits. How do you find which one
# caused it?
#
# git bisect does a binary search: for 200 commits it needs about 8 steps. The
# key is to have a test that reproduces the bug, so I write it first (and it
# stays as a regression test). This is also one reason I want small, atomic
# commits: bisect points you to a 20-line change, not a 3,000-line commit.
# ======================================================================

git bisect start
git bisect bad HEAD              # it is broken now
git bisect good v2.14.0          # it worked at this tag

# automatic: the script returns 0 = good, 1 = bad, 125 = skip
git bisect run pytest tests/test_pricing.py::test_volume_discount -x -q

git bisect reset                 # return to the original HEAD