Skip to content

Property Testing (The Caretaker Who Checks Every Locker)

The School Caretaker

Remember the school from Memory? Lockers in a long hallway (frames), handed out to students (processes) by the principal (the OS), and a supply closet split into little compartments (the slab allocator). Let's give the school a lift too, like the elevator in disk scheduling: it has to stop at every floor someone pressed (the disk arm and its cylinders).

How would you check the school works? The usual way: give one student two lockers, see two lockers with their name on, and smile. That's an example test. It only tells you about the one student you thought of.

Now imagine a caretaker who plays out thousands of made-up school days all night: a student with no lockers, one who asks for more lockers than exist, two sharing one, the lift called to the floor it's on. For each they check a single rule, such as:

Whatever floors are pressed, the lift stops at every one of them exactly once.

That rule is a property. It doesn't say which floor comes first, only something that must be true for every set of button presses.

When the caretaker finds something that breaks the rule, they don't just write "something went wrong on Tuesday, with forty students". They play the day again with less and less -- fewer students, fewer buttons, smaller numbers -- until it's as simple as it can be and still goes wrong. That boiling-down is called shrinking, and the smallest failing input usually points straight at the bug.

In Python, the caretaker is a library called Hypothesis. PyOS's property tests live in tests/test_properties.py, next to the ordinary example tests, and the model check (more on that below) is in tests/test_model.py.

A Property from PyOS

PyOS has four disk scheduling policies: FCFS, SSTF, SCAN and C-SCAN (SCAN and C-SCAN can each start by sweeping up or down, so that's six to test). Each picks a different order for the disk arm to visit the cylinders it's been asked for, but no request may be lost and none may be made up:

# SCANPolicy and CSCANPolicy default to a disk with cylinders 0 to 199.
HIGHEST_CYLINDER = 199
# Every disk-arm policy, by a name Hypothesis can print when a test fails.
DISK_POLICIES: dict[str, DiskPolicy] = {
    "FCFS": FCFSPolicy(),
    "SSTF": SSTFPolicy(),
    "SCAN up": SCANPolicy(direction="up"),
    "SCAN down": SCANPolicy(direction="down"),
    "C-SCAN up": CSCANPolicy(direction="up"),
    "C-SCAN down": CSCANPolicy(direction="down"),
}


def cylinders() -> st.SearchStrategy[int]:
    """Make a cylinder number somewhere on the disk."""
    return st.integers(min_value=0, max_value=HIGHEST_CYLINDER)


@st.composite
def requests_and_head(draw: st.DrawFn) -> tuple[list[int], int]:
    """Make a list of cylinder requests, and a head position that is often one of them."""
    requests = draw(st.lists(cylinders()))
    # The arm already sitting on a requested cylinder is a classic edge case,
    # but with 200 cylinders a random head only rarely lands on one. So half
    # the time, put the head on a request on purpose.
    if requests and draw(st.booleans()):
        return requests, draw(st.sampled_from(requests))
    return requests, draw(cylinders())


def disk_policy_names() -> st.SearchStrategy[str]:
    """Pick the name of one of the disk-arm policies, sweeping up or down."""
    return st.sampled_from(list(DISK_POLICIES))


@given(policy_name=disk_policy_names(), requests_and_head=requests_and_head())
def test_disk_policies_serve_every_request_exactly_once(
    policy_name: str, requests_and_head: tuple[list[int], int]
) -> None:
    """Whatever the policy and wherever the arm starts, it visits each request once: none lost, none added."""
    requests, head = requests_and_head
    order = DISK_POLICIES[policy_name].schedule(list(requests), head=head)
    assert sorted(order) == sorted(requests)

Line by line:

  • cylinders(), requests_and_head() and disk_policy_names() are strategies: recipes for making random inputs, each a named function so the test reads like a sentence. st.integers(min_value=0, max_value=199) makes any cylinder, and st.lists(...) makes a list of them: empty, one request, the same cylinder twice, forty of them.
  • DISK_POLICIES names each policy ("SCAN up"), so a failing test can print which one it was.
  • @st.composite lets one strategy use another. requests_and_head() draws the requests, then the head -- half the time on one of the requests. A version with a plain random head missed the bug in "Try It Yourself" below in its normal 100-example run, because a random head hardly ever lands on a requested cylinder. A property only finds what its strategy can make.
  • @given(...) runs the test many times (100 in a normal run), each time with a fresh policy, list and head.
  • The docstring says the rule in plain English. If you can't say it in one sentence, it probably isn't a good property yet.
  • The assert is the rule itself. Sorting both lists throws away the order, leaving "the same cylinders, the same number of times". sorted is Python's own sort, which we trust, and it's not how any policy works.

Notice what the test doesn't say: never "with the head at 53, SSTF goes to 65 first". It needs no answers, only a rule that's always true. FCFS, which just keeps the arrival order, passes it, and that's fine: being clever is a different rule. A do-nothing scheduler that returns an empty list fails as soon as there's one request.

PyOS has three more properties:

Rule Test
Write each object into its own new slot; each slot reads back its bytes (as bytes), padded with zeros test_slab_write_then_read_gives_back_each_objects_bytes
After a cache is destroyed, memory looks exactly as it did before the cache was made test_destroying_a_slab_cache_gives_back_every_frame
Picking until the queue is empty gives the processes in priority order, oldest first on a tie test_priority_policy_picks_most_important_first_and_oldest_on_ties

The first is a round-trip: what you write in, you read back out. The last checks the priority scheduler against Python's sorted, with no kernel boot, no clock and no waiting.

Checking Against a Model

A property checks one action. But the memory manager changes step by step: hand out frames, share one on fork, take one back, free everything on exit. Bugs often hide in the order of steps -- freeing a process that never had memory, or sharing a frame that's already been given back.

So the caretaker carries a clipboard, noting by hand which student holds which locker, and after every step compares the lockers with it. If the hallway ever disagrees with the clipboard, something's wrong with the principal.

That clipboard is a model: something simple we already trust. For MemoryManager, the model is a plain Python dict from each PID to the list of frames in its page table. Here's the heart of tests/test_model.py:

@settings(stateful_step_count=20)
class MemoryManagerMachine(RuleBasedStateMachine):
    frames = Bundle("frames")

    def __init__(self) -> None:
        """Start with all frames free and nobody holding anything."""
        super().__init__()
        self.memory = MemoryManager(total_frames=TOTAL_FRAMES)
        self.model: dict[int, list[int]] = {}

    def frames_in_use(self) -> Counter[int]:
        return Counter(frame for table in self.model.values() for frame in table)

    @rule(target=frames, pid=pids())
    def allocate_one(self, pid: int) -> MultipleResults[int]:
        """Give a process one new frame, or refuse with OutOfMemoryError if none is free."""
        if len(self.frames_in_use()) == TOTAL_FRAMES:
            with pytest.raises(OutOfMemoryError):
                self.memory.allocate_one(pid)
            return multiple()
        frame = self.memory.allocate_one(pid)
        assert frame not in self.frames_in_use()
        self.model.setdefault(pid, []).append(frame)
        return multiple(frame)

    @invariant()
    def frames_match_the_dict(self) -> None:
        """After every step, page tables, refcounts and the free count all match the dict."""
        in_use = self.frames_in_use()
        for pid in range(HIGHEST_PID + 1):
            assert self.memory.pages_for(pid) == self.model.get(pid, [])
        for frame in range(TOTAL_FRAMES):
            assert self.memory.refcount(frame) == in_use[frame]
        assert self.memory.free_frames == TOTAL_FRAMES - len(in_use)


TestMemoryManagerModel = cast(
    "type[unittest.TestCase]",
    MemoryManagerMachine.TestCase,  # pyright: ignore[reportUnknownMemberType]
)

Piece by piece:

  • RuleBasedStateMachine is Hypothesis's tool for things that change step by step: it makes up a random story of steps and plays it.
  • self.memory and self.model are the hallway and the clipboard, and start with all 8 frames free and an empty dict.
  • frames_in_use() works out everything else from the clipboard: if processes 1 and 2 both list frame 3, its refcount must be 2, and a frame nobody lists must be free.
  • Each @rule is one kind of step, done to both the manager and the dict. allocate_one also checks the manager's answer: the frame must be one nobody holds, and when memory is full, pytest.raises checks the manager refuses with OutOfMemoryError. The real file has rules to allocate several frames, share a frame (as fork does), unshare one, and free a whole process.
  • frames = Bundle("frames") is a bag of frame numbers handed out earlier in the story. allocate_one puts its frame in the bag (target=frames, and multiple(frame) means "add this one"; multiple() means "add nothing"). The share and unshare rules pick either a frame from the bag or any frame at all, so they try frames that are in use and frames that aren't. Two more rules, share_again and unshare_held, pick a frame the process already holds: random picks rarely give one process the same frame twice, and an unshare_frame that took away every copy slipped past without them.
  • The @invariant() runs after every step and compares every page table, every refcount, and the free count. (The real one also checks shared_frame_count.)
  • TestMemoryManagerModel = ...TestCase turns the machine into a test pytest can run: 100 random stories. @settings(stateful_step_count=20) keeps each story to 20 steps, which find the same bugs as the default 50 in a fraction of the time.

The dict is a good model because it's obviously right and it's not MemoryManager's code: no free set, no refcount table, it just counts. A model copied from the manager would share its bugs, and the two would agree perfectly about the wrong answer.

Try It Yourself

Let's break the SCAN policy on purpose and watch the caretaker find it.

Open src/py_os/io/disk.py, find SCANPolicy.schedule, and change the first line of the "up" sweep, as if you'd typed > instead of >=:

            up = sorted(r for r in requests if r > head)  # was: r >= head

Run the property tests and the model check:

uv run pytest tests/test_properties.py tests/test_model.py --no-cov

Hypothesis reports this (trimmed to the interesting part):

policy_name = 'SCAN up', requests_and_head = ([0], 0)

    @given(policy_name=disk_policy_names(), requests_and_head=requests_and_head())
    def test_disk_policies_serve_every_request_exactly_once(
        policy_name: str, requests_and_head: tuple[list[int], int]
    ) -> None:
        """Whatever the policy and wherever the arm starts, it visits each request once: none lost, none added."""
        requests, head = requests_and_head
        order = DISK_POLICIES[policy_name].schedule(list(requests), head=head)
>       assert sorted(order) == sorted(requests)
E       assert [] == [0]
E
E         Right contains one more item: 0
E       Failing test case: test_disk_policies_serve_every_request_exactly_once(
E           policy_name='SCAN up',
E           requests_and_head=([0], 0),
E       )

Look how small that is. Hypothesis tried long lists of requests on all six policies, then shrank the failure to one request, on the cylinder the arm is already on: the lift is on floor 0, someone on floor 0 presses the button, and the doors never open. The up sweep now takes only requests above the head, the down sweep only those below it, and a request at the head falls in the gap.

Only test_disk_policies_serve_every_request_exactly_once fails; the rest never touch the disk. The example tests all still pass. Only the property notices. Every example in tests/io/test_disk_scheduling.py uses a head that isn't one of the requests, like the textbook's head at 53.

Put the line back to r >= head and run the tests again. They pass.

Older Hypothesis versions

Older versions of Hypothesis print Falsifying example: instead of Failing test case:. It means the same thing.

A Bug We Found

When we first ran test_destroying_a_slab_cache_gives_back_every_frame, it failed on PyOS's real code -- no deliberate bug needed:

E       assert [0] == []
E
E         Left contains one more item: 0
E       Failing test case: test_destroying_a_slab_cache_gives_back_every_frame(
E           obj_size=1,  # or any other generated value
E           how_many=1,
E       )

Make a slab cache, allocate one object, destroy the cache, and the kernel's page table still lists frame 0, whatever the object size.

Here's why. A slab cache borrows its frames from the memory manager, charged to the kernel's PID. destroy_cache dropped each one's refcount, which put it back on the free list, but never took it out of the kernel's page table. So frame 0 was free and claimed by the kernel at the same time: a locker that's empty on the principal's list but still has the kernel's name on the door. If the kernel's memory were ever freed, it would "free" frame 0 again, even if a process had been given it in the meantime.

The example test did check that frames came back: mm.free_frames > free_before. And they did! What was wrong was a different part of the picture, which no example asked about. The property's rule was "memory looks exactly as it did before", so it checked both.

The fix was one line, the same pair of calls that shared memory already uses when it gives frames back:

            self._memory.decrement_refcount(frame)
            self._memory.unshare_frame(pid=self._kernel_pid, frame=frame)  # was: missing

The shrunk input also became an ordinary example test in tests/memory/test_slab.py, so this bug can never sneak back in, even if someone deletes the property:

    def test_destroy_cache_removes_frames_from_kernel_page_table(self) -> None:
        mm = MemoryManager(total_frames=TOTAL_FRAMES)
        alloc = SlabAllocator(memory=mm, page_size=PAGE_SIZE, kernel_pid=KERNEL_PID)
        alloc.create_cache("tmp", obj_size=1)
        alloc.allocate("tmp")
        alloc.destroy_cache("tmp")
        assert mm.pages_for(KERNEL_PID) == []
        mm.free(KERNEL_PID)
        assert mm.free_frames == TOTAL_FRAMES

Hunting for more bugs like it turned up a second one with the same missing unshare_frame, in copy-on-write after fork. Reviewing that fix found a third: a write to a copy-on-write page that nobody else held any more (its refcount already 1) still made a copy, and failed with OutOfMemoryError when memory was full. Now it writes in place. All three are fixed.

Run It Yourself

Run just the property tests and the model check (100 examples each, the same every time):

uv run pytest tests/test_properties.py tests/test_model.py --no-cov

--no-cov turns off the coverage check, which expects the whole suite and would fail on two files.

Run them the long way (2,000 random examples each, different every run) to hunt for rarer bugs. It takes about 40 seconds:

uv run pytest tests/test_properties.py tests/test_model.py --no-cov --hypothesis-profile=thorough

The normal run uses a fixed seed, so it tries the same inputs every time and a passing build means the same on every computer. The thorough run is random: each time, the caretaker plays a brand-new set of school days. Both settings live in conftest.py at the top of the project.

Your Turn

Try writing your own property. SSTF ("shortest seek time first") always goes to the nearest request next. So: whatever the requests and wherever the head starts, SSTF's first stop is as close to the head as any request. What's the strategy? What's the assert? (Careful: an empty list has no first stop.) Then break SSTFPolicy on purpose and see what Hypothesis shrinks it to.