Confidence: Solutions

If you keep assert parallel == serial from parallel_pure.py, the script stops with a bare AssertionError before it prints a line. Each pair now carries the ID of the process that computed it, so no parallel result can equal its serial twin. The solution compares the counts alone, the part a pure function guarantees.

import os
from concurrent.futures import ProcessPoolExecutor

def count_primes(limit: int) -> tuple[int, int]:
    count = 0
    for n in range(2, limit):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            count += 1
    return count, os.getpid()

def main() -> None:
    limits = [200_000, 400_000, 600_000, 800_000]
    serial = list(map(count_primes, limits))
    with ProcessPoolExecutor() as pool:
        parallel = list(pool.map(count_primes, limits))
    counts = [count for count, _ in parallel]
    assert counts == [count for count, _ in serial]
    print(counts)
    print("distinct process IDs:",
          len({pid for _, pid in parallel}))
    print("cores:", os.process_cpu_count())

if __name__ == "__main__":
    main()

Keep the worker function importable. ProcessPoolExecutor needs count_primes picklable and importable from __main__ in a worker process. Only a real script file meets that requirement, not a fenced block executed in place.

Check that the answers agree. The assertion compares the counts alone, since the serial run carries the parent’s process ID and the parallel run carries the workers’. The counts stay the same, [17984, 33860, 49098, 63951]: the same pure function gives the same answers wherever it runs, which is the point of parallel_pure.py.

Count the processes that answered. The interesting number is the second line. Three consecutive runs on one 32-core machine reported 4 distinct process IDs each time.

Two things follow, and neither is the one most people predict. The count is greater than one, so the work left the main process, and assert parallel == serial alone could not show where the work ran. But the count also sits far below thirty-two. ProcessPoolExecutor allows one worker per core, but it starts workers on demand: a submitted task starts a new worker only when no existing worker is idle. Four tasks therefore start at most four processes, not thirty-two.

The count can also fall below four. With limits twenty times smaller, [10_000, 20_000, 30_000, 40_000], the same machine reported 3 on each of three runs. Starting a process then takes longer than a task runs, so the first worker up finishes its task and takes the next one from the queue before the last worker is ready. A distinct-ID count shows that the work left the main process, not how many processes the pool started.

The number depends on the core count, the task sizes, and scheduling, so it is reproducible on your machine and nowhere else. That is why it does not belong in a #: marker in the book.

import os
from concurrent.futures import ThreadPoolExecutor

def count_primes(limit: int) -> tuple[int, int]:
    count = 0
    for n in range(2, limit):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            count += 1
    return count, os.getpid()

def main() -> None:
    limits = [10_000, 20_000, 30_000, 40_000]
    with ThreadPoolExecutor() as pool:
        results = list(pool.map(count_primes, limits))
    print([count for count, _ in results])
    print("distinct process IDs:",
          len({pid for _, pid in results}))

if __name__ == "__main__":
    main()

The thread pool reports exactly 1. Threads share their process, so os.getpid() returns the same value in every one of them. Counting distinct process IDs revealed process parallelism in the previous exercise, and it says nothing about thread parallelism. threading.get_ident() is the equivalent for threads. ProcessPoolExecutor and ThreadPoolExecutor present the identical map() interface and differ this fundamentally underneath. That contrast is the substitutable-backend point from Concurrency.

# test_sorted_laws.py
from hypothesis import given, strategies

def insertion_sort(xs: list[int]) -> list[int]:
    ("The obviously correct version, "
     "for the oracle property.")
    result: list[int] = []
    for x in xs:
        position = 0
        while (position < len(result)
               and result[position] <= x):
            position += 1
        result.insert(position, x)
    return result

numbers = strategies.lists(strategies.integers())

@given(numbers)
def test_output_is_ordered(xs: list[int]) -> None:
    ("Invariant: every adjacent pair "
     "of the output is ordered.")
    output = sorted(xs)
    assert all(a <= b for a, b in zip(output, output[1:]))

@given(numbers)
def test_sorting_is_idempotent(xs: list[int]) -> None:
    "Idempotence: sorting a sorted list changes nothing."
    once = sorted(xs)
    assert sorted(once) == once

@given(strategies.lists(strategies.integers(), max_size=8))
def test_agrees_with_insertion_sort(xs: list[int]) -> None:
    "Oracle: the fast version matches the simple one."
    assert sorted(xs) == insertion_sort(xs)

State a fact about every output. The invariant is the weakest of the three, and the interesting part is how weak. A function that ignores its argument and returns [] passes test_output_is_ordered() on every input, and so does one that returns the first element alone. “Ordered” says nothing about the elements being the same ones you supplied.

Check that repeating changes nothing. Idempotence is weaker still on its own: the same []-returning function passes it too. Idempotence adds a different kind of check, one about the operation rather than the output. It catches a sort that drops the last element. That sort’s output is always ordered, so the invariant passes, but running it on its own output drops another element, so twice and once disagree on every list of two or more.

Compare against an independent version. The oracle closes the gap. insertion_sort() is slow and simple enough to check by reading, so asserting that it agrees with sorted() pins down the elements, their multiplicities, and their order at once. The oracle earns its place because it repeats no part of sorted()’s implementation. It arrives at the same answer by a different route. That independence makes an oracle worth having, and makes assert sorted(xs) == sorted(xs) worthless. Capping the list length keeps the quadratic oracle cheap, since the bugs it catches show up on short inputs.

from hypothesis import given, strategies

@given(strategies.text())
def test_upper_lower_agrees_with_lower(s: str) -> None:
    assert s.upper().lower() == s.lower()

Hypothesis falsifies it in well under a second and shrinks to a one-character string:

s = 'µ'

    @given(strategies.text())
    def test_upper_lower_agrees_with_lower(s: str) -> None:
>       assert s.upper().lower() == s.lower()
E       AssertionError: assert 'μ' == 'µ'
E         - µ
E         + μ
E       Failing test case: test_upper_lower_agrees_with_lower(
E           s='µ',
E       )

The two sides print almost identically, so the failure hides until you look at the code points.

Most runs shrink to 'µ'. Now and then a run stops at 'ß' instead, since the shrinker does not always find the smallest failing character; ß breaks the law for a different reason, covered below. Once a run fails, Hypothesis stores the counterexample in .hypothesis/ and replays it first, so later runs report the same character until you delete that directory.

# test_case_mapping.py
import unicodedata

MICRO = "µ"

def test_upper_leaves_the_micro_sign_in_the_greek_block(
) -> None:
    assert unicodedata.name(MICRO) == "MICRO SIGN"
    assert (unicodedata.name(MICRO.upper())
            == "GREEK CAPITAL LETTER MU")
    assert (unicodedata.name(MICRO.upper().lower())
            == "GREEK SMALL LETTER MU")
    # Already lowercase, so unchanged
    assert MICRO.lower() == MICRO
    assert MICRO.upper().lower() != MICRO.lower()

Trace the round trip. µ is U+00B5 MICRO SIGN, a character Latin-1 kept separate from the Greek letter it resembles. µ is lowercase, so .lower() returns it unchanged. But it has no uppercase form of its own, so .upper() maps it to U+039C GREEK CAPITAL LETTER MU, and lowering that gives U+03BC GREEK SMALL LETTER MU. The round trip ends one block away from where it started.

Unicode case mapping is not a pair of inverse functions. It is a many-to-one mapping in each direction, over a repertoire containing characters that are lowercase without being the lowercase of anything. ß breaks the same law from the other side: "ß".upper() is "SS", two characters, so uppercasing can change a string’s length. For case-insensitive comparison Python provides str.casefold() rather than str.lower(), and casefold() has no inverse either.

A hand-written loop over "abcde" does not reach µ. The generated strings reach the parts of the repertoire nobody thinks to type, and that reach is the argument for property testing in one example.

If you compare placed == names without sorting either side, the property fails against the correct group_rounds(), and Hypothesis shrinks the failure to names=['a', 'b'], size=2. group_rounds() shuffles the pool before it forms each round, so the students come back in an order of their own. The property concerns which students each round places, not their order, so the solution sorts both sides before comparing them.

# test_group_rounds.py
import random
from collections import Counter
from collections.abc import Iterator
from itertools import combinations, islice
from hypothesis import given, strategies

type Group = tuple[str, ...]
type Round = list[Group]

def group_rounds(
    students: list[str], size: int, seed: int = 0
) -> Iterator[Round]:
    history: Counter[frozenset[str]] = Counter()
    rng = random.Random(seed)

    def met(group: list[str], candidate: str) -> int:
        return sum(history[frozenset((m, candidate))]
                   for m in group)

    while True:
        pool = list(students)
        rng.shuffle(pool)
        groups: list[list[str]] = []
        while len(pool) >= size:
            leader = pool.pop()
            group = [leader]
            while len(group) < size:
                stranger = min(pool,
                               key=lambda c: met(group, c))
                pool.remove(stranger)
                group.append(stranger)
            groups.append(group)
        # Roster smaller than one group
        if pool and not groups:
            groups.append([])
        # Too few left for a full group of `size`
        for extra in pool:
            host = min(groups, key=lambda g: met(g, extra))
            host.append(extra)
        round_result: Round = [tuple(g) for g in groups]
        for g in round_result:
            for pair in combinations(g, 2):
                history[frozenset(pair)] += 1
        yield round_result

rosters = strategies.lists(
    strategies.text("abcdefghij", min_size=1, max_size=3),
    min_size=2, max_size=12, unique=True)

@given(rosters,
       strategies.integers(min_value=2, max_value=5))
def test_every_student_appears_once_per_round(
        names: list[str], size: int) -> None:
    for grouping in islice(group_rounds(names, size), 3):
        placed = [*group for group in grouping]
        assert sorted(placed) == sorted(names)

Handle a roster smaller than a group. The two lines guarding an empty groups are the interesting part, because the property test finds the need for them. Against the version without them, Hypothesis reports a two-name roster with size=3, such as names=['a', 'b'], and a ValueError: min() iterable argument is empty. With fewer students than the group size, the while len(pool) >= size loop runs zero times and groups stays empty. The leftover loop then asks min() for the smallest of nothing.

The crash is a real defect rather than an unstated precondition. group_rounds() already keeps everyone in a group when a roster divides unevenly, folding the leftovers into existing groups, so the answer for a roster of two and a size of five is one group of two. Crashing is the one answer inconsistent with what the function does everywhere else. The group_rounds() in Toolkits includes the guard, so test_group_rounds.py passes with no assume().

Finding the defect takes no cleverness and no thought about edge cases. The strategy generates small rosters because Hypothesis prefers small examples, so a size of 3 against a roster of 2 comes up on its own. The property says what should be true for every roster.

State the precondition in the strategy. The unique=True on the roster strategy does real work. group_rounds() keys its history by frozenset of names, so two students sharing a name are one student to the algorithm. The property still passes on such a roster, because every name appears in one group, but the schedule the property checks counts the two as one student when group_rounds() avoids repeat meetings. Generating distinct names states group_rounds()’s precondition where the test can see it.

Breaking the function on purpose is the other half of the exercise. Delete the loop that places leftovers:

        # Too few left for a full group of `size`
        for extra in pool:
            host = min(groups, key=lambda g: met(g, extra))
            host.append(extra)

group_rounds() now drops the students who do not fill a whole group, and the property reports the loss at once:

E           AssertionError: assert ['a', 'b'] == ['a', 'aa', 'b']
E             At index 1 diff: 'b' != 'aa'
E             Right contains one more item: 'b'
E           Failing test case: test_every_student_appears_once_per_round(
E               names=['a', 'b', 'aa'],
E               size=2,
E           )

Three students in groups of two leaves one student over, and the report shows that shrunk case rather than whatever wide random roster failed first. The names vary from run to run: Hypothesis shrinks a generated string toward a longer run of a before it reaches a third letter, so 'aa' arrives as readily as 'c' would. Some runs shrink to a different shape, a roster too small to fill one group: names=['a', 'aa'], size=3, where the empty group the guard adds collects nobody and the left side of the assertion is [].

A second run reports the identical counterexample, and reports it noticeably faster. Hypothesis writes each failing case into .hypothesis/examples/ and replays that database before generating anything new, so it re-finds the failure you are in the middle of fixing instead of leaving it to chance. The shrunk case therefore behaves like a regression test you did not write. It keeps failing until you fix the bug, then rejoins the pool of examples and comes up again on every later run.