#!/usr/bin/env python3
"""flexcache 0.3 probe -- ONE python3 start, every measurement 99-python3-flexcache.test scores.

Emits `key=value' lines on stdout, one per line, no blank lines. The suite greps
each key out and compares it against a LITERAL measured on an x86-64 control
(CPython 3.12.3, against the module unpacked from the BUILT .deb) -- so a red
here is a difference between this arm and that control, not a difference from
something read off the source.

WHY THIS PACKAGE NEEDS A PROBE AT ALL, rather than a handful of shell cases.
flexcache's whole contract is a FILESYSTEM contract: a cache is correct only if
it is used when it should be and ignored when it should not, and neither half is
observable from a single call. Every case below is therefore a SEQUENCE -- build
a cache, load it twice with the conversions counted, mutate the source, load
again -- and a sequence cannot be split across interpreter starts without losing
the counter that proves the second load did not re-parse.

The count is the point. `cache-hit' asserted by timing, or by the cache file
merely existing, would pass on an arm where the cache is written and then
ignored on every load; the conversion counter makes the hit a MEASUREMENT.

TWO THINGS ARE DELIBERATELY NOT ASSERTED AS LITERALS.

  - The DiskCacheByMTime cache stem. It is sha1(converter_id + the RESOLVED
    source path), so it moves with the workdir this probe was given, and a
    literal would assert the runner's TMPDIR rather than flexcache's naming
    rule. `bymtime_stem_is_sha1_of_path' recomputes the documented digest here
    and compares, which asserts the RULE and is portable.
  - platform.python_version(). The control ran 3.12.3 and this image ships
    3.12.14, so it is emitted as a value for the record and the suite asserts
    only the 3.12 series.

The DiskCacheByHash stem IS a literal, because sha1(converter_id + the source
file's CONTENT) depends on neither path nor interpreter.
"""

import dataclasses
import hashlib
import json
import os
import pathlib
import pickle
import platform
import shutil
import sys
import tempfile
import time

OUT = []


def emit(key, value):
    """One `key=value' line, ALWAYS one line.

    A converted value here is the source file's text, which ends in a newline --
    and a raw newline would land as one counted result plus a stray line the
    runner cannot attribute, so the verdict would survive and the evidence would
    not. Backslash and newline are escaped, which is why the expectations in the
    suite read `HELLO FLEXCACHE\\n'."""
    text = str(value).replace("\\", "\\\\").replace("\n", "\\n").replace("\r", "\\r")
    OUT.append("%s=%s" % (key, text))


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


def attempt(fn):
    """`!ExceptionName' for a raise, else str(result).

    By exception NAME, not message: the contract a caller codes against is the
    class, and CPython's wording for several of these has moved between
    releases.
    """
    try:
        return str(fn())
    except BaseException as exc:          # noqa: BLE001 -- the class is the answer
        return "!" + type(exc).__name__


# ---------------------------------------------------------------- identity ---
import flexcache                                                     # noqa: E402
from flexcache import flexcache as fc_mod                            # noqa: E402

emit("module_file", getattr(flexcache, "__file__", ""))
emit("is_package", yn(hasattr(flexcache, "__path__")))
emit("version_attr", flexcache.__version__)

# __version__ comes from importlib.metadata.version(), NOT from a literal in the
# source -- so it is a statement about the INSTALLED dist-info, and it is also
# the line that drags python3-email and python3-compression in. `version_unknown'
# is the fallback the module takes when the dist-info cannot be read; it must be
# `no' here, or the package is installed in a way importlib.metadata cannot see.
emit("version_unknown", yn(flexcache.__version__ == "unknown"))
try:
    from importlib.metadata import version as _md_version
    emit("version_metadata", _md_version("flexcache"))
except Exception as exc:                                             # noqa: BLE001
    emit("version_metadata", "!" + type(exc).__name__)

emit("has___all__", yn(hasattr(flexcache, "__all__")))
emit("all_count", len(flexcache.__all__))
emit("all_sorted_joined", ",".join(sorted(flexcache.__all__)))
emit("submodule_name", fc_mod.__name__)

# The three platform facts BasicPythonHeader records into every header it names
# a cache file by. `python_version' is emitted rather than asserted -- see the
# docstring.
emit("plat_system", platform.system())
emit("plat_implementation", platform.python_implementation())
emit("plat_python_version", platform.python_version())
emit("plat_python_series", ".".join(platform.python_version().split(".")[:2]))
emit("sys_platform", sys.platform)

# NOT REACHED ON THIS INTERPRETER, AND THE SUITE SAYS SO. flexcache takes
# TypeAlias from typing_extensions only below 3.10 and Self only below 3.11;
# this image is 3.12, so neither shim fires and python3-typing-extensions --
# which the recipe DOES declare -- is not on the import path here at all.
emit("typing_extensions_imported", yn("typing_extensions" in sys.modules))

# ------------------------------------------------------------- the workdir ---
# Everything below happens under one temporary tree, removed at the end.
ROOT = tempfile.mkdtemp(prefix="bst-flexcache.")
emit("workdir_created", yn(os.path.isdir(ROOT)))

# THE cwd `cache' PATH IS SAMPLED BEFORE ANY FLEXCACHE CALL. This image ships a
# top-level /cache directory and pid 1 runs with cwd `/', so its mere presence
# says nothing about what this probe wrote; only a CHANGE across the run does.
CWD_CACHE = os.path.join(os.getcwd(), "cache")
CWD_CACHE_BEFORE = os.path.exists(CWD_CACHE)
emit("cwd_cache_preexisting", yn(CWD_CACHE_BEFORE))


def W(*parts):
    return os.path.join(ROOT, *parts)


# The conversion counter every cache-hit assertion below rests on.
CALLS = {}


def upper(path):
    """The `expensive parse'. Its __name__ IS the converter_id, so it is part of
    the cache name -- which is why it is a named def and not a lambda."""
    CALLS["upper"] = CALLS.get("upper", 0) + 1
    return pathlib.Path(path).read_text().upper()


def calls(name="upper"):
    return CALLS.get(name, 0)


# ------------------------------------------- DiskCacheByHash: the happy path --
# Named by the source's CONTENT, invalidated by the cache file merely EXISTING.
SRC = W("src.txt")
os.makedirs(W("byhash"), exist_ok=True)
with open(SRC, "w") as fo:
    fo.write("hello flexcache\n")

dc = flexcache.DiskCacheByHash(W("byhash", "cache"))
emit("byhash_folder_created", yn(os.path.isdir(W("byhash", "cache"))))
emit("byhash_folder_is_path", yn(isinstance(dc.cache_folder, pathlib.Path)))

v1, stem1 = dc.load(SRC, converter=upper)
emit("byhash_first_value", v1)
emit("byhash_calls_after_first", calls())
emit("byhash_stem_len", len(stem1))
emit("byhash_stem_is_hex", yn(all(c in "0123456789abcdef" for c in stem1)))

# THE LITERAL. sha1(converter_id + the file's bytes) -- see the docstring for why
# this one is portable and the by-mtime stem is not.
emit("byhash_stem", stem1)
emit("byhash_stem_recomputed",
     yn(stem1 == hashlib.sha1(b"upper" + b"hello flexcache\n").hexdigest()))

# THE DOCUMENTED PATH AND NAMING: <cache_folder>/<stem>.pickle, with the header
# beside it as <stem>.json because DiskCache._store_header is True.
cache_pickle = W("byhash", "cache", stem1 + ".pickle")
cache_json = W("byhash", "cache", stem1 + ".json")
emit("byhash_pickle_exists", yn(os.path.isfile(cache_pickle)))
emit("byhash_json_exists", yn(os.path.isfile(cache_json)))
emit("byhash_store_header_flag", yn(flexcache.DiskCache._store_header))
emit("byhash_hasher", flexcache.DiskCache._hasher().name)
emit("byhash_cache_entries", len(sorted(os.listdir(W("byhash", "cache")))))

# The pickle is the CONVERTED object, not the source.
with open(cache_pickle, "rb") as fi:
    emit("byhash_pickle_payload", pickle.load(fi))

# The sidecar is JSON with every header field STRINGIFIED -- `{k: str(v)}' in
# rawsave -- so `source' comes back as a path string and not as a Path.
with open(cache_json) as fi:
    hdr = json.load(fi)
emit("byhash_json_keys", ",".join(sorted(hdr)))
emit("byhash_json_converter_id", hdr["converter_id"])
emit("byhash_json_source_basename", os.path.basename(hdr["source"]))
emit("byhash_json_source_is_srcpath", yn(hdr["source"] == SRC))
emit("byhash_json_source_is_str", yn(isinstance(hdr["source"], str)))

# THE CACHE HIT, PROVED BY THE COUNTER. The converter passed here would raise if
# it ran; `byhash_calls_after_second' staying at 1 is the assertion, and the
# raising converter is the belt to its braces.
def _must_not_run(_p):
    raise AssertionError("converter ran on what should have been a cache hit")


_must_not_run.__name__ = "upper"          # same converter_id => same cache name
v2, stem2 = dc.load(SRC, converter=_must_not_run)
emit("byhash_second_value", v2)
emit("byhash_calls_after_second", calls())
emit("byhash_stem_stable", yn(stem1 == stem2))

# A SECOND DiskCacheByHash OVER THE SAME FOLDER HITS THE SAME FILE. The cache is
# on disk, not in the instance -- which is the whole point of the package.
dc_b = flexcache.DiskCacheByHash(W("byhash", "cache"))
v3, stem3 = dc_b.load(SRC, converter=_must_not_run)
emit("byhash_crossinstance_value", v3)
emit("byhash_crossinstance_stem_same", yn(stem3 == stem1))
emit("byhash_calls_after_crossinstance", calls())

# ------------------------------------------ DiskCacheByHash: INVALIDATION -----
# Naming is by CONTENT, so editing the source does not invalidate the old entry
# -- it names a DIFFERENT one. The old file stays on disk, which is the part a
# caller who expects an eviction gets wrong.
with open(SRC, "w") as fo:
    fo.write("goodbye flexcache\n")

v4, stem4 = dc.load(SRC, converter=upper)
emit("byhash_after_edit_value", v4)
emit("byhash_after_edit_reparsed", yn(calls() == 2))
emit("byhash_after_edit_calls", calls())
emit("byhash_after_edit_stem_differs", yn(stem4 != stem1))
emit("byhash_after_edit_stem", stem4)
emit("byhash_old_pickle_still_there", yn(os.path.isfile(cache_pickle)))
emit("byhash_entries_after_edit", len(sorted(os.listdir(W("byhash", "cache")))))

# ...AND THE ORIGINAL CONTENT STILL HITS ITS ORIGINAL ENTRY. Restoring the bytes
# restores the cache hit, with no re-parse -- the counter is what says so.
with open(SRC, "w") as fo:
    fo.write("hello flexcache\n")
v5, stem5 = dc.load(SRC, converter=_must_not_run)
emit("byhash_restored_value", v5)
emit("byhash_restored_stem_same", yn(stem5 == stem1))
emit("byhash_calls_after_restore", calls())

# MTIME IS NOT CONSULTED AT ALL by InvalidateByExist. A source made far NEWER
# than its cache file still hits.
os.utime(SRC, (2000000000, 2000000000))
v6, _ = dc.load(SRC, converter=_must_not_run)
emit("byhash_newer_source_still_hits", yn(v6 == "HELLO FLEXCACHE\n"))
emit("byhash_calls_after_newer_source", calls())

# ------------------------------------------ DiskCacheByMTime: naming by path --
# Named by the RESOLVED path, invalidated by `cache mtime > source mtime'.
os.makedirs(W("bymtime"), exist_ok=True)
MSRC = W("m-src.txt")
with open(MSRC, "w") as fo:
    fo.write("mtime source\n")

dm = flexcache.DiskCacheByMTime(W("bymtime", "cache"))
CALLS["upper"] = 0
m1, mstem1 = dm.load(MSRC, converter=upper)
emit("bymtime_first_value", m1)
emit("bymtime_calls_after_first", calls())
emit("bymtime_stem_len", len(mstem1))
emit("bymtime_stem_is_sha1_of_path",
     yn(mstem1 == hashlib.sha1(
         b"upper" + bytes(pathlib.Path(MSRC).resolve())).hexdigest()))
emit("bymtime_pickle_exists",
     yn(os.path.isfile(W("bymtime", "cache", mstem1 + ".pickle"))))

# THE SOURCE IS PUSHED INTO THE PAST BEFORE THE HIT IS MEASURED, DELIBERATELY.
# is_valid is a STRICT `>' on st_mtime, so on a filesystem whose mtime is
# coarse the cache written moments after the source can carry an EQUAL stamp and
# the cache never hits. That is a real risk on this image's virtiofs root and it
# is measured separately, below, as `bymtime_natural_hit' -- this case sets the
# source's mtime a day back so that it asserts flexcache's rule rather than the
# filesystem's timestamp resolution.
mcache = W("bymtime", "cache", mstem1 + ".pickle")
old = os.stat(mcache).st_mtime - 86400
os.utime(MSRC, (old, old))
m2, mstem2 = dm.load(MSRC, converter=_must_not_run)
emit("bymtime_second_value", m2)
emit("bymtime_calls_after_second", calls())
emit("bymtime_stem_stable", yn(mstem1 == mstem2))

# INVALIDATION: a source made NEWER than the cache re-parses, and rewrites the
# SAME file -- the opposite of the by-hash case, where a new name is minted.
newer = os.stat(mcache).st_mtime + 86400
with open(MSRC, "w") as fo:
    fo.write("mtime source edited\n")
os.utime(MSRC, (newer, newer))
m3, mstem3 = dm.load(MSRC, converter=upper)
emit("bymtime_after_touch_value", m3)
emit("bymtime_after_touch_reparsed", yn(calls() == 2))
emit("bymtime_after_touch_stem_same", yn(mstem3 == mstem1))
emit("bymtime_entries", len(sorted(os.listdir(W("bymtime", "cache")))))
with open(mcache, "rb") as fi:
    emit("bymtime_pickle_rewritten", pickle.load(fi))

# EQUAL MTIMES ARE NOT VALID -- the comparison is `>' and not `>='. Asserted
# directly, because it is what makes the coarse-timestamp risk above real.
same = os.stat(mcache).st_mtime
os.utime(MSRC, (same, same))
CALLS["upper"] = 0
m4, _ = dm.load(MSRC, converter=upper)
emit("bymtime_equal_mtime_reparses", yn(calls() == 1))
emit("bymtime_equal_mtime_value", m4)

# ----------------------------------- the timestamp granularity that bites -----
# THE STRICT `>' MEETS THE FILESYSTEM'S TIMESTAMP TICK, AND THIS IS THE ONE
# PLATFORM RISK IN THE PACKAGE.
#
# InvalidateByPathMTime is `cache.st_mtime > source.st_mtime'. A caller's normal
# sequence -- write the source, build the cache from it microseconds later --
# puts BOTH stamps in the same filesystem tick whenever the tick is coarser than
# the gap, which makes them EQUAL, which is not greater, which means the cache
# NEVER HITS and the expensive parse runs on every single load. Nothing raises
# and nothing is corrupt; the package just silently stops being a cache.
#
# THIS IS MEASURED, AND IT FIRED ON THE CONTROL. On the x86-64 build host's
# /tmp the tick is 1 ms, and a back-to-back write/populate/load landed both
# stamps in the same millisecond on three runs out of six -- so the naive case
# is a COIN FLIP there, not a pass. A suite case that simply did the naive
# sequence would therefore flap red on some images and green on others while
# saying nothing about either.
#
# So the tick is MEASURED FIRST and the case then settles past it. That makes
# the assertion deterministic on any filesystem, and leaves the tick itself --
# `mtime_granularity_ns' -- as the number that actually describes this image.
def _granularity_ns(path, budget=200):
    """Smallest non-zero mtime step this filesystem reports, in ns.

    Touch a file repeatedly and keep the smallest positive difference seen. A
    filesystem with nanosecond stamps answers a few hundred ns; one with a 1 ms
    tick answers 1000000; a 1-second one answers 1000000000. Returns 0 if the
    stamp never moved inside the budget, which is itself the answer -- the clock
    is coarser than the whole loop.
    """
    best = 0
    prev = None
    for _ in range(budget):
        with open(path, "w") as _fo:
            _fo.write("t")
        now = os.stat(path).st_mtime_ns
        if prev is not None and now > prev:
            d = now - prev
            if best == 0 or d < best:
                best = d
        prev = now
    return best


TICKF = W("tick.txt")
GRAN = _granularity_ns(TICKF)
emit("mtime_granularity_ns", GRAN)
emit("mtime_granularity_measured", yn(GRAN > 0))
emit("mtime_subsecond", yn(0 < GRAN < 1000000000))

os.makedirs(W("natural"), exist_ok=True)
NSRC = W("n-src.txt")
with open(NSRC, "w") as fo:
    fo.write("natural\n")

# Settle past ONE measured tick -- capped at 2s so a filesystem with a wild
# answer cannot make this suite the batch's long pole -- and only then populate.
# The two stamps are now unambiguously ordered and the hit is a statement about
# flexcache.
_settle = min(2.0, (GRAN or 1000000000) / 1e9)
time.sleep(_settle)
emit("settle_seconds", "%.3f" % _settle)

dn = flexcache.DiskCacheByMTime(W("natural", "cache"))
CALLS["upper"] = 0
dn.load(NSRC, converter=upper)
dn.load(NSRC, converter=upper)
emit("bymtime_settled_hit", yn(calls() == 1))
emit("bymtime_settled_calls", calls())

# AND THE SAME SEQUENCE WITHOUT THE SETTLE, reported as a COUNT rather than as a
# verdict. Five write/populate/load rounds back to back: `5' means every one hit
# and the tick is finer than the round; anything less is the tick biting, and
# `0' means it bites every time. It is emitted for the record and the suite
# scores only that it is in range, because the number is a property of the
# filesystem and not of the port.
_hits = 0
for _i in range(5):
    _d = W("race%d" % _i)
    os.makedirs(_d, exist_ok=True)
    _s = os.path.join(_d, "s.txt")
    with open(_s, "w") as fo:
        fo.write("race\n")
    _dc = flexcache.DiskCacheByMTime(os.path.join(_d, "cache"))
    CALLS["upper"] = 0
    _dc.load(_s, converter=upper)
    _dc.load(_s, converter=upper)
    if calls() == 1:
        _hits += 1
emit("bymtime_unsettled_hits_of_5", _hits)
emit("bymtime_unsettled_in_range", yn(0 <= _hits <= 5))

# ------------------------------------------------------ folder construction ---
# `cache_folder.mkdir(parents=True, exist_ok=True)' in __init__ -- so a cache
# directory NESTED SEVERAL LEVELS DEEP under nothing at all is created, and
# constructing over an EXISTING one is not an error.
deep = W("deep", "a", "b", "c")
flexcache.DiskCacheByHash(deep)
emit("deep_folder_created", yn(os.path.isdir(deep)))
flexcache.DiskCacheByHash(deep)
emit("existing_folder_reused", yn(os.path.isdir(deep)))

# A str AND a Path are both accepted for the folder.
emit("folder_from_path_ok",
     yn(isinstance(flexcache.DiskCacheByHash(pathlib.Path(W("frompath"))).cache_folder,
                   pathlib.Path)))

# ------------------------------------------------------------- error paths ---
# A CACHE FOLDER THAT IS AN EXISTING FILE. `exist_ok=True' forgives an existing
# DIRECTORY only, so this raises at CONSTRUCTION -- before any load -- and it is
# FileExistsError rather than NotADirectoryError.
notdir = W("a-file")
with open(notdir, "w") as fo:
    fo.write("x")
emit("folder_over_file", attempt(lambda: flexcache.DiskCacheByHash(notdir)))

# ...and a folder BENEATH a file is the other one: the parent cannot be made.
emit("folder_under_file",
     attempt(lambda: flexcache.DiskCacheByHash(os.path.join(notdir, "sub"))))

# A CORRUPT CACHE FILE IS NOT SURVIVED. is_valid says the file exists, so
# rawload opens it and hands the bytes to pickle -- which raises. flexcache does
# NOT fall back to re-parsing, and a caller who assumed it does loses the call.
#
# CORRUPTION, NOT PERMISSIONS, IS WHAT IS MEASURED HERE. The documented
# unreadable-file path raises PermissionError out of the same open(), but this
# image's suite runs as root, where mode 000 does not refuse a read -- so the
# permission case would silently measure nothing. `unreadable_euid' and
# `unreadable_raised' below record that directly rather than leaving it implied.
#
# THE TWO CASES BELOW DAMAGE THIS ENTRY, so the permission case does NOT reuse
# it -- see `unread' further down. Root's DAC override does hold on the /tmp
# tmpfs this probe works in; where it does not reach is the virtio-fs backend
# of the cloud rootfs (brintos/hardwarejs#27), which is not a path this probe
# writes to.
os.makedirs(W("corrupt"), exist_ok=True)
dcor = flexcache.DiskCacheByHash(W("corrupt", "cache"))
CSRC = W("c-src.txt")
with open(CSRC, "w") as fo:
    fo.write("corrupt me\n")
_, cstem = dcor.load(CSRC, converter=upper)
with open(W("corrupt", "cache", cstem + ".pickle"), "wb") as fo:
    fo.write(b"not a pickle at all")
emit("corrupt_cache", attempt(lambda: dcor.load(CSRC, converter=upper)))

# An EMPTY cache file is the same path with a different exception out of pickle,
# and both are asserted because they are what a truncated write leaves behind.
with open(W("corrupt", "cache", cstem + ".pickle"), "wb"):
    pass
emit("empty_cache", attempt(lambda: dcor.load(CSRC, converter=upper)))

# THIS IMAGE RUNS ITS SELFTESTS AS ROOT, where mode 000 does not refuse a read,
# so the documented PermissionError cannot be provoked here. Both halves are
# emitted: the euid, and a derived key that is `yes\' on either kind of host --
# root reading it fine, or a normal account being refused -- so the case scores
# the RELATIONSHIP rather than an outcome that depends on who ran it.
#
# IT GETS ITS OWN, INTACT ENTRY. Reusing the entry the two cases above damaged
# made this measurement impossible to pass on a root image: that file is EMPTY,
# so the load raises EOFError whether or not the open was refused -- which is
# exactly what `empty_cache' above asserts -- and `unreadable_matches_euid'
# then reads a pickle failure as a permission one. `unreadable_entry_ok' is the
# guard: it says the entry this case chmods is a GOOD one, so a raise below is
# about the mode bits and nothing else.
os.makedirs(W("unread"), exist_ok=True)
dunr = flexcache.DiskCacheByHash(W("unread", "cache"))
USRC = W("u-src.txt")
with open(USRC, "w") as fo:
    fo.write("unreadable me\n")
_, ustem = dunr.load(USRC, converter=upper)
UPICKLE = W("unread", "cache", ustem + ".pickle")
emit("unreadable_entry_ok", yn(os.path.getsize(UPICKLE) > 0))

emit("unreadable_euid", os.geteuid())
os.chmod(UPICKLE, 0)
# A FRESH INSTANCE, so the answer comes off disk rather than out of anything the
# first load may still be holding.
_unreadable = attempt(
    lambda: flexcache.DiskCacheByHash(W("unread", "cache")).load(
        USRC, converter=upper))
emit("unreadable_raised", yn(_unreadable.startswith("!")))
emit("unreadable_result", _unreadable)
emit("unreadable_matches_euid",
     yn((os.geteuid() == 0) != _unreadable.startswith("!")))
os.chmod(UPICKLE, 0o644)

# A SOURCE THAT DISAPPEARS BETWEEN LOADS raises FileNotFoundError from BOTH
# strategies, and from DIFFERENT places: by-hash reads the file to NAME the
# entry, by-mtime stats it to VALIDATE one. Neither serves the cached value it
# still has on disk.
os.makedirs(W("gone"), exist_ok=True)
GSRC = W("g-src.txt")
with open(GSRC, "w") as fo:
    fo.write("here for now\n")
dg_hash = flexcache.DiskCacheByHash(W("gone", "byhash"))
dg_mtime = flexcache.DiskCacheByMTime(W("gone", "bymtime"))
dg_hash.load(GSRC, converter=upper)
dg_mtime.load(GSRC, converter=upper)
os.unlink(GSRC)
emit("gone_byhash", attempt(lambda: dg_hash.load(GSRC, converter=upper)))
emit("gone_bymtime", attempt(lambda: dg_mtime.load(GSRC, converter=upper)))

# A SOURCE THAT NEVER EXISTED, for the same two, so a red above can be told
# apart from a naming failure.
emit("absent_byhash", attempt(lambda: dg_hash.load(W("never.txt"), converter=upper)))
emit("absent_bymtime", attempt(lambda: dg_mtime.load(W("never.txt"), converter=upper)))

# --------------------------------------------------------- documented quirks --
# A FALSY CONVERTED OBJECT IS NEVER A CACHE HIT. `if converted_object:' in
# load() tests the VALUE, so an empty string, an empty list and 0 are all
# re-converted on every call even though the cache file was written and is
# valid. That is the single most surprising thing in the package: the cache
# silently does nothing for a whole class of results.
os.makedirs(W("falsy"), exist_ok=True)
dfal = flexcache.DiskCacheByHash(W("falsy", "cache"))
FSRC = W("f-src.txt")
with open(FSRC, "w") as fo:
    fo.write("falsy\n")


def to_empty(_p):
    CALLS["empty"] = CALLS.get("empty", 0) + 1
    return ""


fv1, fstem = dfal.load(FSRC, converter=to_empty)
fv2, _ = dfal.load(FSRC, converter=to_empty)
emit("falsy_value_repr", repr(fv2))
emit("falsy_calls", calls("empty"))
emit("falsy_reconverts_every_time", yn(calls("empty") == 2))
emit("falsy_cache_file_written",
     yn(os.path.isfile(W("falsy", "cache", fstem + ".pickle"))))

# ...and the same source with a TRUTHY converter is cached normally, so the
# difference is the value and not the file.
def to_nonempty(_p):
    CALLS["nonempty"] = CALLS.get("nonempty", 0) + 1
    return "x"


dfal.load(FSRC, converter=to_nonempty)
dfal.load(FSRC, converter=to_nonempty)
emit("truthy_calls", calls("nonempty"))

# NO CONVERTER AND NO CACHE RETURNS (None, stem) RATHER THAN RAISING -- and the
# stem is still computed, so a caller can name the entry it did not get.
os.makedirs(W("noconv"), exist_ok=True)
dnc = flexcache.DiskCacheByHash(W("noconv", "cache"))
NSRC2 = W("nc-src.txt")
with open(NSRC2, "w") as fo:
    fo.write("no converter\n")
nv, nstem = dnc.load(NSRC2)
emit("noconverter_value", repr(nv))
emit("noconverter_stem_len", len(nstem))
emit("noconverter_wrote_nothing", yn(os.listdir(W("noconv", "cache")) == []))

# A CONVERTER GIVEN AS A STRING IS A NAME, NOT A CALLABLE: it sets the
# converter_id and then load() behaves as though no converter were passed. Two
# different names over one source therefore name two different entries.
sv_a, sa = dnc.load(NSRC2, converter="alpha")
sv_b, sb = dnc.load(NSRC2, converter="beta")
emit("strconv_value", repr(sv_a))
emit("strconv_names_differ", yn(sa != sb))
emit("strconv_alpha_stem_recomputed",
     yn(sa == hashlib.sha1(b"alpha" + b"no converter\n").hexdigest()))

# THE converter_id IS getattr(converter, "__name__", ""), AND A LAMBDA HAS ONE.
# It is the string `<lambda>\' -- so every lambda a caller ever passes for a
# given source shares ONE cache entry, and swapping one lambda for another with
# different behaviour silently serves the first one\'s result. MEASURED rather
# than reasoned: the first version of this probe asserted the empty string here
# and was wrong.
emit("lambda_converter_id_is_lambda",
     yn(dnc.load(NSRC2, converter=lambda p: "L")[1]
        == hashlib.sha1(b"<lambda>" + b"no converter\n").hexdigest()))


class _CallableNoName:
    def __call__(self, _p):
        return "C"


# A callable OBJECT has no __name__, so it does fall to "" -- which is the case
# the lambda one looks like but is not, and the two share nothing.
emit("callable_object_empty_id",
     yn(dnc.load(NSRC2, converter=_CallableNoName())[1]
        == hashlib.sha1(b"" + b"no converter\n").hexdigest()))

# `pass_hash=True' HANDS THE STEM TO THE CONVERTER, which is how a caller names
# side files beside the cache entry.
def with_hash(_p, stem):
    return "stem:" + stem


ph_v, ph_stem = dnc.load(NSRC2, converter=with_hash, pass_hash=True)
emit("pass_hash_value_matches_stem", yn(ph_v == "stem:" + ph_stem))

# `save()' STORES A VALUE THE CACHE NEVER CONVERTED, and returns the stem.
sav_stem = dnc.save("PRECOMPUTED", NSRC2, "upper")
emit("save_stem_recomputed",
     yn(sav_stem == hashlib.sha1(b"upper" + b"no converter\n").hexdigest()))
emit("save_then_load",
     dnc.load(NSRC2, converter=_must_not_run)[0])

# --------------------------------------------------------------- the types ---
# THE BASE DiskCache REGISTERS NO HEADER CLASS, so it cannot be used directly --
# `_get_header_class' raises TypeError. A caller who reaches for DiskCache
# instead of one of the two convenience subclasses gets this and nothing else.
dbase = flexcache.DiskCache(W("base"))
emit("base_diskcache_load", attempt(lambda: dbase.load(SRC, converter=upper)))
emit("base_header_classes_empty", yn(dbase._header_classes == {}))

# ...and registering one makes it work, which is the documented extension point.
dbase.register_header_class(str, flexcache.DiskCacheByHash.Header.from_string)
CALLS["upper"] = 0
emit("base_registered_value", dbase.load(SRC, converter=upper)[0])
emit("base_registered_calls", calls())

# A SOURCE OF THE WRONG TYPE is refused by the same lookup, not by a check
# inside the header -- so it is TypeError from `_get_header_class'.
emit("int_source", attempt(lambda: dc.load(42, converter=upper)))

# ...but a Path and a str reach the SAME entry, because both are registered and
# NameByFileContent names by content either way.
emit("path_and_str_same_stem",
     yn(dc.load(pathlib.Path(SRC), converter=_must_not_run)[1]
        == dc.load(SRC, converter=_must_not_run)[1]))

# THE HEADERS ARE FROZEN DATACLASSES. Assigning to one raises
# FrozenInstanceError, which is what lets a header be safely used as the cache
# name it was computed from.
hdr_obj = flexcache.DiskCacheByHash.Header(pathlib.Path(SRC), "upper")
emit("header_frozen",
     attempt(lambda: setattr(hdr_obj, "converter_id", "other")))
emit("header_source_type", type(hdr_obj.source).__name__)
emit("header_is_valid_missing",
     yn(hdr_obj.is_valid(pathlib.Path(W("byhash", "cache", "nope.pickle")))))
emit("header_is_valid_present",
     yn(hdr_obj.is_valid(pathlib.Path(cache_pickle))))

# BaseHeader.__post_init__ CHECKS THE SOURCE TYPE, and it is the one place
# flexcache raises TypeError with its own message rather than through a lookup.
emit("header_wrong_source_type",
     attempt(lambda: flexcache.DiskCacheByHash.Header(42, "upper")))

# `NameByFields' NAMES BY EVERY FIELD BUT THE SOURCE, which is what makes
# BasicPythonHeader's cache entries python-version specific. Asserted as a
# RELATIONSHIP -- two headers differing only in python_version hash apart --
# because the literal would encode this control's interpreter.
@dataclasses.dataclass(frozen=True)
class _PyHeader(flexcache.NameByFields, flexcache.InvalidateByExist,
                flexcache.BasicPythonHeader):
    pass


_h1 = _PyHeader("s", "cid", "Linux", "CPython", "3.12.3")
_h2 = _PyHeader("s", "cid", "Linux", "CPython", "3.12.14")
_dcf = flexcache.DiskCache(W("fields"))
emit("namebyfields_version_changes_name",
     yn(_dcf.cache_stem_for(_h1) != _dcf.cache_stem_for(_h2)))
emit("namebyfields_stable",
     yn(_dcf.cache_stem_for(_h1) == _dcf.cache_stem_for(_PyHeader(
         "other-source", "cid", "Linux", "CPython", "3.12.3"))))

# The exported names are all classes, and the two cache classes are DiskCache
# subclasses -- a shape assertion, so a release that turned one into a factory
# function says so here.
emit("byhash_is_diskcache",
     yn(issubclass(flexcache.DiskCacheByHash, flexcache.DiskCache)))
emit("bymtime_is_diskcache",
     yn(issubclass(flexcache.DiskCacheByMTime, flexcache.DiskCache)))
emit("byhash_header_bases",
     ",".join(b.__name__ for b in flexcache.DiskCacheByHash.Header.__bases__))
emit("bymtime_header_bases",
     ",".join(b.__name__ for b in flexcache.DiskCacheByMTime.Header.__bases__))

# --------------------------------------------------------------- filesystem --
# NOTHING WAS WRITTEN OUTSIDE THE WORKDIR. Every path above is under ROOT, and
# this is the case that says the package did not reach anywhere else -- the
# cache folder is the only thing it is entitled to create.
#
# `no_cwd_cache' compares the cwd `cache' path against the sample taken before
# the first flexcache call, because on this image that path is /cache and the
# ROOTFS SHIPS IT. Asserting its absence asserted a property of the image; what
# this case is for is whether the probe left something new there.
emit("workdir_is_tmp", yn(ROOT.startswith(tempfile.gettempdir())))
emit("no_cwd_cache", yn(os.path.exists(CWD_CACHE) == CWD_CACHE_BEFORE))

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

emit("emitted_count", len(OUT) + 2)
emit("probe_done", "yes")
sys.stdout.write("\n".join(OUT) + "\n")