DEV Community

Dakota Huang
Dakota Huang

Posted on

Pin Path Sets and Duplicate Winners Before One Glob Extract

Live glob order is a flaky golden. Pin path sets, content hashes, and duplicate winners first. Extract one collector only after those contracts are frozen.

Filesystem walk order changes across disks. It also changes across containers. A sorted() cleanup looks like hygiene. It is a behavior change when two files share a key.

Core contract, not the walk

glob.glob does not promise lexical order. Python documents that as platform-dependent. Treat the returned sequence as an implementation accident. Treat membership and collision policy as the product contract.

AI-assisted refactors add sorted() constantly. They also surface swallowed errors. Both edits can be correct later. Neither belongs in the first cut.

Observables: pin or ignore

Score each field before you touch the module. Use the table as a merge gate.

Observable Pin in golden? Why
Relative path set Yes Membership is the collector contract
sha256 of each file Yes Parsers depend on exact bytes
Duplicate-key winner Yes Hidden order dependency
Item count and key union Yes Detects dropped or extra records
Raw glob.glob order No Varies by inode layout
mtime, ctime, inode No CI checkouts rewrite them
Absolute prefixes No Temp roots and hosts differ
Symlink skip policy Yes Changes which bytes get read
Decode and JSON errors Yes Current code swallows them

Do not freeze wall-clock stamps. Do not freeze device numbers. Do not freeze live walk order on a real disk.

Example messy module

This fixture is a teaching module. It is unlabeled production code. Treat it as unexecuted until you run it.

# inventory.py
from __future__ import annotations

import glob
import json
import os
from typing import Any

ROOT = os.environ.get("INV_ROOT", "data")


def load_items() -> list[dict[str, Any]]:
    pattern = os.path.join(ROOT, "**", "*.json")
    paths = glob.glob(pattern, recursive=True)
    items: list[dict[str, Any]] = []
    for path in paths:
        if os.path.islink(path):
            continue
        try:
            raw = open(path, "rb").read()
            payload = json.loads(raw.decode("utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(payload, dict):
            continue
        sku = payload.get("sku")
        if not sku:
            continue
        payload["_source"] = path
        items.append(payload)
    return items
Enter fullscreen mode Exit fullscreen mode

Callers often take the first matching SKU. That makes glob order a silent policy. The skipped errors are also a policy. Leave both policies intact for the first extract.

Stub glob, do not golden the disk

Live tmp_path layouts still race on order. Stub glob.glob with a tuple you control. Then the loop contract is deterministic. The filesystem remains unpinned noise.

# test_inventory_char.py
from __future__ import annotations

import hashlib
import json
from pathlib import Path

import inventory

GOLDEN = Path(__file__).with_name("inventory_char.json")


def _write_tree(root: Path) -> dict[str, Path]:
    root.mkdir(parents=True, exist_ok=True)
    a = root / "a.json"
    b = root / "nested" / "b.json"
    dup = root / "nested" / "dup.json"
    bad = root / "nested" / "bad.json"
    txt = root / "nested" / "ignore.txt"
    b.parent.mkdir(parents=True, exist_ok=True)
    a.write_text('{"sku": "A", "qty": 1}', encoding="utf-8")
    b.write_text('{"sku": "B", "qty": 2}', encoding="utf-8")
    dup.write_text('{"sku": "A", "qty": 99}', encoding="utf-8")
    bad.write_text("{not-json", encoding="utf-8")
    txt.write_text("nope", encoding="utf-8")
    link = root / "link.json"
    link.symlink_to(a)
    return {"a": a, "b": b, "dup": dup, "bad": bad, "link": link}


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _snapshot(items: list[dict]) -> dict:
    sources = [i.get("_source", "") for i in items]
    by_sku: dict[str, list[int]] = {}
    for i in items:
        by_sku.setdefault(str(i.get("sku")), []).append(int(i.get("qty", -1)))
    first_qty_for_a = next((i.get("qty") for i in items if i.get("sku") == "A"), None)
    return {
        "count": len(items),
        "skus_sorted": sorted(str(i.get("sku")) for i in items),
        "first_sku": items[0]["sku"] if items else None,
        "first_qty_for_sku_A": first_qty_for_a,
        "qty_by_sku": {k: v for k, v in sorted(by_sku.items())},
        "source_names": sorted(Path(s).name for s in sources),
        "key_union": sorted({k for i in items for k in i.keys()}),
    }


def test_load_items_pins_sets_and_duplicate_winner(tmp_path, monkeypatch):
    files = _write_tree(tmp_path)
    monkeypatch.setenv("INV_ROOT", str(tmp_path))
    ordered = [
        str(files["dup"]),
        str(files["link"]),
        str(files["a"]),
        str(files["bad"]),
        str(files["b"]),
    ]
    monkeypatch.setattr(inventory.glob, "glob", lambda *a, **k: list(ordered))

    items = inventory.load_items()
    snap = _snapshot(items)
    snap["hashes"] = {
        name: _sha256(path)
        for name, path in files.items()
        if path.is_file() and not path.is_symlink()
    }

    if not GOLDEN.exists():
        GOLDEN.write_text(json.dumps(snap, indent=2, sort_keys=True) + "\n")
    expected = json.loads(GOLDEN.read_text(encoding="utf-8"))
    assert snap == expected
Enter fullscreen mode Exit fullscreen mode

Record the golden once on a quiet tree. Commit that file in the same change set. Later runs must match it byte for byte after sort_keys.

The stub list is the experiment knob. link.json stays in the list. The loop still skips it. bad.json stays in the list. The loop still swallows the decode error. sku A appears twice. The first remaining row is the winner for naive callers.

Numbered extract sequence

Follow this order. Do not skip a step.

  1. Commit the messy module without edits.
  2. Add the stubbed characterization test.
  3. Generate inventory_char.json on one machine.
  4. Re-run the test on a second path layout.
  5. Confirm hashes and winners still match.
  6. Extract iter_json_paths() as a one-line wrapper.
  7. Keep load_items() iterating that sequence.
  8. Point the stub at iter_json_paths if needed.
  9. Reject sorts, filters, and new error lists.
  10. Stop. Do not fold in extra cleanup.

The extract should look like this. Anything larger is a second change.

def iter_json_paths(root: str) -> list[str]:
    pattern = os.path.join(root, "**", "*.json")
    return glob.glob(pattern, recursive=True)


def load_items() -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    for path in iter_json_paths(ROOT):
        if os.path.islink(path):
            continue
        try:
            raw = open(path, "rb").read()
            payload = json.loads(raw.decode("utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(payload, dict):
            continue
        sku = payload.get("sku")
        if not sku:
            continue
        payload["_source"] = path
        items.append(payload)
    return items
Enter fullscreen mode Exit fullscreen mode

Keep the body of the loop untouched. The helper returns the same list object shape. It does not sort. It does not uniquify SKUs. It does not follow symlinks.

What the golden must reject

Add two negative checks after the extract. They document the change you refused.

def test_reject_sorted_collector(tmp_path, monkeypatch):
    files = _write_tree(tmp_path)
    monkeypatch.setenv("INV_ROOT", str(tmp_path))
    ordered = [str(files["dup"]), str(files["a"]), str(files["b"])]
    monkeypatch.setattr(inventory, "iter_json_paths", lambda root: list(ordered))
    items = inventory.load_items()
    assert items[0]["qty"] == 99

    monkeypatch.setattr(
        inventory, "iter_json_paths", lambda root: sorted(ordered)
    )
    sorted_items = inventory.load_items()
    # lexical a.json before dup.json on this layout
    assert sorted_items[0]["qty"] != items[0]["qty"]


def test_reject_surfacing_bad_json(tmp_path, monkeypatch):
    files = _write_tree(tmp_path)
    monkeypatch.setenv("INV_ROOT", str(tmp_path))
    monkeypatch.setattr(
        inventory,
        "iter_json_paths",
        lambda root: [str(files["bad"]), str(files["b"])],
    )
    items = inventory.load_items()
    assert len(items) == 1
    assert items[0]["sku"] == "B"
Enter fullscreen mode Exit fullscreen mode

The first test proves sorted() flips the SKU A winner. The second test proves a raised parse error would be a new contract. Hold both for a later, explicit change.

Free-model pass after goldens only

A model can propose the helper text. It cannot choose the observables. Write the goldens yourself. Then ask the model for a one-function extract that keeps the loop body identical.

Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode's free model access and free server option can draft that extract against the committed tests. They do not replace the pin step. If the patch adds sorted() or an errors list, the tests fail and the patch is discarded.

Limitations

This harness does not prove thread safety. It does not prove encoding on Windows code pages. It does not prove ** glob behavior on every Python build. glob.glob still ignores some symlink directories depending on platform. Path.rglob is a different iterator and needs its own pins.

Stubbing glob.glob hides real permission errors. Add one integration test on a tiny fixture tree. Assert only the path set and hashes there. Leave order out of that test.

Duplicate winners depend on the stub sequence. Production order remains unstable. If product code needs a stable winner, make that a second change. Use an explicit min(path) or last-write rule. Do not pretend glob order is that rule.

Open file handles use a bare open. That is part of the mess. Do not convert it to a context manager in the same patch. Handle leaks are a separate characterization target.

Who should not use this

Do not use this flow on cryptographic parsers. Do not use it on access-control filters. Do not use it when dropped JSON must become a hard failure. Those cases need designed tests, not captured silence.

Do not use it as a substitute for schema checks. Hashes freeze bytes, not meaning. A valid JSON object with the wrong units will still match.

Skip this extract if the module already has one seam. Skip it if two collectors exist. Join them only after each has goldens.

Stop condition

The change is done when iter_json_paths exists. The loop body must match the prior bytes. Goldens must still pass on the stubbed sequence. No sort. No new logging. No error-channel redesign.

If a later change needs stable order, add a named sort at the call site. Record a new golden for that call site only. Keep the collector unordered. That split is the whole point of the pin.

Top comments (0)