015 — CPython os.walk 3.14.5：一次半途壞掉的走訪，跟一次完整的走訪，是同一個值

{
  "name": "015-cpython-os-walk",
  "upstream": "CPython Lib/os.py — os.walk",
  "examined_version": "measured at run time",
  "license": "PSF-2.0",
  "what_it_is": "The traversal every Python tool reaches for, and what it does with an error it meets on the way.",

  "why_this_one": "Example 015 the same day argues that working and absent are not the two options — a capability can be present and failing. os.walk is where that state is produced constantly and named nowhere: its default swallows errors, continues, and returns a result that looks complete.",

  "modes": {
    "silent": {"onerror": null, "reports": false, "note": "the default"},
    "reporting": {"onerror": "a callback", "reports": true}
  },

  "the_finding": "A subdirectory that disappears between the top-level listing and its visit produces the SAME file list under both modes. The only difference is whether anyone was told. A partial traversal and a complete one are the same value.",

  "the_second_finding": "'The walk found nothing' has two causes — an empty directory and a path that does not exist — and they DO differ: one yields a row, the other yields none. But the idiomatic `for _, _, files in os.walk(top)` collects files, and both give zero files. The discriminator exists and the ordinary way of using it destroys it.",

  "sets": {
    "FMS": "this file: the two modes, what each reports, and the units map",
    "SCL": "which mode this deployment uses, and whether a silent traversal is fatal",
    "SMS": "mode resolution by id and the probes against the real os.walk",
    "TMS": "one file per mode — each declares whether anyone is told, and reaches no sibling set",
    "DMS": "the two causes, the two modes, and the gaps"
  },

  "units": {"TMS/modes": ["reporting.py", "silent.py"]},

  "non_goals": [
    "Saying the default is wrong. A traversal that raised on every unreadable directory would be unusable on a real filesystem, and onerror exists because no single default suits everyone.",
    "Reimplementing traversal. Every number comes from the platform's own os.walk.",
    "Any claim about how often this loses data in the wild."
  ]
}

{
  "mode": "reporting",
  "a_silent_traversal_is_fatal": true,
  "_note": "main.py runs both modes regardless of what this says."
}

"""Which traversal mode this deployment uses."""
import json
import pathlib

_C = json.loads((pathlib.Path(__file__).parent / "policy.json").read_text(encoding="utf-8"))

mode = lambda: _C["mode"]                                              # noqa: E731
silent_is_fatal = lambda: bool(_C["a_silent_traversal_is_fatal"])      # noqa: E731

"""What os.walk actually does, measured on this interpreter.

Nothing here needs special permissions: a directory that does not exist, and a
directory that disappears between the top-level listing and the visit. Both are
ordinary in a running system.
"""
import inspect
import os
import shutil
import tempfile


def build_tree(names=("a", "b", "c")):
    root = tempfile.mkdtemp()
    for name in names:
        os.makedirs(os.path.join(root, name))
        open(os.path.join(root, name, f"{name}.txt"), "w").close()
    return root


def empty_versus_missing():
    """Two causes of 'the walk found nothing', and what survives the idiomatic loop."""
    root = tempfile.mkdtemp()
    empty = os.path.join(root, "empty_dir")
    os.makedirs(empty)
    missing = os.path.join(root, "does-not-exist")

    def collected(top):
        found = []
        for _, _, files in os.walk(top):
            found.extend(files)
        return found

    return {
        "empty": {"rows": len(list(os.walk(empty))), "files": len(collected(empty))},
        "missing": {"rows": len(list(os.walk(missing))), "files": len(collected(missing))},
        "raised": None,
    }


def vanishing_subdirectory(onerror=None):
    """Start the walk, delete a subdirectory before it is visited, keep walking."""
    root = build_tree()
    errors = []
    walker = os.walk(root, onerror=onerror(errors) if onerror else None)
    next(walker)                                     # the top level lists a, b, c
    shutil.rmtree(os.path.join(root, "b"))           # b is gone before its turn
    found = []
    for _, _, files in walker:
        found.extend(files)
    return {"files": sorted(found), "errors": errors, "raised": None}


def signature():
    return str(inspect.signature(os.walk))


def onerror_lines():
    source = inspect.getsource(os.walk)
    return [line.strip() for line in source.splitlines()
            if "onerror" in line and line.strip().startswith(("if", "return"))]

"""The re-cut: a traversal mode declares whether anyone is told."""
import importlib

MODES = ["silent", "reporting"]


def load():
    loaded, problems = {}, []
    for module_name in MODES:
        try:
            module = importlib.import_module(f"TMS.modes.{module_name}")
        except ModuleNotFoundError:
            problems.append(f'mode "{module_name}" has no module - fail closed')
            continue
        for attribute in ("NAME", "REPORTS_ERRORS", "WHAT_THE_CALLER_SEES"):
            if not hasattr(module, attribute):
                problems.append(f"{module_name} does not declare {attribute}")
        loaded[module.NAME] = module
    return loaded, problems


def resolve(name, loaded):
    module = loaded.get(name)
    if module is None:
        return None, f'mode "{name}" has no implementation - fail closed (known: {", ".join(sorted(loaded))})'
    return module, None

"""os.walk with an onerror callback.

The same traversal, the same files, and one difference: somebody is told. The
files that come back are identical either way, which is the whole finding.
"""
NAME = "reporting"
REPORTS_ERRORS = True
WHAT_THE_CALLER_SEES = "the same list, plus the errors that shortened it"


def make_collector(into):
    return lambda error: into.append(f"{type(error).__name__} on {error.filename}")

"""os.walk's default: onerror=None.

An error during the traversal is discarded. The walk continues, the caller gets
a result that looks complete, and nothing anywhere says a directory was skipped.
"""
NAME = "silent"
ONERROR = None
REPORTS_ERRORS = False
WHAT_THE_CALLER_SEES = "a shorter list, indistinguishable from a smaller tree"

"""The two causes, the two modes, and what none of it can see."""


def causes(measured, out):
    out("\n  'the walk found nothing' — two causes:")
    out(f"    {'':<22} {'rows yielded':<14} files collected")
    for label in ("empty", "missing"):
        row = measured[label]
        out(f"    {label:<22} {row['rows']:<14} {row['files']}")
    out("\n    The rows differ. The files do not — and files is what the idiomatic")
    out("    `for _, _, files in os.walk(top)` collects. The discriminator exists")
    out("    and the ordinary way of using it destroys it.")


def modes(silent, reporting, out):
    out("\n  a subdirectory that disappears mid-walk:")
    out(f"    {'mode':<12} {'files':<26} errors")
    out(f"    {'silent':<12} {str(silent['files']):<26} {silent['errors'] or 'none reported'}")
    out(f"    {'reporting':<12} {str(reporting['files']):<26} {reporting['errors']}")
    out("\n    Same files. The only difference is whether anybody was told.")


def gaps(out):
    out("\n  measurable, not measured here:")
    out("    - how often a real traversal loses a directory to a permission error")
    out("    - how many callers pass onerror at all")
    out("\n  not measurable by this entry at all:")
    out("    - whether the default is wrong. A traversal that raised on every")
    out("      unreadable directory would be unusable on a real filesystem, and")
    out("      onerror exists precisely because the default cannot suit everyone.")
    out("    - what any caller believed an empty result meant.")

"""The island test.

    python src/island_test.py

Section 2 is the finding: a partial traversal and a complete one are the same
value, and the only thing that differs is whether anyone was told.
"""
import json
import pathlib
import re
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from SCL import policy  # noqa: E402
from SMS import upstream, walks  # noqa: E402
from TMS.modes import reporting as reporting_mode  # noqa: E402

ARCH = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))
FAILURES = []


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


LOADED, PROBLEMS = walks.load()

print("\n== 1. every mode is an island, declares whether anyone is told, and FMS matches the tree")
check("both modes loaded with no problems", not PROBLEMS, "; ".join(PROBLEMS))
mode_dir = HERE / "TMS" / "modes"
for name in sorted(f.name for f in mode_dir.iterdir()
                   if f.suffix == ".py" and f.name != "__init__.py"):
    reaches = re.findall(r"^\s*(?:from|import)\s+(\S+)", (mode_dir / name).read_text(encoding="utf-8"), re.M)
    check(f"{name} imports nothing", not reaches, ", ".join(reaches) or "no imports at all")
for where, expected in ARCH["units"].items():
    on_disk = sorted(f.name for f in (HERE / where).iterdir()
                     if f.suffix == ".py" and f.name != "__init__.py")
    check(f"{where}: FMS declares {len(expected)}, on disk {len(on_disk)}",
          on_disk == sorted(expected), ", ".join(on_disk))
check("exactly one mode reports errors",
      sum(1 for m in LOADED.values() if m.REPORTS_ERRORS) == 1,
      ", ".join(f"{m.NAME}={m.REPORTS_ERRORS}" for m in sorted(LOADED.values(), key=lambda m: m.NAME)))

print("\n== 2. a partial traversal and a complete one are the same value")
silent = upstream.vanishing_subdirectory()
loud = upstream.vanishing_subdirectory(onerror=reporting_mode.make_collector)
check("the same files come back either way", silent["files"] == loud["files"],
      str(silent["files"]))
check("neither run raised", silent["raised"] is None and loud["raised"] is None)
check("the silent run reports nothing", silent["errors"] == [], str(silent["errors"]))
check("the reporting run names the directory that vanished",
      len(loud["errors"]) == 1 and "FileNotFoundError" in loud["errors"][0],
      loud["errors"][0])
check("so the only difference is whether anybody was told",
      silent["files"] == loud["files"] and bool(loud["errors"]) != bool(silent["errors"]))

print("\n== 3. the discriminator exists and the idiomatic loop destroys it")
measured = upstream.empty_versus_missing()
check("an empty directory and a missing one yield different row counts",
      measured["empty"]["rows"] != measured["missing"]["rows"],
      f"{measured['empty']['rows']} vs {measured['missing']['rows']}")
check("and the same number of files - which is what the usual loop collects",
      measured["empty"]["files"] == measured["missing"]["files"],
      f"{measured['empty']['files']} both")
check("neither raised", measured["raised"] is None,
      "os.walk on a path that does not exist is not an error, it is an empty walk")

print("\n== 3b. the drill: a mode that claims to report and does not")


class Overclaiming:
    NAME = "claims-to-report"
    REPORTS_ERRORS = True
    WHAT_THE_CALLER_SEES = "nothing extra, despite the declaration"

    @staticmethod
    def make_collector(into):
        return None                        # the lie: no callback is installed


probe = upstream.vanishing_subdirectory(onerror=Overclaiming.make_collector)
check("a mode declaring REPORTS_ERRORS=True that reports nothing is caught",
      Overclaiming.REPORTS_ERRORS and probe["errors"] == [],
      "declared it reports, reported nothing")

print("\n== 4. fail closed")
_, problem = walks.resolve("raise-on-anything", LOADED)
check("an unresolvable mode stops the run", problem is not None, problem or "resolved anyway")
check("SCL names a mode that exists", policy.mode() in LOADED, policy.mode())
check("and this deployment picked the one that reports",
      LOADED[policy.mode()].REPORTS_ERRORS, policy.mode())

print("\n== 5. what this entry cannot see")
print("        MEASURABLE, NOT MEASURED")
print("          - how often a real traversal loses a directory to a permission error")
print("          - how many callers pass onerror at all")
print("        NOT MEASURABLE HERE")
print("          - whether the default is wrong. A walk that raised on every")
print("            unreadable directory would be unusable on a real filesystem,")
print("            and onerror exists because no default suits everyone.")
print("          - what any caller believed an empty result meant.")

print(f"\n{len(FAILURES)} failure(s)" if FAILURES else "\nall checks passed")
for f in FAILURES:
    print(f"  - {f}")
sys.exit(1 if FAILURES else 0)

"""os.walk: what it does with an error it meets on the way.

    python src/main.py            the two causes, the two modes
    python src/main.py --strict   exit 1 if this deployment walks silently
"""
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from DMS import report  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import upstream, walks  # noqa: E402
from TMS.modes import reporting as reporting_mode  # noqa: E402

ARCH = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))


def main(argv):
    out = lambda line="": sys.stdout.write(line + "\n")  # noqa: E731
    loaded, problems = walks.load()
    if problems:
        for problem in problems:
            out(f"  !! {problem}")
        return 1

    out(f"\n== os.walk{upstream.signature()}  [python {sys.version.split()[0]}]")
    report.causes(upstream.empty_versus_missing(), out)
    report.modes(upstream.vanishing_subdirectory(),
                 upstream.vanishing_subdirectory(onerror=reporting_mode.make_collector), out)

    out("\n== what the source says about onerror")
    for line in upstream.onerror_lines():
        out(f"    {line}")
    out("    one branch. Absent it, the error is discarded and the walk continues.")

    module, problem = walks.resolve(policy.mode(), loaded)
    if problem:
        out(f"\n  !! {problem}")
        return 1
    out(f"\n== this deployment walks in `{module.NAME}` mode")
    out(f"    reports errors: {module.REPORTS_ERRORS}")
    out(f"    the caller sees: {module.WHAT_THE_CALLER_SEES}")

    report.gaps(out)

    if "--strict" in argv and not module.REPORTS_ERRORS and policy.silent_is_fatal():
        return 1
    return 0


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