017 — CPython zlib：半個串流跟一個完整串流，解出來是同一串位元組

{
  "name": "017-cpython-zlib",
  "upstream": "CPython zlib, measured at run time",
  "what_is_being_examined": "Two ways of decompressing the same truncated stream, and what each one lets a caller ask afterwards.",

  "the_finding": "A truncated stream and a complete one decompress to BYTE-IDENTICAL output through decompressobj, and the only field that separates them is `.eof`. The one-shot zlib.decompress raises on the same input. Same data, two APIs in one module, one refuses and one returns a prefix in silence.",

  "second_finding": "`.eof` lives on the decompressor object and the bytes are the return value. The flag does not travel with the data, so a function that decompresses and returns bytes has destroyed the distinction by the time its caller sees it — the same shape example 016 named when it made the outcome travel WITH the records.",

  "third_finding": "The truncation cuts mid-record. A caller doing the idiomatic split on a newline gets a run of whole records and one fragment, and the fragment is well-formed enough to be mistaken for a record.",

  "readers": {
    "zlib.decompress": {"raises_on_truncation": true,  "reports_completeness": false},
    "decompressobj":   {"raises_on_truncation": false, "reports_completeness": true}
  },

  "the_control": "A COMPLETE stream whose payload is exactly what the truncated read produced. Without it, 'the truncated read returned 218 bytes' is not evidence of anything; with it, the two reads are byte-identical and only `.eof` differs.",

  "sets": {
    "FMS": "this file: what each reader raises and reports, and the units map",
    "SCL": "which reader this deployment uses, and whether not reaching eof is fatal here",
    "SMS": "the probes that run the real zlib, including the control stream",
    "TMS": "one file per reader — each declares what it raises and what it reports, and imports nothing",
    "DMS": "what the caller holds, what it can still ask, and where the answer lives"
  },

  "units": {"TMS/readers": ["incremental.py", "one_shot.py"]},

  "the_recut_adds": "A reader declares whether it reports completeness, and the declaration is checked by running it. Section 3b is the drill: a reader claiming to report completeness while returning only bytes must be caught."
}

{
  "deployment": "log-shipper",
  "reader": "decompressobj",
  "a_stream_that_did_not_reach_eof_is": "fatal",
  "why": "A log shipper streams batches too large to hold whole, so it needs the incremental reader — and a truncated batch that ships as a short batch is a silent gap in an audit trail. It reads `.eof` after the final call and refuses when it is False.",
  "what_this_costs": "One attribute read that nothing in the API requires and no exception forces. Every caller that does not perform it gets a prefix that looks like a complete short message.",
  "not_a_general_rule": "A best-effort preview pane is right to render whatever decompressed. SCL is where that position lives."
}

"""Probes that run the real zlib. Nothing here is simulated.

The payload is deliberately line-oriented, because the thing a caller does with
a decompressed blob is split it, and the split is where the evidence dies.
"""
import zlib

PAYLOAD = b"".join(b"record-%03d\n" % n for n in range(60))


def versions():
    return f"zlib {zlib.ZLIB_VERSION} (runtime {zlib.ZLIB_RUNTIME_VERSION})"


def compressed():
    return zlib.compress(PAYLOAD, 9)


def truncated_stream():
    """Half a valid stream. A dropped connection produces exactly this."""
    whole = compressed()
    return whole[: len(whole) // 2]


def one_shot(data):
    """zlib.decompress — the function everybody reaches for."""
    try:
        return {"bytes": zlib.decompress(data), "raised": None, "eof": None}
    except zlib.error as raised:
        return {"bytes": None, "raised": str(raised), "eof": None}


def incremental(data, max_length=None):
    """decompressobj — the streaming path, and the one with a completeness flag."""
    decompressor = zlib.decompressobj()
    try:
        out = (decompressor.decompress(data) if max_length is None
               else decompressor.decompress(data, max_length))
    except zlib.error as raised:
        return {"bytes": None, "raised": str(raised), "eof": None, "unconsumed": None}
    return {"bytes": out, "raised": None, "eof": decompressor.eof,
            "unconsumed": len(decompressor.unconsumed_tail)}


def control_for(output):
    """A COMPLETE stream whose payload is exactly `output`.

    This is what makes the entry able to say anything: without it, "the
    truncated read returned 218 bytes" is not evidence that a truncated read is
    indistinguishable from an honest short one.
    """
    return zlib.compress(output, 9)


def whole_records(data):
    """What a caller gets after the idiomatic split."""
    lines = data.split(b"\n")
    complete = [line for line in lines[:-1]]
    trailing = lines[-1]
    return complete, trailing

"""decompressobj().decompress, wrapped as a declaring unit.

It does not raise on a truncated stream — it cannot, because "not finished yet"
is its normal state between calls. It reports completeness instead, on `.eof`.

The catch, which is the whole entry: `.eof` is on the OBJECT and the bytes are
the VALUE. The moment the bytes leave the decompressor the flag does not go
with them.
"""
READER = "decompressobj"
RAISES_ON_TRUNCATION = False
REPORTS_COMPLETENESS = True
COMPLETENESS_LIVES_ON = "the decompressor object, not the returned bytes"

"""zlib.decompress, wrapped as a declaring unit.

It cannot report completeness because it has nowhere to report it from — it
returns bytes and nothing else. What it does instead is refuse: a truncated
stream raises. The declaration is verified by running it, not by reading here.
"""
READER = "zlib.decompress"
RAISES_ON_TRUNCATION = True
REPORTS_COMPLETENESS = False

"""What a person is shown.

The byte counts are never printed without the `eof` column beside them, because
the byte count is the number the two situations share.
"""


def reads(rows):
    lines = ["    stream        reader           bytes   eof     raised"]
    for row in rows:
        lines.append(
            f'    {row["stream"]:<13} {row["reader"]:<16} '
            f'{("-" if row["bytes"] is None else len(row["bytes"])):<7} '
            f'{str(row["eof"]) if row["eof"] is not None else "-":<7} {row["raised"] or ""}'.rstrip())
    return "\n".join(lines)


def identity(truncated, control):
    return "\n".join([
        f'    truncated stream -> {len(truncated["bytes"])} bytes  eof = {truncated["eof"]}',
        f'    complete  stream -> {len(control["bytes"])} bytes  eof = {control["eof"]}',
        f'    byte-identical   : {truncated["bytes"] == control["bytes"]}',
        f'    told apart by any returned value: {truncated["bytes"] != control["bytes"]}',
        f'    told apart by .eof              : {truncated["eof"] != control["eof"]}',
    ])


def after_the_split(complete, trailing):
    return "\n".join([
        f"    whole records the caller collects: {len(complete)}",
        f"    and one trailing fragment:         {trailing!r}",
        "    the fragment is well formed enough to be mistaken for a record",
    ])

"""The island test, run against the real zlib.

    python src/island_test.py

Section 3 is the control. Section 3b is the drill: a reader that DECLARES it
reports completeness while returning only bytes must be caught by running it.
"""
import json
import pathlib
import re
import sys
import zlib

from SMS import upstream
from TMS.readers import incremental, one_shot

HERE = pathlib.Path(__file__).parent
FMS = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))
POLICY = json.loads((HERE / "SCL" / "policy.json").read_text(encoding="utf-8"))
FAILURES = []
RAN = [0]


def check(label, ok, detail=""):
    RAN[0] += 1
    print(f'  {"PASS" if ok else "FAIL"}  {label}{f" - {detail}" if detail else ""}')
    if not ok:
        FAILURES.append(label)


print(f"\n  {upstream.versions()}")
whole, trunc = upstream.compressed(), upstream.truncated_stream()

print("\n== 1. each reader is an island, and FMS matches the tree")
for unit, declared in FMS["units"].items():
    directory = HERE.joinpath(*unit.split("/"))
    on_disk = sorted(p.name for p in directory.glob("*.py") if p.name != "__init__.py")
    check(f"{unit}: FMS declares what is on disk", on_disk == sorted(declared),
          f'disk {", ".join(on_disk)} | FMS {", ".join(sorted(declared))}')
    for name in on_disk:
        body = directory.joinpath(name).read_text(encoding="utf-8")
        check(f"{unit}/{name} imports nothing at all",
              not re.search(r"^\s*(import|from)\s", body, re.M))
check("both readers declare what they raise and what they report",
      all(isinstance(m.RAISES_ON_TRUNCATION, bool) and isinstance(m.REPORTS_COMPLETENESS, bool)
          for m in (one_shot, incremental)))

print("\n== 2. the same truncated stream, two readers from one module")
shot = upstream.one_shot(trunc)
step = upstream.incremental(trunc)
check("zlib.decompress raises", shot["raised"] is not None, shot["raised"])
check("  and the message names truncation", "truncated" in (shot["raised"] or ""))
check("  and it returns no bytes at all", shot["bytes"] is None)
check("decompressobj does NOT raise", step["raised"] is None)
check("  and returns real bytes", step["bytes"] is not None and len(step["bytes"]) > 0,
      f'{len(step["bytes"] or b"")} bytes')
check("  and reports eof = False", step["eof"] is False)
check("on a complete stream both succeed",
      upstream.one_shot(whole)["raised"] is None and upstream.incremental(whole)["raised"] is None)
check("and only then is eof True", upstream.incremental(whole)["eof"] is True)

print("\n== 3. the control - a complete stream carrying exactly those bytes")
control = upstream.incremental(upstream.control_for(step["bytes"]))
check("the control stream is complete", control["eof"] is True)
check("it decompresses to the same number of bytes", len(control["bytes"]) == len(step["bytes"]),
      f'{len(control["bytes"])} vs {len(step["bytes"])}')
check("BYTE-IDENTICAL to the truncated read", control["bytes"] == step["bytes"])
check("so no returned value can tell them apart", not (control["bytes"] != step["bytes"]))
check("and .eof can", control["eof"] != step["eof"], f'{control["eof"]} vs {step["eof"]}')
check("which means the discriminator is on the object, not on the value",
      incremental.COMPLETENESS_LIVES_ON.startswith("the decompressor object"))

print("\n== 3b. DRILL - a reader that overclaims must be caught by running it")


def reports_completeness(read):
    """Run the reader on a truncated stream and see whether the result carries
    a completeness answer at all."""
    return read(trunc)["eof"] is not None


liar_declares = True  # a unit claiming decompressobj's flag while returning only bytes
check("the drill unit declares it reports completeness", liar_declares is True)
check("running it says otherwise", reports_completeness(upstream.one_shot) is False)
check("so the declaration is refused", reports_completeness(upstream.one_shot) != liar_declares)
check("and the honest declarations hold under the same probe",
      reports_completeness(upstream.incremental) is incremental.REPORTS_COMPLETENESS
      and reports_completeness(upstream.one_shot) is one_shot.REPORTS_COMPLETENESS)
check("the raise-on-truncation declarations hold too",
      (upstream.one_shot(trunc)["raised"] is not None) is one_shot.RAISES_ON_TRUNCATION
      and (upstream.incremental(trunc)["raised"] is not None) is incremental.RAISES_ON_TRUNCATION)

print("\n== 4. a deliberate partial read lands in the same state as an accidental one")
capped = upstream.incremental(whole, max_length=100)
check("max_length returns exactly the cap", len(capped["bytes"]) == 100)
check("with eof False", capped["eof"] is False)
check("and an unconsumed tail", capped["unconsumed"] > 0, f'{capped["unconsumed"]} bytes')
check("the accidental truncation has eof False as well", step["eof"] is False)
check("so eof alone does not say WHY, only that it is not finished",
      capped["eof"] == step["eof"] and capped["unconsumed"] != (step["unconsumed"] or 0))

print("\n== 5. the truncation cuts mid-record")
complete_records, trailing = upstream.whole_records(step["bytes"])
check("the caller collects whole records", len(complete_records) > 0, f"{len(complete_records)}")
check("and one trailing fragment", len(trailing) > 0, repr(trailing))
check("the fragment starts like a record", trailing.startswith(b"record-"))
check("and is shorter than one", len(trailing) < len(b"record-000"))
check("a complete payload leaves no fragment",
      upstream.whole_records(upstream.PAYLOAD)[1] == b"")

print("\n== 6. what this does not change")
check("the incremental reader is right not to raise - not-finished-yet is its normal state",
      upstream.incremental(whole[: len(whole) // 4])["raised"] is None)
check("feeding it the rest still finishes",
      zlib.decompressobj().decompress(whole)[-11:] == b"record-059\n")
check("SCL names the cost in words",
      "one attribute read" in POLICY["what_this_costs"].lower())
check("and this deployment does refuse", POLICY["a_stream_that_did_not_reach_eof_is"] == "fatal")

print()
if FAILURES:
    print(f'  {len(FAILURES)} FAILED: {" | ".join(FAILURES)}')
    sys.exit(1)
print(f"  {RAN[0]} checks passed - every probe ran the real zlib")

"""One truncated stream, two readers from the same module.

    python src/main.py            what each reader returns and what it lets you ask
    python src/main.py --strict   exit 1 if this deployment read a stream that never reached eof
"""
import json
import pathlib
import sys

from DMS import report
from SMS import upstream
from TMS.readers import incremental, one_shot

POLICY = json.loads((pathlib.Path(__file__).parent / "SCL" / "policy.json").read_text(encoding="utf-8"))
READERS = {module.READER: module for module in (one_shot, incremental)}


def main(argv):
    whole, trunc = upstream.compressed(), upstream.truncated_stream()
    print(f"\n  {upstream.versions()}")
    print(f'  {POLICY["deployment"]}: read with {POLICY["reader"]}\n')
    print(f"  payload {len(upstream.PAYLOAD)} bytes -> {len(whole)} compressed, "
          f"truncated to {len(trunc)}\n")

    rows = [
        {"stream": "truncated", "reader": "zlib.decompress", **upstream.one_shot(trunc)},
        {"stream": "truncated", "reader": "decompressobj", **upstream.incremental(trunc)},
        {"stream": "complete", "reader": "zlib.decompress", **upstream.one_shot(whole)},
        {"stream": "complete", "reader": "decompressobj", **upstream.incremental(whole)},
    ]
    print(report.reads(rows))
    print("\n  Same input. One refuses, the other returns a prefix and says nothing.\n")

    partial = upstream.incremental(trunc)
    control = upstream.incremental(upstream.control_for(partial["bytes"]))
    print("  the control - a COMPLETE stream carrying exactly those bytes:")
    print(report.identity(partial, control))
    print("\n  The only field that separates them is .eof, and it is on the decompressor,")
    print("  not on the bytes. A function that returns the bytes has thrown it away.\n")

    capped = upstream.incremental(whole, max_length=100)
    print(f'  decompress(complete, max_length=100) -> {len(capped["bytes"])} bytes, '
          f'eof = {capped["eof"]}, unconsumed_tail = {capped["unconsumed"]}')
    print("  a deliberate partial read lands in the same state as an accidental one\n")

    complete_records, trailing = upstream.whole_records(partial["bytes"])
    print("  what the idiomatic split leaves the caller:")
    print(report.after_the_split(complete_records, trailing))

    if "--strict" in argv and partial["eof"] is False \
            and POLICY["a_stream_that_did_not_reach_eof_is"] == "fatal":
        print(f'\n  --strict: the stream never reached eof and this deployment calls that fatal')
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))