Code Quality and Defensive Programming Questions

Covers writing clean, maintainable, and readable code together with proactive techniques to prevent failures and handle unexpected inputs. Topics include naming and structure, modular design, consistent style, comments and documentation, and making code testable and observable. Defensive practices include explicit input validation, boundary checks, null and error handling, assertions, graceful degradation, resource management, and clear error reporting. Candidates should demonstrate thinking through edge cases such as empty inputs, single element cases, duplicates, very large inputs, integer overflow and underflow, null pointers, timeouts, race conditions, buffer overflows in system or embedded contexts, and other hardware specific failures. Also evaluate use of static analysis, linters, unit tests, fuzzing, property based tests, code reviews, logging and monitoring to detect and prevent defects, and tradeoffs between robustness and performance.

HardTechnical

24 practiced

Your company uses a large monorepo with thousands of ML projects. Propose a scalable static analysis and linting approach that gives fast feedback to developers without running full checks on unrelated projects. Cover incremental analysis, per-PR checks, baseline suppression, caching analysis artifacts, and developer tooling for local fast checks.

Sample Answer

Approach overview — build a change-aware, multi-stage static-analysis pipeline that scopes work to the minimal impacted graph, provides fast local feedback, and escalates to heavier checks only when needed.

Key techniques- Change graph + impact analysis: maintain a repository dependency graph (files → targets → datasets → models). On each commit/PR, compute changed files via git and walk the graph to determine affected targets. Use build systems like Bazel/Pants or a custom graph store to make this fast and reproducible.- Incremental analysis: store analysis artifacts (ASTs, type-check caches, linter results, model schema fingerprints) keyed by content hashes. Re-run analyzers only for nodes with changed inputs or invalidated caches. Use content-addressable remote cache to share artifacts across CI and dev machines.- Per-PR staged checks: - Fast pre-commit/pre-submit: syntax, flake8/ruff, pytype/mypy incremental check on changed files, simple model-schema validation, small smoke inference tests using minimized fixture data. These run locally and in lightweight CI runners. - Medium CI: unit tests for affected targets, dependency vulnerability scan, sanity checks on model size and serialization. - Full checks on merge/main or explicit opt-in: full integration tests, long retrains, heavy static analyzers (e.g., whole-target type flow, data-lineage analysis).- Baseline suppression & debt management: capture existing findings into per-target baseline files (with hashes and timestamps). New findings in modified code surface as failures; pre-existing ones are informational. Provide a path to "adopt and fix": allow creating focused suppression entries and require owners to reduce baseline size over time via metrics and sprint goals.- Caching analysis artifacts: persist analyzer outputs (ASTs, type-check caches, taint-analysis state) to remote cache and local disk. Use deterministic hashes of source + deps + config to ensure reuse. For ML models, cache model fingerprints, precomputed shape inference, and small compiled artifacts.- Developer tooling for fast local checks: - CLI tool (repo-check) to compute affected targets and run the relevant fast checks locally, with options: --quick (only linters/types), --smoke (plus inference on tiny fixtures), --full-target. - IDE plugins to run incremental linters on save using the same cache and graph. - Remote execution for heavier local runs: developers can offload expensive checks to shared remote executors that return results quickly using cached artifacts. - Minimal test fixtures: keep lightweight representative datasets for fast smoke tests; parameterize to run subset locally and full in CI.- Observability and governance: collect metrics (time-to-first-feedback, cache hit rate, baseline size), alert when a PR touches many targets, and surface owners for slow/large changes.

Why this works- Limits work to true impact surface via the dependency graph, dramatically reducing unnecessary runs.- Remote cache + deterministic keys give near-instant reuse across machines.- Staged checks balance developer productivity with safety: fast feedback prevents regressions; heavy checks protect production.- Baseline suppression avoids noisy regressions while enforcing continuous improvement.

Trade-offs & considerations- Building/maintaining the graph and deterministic hashing adds engineering cost.- Requires buy-in to keep fixtures, baselines, and ownership metadata up to date.- Security: ensure cached analysis artifacts are validated and signed.

Example tools & stack- Bazel/Pants, Content-addressable remote cache (e.g., Buildbarn/remote cache), Ruff/Flake8, MyPy/pytype, Semgrep/SonarQube for ML-specific patterns, custom repo-check CLI, remote execution (BuildGrid/remoteexecution or cloud CI executors).

This design gives ML engineers fast local feedback for everyday edits while scaling analyses across thousands of projects without redundant work.

HardTechnical

28 practiced

Using Hypothesis or a custom generator, design property-based tests for a deep learning training loop that assert important invariants: (1) loss on a tiny synthetic dataset decreases after a few steps, (2) gradients and parameters remain finite, (3) a saved checkpoint can be loaded and reproduce inference outputs. Provide code sketches and explain strategies to reduce flakiness from randomness.

Sample Answer

Approach: use Hypothesis to generate small deterministic training scenarios (model width, lr, seed, batch size) and run a short training loop on a tiny synthetic dataset. Assert (1) loss decreases after a few steps, (2) grads and params are finite, (3) checkpointing reproduces inference outputs. Reduce flakiness with fixed seeds, deterministic CuDNN or CPU-only, small batches, and tolerance thresholds.

python

# python
from hypothesis import given, settings, strategies as st
import torch, tempfile, os

def make_toy_data(seed, batch, dim=4):
    torch.manual_seed(seed)
    X = torch.randn(batch, dim)
    w = torch.randn(dim,1)
    y = X @ w + 0.1*torch.randn(batch,1)
    return X, y

@st.composite
def scenarios(draw):
    seed = draw(st.integers(0, 2**31-1))
    lr = draw(st.floats(1e-4, 1e-1))
    batch = draw(st.integers(4, 16))
    width = draw(st.integers(1, 16))
    return dict(seed=seed, lr=lr, batch=batch, width=width)

@given(scenarios())
@settings(max_examples=50, deadline=None)
def test_training_invariants(s):
    torch.manual_seed(s['seed'])
    torch.use_deterministic_algorithms(True)  # reduce nondeterminism
    X, y = make_toy_data(s['seed']+1, s['batch'], dim=4)
    model = torch.nn.Sequential(torch.nn.Linear(4, s['width']), torch.nn.ReLU(), torch.nn.Linear(s['width'],1))
    opt = torch.optim.SGD(model.parameters(), lr=s['lr'])
    loss_fn = torch.nn.MSELoss()

    # warm start outputs
    model.eval()
    with torch.no_grad():
        baseline = model(X).clone()

    losses = []
    model.train()
    for step in range(6):
        opt.zero_grad()
        out = model(X)
        loss = loss_fn(out, y)
        losses.append(loss.item())
        loss.backward()
        # check grads finite
        for p in model.parameters():
            assert torch.all(torch.isfinite(p.grad)), "non-finite grad"
        opt.step()
        # check params finite
        for p in model.parameters():
            assert torch.all(torch.isfinite(p)), "non-finite param"

    # loss decreased
    assert losses[-1] <= losses[0] + 1e-6 or losses[-1] < sum(losses[:3])/3, "loss didn't decrease"

    # checkpoint and reproduce
    tmp = tempfile.NamedTemporaryFile(delete=False)
    torch.save({'model': model.state_dict()}, tmp.name)
    tmp.close()
    m2 = torch.nn.Sequential(torch.nn.Linear(4, s['width']), torch.nn.ReLU(), torch.nn.Linear(s['width'],1))
    ckpt = torch.load(tmp.name, map_location='cpu')
    m2.load_state_dict(ckpt['model'])
    m2.eval()
    with torch.no_grad():
        out1 = model(X)
        out2 = m2(X)
    os.unlink(tmp.name)
    assert torch.allclose(out1, out2, atol=1e-6), "checkpoint mismatch"

Key strategies to reduce flakiness:- Seed everything (torch, numpy, Python random).- Use deterministic algorithms (torch.use_deterministic_algorithms(True)) or run on CPU.- Keep datasets tiny and gradients stable (small lr range).- Allow small tolerances for numerical differences and avoid single-step flaky assertions by checking trend over multiple steps.- Limit Hypothesis examples and set deadline=None to avoid false failures from timeouts.

MediumSystem Design

27 practiced

You're training a large model on a shared GPU cluster. Describe defensive strategies to avoid OOM and job eviction: dynamic batch sizing, gradient accumulation, memory-aware data loaders, checkpointing on OOM, automatic retries on preemption, and graceful degradation to CPU for small jobs. Sketch a Python training loop with safe checkpointing and OOM handling patterns.

Sample Answer

Requirements:- Robust training on shared GPU cluster: avoid OOM, survive preemption/eviction, allow small CPU fallback.- Minimal model/data changes, graceful retries, safe checkpointing.

Defensive strategies (brief):- Dynamic batch sizing: probe/adjust batch until fits.- Gradient accumulation: simulate large batch with smaller micro-batches.- Memory-aware dataloader: limit prefetch, pin_memory=False if needed.- Checkpoint-on-OOM: catch OOM, save minimal state (optimizer state if possible) to resume.- Automatic retries on preemption: detect signals (e.g., SIGTERM) and checkpoint then retry.- Graceful CPU fallback: for tiny experiments, move model/data to CPU.

High-level components:1. Probe/auto-tuner for batch size before heavy training.2. Training loop with micro-batches + accumulation.3. try/except for torch.cuda.OutOfMemoryError and signal handler for preemption.4. Atomic checkpointing (write to tmp then rename).5. Retry/backoff policy.

Example Python (PyTorch) training loop with safe checkpointing and OOM handling:

python

import os, signal, time, torch
from pathlib import Path

CHECKPOINT_DIR = Path("/checkpoints")
CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True)

def atomic_save(state, path):
    tmp = path.with_suffix(".tmp")
    torch.save(state, tmp)
    tmp.replace(path)

def signal_handler(signum, frame):
    # global flag to trigger checkpoint at safe point
    global PREEMPTED
    PREEMPTED = True

signal.signal(signal.SIGTERM, signal_handler)
PREEMPTED = False

def train(model, optimizer, dataloader, device, start_step=0, accum_steps=4, max_retries=3):
    model.to(device)
    scaler = torch.cuda.amp.GradScaler(enabled=(device.type=='cuda'))
    step = start_step
    retries = 0
    while True:
        try:
            for batch in dataloader:
                if PREEMPTED:
                    # checkpoint and exit loop to allow scheduler to resubmit
                    atomic_save({'step': step, 'model': model.state_dict(), 'opt':optimizer.state_dict()}, CHECKPOINT_DIR/"preempt.pt")
                    raise SystemExit("Preempted - checkpointed")

                inputs, targets = batch
                inputs, targets = inputs.to(device, non_blocking=True), targets.to(device, non_blocking=True)

                with torch.cuda.amp.autocast(enabled=(device.type=='cuda')):
                    outputs = model(inputs)
                    loss = torch.nn.functional.cross_entropy(outputs, targets) / accum_steps

                scaler.scale(loss).backward()

                if (step + 1) % accum_steps == 0:
                    scaler.step(optimizer)
                    scaler.update()
                    optimizer.zero_grad()

                step += 1

            # epoch done -> checkpoint
            atomic_save({'step': step, 'model': model.state_dict(), 'opt':optimizer.state_dict()}, CHECKPOINT_DIR/"latest.pt")
            break

        except torch.cuda.OutOfMemoryError as e:
            torch.cuda.empty_cache()
            retries += 1
            if retries > max_retries:
                # save minimal state and escalate
                atomic_save({'step': step}, CHECKPOINT_DIR/"oom_fatal.pt")
                raise
            # adaptive reaction: reduce batch size or increase accum_steps
            accum_steps = min(accum_steps * 2, 64)
            # optionally shrink dataloader batch size externally and recreate dataloader
            print(f"OOM occurred, retry {retries}, increasing accum_steps to {accum_steps} and retrying")
            time.sleep(5)

        except Exception as e:
            # other failures -> checkpoint minimal info and re-raise
            atomic_save({'step': step}, CHECKPOINT_DIR/"error.pt")
            raise

Key notes:- Probe GPU memory beforehand (torch.cuda.get_device_properties) to choose initial batch size.- Keep dataloader num_workers small, drop_last=True to avoid variable micro-batches.- Use atomic saves to avoid corrupted checkpoints; prefer object storage or network atomic writes if on NFS.- Use exponential backoff for retries; log OOM occurrences and adapt policies (reduce model size, mixed precision, gradient checkpointing).- For CPU fallback: detect unavailable GPU or persistent OOM and set device='cpu' and reduce batch size accordingly.

EasyTechnical

30 practiced

Explain what fuzz testing is and how it applies to ML pipelines. Give concrete examples of inputs to fuzz (image bytes, CSV lines, serialized protobufs) and list tools suitable for Python/ML ecosystems. Describe how to integrate simple fuzzing runs into CI for preprocessing modules to catch crashes from corrupted inputs.

Sample Answer

Fuzz testing = automated, often random or mutation-based injection of unexpected inputs to find crashes, exceptions, memory errors, or logic bugs. In ML pipelines it targets data preprocessing, feature extraction, deserialization, and model serving layers where corrupted or adversarial inputs can cause crashes, silent data corruption, or security issues.

Concrete inputs to fuzz:- Image bytes: corrupted JPEG/PNG byte streams, truncated files, wrong headers- CSV/TSV lines: malformed rows, extra delimiters, non-UTF8, very long fields, missing columns- Serialized protobufs/Avro/MsgPack: wrong field tags, truncated messages, unexpected enum values- JSON: deep nesting, large arrays, unexpected types- Binary model weights or ONNX blobs: truncated or modified sections

Tools suitable for Python/ML:- Hypothesis — property-based testing for structured inputs (CSV rows, arrays, numeric ranges)- Atheris (Google) — byte-level coverage-guided fuzzer for Python extensions and native code paths- python-afl / AFL++ — for compiled C/C++ extensions called from Python- Radamsa — mutation-based generator, great for corpora mutation- protobuf-fuzzers / libprotobuf-mutator — for schema-aware protobuf mutations- Fuzzing harnesses built with pytest (pytest + Hypothesis) for easy CI integration

Example simple fuzz test (Hypothesis) for a preprocessing function that parses CSV lines:

python

from hypothesis import given, strategies as st
import mypipeline.preprocessing as pp

@given(st.binary())
def test_parse_csv_line_no_crash(b):
    # feed arbitrary bytes; ensure no uncaught exceptions
    try:
        line = b.decode('utf-8', errors='replace')
        pp.parse_csv_line(line)
    except pp.ExpectedRecoverableError:
        # known recoverable error is OK
        pass

CI integration for preprocessing modules:- Add lightweight fuzz tests (Hypothesis or small corpus + Radamsa/Atheris) to unit test suite.- Limit runtime/iterations in CI (e.g., Hypothesis max_examples=200; Atheris short mutational budget) to keep CI fast.- Fail build on uncaught exceptions or assertion of invariants (e.g., shape checks, type checks).- Persist any failing inputs/artifacts as CI artifacts for debugging.- Schedule longer nightly fuzz runs (AFL++, Atheris) against larger corpora to find deeper bugs.

Best practices:- Start with known-good corpus (sample images/CSV/protobufs) and mutate them.- Focus on boundaries: very large/small values, truncated streams, wrong encodings.- Make preprocessors defensive: explicit validation, schema checks, length limits, clear exceptions.- Combine schema-aware fuzzing (protobuf mutators) with general mutation tools for coverage.

MediumTechnical

25 practiced

Write property-based tests using Hypothesis for a normalization function normalize_vector(x) that takes a 1D numpy array and returns a unit-norm array. Tests should assert invariants: (1) non-zero vectors yield norm ≈ 1, (2) zero vector returns zero vector, (3) function preserves dtype within numeric types, (4) NaN/inf handling policy. Provide example Hypothesis strategies for numeric arrays and explain tolerances.

Sample Answer

Approach: use Hypothesis to generate 1D numpy arrays across dtypes and sizes; treat NaN/inf separately. Assert invariants: non-zero vectors are normalized to unit L2 norm within tolerance, zero vectors stay zero, dtype preserved for numeric dtypes, and NaN/inf policy verified (here we choose: propagate NaNs and return array with NaN/inf unchanged; alternatively could raise).

python

import numpy as np
from hypothesis import given, strategies as st, assume
from hypothesis.extra.numpy import arrays, array_shapes, dtypes
from math import isfinite

# Example normalize_vector implementation to test
def normalize_vector(x: np.ndarray) -> np.ndarray:
    x = np.asarray(x)
    norm = np.linalg.norm(x)
    if norm == 0 or np.isnan(norm):
        return np.zeros_like(x) if norm == 0 else x.copy()
    return x / norm

# Strategy: numeric dtypes and 1D arrays of length 0..100
numeric_dtypes = dtypes(available_types=[np.float16, np.float32, np.float64, np.int32, np.int64])
vecs = lambda dt: arrays(dtype=dt, shape=array_shapes(min_dims=1, max_dims=1, min_side=0, max_side=50),
                        elements=st.floats(allow_nan=True, allow_infinity=True, width=None))

# 1) Non-zero vectors -> norm ≈ 1
@given(dt=numeric_dtypes, x=st.data())
def test_non_zero_normalizes(dt, x):
    arr = x.draw(vecs(dt))
    # cast ints to float for norm checking if needed
    arr = np.asarray(arr, dtype=dt)
    assume(np.linalg.norm(np.nan_to_num(arr, nan=0.0, posinf=0.0, neginf=0.0)) != 0)
    out = normalize_vector(arr)
    # tolerance: relative tolerance 1e-6 for float64, 1e-3 for float32, 1e-2 for float16
    tol = {np.float64:1e-6, np.float32:1e-3, np.float16:1e-2}.get(out.dtype.type,1e-6)
    if np.any(~np.isfinite(arr)):
        # if input had NaN/inf, behavior documented separately
        return
    assert abs(np.linalg.norm(out) - 1.0) <= tol

# 2) Zero vector returns zero vector
@given(dt=numeric_dtypes, n=st.integers(min_value=0, max_value=50))
def test_zero_vector_preserved(dt, n):
    z = np.zeros((n,), dtype=dt)
    out = normalize_vector(z)
    assert np.array_equal(out, np.zeros_like(z))

# 3) Dtype preserved for numeric types
@given(dt=numeric_dtypes, arr=vecs(dt))
def test_dtype_preserved(dt, arr):
    arr = np.asarray(arr, dtype=dt)
    out = normalize_vector(arr)
    # for integer input, many implementations cast to float; choose policy: preserve dtype if float, else float64
    if np.issubdtype(dt, np.floating):
        assert out.dtype == dt
    else:
        assert np.issubdtype(out.dtype, np.floating)

# 4) NaN/inf handling policy: here we expect NaNs/inf to propagate (or documented behavior)
@given(arr=arrays(dtype=np.float64, shape=array_shapes(min_dims=1, max_dims=1, min_side=1, max_side=20),
                 elements=st.floats(allow_nan=True, allow_infinity=True)))
def test_nan_inf_policy(arr):
    out = normalize_vector(arr)
    # If any NaN present, ensure output contains NaN at same positions (propagation)
    nan_mask = np.isnan(arr)
    assert np.array_equal(np.isnan(out), nan_mask)
    # For infinite entries, ensure not silently turned finite (policy: preserved or turned into NaN)
    inf_mask = np.isinf(arr)
    assert np.array_equal(np.isinf(out), inf_mask) or np.any(np.isnan(out[inf_mask]))

Tolerances: use relative tolerances depending on dtype precision; compare norms with absolute tolerance appropriate to magnitude and dtype. Clarify NaN/inf policy in tests to match intended implementation (propagate vs. sanitize).

Unlock Full Question Bank

Get access to hundreds of Code Quality and Defensive Programming interview questions and detailed answers.

Join thousands of developers preparing for their dream job.