Mock CTO Interview

# ======================================================================
# Q: CTO: "The /api/orders/ endpoint makes 400 queries. How do you fix it?"
#
# You: "It is an N+1 caused by nested serializers. I confirm it with django-
# debug-toolbar or silk. I add select_related for the FKs, Prefetch for the
# items together with their product, annotate for the counts instead of a
# SerializerMethodField that runs a query, and I lock the number of queries with
# a test so the problem does not come back." Chapter 6.
# ======================================================================

def get_queryset(self):
    return (Order.objects
            .filter(company_id=self.request.user.company_id)
            .select_related('customer', 'shipping_address')
            .prefetch_related(Prefetch(
                'items', queryset=OrderItem.objects.select_related('product')))
            .annotate(items_count=Count('items')))

def test_orders_list_query_count(api_client, user, django_assert_max_num_queries):
    OrderFactory.create_batch(50, company=user.company)
    api_client.force_authenticate(user)
    with django_assert_max_num_queries(4):   # constant, regardless of the number of rows
        api_client.get('/api/v1/orders/')

# ======================================================================
# Q: CTO: "Two customers buy the last product at the same time. What happens in
# your system?"
#
# You: "If the code reads the stock in Python, checks it and then saves, we get
# a lost update and overselling. I let the database guarantee it with a
# conditional atomic UPDATE: if 0 rows are affected, the product is sold out and
# the second user sees a friendly message. As a safety net I have a
# CheckConstraint so the stock can never become negative. If the flow is more
# complex (multiple products, coupons), I lock with select_for_update in a fixed
# order." Chapter 4.
# ======================================================================

def reserve(product_id, qty):
    updated = (Product.objects
               .filter(pk=product_id, stock__gte=qty)
               .update(stock=F('stock') - qty))
    if not updated:
        raise OutOfStock(product_id)

class Product(models.Model):
    stock = models.IntegerField()
    class Meta:
        constraints = [models.CheckConstraint(
            condition=Q(stock__gte=0), name='product_stock_non_negative')]

# ======================================================================
# Q: CTO: "How did the synchronization with the ERP work?"
#
# You: "With Celery tasks. Pull for stock and prices every few minutes with
# Celery beat, push for orders via transaction.on_commit. The tasks were
# idempotent with a unique external_id, had retries with exponential backoff,
# and raised an alert when they failed repeatedly. A nightly reconciliation job
# found discrepancies."
# ======================================================================

@shared_task
def pull_stock_from_erp():
    rows = erp.fetch_stock(changed_since=cache.get('erp_stock_cursor'))
    by_sku = {r.sku: r.qty for r in rows}
    products = list(Product.objects.filter(sku__in=by_sku))
    for p in products:
        p.stock = by_sku[p.sku]
    Product.objects.bulk_update(products, ['stock'], batch_size=1000)
    cache.set('erp_stock_cursor', rows.cursor, None)

# ======================================================================
# Q: CTO: "How did you migrate users with their old passwords?"
#
# You: "I imported the legacy hashes with an algorithm prefix and wrote a custom
# password hasher. Because the modern hasher is first in PASSWORD_HASHERS,
# Django automatically rehashes on the first successful login. So no mass
# password reset was needed, and over time the old hashes disappeared. Those who
# never logged in received a reset email after six months."
# ======================================================================

class LegacyASPHasher(BasePasswordHasher):
    algorithm = 'legacy_md5'          # stored as legacy_md5$salt$hash

    def verify(self, password, encoded):
        _, salt, hash_ = encoded.split('$', 2)
        candidate = hashlib.md5((salt + password).encode()).hexdigest()
        return constant_time_compare(candidate, hash_)

    def safe_summary(self, encoded):
        return {'algorithm': self.algorithm}

PASSWORD_HASHERS = [
    'django.contrib.auth.hashers.Argon2PasswordHasher',   # preferred -> rehash
    'django.contrib.auth.hashers.PBKDF2PasswordHasher',
    'accounts.hashers.LegacyASPHasher',
]

# ======================================================================
# Q: CTO: "Tell me about the migration from ASP to Django. What was the biggest
# risk?"
#
# You: "The biggest risk was hidden business logic, for example B2B pricing
# rules written in stored procedures and in VBScript. I handled it with
# characterization tests: I recorded real scenarios from the old system
# (customer, SKU, quantity -> final price) and the new system had to produce
# exactly the same results. The second risk was the data: encodings
# (windows-1253), duplicates, orphaned records. I wrote an idempotent ETL with
# reconciliation reports. We did the cutover module by module behind nginx, with
# a rollback plan for every step."
# ======================================================================

# The cases were extracted from the legacy system (inputs + the price the ASP produced)
@pytest.mark.parametrize('case', load_cases('legacy_b2b_pricing.json'))
def test_b2b_price_matches_legacy(case):
    price = pricing.calculate(customer_id=case['customer'],
                              sku=case['sku'], qty=case['qty'])
    assert price == Decimal(case['legacy_price'])