os/test_models.py

配套源码，运行方法见同目录 README。返回实验总览。

import unittest

from models import MAXVA, PAGE_SIZE, MemoryFault, OutOfMemory, PhysicalMemory, Process, Sv39, split_va


class AddressTranslationTests(unittest.TestCase):
    def test_indexes_are_nine_bit_fields(self):
        va = (3 << 30) | (5 << 21) | (7 << 12) | 0xABC
        self.assertEqual(split_va(va), (3, 5, 7, 0xABC))

    def test_four_kib_page_offset_is_preserved(self):
        table = Sv39()
        table.map(0x4000, 0x80002000, write=True)
        self.assertEqual(table.translate(0x4123, "write"), 0x80002123)

    def test_two_mib_leaf_preserves_twenty_one_bit_offset(self):
        table = Sv39()
        table.map(0x200000, 0x80400000, level=1)
        self.assertEqual(table.translate(0x3FFFFF), 0x805FFFFF)

    def test_one_gib_leaf(self):
        table = Sv39()
        table.map(0x40000000, 0x80000000, level=2)
        self.assertEqual(table.translate(0x456789AB), 0x856789AB)

    def test_permissions_and_user_bit(self):
        table = Sv39()
        table.map(0, 0x80000000, read=False, execute=True)
        table.map(PAGE_SIZE, 0x80001000, user=False)
        self.assertEqual(table.translate(0, "execute"), 0x80000000)
        with self.assertRaises(MemoryFault):
            table.translate(0, "read")
        with self.assertRaises(MemoryFault):
            table.translate(PAGE_SIZE)
        self.assertEqual(table.translate(PAGE_SIZE, from_user=False), 0x80001000)

    def test_missing_and_out_of_range_addresses(self):
        table = Sv39()
        for address in (0, -1, MAXVA):
            with self.assertRaises(MemoryFault):
                table.translate(address)

    def test_alignment_and_illegal_pte_permissions(self):
        table = Sv39()
        with self.assertRaises(MemoryFault):
            table.map(0, PAGE_SIZE, level=1)
        with self.assertRaises(MemoryFault):
            table.map(0, 0, read=False, write=True)

    def test_huge_page_overlap_cannot_replace_mapping(self):
        table = Sv39()
        table.map(0, 0x80000000, level=1)
        with self.assertRaises(MemoryFault):
            table.map(PAGE_SIZE, 0x90000000)
        self.assertEqual(table.translate(PAGE_SIZE), 0x80001000)


class CopyOnWriteTests(unittest.TestCase):
    def setUp(self):
        self.memory = PhysicalMemory(capacity=8)
        self.parent = Process(self.memory)
        self.parent.map_zero(0)
        self.parent.write(0, b"initial")

    def test_fork_does_not_copy_data_pages(self):
        child = self.parent.fork()
        self.assertEqual(len(self.memory.pages), 1)
        self.assertEqual(self.memory.copied_bytes, 0)
        self.assertEqual(child.read(0, 7), b"initial")
        self.assertEqual(list(self.memory.refs.values()), [2])
        self.assertFalse(self.parent.mappings[0].writable)

    def test_child_write_preserves_parent_and_untouched_bytes(self):
        child = self.parent.fork()
        child.write(0, b"new")
        self.assertEqual(child.read(0, 7), b"newtial")
        self.assertEqual(self.parent.read(0, 7), b"initial")
        self.assertEqual(self.memory.copied_bytes, PAGE_SIZE)
        child.write(0, b"abc")
        self.assertEqual(self.memory.copied_bytes, PAGE_SIZE)

    def test_parent_write_preserves_child(self):
        child = self.parent.fork()
        self.parent.write(0, b"P")
        self.assertEqual(child.read(0, 1), b"i")

    def test_fork_of_already_cow_mapping_preserves_cow_bit(self):
        child = self.parent.fork()
        grandchild = child.fork()
        self.assertEqual(list(self.memory.refs.values()), [3])
        grandchild.write(0, b"G")
        self.assertEqual(self.parent.read(0, 1), b"i")
        self.assertEqual(child.read(0, 1), b"i")

    def test_original_readonly_page_remains_readonly_after_fork(self):
        self.parent.map_zero(PAGE_SIZE, writable=False)
        child = self.parent.fork()
        self.assertFalse(child.mappings[1].cow)
        with self.assertRaises(MemoryFault):
            child.write(PAGE_SIZE, b"x")

    def test_last_reference_can_become_writable_without_copy(self):
        child = self.parent.fork()
        child.close()
        self.parent.write(0, b"P")
        self.assertEqual(self.memory.copied_bytes, 0)
        self.assertEqual(len(self.memory.pages), 1)

    def test_exit_order_never_frees_shared_live_page(self):
        child = self.parent.fork()
        self.parent.close()
        self.assertEqual(child.read(0, 7), b"initial")
        child.close()
        self.assertEqual(self.memory.pages, {})
        self.assertEqual(self.memory.refs, {})
        child.close()  # Idempotent teardown.

    def test_oom_during_cow_preserves_mapping_and_data(self):
        self.memory.capacity = 1
        child = self.parent.fork()
        old_page = child.mappings[0].page
        with self.assertRaises(OutOfMemory):
            child.write(0, b"X")
        self.assertEqual(child.mappings[0].page, old_page)
        self.assertEqual(child.read(0, 7), b"initial")
        self.assertEqual(self.memory.refs[old_page], 2)

    def test_kernel_copyout_resolves_cow_on_each_crossed_page(self):
        self.parent.map_zero(PAGE_SIZE)
        child = self.parent.fork()
        child.copyout(PAGE_SIZE - 2, b"ABCD")
        self.assertEqual(child.read(PAGE_SIZE - 2, 4), b"ABCD")
        self.assertEqual(self.parent.read(PAGE_SIZE - 2, 4), bytes(4))
        self.assertEqual(self.memory.copied_bytes, 2 * PAGE_SIZE)

    def test_unmapped_write_does_not_allocate(self):
        before = self.memory.allocations
        with self.assertRaises(MemoryFault):
            self.parent.write(3 * PAGE_SIZE, b"x")
        self.assertEqual(self.memory.allocations, before)

    def test_repeated_forks_writes_and_exits_reclaim_every_page(self):
        children = [self.parent.fork() for _ in range(4)]
        for index, child in enumerate(children):
            child.write(0, bytes([index]))
        self.assertEqual(self.parent.read(0, 7), b"initial")
        for child in children:
            child.close()
        self.parent.close()
        self.assertEqual(len(self.memory.pages), 0)
        self.assertEqual(self.memory.refs, {})


if __name__ == "__main__":
    unittest.main()