Update coverage badges [skip ci]

tests/layout/test_page_buffer.py

@@ -0,0 +1,237 @@
+"""
+Tests for the page caching layer.
+
+Covers PageBuffer's LRU behaviour and BufferedPageRenderer's cache hits, plus
+regressions for S12/R1/R2: the module must not start worker processes and must
+not do blocking work in a finaliser.
+"""
+
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+from pyWebLayout.layout.page_buffer import PageBuffer, BufferedPageRenderer
+from pyWebLayout.layout.ereader_layout import RenderingPosition
+from pyWebLayout.abstract.block import Paragraph
+from pyWebLayout.abstract.inline import Word
+from pyWebLayout.style import Font
+from pyWebLayout.style.page_style import PageStyle
+
+
+# ============================================================================
+# Fixtures
+# ============================================================================
+
+@pytest.fixture
+def sample_blocks():
+    """A document long enough to paginate over several pages."""
+    font = Font()
+    blocks = []
+    for p in range(6):
+        para = Paragraph(style=font)
+        for w in range(120):
+            para.add_word(Word(f"p{p}w{w}", font))
+        blocks.append(para)
+    return blocks
+
+
+@pytest.fixture
+def renderer(sample_blocks):
+    return BufferedPageRenderer(sample_blocks, PageStyle(), buffer_size=3, page_size=(800, 600))
+
+
+# ============================================================================
+# PageBuffer
+# ============================================================================
+
+class TestPageBuffer:
+    def test_get_page_misses_when_empty(self):
+        buf = PageBuffer(buffer_size=3)
+        assert buf.get_page(RenderingPosition()) is None
+
+    def test_cache_page_round_trips(self):
+        buf = PageBuffer(buffer_size=3)
+        pos, nxt = RenderingPosition(block_index=0), RenderingPosition(block_index=1)
+        sentinel = object()
+
+        buf.cache_page(pos, sentinel, nxt)
+
+        assert buf.get_page(pos) is sentinel
+        assert buf.position_map[pos] == nxt
+
+    def test_lru_evicts_oldest_and_cleans_position_map(self):
+        buf = PageBuffer(buffer_size=2)
+        positions = [RenderingPosition(block_index=i) for i in range(4)]
+        for i, pos in enumerate(positions):
+            buf.cache_page(pos, object(), RenderingPosition(block_index=i + 1))
+
+        assert buf.get_page(positions[0]) is None, "oldest should have been evicted"
+        assert positions[0] not in buf.position_map, "position map must not leak evicted entries"
+        assert buf.get_page(positions[-1]) is not None
+
+    def test_get_page_refreshes_lru_order(self):
+        buf = PageBuffer(buffer_size=2)
+        a, b, c = (RenderingPosition(block_index=i) for i in range(3))
+        buf.cache_page(a, object())
+        buf.cache_page(b, object())
+
+        buf.get_page(a)          # a becomes most recently used
+        buf.cache_page(c, object())
+
+        assert buf.get_page(a) is not None, "recently used entry should survive"
+        assert buf.get_page(b) is None, "least recently used entry should be evicted"
+
+    def test_backward_pages_land_in_the_backward_buffer(self):
+        buf = PageBuffer(buffer_size=3)
+        start, end = RenderingPosition(block_index=1), RenderingPosition(block_index=2)
+
+        buf.cache_page(start, object(), end, is_backward=True)
+
+        assert start in buf.backward_buffer
+        assert start not in buf.forward_buffer
+        assert buf.reverse_position_map[end] == start
+
+    def test_font_scale_change_invalidates(self):
+        buf = PageBuffer(buffer_size=3)
+        pos = RenderingPosition()
+        buf.cache_page(pos, object(), RenderingPosition(block_index=1))
+
+        buf.set_font_scale(1.5)
+
+        assert buf.get_page(pos) is None
+
+    def test_same_font_scale_keeps_cache(self):
+        buf = PageBuffer(buffer_size=3)
+        pos = RenderingPosition()
+        buf.cache_page(pos, object(), RenderingPosition(block_index=1))
+
+        buf.set_font_scale(1.0)
+
+        assert buf.get_page(pos) is not None
+
+    def test_shutdown_is_idempotent(self):
+        buf = PageBuffer(buffer_size=3)
+        buf.cache_page(RenderingPosition(), object())
+
+        buf.shutdown()
+        buf.shutdown()
+
+        assert buf.get_cache_stats()['forward_buffer_size'] == 0
+
+
+# ============================================================================
+# BufferedPageRenderer
+# ============================================================================
+
+class TestBufferedPageRenderer:
+    def test_render_page_returns_a_page_and_advances(self, renderer):
+        page, next_pos = renderer.render_page(RenderingPosition(), 1.0)
+
+        assert page is not None
+        assert next_pos != RenderingPosition()
+
+    def test_second_render_of_same_position_is_served_from_cache(self, renderer):
+        pos = RenderingPosition()
+        first, first_next = renderer.render_page(pos, 1.0)
+        second, second_next = renderer.render_page(pos, 1.0)
+
+        assert second is first, "identical page object means it came from the cache"
+        assert second_next == first_next
+
+    def test_font_scale_change_forces_a_re_render(self, renderer):
+        pos = RenderingPosition()
+        first, _ = renderer.render_page(pos, 1.0)
+        scaled, _ = renderer.render_page(pos, 1.5)
+
+        assert scaled is not first
+
+    def test_backward_render_round_trips_to_the_original_position(self, renderer):
+        start = RenderingPosition()
+        _, second_page_pos = renderer.render_page(start, 1.0)
+
+        _, back_to = renderer.render_page_backward(second_page_pos, 1.0)
+
+        assert back_to == start
+
+    def test_shutdown_clears_the_cache(self, renderer):
+        renderer.render_page(RenderingPosition(), 1.0)
+        renderer.shutdown()
+
+        assert renderer.get_cache_stats()['forward_buffer_size'] == 0
+
+
+# ============================================================================
+# S12 / R1 / R2 regressions
+# ============================================================================
+
+class TestNoBackgroundProcesses:
+    """
+    The process pool that used to live here never produced a usable page (a Page
+    holds a live PIL canvas and cannot be pickled), and on Python 3.14's
+    forkserver default it raised when driven from module-level code.
+    """
+
+    def test_module_declares_no_process_pool(self):
+        import pyWebLayout.layout.page_buffer as page_buffer
+
+        source = page_buffer.__file__
+        assert not hasattr(page_buffer, '_render_page_worker')
+        assert not hasattr(PageBuffer(), 'executor')
+        with open(source, encoding='utf-8') as fh:
+            body = fh.read().split('"""', 2)[-1]  # skip the module docstring
+        assert 'ProcessPoolExecutor' not in body
+        assert 'pickle' not in body
+
+    def test_page_buffer_has_no_finaliser(self):
+        """
+        PageBuffer.__del__ called executor.shutdown(wait=True), which deadlocked
+        the interpreter at exit. Cleanup must be explicit.
+        """
+        assert '__del__' not in vars(PageBuffer)
+
+    def test_navigation_works_without_a_main_guard(self, tmp_path):
+        """
+        R1: EreaderLayoutManager raised RuntimeError when used from module-level
+        script code, because submitting to a ProcessPoolExecutor under a
+        non-fork start method requires an `if __name__ == "__main__"` guard.
+        """
+        script = textwrap.dedent(f"""
+            from pyWebLayout.io.readers.html_extraction import parse_html_string
+            from pyWebLayout.layout.ereader_manager import EreaderLayoutManager
+
+            blocks = parse_html_string("<p>" + " ".join(f"w{{i}}" for i in range(2000)) + "</p>")
+            m = EreaderLayoutManager(blocks, page_size=(800, 600),
+                                     bookmarks_dir={str(tmp_path)!r})
+            m.get_current_page()
+            m.next_page()
+            m.previous_page()
+            m.shutdown()
+            print("OK")
+        """)
+        result = subprocess.run([sys.executable, "-c", script],
+                                capture_output=True, text=True, timeout=120)
+
+        assert result.returncode == 0, result.stderr
+        assert "OK" in result.stdout
+
+    def test_interpreter_exits_without_explicit_shutdown(self, tmp_path):
+        """
+        R2: a manager left to be finalised at exit must not hang. The timeout is
+        the assertion.
+        """
+        script = textwrap.dedent(f"""
+            from pyWebLayout.io.readers.html_extraction import parse_html_string
+            from pyWebLayout.layout.ereader_manager import EreaderLayoutManager
+
+            blocks = parse_html_string("<p>" + " ".join(f"w{{i}}" for i in range(500)) + "</p>")
+            m = EreaderLayoutManager(blocks, page_size=(800, 600),
+                                     bookmarks_dir={str(tmp_path)!r})
+            m.get_current_page()
+            # deliberately no shutdown() - rely on interpreter teardown
+        """)
+        result = subprocess.run([sys.executable, "-c", script],
+                                capture_output=True, text=True, timeout=60)
+
+        assert result.returncode == 0, result.stderr