"""test_train.py — the verdict for train.py. You do not write this file."""
import copy
import io
import unittest
from contextlib import redirect_stdout
import torch
import torch.nn.functional as F
from torch import nn
from model import MLP, LinearClassifier, count_parameters
from train import build_model, evaluate, fit, make_loader
def two_blobs(n: int = 200) -> tuple[torch.Tensor, torch.Tensor]:
"""Two clouds of 2-D points, one around (-1, -1) and one around (+1, +1)."""
g = torch.Generator().manual_seed(0)
y = torch.arange(n) % 2
centre = (y * 2 - 1).float().unsqueeze(1).repeat(1, 2)
return centre + 0.5 * torch.randn(n, 2, generator=g), y
def noisy_blobs(n: int, seed: int) -> tuple[torch.Tensor, torch.Tensor]:
"""Two heavily overlapping clouds: easy to memorise when n is small."""
g = torch.Generator().manual_seed(seed)
y = torch.arange(n) % 2
centre = (y * 2 - 1).float().unsqueeze(1).repeat(1, 2)
return centre + 1.2 * torch.randn(n, 2, generator=g), y
class Spy(nn.Module):
"""A linear model that records whether it was in training mode on each call."""
def __init__(self) -> None:
super().__init__()
self.fc = nn.Linear(2, 2)
self.modes: list[bool] = []
def forward(self, x: torch.Tensor) -> torch.Tensor:
self.modes.append(self.training)
return self.fc(x)
def run_fit(model: nn.Module, loader, epochs: int, lr: float) -> str:
"""Call fit() and return what it printed."""
out = io.StringIO()
with redirect_stdout(out):
fit(model, loader, epochs, lr)
return out.getvalue()
class TestMakeLoader(unittest.TestCase):
def test_batches_cover_every_row_in_order(self):
x = torch.arange(10).float().unsqueeze(1)
y = torch.arange(10)
loader = make_loader(x, y, batch_size=4, shuffle=False)
sizes = [len(yb) for _, yb in loader]
self.assertEqual(sizes, [4, 4, 2])
self.assertEqual(torch.cat([yb for _, yb in loader]).tolist(), list(range(10)))
class TestEvaluate(unittest.TestCase):
def test_loss_is_a_per_image_average_even_with_uneven_batches(self):
model = nn.Linear(2, 2)
with torch.no_grad():
model.weight.copy_(torch.eye(2))
model.bias.zero_()
x = torch.tensor([[2.0, 0.0], [0.0, 3.0], [1.0, 0.0], [0.0, 0.5]])
y = torch.tensor([0, 1, 1, 0]) # the scores are x itself: 2 of 4 are right
loss, acc = evaluate(model, make_loader(x, y, batch_size=3))
self.assertAlmostEqual(acc, 0.5)
self.assertAlmostEqual(loss, F.cross_entropy(x, y).item(), places=5)
class TestFit(unittest.TestCase):
def setUp(self):
self.x, self.y = two_blobs()
self.loader = make_loader(self.x, self.y, batch_size=50) # 4 batches
def test_first_step_moves_every_number_by_about_lr_as_adam_does(self):
# Lesson 5 asserted SGD's arithmetic here. fit() now uses Adam, whose
# very first step moves every number by roughly lr, whatever its gradient.
torch.manual_seed(1)
model = nn.Linear(2, 2)
before = copy.deepcopy(model.state_dict())
one_batch = make_loader(self.x, self.y, batch_size=len(self.y))
run_fit(model, one_batch, epochs=1, lr=0.01)
for name, p in model.named_parameters():
moved = (p.detach() - before[name]).abs()
self.assertTrue(
torch.allclose(moved, torch.full_like(moved, 0.01), atol=1e-4),
f"{name} moved by {moved.flatten().tolist()}, want about 0.01 each",
)
def test_accuracy_rises_on_separable_data(self):
torch.manual_seed(1)
model = nn.Linear(2, 2)
_, before = evaluate(model, self.loader)
run_fit(model, self.loader, epochs=5, lr=0.1)
_, after = evaluate(model, self.loader)
self.assertGreater(after, 0.9, f"accuracy {before:.1%} -> {after:.1%}")
def test_prints_one_line_per_epoch(self):
torch.manual_seed(1)
text = run_fit(nn.Linear(2, 2), self.loader, epochs=3, lr=0.1)
lines = text.strip().splitlines()
self.assertEqual(len(lines), 3)
self.assertIn("train acc", lines[0])
class TestLinearClassifier(unittest.TestCase):
def test_maps_784_pixels_to_10_scores(self):
out = LinearClassifier()(torch.zeros(5, 784))
self.assertEqual(tuple(out.shape), (5, 10))
def test_is_an_nn_module_with_lesson_4s_7850_numbers(self):
model = LinearClassifier()
self.assertIsInstance(model, nn.Module)
self.assertEqual(sum(p.numel() for p in model.parameters()), 7850)
def test_other_sizes(self):
out = LinearClassifier(in_features=2, num_classes=3)(torch.zeros(4, 2))
self.assertEqual(tuple(out.shape), (4, 3))
def test_returns_raw_scores_not_probabilities(self):
torch.manual_seed(0)
out = LinearClassifier()(torch.rand(8, 784))
self.assertTrue((out < 0).any(), "no negative score: did you apply softmax?")
self.assertFalse(torch.allclose(out.sum(dim=1), torch.ones(8)))
def test_learns_with_fit(self):
x, y = two_blobs()
loader = make_loader(x, y, batch_size=50)
torch.manual_seed(1)
model = LinearClassifier(in_features=2, num_classes=2)
run_fit(model, loader, epochs=5, lr=0.1)
_, acc = evaluate(model, loader)
self.assertGreater(acc, 0.9)
class TestBuildModel(unittest.TestCase):
def test_linear(self):
model = build_model("linear", (128,), 0.0)
self.assertIsInstance(model, LinearClassifier)
self.assertEqual(count_parameters(model), 7850)
def test_mlp_uses_the_hidden_widths(self):
self.assertIsInstance(build_model("mlp", (128,), 0.0), MLP)
self.assertEqual(count_parameters(build_model("mlp", (128,), 0.0)), 101_770)
self.assertEqual(count_parameters(build_model("mlp", (128, 64), 0.0)), 109_386)
def test_mlp_passes_dropout_on(self):
model = build_model("mlp", (128,), 0.3)
drops = [m for m in model.modules() if isinstance(m, nn.Dropout)]
self.assertEqual([d.p for d in drops], [0.3])
class TestEvaluateMode(unittest.TestCase):
def test_evaluate_switches_dropout_off(self):
x, y = two_blobs()
loader = make_loader(x, y, batch_size=200)
model = nn.Sequential(nn.Linear(2, 2), nn.Dropout(0.9))
model.train()
first, second = evaluate(model, loader), evaluate(model, loader)
self.assertEqual(first, second, "same model, same data, two different answers")
self.assertFalse(model.training, "evaluate() left the model in training mode")
class TestFitWithValidation(unittest.TestCase):
def setUp(self):
self.x, self.y = noisy_blobs(30, seed=0)
self.xv, self.yv = noisy_blobs(200, seed=1)
self.loader = make_loader(self.x, self.y, batch_size=10, shuffle=True)
self.val = make_loader(self.xv, self.yv, batch_size=200)
def fit_quietly(self, model, **kw):
out = io.StringIO()
with redirect_stdout(out):
history = fit(model, self.loader, val_loader=self.val, **kw)
return history, out.getvalue()
def test_one_row_per_epoch_with_train_and_val_numbers(self):
torch.manual_seed(0)
history, _ = self.fit_quietly(MLP(2, (8,), 2), epochs=4, lr=0.01)
self.assertEqual([r["epoch"] for r in history], [1, 2, 3, 4])
for key in ("train_loss", "train_acc", "val_loss", "val_acc"):
self.assertTrue(all(key in r for r in history), f"missing {key}")
def test_without_a_val_loader_there_are_no_val_keys(self):
torch.manual_seed(0)
out = io.StringIO()
with redirect_stdout(out):
history = fit(MLP(2, (8,), 2), self.loader, epochs=2, lr=0.01)
self.assertEqual(len(history), 2)
self.assertNotIn("val_loss", history[0])
def test_best_marks_every_new_lowest_val_loss(self):
torch.manual_seed(0)
history, _ = self.fit_quietly(MLP(2, (8,), 2), epochs=6, lr=0.01)
lowest = float("inf")
for row in history:
improved = row["val_loss"] < lowest
self.assertEqual(row.get("best", False), improved, f"epoch {row['epoch']}")
lowest = min(lowest, row["val_loss"])
def test_verbose_false_prints_nothing_and_true_prints_a_line_per_epoch(self):
torch.manual_seed(0)
_, quiet = self.fit_quietly(MLP(2, (8,), 2), epochs=3, lr=0.01, verbose=False)
self.assertEqual(quiet, "")
_, loud = self.fit_quietly(MLP(2, (8,), 2), epochs=3, lr=0.01)
lines = loud.strip().splitlines()
self.assertEqual(len(lines), 3)
self.assertIn("val acc", lines[0])
def test_dropout_is_on_while_learning_in_every_epoch(self):
spy = Spy()
loader = make_loader(self.x, self.y, batch_size=10) # 3 batches
out = io.StringIO()
with redirect_stdout(out):
fit(spy, loader, epochs=2, lr=0.01)
# per epoch: 3 training batches, then 3 measuring batches
self.assertEqual(spy.modes, ([True] * 3 + [False] * 3) * 2)
def test_patience_stops_early_when_nothing_improves(self):
torch.manual_seed(0)
# lr = 0: the weights never move, so val loss improves only in epoch 1.
history, _ = self.fit_quietly(MLP(2, (8,), 2), epochs=50, lr=0.0, patience=3)
self.assertEqual(len(history), 4)
def test_patience_puts_the_best_weights_back(self):
torch.manual_seed(0)
model = MLP(2, (64, 64), 2)
history, _ = self.fit_quietly(model, epochs=100, lr=0.01, patience=5)
best = min(r["val_loss"] for r in history)
self.assertLess(len(history), 100, "never stopped early")
self.assertGreater(history[-1]["val_loss"], best + 0.05, "no overfitting to undo")
restored, _ = evaluate(model, self.val)
self.assertAlmostEqual(restored, best, places=4)
def test_weight_decay_shrinks_the_weights(self):
def total_size(weight_decay: float) -> float:
torch.manual_seed(1)
model = MLP(2, (16,), 2)
out = io.StringIO()
with redirect_stdout(out):
fit(model, self.loader, epochs=30, lr=0.01, weight_decay=weight_decay)
return sum(p.abs().sum().item() for p in model.parameters())
self.assertLess(total_size(0.1), 0.8 * total_size(0.0))
if __name__ == "__main__":
unittest.main()