#!/usr/bin/env python3
# python3-smmap-probe.py -- one python3 start, every measurement this suite needs.
#
# Emitted as `key=value' lines on stdout, one per line, read back by
# tests/99-python3-smmap.test through its `pv' helper. Nothing here prints to
# stderr on a healthy run: the suite asserts that stderr is empty, so a warning
# from the interpreter is a red rather than something nobody notices.
#
# WHY THE EXPECTATIONS ARE COMPUTED HERE RATHER THAN STORED.
#
# Every backing file this probe maps is written BY this probe, with the byte at
# offset i set to pat(i) = (i * 37 + 11) & 0xFF. The value at any offset is
# therefore a function of the offset alone, so a window read back at [i, j) can
# be checked against a value COMPUTED from i and j -- there is no stored copy to
# drift, and a window of any size is as cheap to assert as one byte.
#
# That makes most emissions `yes'/`no' rather than data, which on its own would
# be a suite that cannot fail: a broken comparator would emit `yes' everywhere.
# Two things stop that. `pattern_at_*' emit the generator's own output at fixed
# offsets, which the suite pins against literals; and `control_*' deliberately
# compares a correct read against the WRONG bytes and must report a mismatch. A
# comparator that cannot fail fails those.
#
# WHAT IS ARM-DEPENDENT AND WHAT IS NOT.
#
# mmap.ALLOCATIONGRANULARITY and sys.maxsize are facts about the target, and
# smmap derives its default window size and memory budget from the second of
# them. They are emitted raw AND checked for internal consistency, so the
# consistency cases hold on any arm while the raw values stay visible.
#
# The mapped files are a FIXED 200000 bytes and the managers are built with a
# FIXED 65536-byte window rather than sizes derived from the granularity, so the
# offsets, lengths and region counts below are literals on any arm whose
# granularity divides 65536.
import os
import sys
import tempfile
import shutil

EMITTED = 0
_seen = set()


def emit(k, v):
    global EMITTED
    if k in _seen:
        raise AssertionError("duplicate probe key %s" % k)
    _seen.add(k)
    s = str(v).replace("\n", " ").replace("\r", " ")
    print("%s=%s" % (k, s))
    EMITTED += 1


def yn(b):
    return "yes" if b else "no"


def attempt(k, fn):
    """Emit fn()'s value, or the exception type it raised as `!<Type>'."""
    try:
        emit(k, fn())
    except Exception as e:              # noqa: BLE001 - the type IS the measurement
        emit(k, "!" + type(e).__name__)


def pat(i):
    return (i * 37 + 11) & 0xFF


def expected(off, n):
    return bytes(pat(off + k) for k in range(n))


def tobytes(x):
    return x.tobytes() if hasattr(x, "tobytes") else bytes(x)


# --------------------------------------------------------------------------
# import
# --------------------------------------------------------------------------
import smmap                                                    # noqa: E402
import smmap.util as U                                          # noqa: E402
import smmap.mman as M                                          # noqa: E402
import smmap.buf as B                                           # noqa: E402
from mmap import ALLOCATIONGRANULARITY as G                     # noqa: E402

emit("module_file", smmap.__file__)
emit("is_package", yn(hasattr(smmap, "__path__")))
emit("version", smmap.__version__)
emit("version_info", ".".join(str(i) for i in smmap.version_info))
emit("version_info_is_tuple", yn(isinstance(smmap.version_info, tuple)))
emit("author", smmap.__author__)
emit("contact", smmap.__contact__)
emit("homepage", smmap.__homepage__)
emit("submodule_util_file", U.__file__)
emit("submodule_mman_file", M.__file__)
emit("submodule_buf_file", B.__file__)

# The four names `from .mman import *' and `from .buf import *' actually bring
# in, which is what __all__ in those two modules says and nothing more.
for n in ("StaticWindowMapManager", "SlidingWindowMapManager",
          "WindowCursor", "SlidingWindowMapBuffer"):
    emit("root_has_" + n.lower(), yn(hasattr(smmap, n)))

# THE TRAP: util's names are NOT re-exported at the package root. mman imports
# them, but a star import obeys mman.__all__, which does not list them. So
# `smmap.align_to_mmap' is an AttributeError while `smmap.util.align_to_mmap'
# is the function -- and the README's own vocabulary is util's.
for n in ("align_to_mmap", "is_64_bit", "MapWindow", "MapRegion",
          "MapRegionList", "ALLOCATIONGRANULARITY"):
    emit("root_has_" + n.lower(), yn(hasattr(smmap, n)))
    emit("util_has_" + n.lower(), yn(hasattr(U, n)))

emit("mman_all", ",".join(M.__all__))
emit("buf_all", ",".join(B.__all__))
emit("util_all", ",".join(U.__all__))
emit("root_has_all_attr", yn(hasattr(smmap, "__all__")))
# mman DOES hold the util names as module globals -- it imported them by name.
# This is what makes the root omission an export decision rather than an
# absence.
emit("mman_has_is_64_bit", yn(hasattr(M, "is_64_bit")))
emit("mman_has_mapwindow", yn(hasattr(M, "MapWindow")))

# the shipped test subpackage: find_packages() picks up smmap.test, so it is on
# the image. Namespaced under smmap, unlike a top-level one.
try:
    import smmap.test as T
    emit("test_subpackage_importable", "yes")
    emit("test_subpackage_is_package", yn(hasattr(T, "__path__")))
except Exception as e:                                          # noqa: BLE001
    emit("test_subpackage_importable", "!" + type(e).__name__)
    emit("test_subpackage_is_package", "no")

# --------------------------------------------------------------------------
# the arm
# --------------------------------------------------------------------------
emit("allocation_granularity", G)
emit("granularity_is_power_of_two", yn(G > 0 and (G & (G - 1)) == 0))
emit("granularity_divides_65536", yn(65536 % G == 0))
emit("util_granularity_is_mmap_granularity", yn(U.ALLOCATIONGRANULARITY == G))
emit("sys_maxsize", sys.maxsize)
# Not a guess at which one this arm is -- the assertion is that it is one of
# the two a CPython build can report, so the case holds on either and a third
# answer is the thing worth a red.
emit("maxsize_is_31_or_63_bit",
     yn(sys.maxsize in ((1 << 31) - 1, (1 << 63) - 1)))
emit("is_64_bit", yn(U.is_64_bit()))
# The definition, restated: this is what makes the two cases above one fact
# rather than two, on whichever arm this runs.
emit("is_64_bit_consistent", yn(U.is_64_bit() == (sys.maxsize > (1 << 32) - 1)))

MB = 1024 * 1024
_coeff_w = 1024 if U.is_64_bit() else 64
_coeff_m = 8192 if U.is_64_bit() else 1024

# SlidingWindowMapManager defaults window_size to -1, which means "pick from the
# architecture"; StaticWindowMapManager defaults it to 0, which means "map the
# whole file". Two different defaults on the same keyword.
_sl = M.SlidingWindowMapManager()
_st = M.StaticWindowMapManager()
emit("sliding_default_window_size", _sl.window_size())
emit("sliding_default_window_matches_arch", yn(_sl.window_size() == _coeff_w * MB))
emit("static_default_window_size", _st.window_size())
emit("static_default_window_is_zero", yn(_st.window_size() == 0))
emit("sliding_default_max_memory", _sl.max_mapped_memory_size())
emit("sliding_default_max_memory_matches_arch",
     yn(_sl.max_mapped_memory_size() == _coeff_m * MB))
emit("static_default_max_memory_matches_arch",
     yn(_st.max_mapped_memory_size() == _coeff_m * MB))
emit("static_explicit_minus_one_window_matches_arch",
     yn(M.StaticWindowMapManager(window_size=-1).window_size() == _coeff_w * MB))
emit("default_max_handles_is_maxsize", yn(_sl.max_file_handles() == sys.maxsize))
emit("fresh_manager_memory_size", _sl.mapped_memory_size())
emit("fresh_manager_handles", _sl.num_file_handles())
emit("fresh_manager_open_files", _sl.num_open_files())

# --------------------------------------------------------------------------
# align_to_mmap -- expressed in multiples of G, so the literals are the arm's
# own granularity counted out rather than a page size guessed in advance.
# --------------------------------------------------------------------------
_align = [
    ("0_down", 0, 0, 0), ("0_up", 0, 1, 0),
    ("1_down", 1, 0, 0), ("1_up", 1, 1, 1),
    ("g_minus_1_down", G - 1, 0, 0), ("g_minus_1_up", G - 1, 1, 1),
    ("g_down", G, 0, 1), ("g_up", G, 1, 1),
    ("g_plus_1_down", G + 1, 0, 1), ("g_plus_1_up", G + 1, 1, 2),
    ("3g_down", 3 * G, 0, 3), ("3g_up", 3 * G, 1, 3),
    ("3g_plus_7_up", 3 * G + 7, 1, 4),
]
_align_ok = True
for name, num, up, mult in _align:
    got = U.align_to_mmap(num, up)
    emit("align_%s_ratio" % name, got // G if got % G == 0 else "not-a-multiple:%d" % got)
    _align_ok = _align_ok and got == mult * G
emit("align_all_ok", yn(_align_ok))
# round_up is truthy-tested, not compared to True.
emit("align_round_up_is_truthy_test",
     yn(U.align_to_mmap(1, "x") == G and U.align_to_mmap(1, 0) == 0))

# --------------------------------------------------------------------------
# MapWindow -- pure arithmetic, no file, no arm.
# --------------------------------------------------------------------------
w = U.MapWindow(10, 20)
emit("window_ofs", w.ofs)
emit("window_size", w.size)
emit("window_ofs_end", w.ofs_end())
emit("window_repr", repr(w))
emit("window_has_slots", yn(hasattr(U.MapWindow, "__slots__")))
attempt("window_no_dict", lambda: yn(not hasattr(U.MapWindow(0, 0), "__dict__")))

# align(): ofs down to a G boundary, size up, and the original area stays
# covered. Asserted as the covering property, which is arm-independent.
w2 = U.MapWindow(G + 100, 50)
_o0, _e0 = w2.ofs, w2.ofs + w2.size
w2.align()
emit("window_align_ofs_ratio", w2.ofs // G if w2.ofs % G == 0 else "unaligned")
emit("window_align_size_ratio", w2.size // G if w2.size % G == 0 else "unaligned")
emit("window_align_still_covers", yn(w2.ofs <= _o0 and w2.ofs_end() >= _e0))

w3 = U.MapWindow(1000, 100)
w3.extend_left_to(U.MapWindow(0, 200), 10000)
emit("window_extend_left_ofs", w3.ofs)
emit("window_extend_left_size", w3.size)
w4 = U.MapWindow(1000, 100)
w4.extend_left_to(U.MapWindow(0, 200), 300)
emit("window_extend_left_capped_ofs", w4.ofs)
emit("window_extend_left_capped_size", w4.size)
w5 = U.MapWindow(1000, 100)
w5.extend_right_to(U.MapWindow(5000, 10), 10000)
emit("window_extend_right_size", w5.size)
w6 = U.MapWindow(1000, 100)
w6.extend_right_to(U.MapWindow(5000, 10), 200)
emit("window_extend_right_capped_size", w6.size)

# --------------------------------------------------------------------------
# the pattern generator itself -- the literals that stop a broken comparator
# emitting `yes' for everything below.
# --------------------------------------------------------------------------
for off in (0, 1, 2, 100, 255, 256, 4095, 65535, 199999):
    emit("pattern_at_%d" % off, pat(off))
emit("pattern_first16_hex", expected(0, 16).hex())
emit("pattern_at_65536_16_hex", expected(65536, 16).hex())
emit("pattern_period", next(p for p in range(1, 1000) if all(pat(i) == pat(i + p) for i in range(500))))

# --------------------------------------------------------------------------
# the files
# --------------------------------------------------------------------------
FSIZE = 200000
WIN = 65536
BUDGET = 2 * WIN

tmpdir = tempfile.mkdtemp(prefix="smmap-probe.")
big = os.path.join(tmpdir, "big")
empty = os.path.join(tmpdir, "empty")
one = os.path.join(tmpdir, "one")
missing = os.path.join(tmpdir, "no-such-file")

with open(big, "wb") as fh:
    fh.write(bytes(pat(i) for i in range(FSIZE)))
with open(empty, "wb") as fh:
    pass
with open(one, "wb") as fh:
    fh.write(bytes([pat(0)]))

emit("file_size", os.path.getsize(big))
emit("file_size_expected", yn(os.path.getsize(big) == FSIZE))
emit("empty_file_size", os.path.getsize(empty))
emit("one_byte_file_size", os.path.getsize(one))
emit("configured_window", WIN)
emit("configured_budget", BUDGET)
emit("window_divides_file", yn(FSIZE // WIN == 3))


def newman(window=WIN, budget=BUDGET, handles=None):
    if handles is None:
        return M.SlidingWindowMapManager(window_size=window, max_memory_size=budget)
    return M.SlidingWindowMapManager(window_size=window, max_memory_size=budget,
                                     max_open_handles=handles)


def read_via_buffer(man, path, boff, bsize, i, j):
    """Read [i, j) through a SlidingWindowMapBuffer, as bytes."""
    c = man.make_cursor(path)
    buf = B.SlidingWindowMapBuffer(c, boff, bsize)
    try:
        return tobytes(buf[i:j])
    finally:
        buf.end_access()


# --- reads --------------------------------------------------------------
man = newman()
cur = man.make_cursor(big)
emit("cursor_associated", yn(cur.is_associated()))
emit("cursor_valid_before_use", yn(cur.is_valid()))
emit("cursor_file_size", cur.file_size())
emit("cursor_path", yn(cur.path() == big))
attempt("cursor_fd_on_path_cursor", lambda: cur.fd())
emit("cursor_path_or_fd_is_str", yn(isinstance(cur.path_or_fd(), str)))

def _fresh(off=0, size=FSIZE):
    c = man.make_cursor(big)
    return B.SlidingWindowMapBuffer(c, off, size)


buf = B.SlidingWindowMapBuffer(cur, 0, FSIZE)
emit("buffer_len", len(buf))
emit("buffer_len_is_file_size", yn(len(buf) == FSIZE))
emit("buffer_cursor_is_ours", yn(buf.cursor() is cur))

# 1. entirely inside one window.
_r = tobytes(buf[100:1100])
emit("read_inside_len", len(_r))
emit("read_inside_ok", yn(_r == expected(100, 1000)))
emit("read_inside_first8_hex", _r[:8].hex())

# 2. spanning a window boundary: 65536 is where the second region begins.
_r = tobytes(buf[WIN - 500:WIN + 500])
emit("read_span_len", len(_r))
emit("read_span_ok", yn(_r == expected(WIN - 500, 1000)))
emit("read_span_seam_hex", _r[496:504].hex())

# 3. spanning THREE regions.
_r = tobytes(buf[WIN - 10:3 * WIN + 10])
emit("read_three_regions_len", len(_r))
emit("read_three_regions_ok", yn(_r == expected(WIN - 10, 2 * WIN + 20)))

# 4. exact start of file.
_r = tobytes(buf[0:64])
emit("read_at_start_len", len(_r))
emit("read_at_start_ok", yn(_r == expected(0, 64)))
emit("read_at_start_hex", _r[:16].hex())

# 5. exact end of file.
_r = tobytes(buf[FSIZE - 64:FSIZE])
emit("read_at_end_len", len(_r))
emit("read_at_end_ok", yn(_r == expected(FSIZE - 64, 64)))
emit("read_at_end_last8_hex", _r[-8:].hex())

# 6. the whole file in one slice.
_r = tobytes(buf[0:FSIZE])
emit("read_whole_len", len(_r))
emit("read_whole_ok", yn(_r == expected(0, FSIZE)))

# 7. zero-length read, and THE RETURN TYPE, which is not one type.
#
# __getslice__ has two paths and they return different things. The fast path --
# the slice lies wholly inside the region the cursor is already on -- hands back
# a memoryview slice of the mmap. The slow path joins its pieces and hands back
# bytes. So `buf[i:j]' is a memoryview or a bytes depending on where the CURSOR
# happened to be left by the previous read, and `buf[i:j] + b"x"' is a TypeError
# in exactly half the cases. Each is measured on a buffer positioned on purpose,
# because reading them off whatever the last case left behind is how this got
# mislabelled the first time.
_fastbuf = _fresh()                       # begin_access left it on region 0
_fast = _fastbuf[10:20]
emit("read_fastpath_type", type(_fast).__name__)
emit("read_fastpath_ok", yn(tobytes(_fast) == expected(10, 10)))
_slowbuf = _fresh()
_slow = _slowbuf[WIN - 5:WIN + 5]         # crosses out of region 0
emit("read_slowpath_type", type(_slow).__name__)
emit("read_slowpath_ok", yn(tobytes(_slow) == expected(WIN - 5, 10)))
emit("read_paths_differ_in_type", yn(type(_fast).__name__ != type(_slow).__name__))
# The consequence, stated as a measurement rather than as a warning.
try:
    _cat = _fast + b"x"
    emit("read_fastpath_concat", "ok")
except Exception as e:                                          # noqa: BLE001
    emit("read_fastpath_concat", "!" + type(e).__name__)
try:
    _cat = _slow + b"x"
    emit("read_slowpath_concat", "ok")
except Exception as e:                                          # noqa: BLE001
    emit("read_slowpath_concat", "!" + type(e).__name__)

_zf = _fresh()[5:5]                       # zero length, fast path
emit("read_zero_len", len(tobytes(_zf)))
emit("read_zero_is_empty", yn(tobytes(_zf) == b""))
emit("read_zero_type", type(_zf).__name__)
_zb = _fresh()
_zb[WIN - 5:WIN + 5]                      # reposition off region 0
_zs = _zb[5:5]                            # zero length, slow path
emit("read_zero_slowpath_type", type(_zs).__name__)
emit("read_zero_slowpath_is_empty", yn(tobytes(_zs) == b""))

# 8. THE EXACT-FIT READ AT EOF works -- established first, and on this buffer,
# so that what case 9 does to the buffer cannot be confused with EOF being
# unreadable.
_r = tobytes(buf[FSIZE - 10:FSIZE])
emit("read_last_10_len", len(_r))
emit("read_last_10_ok", yn(_r == expected(FSIZE - 10, 10)))

# 9. past the end. MEASURED, AND NOT WHAT THE INTERFACE SUGGESTS: an over-long
# slice does not come back short. __getslice__ loops `while l', consuming what
# each region gives it; when the last region ends at EOF the next use_region is
# past the file, comes back invalid, and the loop's own `assert c.is_valid()' is
# what stops it. So the bounds check on this interface is a debugging aid, and
# on `python3 -O' -- where asserts are compiled out -- that line is gone.
#
# Each of these gets its OWN buffer, because of what case 10 measures.
attempt("read_past_end_overlong", lambda: len(tobytes(_fresh()[FSIZE - 10:FSIZE + 1000])))
attempt("read_slice_at_eof", lambda: len(tobytes(_fresh()[FSIZE:FSIZE + 10])))
attempt("read_slice_past_eof", lambda: len(tobytes(_fresh()[FSIZE + 100:FSIZE + 200])))
attempt("int_index_past_end", lambda: _fresh()[FSIZE])
emit("asserts_enabled", yn(sys.flags.optimize == 0))

# 10. AN OVERRUN POISONS THE BUFFER. The failed use_region leaves the cursor
# with no region and nothing puts one back, so every LATER read on that same
# buffer -- including reads that were fine a moment earlier -- dies on the entry
# assert instead. A caller who catches the first AssertionError and carries on
# gets a second one from a correct read.
_p = _fresh()
emit("poison_read_before_ok", yn(tobytes(_p[100:200]) == expected(100, 100)))
try:
    _p[FSIZE - 10:FSIZE + 1000]
    emit("poison_overrun_raised", "no")
except Exception as e:                                          # noqa: BLE001
    emit("poison_overrun_raised", type(e).__name__)
emit("poison_cursor_invalid_after", yn(not _p.cursor().is_valid()))
attempt("poison_read_after", lambda: len(tobytes(_p[100:200])))
# A fresh buffer over the same file is unaffected -- the damage is per-cursor,
# not per-file and not per-manager.
emit("poison_fresh_buffer_ok", yn(tobytes(_fresh()[100:200]) == expected(100, 100)))

# 9. negative indices are relative to the buffer's size.
_r = tobytes(buf[-16:])
emit("read_negative_len", len(_r))
emit("read_negative_ok", yn(_r == expected(FSIZE - 16, 16)))

# 10. integer indexing. THE TRAP: the docstring says index 0 is the buffer's
# offset, the code indexes ABSOLUTELY into the file. For an offset-0 buffer the
# two agree, which is why it goes unnoticed; `int_index_offset_buffer_*' below
# is the same read on a buffer whose offset is not 0.
emit("int_index_0", buf[0])
emit("int_index_0_ok", yn(buf[0] == pat(0)))
emit("int_index_1000_ok", yn(buf[1000] == pat(1000)))
emit("int_index_across_window_ok", yn(buf[WIN + 7] == pat(WIN + 7)))
emit("int_index_type", type(buf[0]).__name__)
emit("int_index_negative_ok", yn(buf[-1] == pat(FSIZE - 1)))

# the negative control: a correct read compared against the WRONG bytes must
# come back as a mismatch. A comparator that cannot say `no' is a suite that
# cannot fail.
_r = tobytes(buf[100:1100])
emit("control_mismatch_detected", yn(not (_r == expected(101, 1000))))
emit("control_length_mismatch_detected", yn(not (_r == expected(100, 999))))
emit("control_pattern_not_constant", yn(len(set(expected(0, 256))) > 1))

buf.end_access()
emit("buffer_len_after_end_access", len(buf))

# --- the offset buffer: the docstring/code disagreement -------------------
OFF = 1000
c2 = man.make_cursor(big)
buf2 = B.SlidingWindowMapBuffer(c2, OFF, 5000)
emit("offset_buffer_len", len(buf2))
_r = tobytes(buf2[OFF:OFF + 32])
emit("offset_buffer_absolute_slice_ok", yn(_r == expected(OFF, 32)))
emit("int_index_offset_buffer_is_absolute", yn(buf2[OFF] == pat(OFF)))
emit("int_index_offset_buffer_is_not_relative", yn(buf2[0] != pat(OFF)))
emit("int_index_offset_buffer_0_is_file_offset_0", yn(buf2[0] == pat(0)))
buf2.end_access()

# --- accounting ----------------------------------------------------------
man2 = newman()
c3 = man2.make_cursor(big)
emit("acct_handles_before", man2.num_file_handles())
emit("acct_memory_before", man2.mapped_memory_size())
emit("acct_open_files_before", man2.num_open_files())

c3.use_region(0, 16)
emit("acct_valid_after_first_use", yn(c3.is_valid()))
emit("acct_handles_after_1", man2.num_file_handles())
emit("acct_region_size_1", c3.region().size())
emit("acct_region_size_is_window", yn(c3.region().size() == WIN))
# The same fact stated so that it survives a granularity larger than the
# configured window: MapWindow.align() rounds the size UP to a granularity
# multiple, so the region is the larger of the two. On an arm where the
# granularity divides 65536 the two cases are the same case; where it does
# not, this one stays green and names why the one above went red.
emit("acct_region_size_is_window_or_granularity",
     yn(c3.region().size() == max(WIN, G)))
emit("acct_memory_after_1", man2.mapped_memory_size())
emit("acct_memory_after_1_is_window", yn(man2.mapped_memory_size() == WIN))
emit("acct_region_ofs_begin_1", c3.region().ofs_begin())
emit("acct_region_includes_0", yn(c3.region().includes_ofs(0)))
emit("acct_region_excludes_window", yn(not c3.region().includes_ofs(WIN)))
emit("acct_cursor_ofs_begin", c3.ofs_begin())
emit("acct_cursor_size", c3.size())
emit("acct_cursor_includes_0", yn(c3.includes_ofs(0)))
emit("acct_region_client_count_1", c3.region().client_count())
emit("acct_open_files_after_1", man2.num_open_files())

c3.use_region(WIN, 16)
emit("acct_handles_after_2", man2.num_file_handles())
emit("acct_memory_after_2_is_2_windows", yn(man2.mapped_memory_size() == 2 * WIN))
emit("acct_region_ofs_begin_2", c3.region().ofs_begin())

# The budget is two windows. A third distinct region has to evict, so the
# handle count stops climbing -- this is the whole point of the manager.
c3.use_region(2 * WIN, 16)
emit("acct_handles_after_3", man2.num_file_handles())
emit("acct_handles_bounded_by_budget", yn(man2.num_file_handles() <= BUDGET // WIN))
emit("acct_memory_within_budget", yn(man2.mapped_memory_size() <= BUDGET))
emit("acct_region_ofs_begin_3", c3.region().ofs_begin())

# re-reading an EVICTED region gives the same bytes back.
_again = tobytes(c3.use_region(0, 32).buffer())
emit("acct_reread_evicted_len", len(_again))
emit("acct_reread_evicted_ok", yn(_again == expected(0, 32)))
emit("acct_reread_evicted_hex", _again[:16].hex())

c3.unuse_region()
emit("acct_valid_after_unuse", yn(c3.is_valid()))
_freed = man2.collect()
emit("acct_collect_freed_some", yn(_freed > 0))
emit("acct_handles_after_collect", man2.num_file_handles())
emit("acct_memory_after_collect", man2.mapped_memory_size())
emit("acct_handles_return_to_zero", yn(man2.num_file_handles() == 0))
emit("acct_memory_returns_to_zero", yn(man2.mapped_memory_size() == 0))
emit("acct_collect_twice_is_zero", yn(man2.collect() == 0))

# a second, sequential walk of the whole file through a bounded manager: every
# byte correct while never holding more than the budget.
man3 = newman()
c4 = man3.make_cursor(big)
_high = 0
_ok = True
_pos = 0
while _pos < FSIZE:
    c4.use_region(_pos, 4096)
    if not c4.is_valid():
        _ok = False
        break
    _d = tobytes(c4.buffer())
    _ok = _ok and _d == expected(_pos, len(_d))
    _high = max(_high, man3.mapped_memory_size())
    _pos += len(_d)
emit("walk_bytes_ok", yn(_ok))
emit("walk_end_pos", _pos)
emit("walk_covered_whole_file", yn(_pos == FSIZE))
emit("walk_peak_memory", _high)
emit("walk_peak_within_budget", yn(_high <= BUDGET))
c4.unuse_region()
man3.collect()
emit("walk_handles_after_collect", man3.num_file_handles())

# max_open_handles is honoured as its own limit, independently of memory.
man4 = newman(handles=1)
emit("acct_max_handles_1", man4.max_file_handles())
c5 = man4.make_cursor(big)
c5.use_region(0, 16)
c5.use_region(WIN, 16)
c5.use_region(2 * WIN, 16)
emit("acct_handles_capped_at_1", yn(man4.num_file_handles() <= 1))
emit("acct_capped_read_ok", yn(tobytes(c5.use_region(2 * WIN, 32).buffer())
                               == expected(2 * WIN, 32)))
c5.unuse_region()
man4.collect()

# --- the static manager: one region, the whole file ----------------------
mans = M.StaticWindowMapManager()
cs = mans.make_cursor(big)
cs.use_region(0, 16)
emit("static_region_size", cs.region().size())
emit("static_region_is_whole_file", yn(cs.region().size() == FSIZE))
emit("static_handles", mans.num_file_handles())
_r = tobytes(cs.use_region(FSIZE - 32, 32).buffer())
emit("static_read_at_end_ok", yn(_r == expected(FSIZE - 32, 32)))
emit("static_handles_after_second_use", mans.num_file_handles())
emit("static_never_grows", yn(mans.num_file_handles() == 1))
cs.unuse_region()
mans.collect()
emit("static_handles_after_collect", mans.num_file_handles())

# --- a cursor on a file descriptor ---------------------------------------
fd = os.open(big, os.O_RDONLY)
manfd = newman()
cfd = manfd.make_cursor(fd)
cfd.use_region(0, 32)
emit("fd_cursor_valid", yn(cfd.is_valid()))
emit("fd_cursor_read_ok", yn(tobytes(cfd.buffer()) == expected(0, 32)))
emit("fd_cursor_file_size", cfd.file_size())
emit("fd_cursor_fd_matches", yn(cfd.fd() == fd))
attempt("fd_cursor_path_raises", lambda: cfd.path())
emit("fd_cursor_path_or_fd_is_int", yn(isinstance(cfd.path_or_fd(), int)))
cfd.unuse_region()
manfd.collect()
os.close(fd)

# --- error paths ---------------------------------------------------------
# A path that does not exist. make_cursor itself does NOT touch the filesystem,
# so nothing raises until the first use_region asks for the size.
mane = newman()
ce = mane.make_cursor(missing)
emit("missing_make_cursor_ok", "yes")
emit("missing_associated", yn(ce.is_associated()))
attempt("missing_use_region", lambda: ce.use_region(0, 16) and "no-exception")
attempt("missing_file_size", lambda: ce.file_size())
attempt("missing_buffer", lambda: B.SlidingWindowMapBuffer(mane.make_cursor(missing), 0, 16)
        and "no-exception")

# A zero-length file. use_region does NOT raise -- offset 0 is already >= the
# size, so the cursor simply comes back invalid, and it is the BUFFER that
# refuses. A caller who checks only for an exception sees nothing wrong.
manz = newman()
cz = manz.make_cursor(empty)
emit("empty_file_size_via_cursor", cz.file_size())
cz.use_region(0, 16)
emit("empty_use_region_no_raise", "yes")
emit("empty_cursor_invalid", yn(not cz.is_valid()))
emit("empty_handles", manz.num_file_handles())
attempt("empty_buffer", lambda: B.SlidingWindowMapBuffer(manz.make_cursor(empty), 0, 16)
        and "no-exception")

# One byte: the smallest file that maps at all.
mano = newman()
co = mano.make_cursor(one)
co.use_region(0, 16)
emit("one_byte_cursor_valid", yn(co.is_valid()))
emit("one_byte_cursor_size", co.size())
emit("one_byte_region_size", co.region().size())
emit("one_byte_read_ok", yn(tobytes(co.buffer()) == expected(0, 1)))
bo = B.SlidingWindowMapBuffer(mano.make_cursor(one), 0, 1)
emit("one_byte_buffer_len", len(bo))
emit("one_byte_buffer_read_ok", yn(tobytes(bo[0:1]) == expected(0, 1)))
bo.end_access()
co.unuse_region()
mano.collect()

# offset at and past the end of a real file.
manp = newman()
cp = manp.make_cursor(big)
cp.use_region(FSIZE, 16)
emit("offset_at_eof_invalid", yn(not cp.is_valid()))
cp.use_region(FSIZE + 4096, 16)
emit("offset_past_eof_invalid", yn(not cp.is_valid()))
emit("offset_past_eof_no_handle", yn(manp.num_file_handles() == 0))
cp.use_region(FSIZE - 1, 16)
emit("offset_last_byte_valid", yn(cp.is_valid()))
emit("offset_last_byte_size", cp.size())
attempt("buffer_offset_past_eof",
        lambda: B.SlidingWindowMapBuffer(manp.make_cursor(big), FSIZE + 10, 16)
        and "no-exception")
cp.unuse_region()
manp.collect()

# a directory is not a mappable file
mand = newman()
cd = mand.make_cursor(tmpdir)
attempt("directory_use_region", lambda: cd.use_region(0, 16) and "no-exception")

# --- cursor copy semantics -----------------------------------------------
manc = newman()
cc = manc.make_cursor(big)
cc.use_region(0, 32)
import copy as _copy                                            # noqa: E402
cc2 = _copy.copy(cc)
emit("copy_valid", yn(cc2.is_valid()))
emit("copy_read_ok", yn(tobytes(cc2.buffer()) == expected(0, 32)))
emit("copy_is_distinct_object", yn(cc2 is not cc))
emit("copy_client_count", cc.region().client_count())
cc2.unuse_region()
cc.unuse_region()
manc.collect()

emit("region_has_slots", yn(hasattr(U.MapRegion, "__slots__")))
emit("cursor_has_slots", yn(hasattr(M.WindowCursor, "__slots__")))
emit("buffer_has_slots", yn(hasattr(B.SlidingWindowMapBuffer, "__slots__")))
emit("sliding_is_static_subclass",
     yn(issubclass(M.SlidingWindowMapManager, M.StaticWindowMapManager)))
emit("regionlist_is_list_subclass", yn(issubclass(U.MapRegionList, list)))

# force_map_handle_removal_win is a no-op off Windows and returns None.
emit("sys_platform", sys.platform)
emit("force_removal_win_is_none",
     yn(_sl.force_map_handle_removal_win("/") is None))

shutil.rmtree(tmpdir, ignore_errors=True)
emit("tmpdir_removed", yn(not os.path.exists(tmpdir)))

emit("emitted_count", EMITTED + 2)
emit("probe_done", "yes")