DEV Community

Dakota Huang
Dakota Huang

Posted on

Pin Import-Time Open Calls Before One Package Move

Do not relocate a messy package before pinning imports.
Green unit tests can miss import-time file creation.
Record every open call during the first import.

Why layout changes fail

Messy packages often write files during import.
__init__.py may create caches, logs, or lockfiles.
A rename then points those writes at new paths.

Test files usually import the package very late.
They never replay a cold process start sequence.
Production still executes the hidden import-time path.

Large model diffs like to move many modules.
That move also reorders plugin registration side effects.
The open log is the cheaper oracle here.

Observables to pin

Pin three observables before touching package layout.
First, record ordered path and mode pairs from open.
Second, record directories created by mkdir calls.

Third, record one public function result after import.
Keep that function on its original qualified name.
Do not change call signatures during the tape stage.

Hash file bytes after the import returns control.
A path-only log misses truncated or empty writes.
Size plus sha256 catches silent format drift.

Artifact: import open tape

Use a process-local wrapper, not a kernel tracer.
Wrap open, mkdir, and Path.write_text during import.
Restore the originals in a finally block always.

The following harness is an unexecuted local example.
Run it inside a throwaway clone with fake HOME.
Do not point it at a production working tree.

# characterize_import_opens.py
from __future__ import annotations

import builtins
import hashlib
import importlib
import json
import os
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any


@dataclass(frozen=True)
class OpenEvent:
    order: int
    op: str
    path: str
    mode: str
    size: int | None
    sha256: str | None


class ImportOpenTape:
    def __init__(self, root: Path, home: Path) -> None:
        self.root = root.resolve()
        self.home = home.resolve()
        self.events: list[OpenEvent] = []
        self._order = 0
        self._real_open = builtins.open
        self._real_os_mkdir = os.mkdir
        self._real_makedirs = os.makedirs
        self._real_path_mkdir = Path.mkdir
        self._real_write_text = Path.write_text

    def _rel(self, path: str | os.PathLike[str]) -> str | None:
        p = Path(path).resolve()
        if "__pycache__" in p.parts or p.suffix == ".pyc":
            return None
        try:
            return str(p.relative_to(self.root))
        except ValueError:
            pass
        try:
            return "$HOME/" + str(p.relative_to(self.home))
        except ValueError:
            return None

    def _stat(self, path: str | os.PathLike[str]) -> tuple[int | None, str | None]:
        p = Path(path)
        if not p.is_file():
            return None, None
        data = p.read_bytes()
        return len(data), hashlib.sha256(data).hexdigest()

    def _record(self, op: str, path: str, mode: str) -> None:
        rel = self._rel(path)
        if rel is None:
            return
        size, digest = self._stat(path)
        self._order += 1
        self.events.append(OpenEvent(self._order, op, rel, mode, size, digest))

    def install(self) -> None:
        tape = self

        def wrapped_open(file: Any, mode: str = "r", *args: Any, **kwargs: Any) -> Any:
            fh = tape._real_open(file, mode, *args, **kwargs)
            if not isinstance(file, int):
                tape._record("open", str(file), mode)
            return fh

        def wrapped_os_mkdir(path: Any, mode: int = 0o777) -> None:
            tape._real_os_mkdir(path, mode)
            tape._record("os.mkdir", str(path), oct(mode))

        def wrapped_makedirs(name: Any, mode: int = 0o777, exist_ok: bool = False) -> None:
            tape._real_makedirs(name, mode=mode, exist_ok=exist_ok)
            tape._record("makedirs", str(name), oct(mode))

        def wrapped_path_mkdir(
            self_path: Path,
            mode: int = 0o777,
            parents: bool = False,
            exist_ok: bool = False,
        ) -> None:
            tape._real_path_mkdir(self_path, mode=mode, parents=parents, exist_ok=exist_ok)
            tape._record("Path.mkdir", str(self_path), oct(mode))

        def wrapped_write_text(
            self_path: Path,
            data: str,
            encoding: str | None = None,
            errors: str | None = None,
            newline: str | None = None,
        ) -> int:
            n = tape._real_write_text(
                self_path,
                data,
                encoding=encoding,
                errors=errors,
                newline=newline,
            )
            tape._record("write_text", str(self_path), "w")
            return n

        builtins.open = wrapped_open  # type: ignore[assignment]
        os.mkdir = wrapped_os_mkdir  # type: ignore[assignment]
        os.makedirs = wrapped_makedirs  # type: ignore[assignment]
        Path.mkdir = wrapped_path_mkdir  # type: ignore[assignment]
        Path.write_text = wrapped_write_text  # type: ignore[assignment]

    def restore(self) -> None:
        builtins.open = self._real_open
        os.mkdir = self._real_os_mkdir
        os.makedirs = self._real_makedirs
        Path.mkdir = self._real_path_mkdir
        Path.write_text = self._real_write_text


def import_once(module_name: str, root: Path, home: Path) -> list[OpenEvent]:
    if module_name in sys.modules:
        raise RuntimeError("module already loaded; start a fresh process")
    tape = ImportOpenTape(root, home)
    tape.install()
    try:
        importlib.import_module(module_name)
        return list(tape.events)
    finally:
        tape.restore()


def main() -> None:
    module_name = os.environ.get("TAPE_MODULE", "messy_pkg")
    root = Path(os.environ.get("TAPE_ROOT", ".")).resolve()
    home = Path(os.environ.get("HOME", ".")).resolve()
    out = Path(os.environ.get("TAPE_OUT", "import_open_tape.json"))
    events = import_once(module_name, root, home)
    payload = json.dumps([asdict(e) for e in events], indent=2) + "\n"
    out.write_text(payload, encoding="utf-8")
    print(f"wrote {len(events)} events to {out}")


if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

Start each tape run in a brand new process.
Reusing sys.modules hides the original import side effects.
A pytest fixture cannot safely share that process.

Drop pycache and pyc events from the golden tape.
Those files change with interpreter minor versions.
Keep only clone-relative paths and HOME tokens.

A tiny messy package

The next package is a labeled example only.
It writes a JSON cache during the first import.
It also registers a plugin name on a module list.

# messy_pkg/__init__.py
from __future__ import annotations

import json
import os
from pathlib import Path

PLUGIN_NAMES: list[str] = []
_CACHE_DIR = Path(os.environ.get("HOME", ".")) / ".messy_pkg"
_CACHE_FILE = _CACHE_DIR / "cache.json"


def _ensure_cache() -> dict:
    _CACHE_DIR.mkdir(parents=True, exist_ok=True)
    if not _CACHE_FILE.exists():
        payload = {"version": 1, "items": []}
        _CACHE_FILE.write_text(json.dumps(payload), encoding="utf-8")
    return json.loads(_CACHE_FILE.read_text(encoding="utf-8"))


CACHE = _ensure_cache()
PLUGIN_NAMES.append("default")


def public_count() -> int:
    return len(CACHE.get("items", []))
Enter fullscreen mode Exit fullscreen mode

Capture the tape with a one-line command.
Keep HOME inside a temporary directory for the run.
Otherwise user-level caches pollute the golden file.

export HOME="$(mktemp -d)"
export TAPE_MODULE=messy_pkg
export TAPE_ROOT="$PWD"
export PYTHONPATH="$PWD"
python characterize_import_opens.py
Enter fullscreen mode Exit fullscreen mode

Expected golden tape

The golden file must list opens in order.
Relative paths keep the tape portable across clones.
sha256 values pin the cache bytes, not only names.

[
  {"order": 1, "op": "Path.mkdir", "path": "$HOME/.messy_pkg", "mode": "0o777", "size": null, "sha256": null},
  {"order": 2, "op": "write_text", "path": "$HOME/.messy_pkg/cache.json", "mode": "w", "size": 28, "sha256": "<hash>"}
]
Enter fullscreen mode Exit fullscreen mode

Replace <hash> with the real sha256 from your run.
Commit that JSON beside tests as the golden oracle.
Do not hand-edit event order to make a later patch pass.

Path.write_text also calls open under this wrapper.
Expect paired events for the same cache path.
Do not drop the inner open to beautify the tape.

Normalize volatile paths

Temp directories change on every tape process start.
Store paths relative to the clone root always.
Rewrite HOME-relative paths to a stable token.

Do not hash files outside the clone root.
Those files include interpreter bytecode and site packages.
Filter events whose resolved path leaves both roots.

Open in write mode records size before writes finish.
Flush and close before hashing if you wrap writes.
The sample harness hashes after each successful open.

That timing misses later writes on the same handle.
Add a close wrapper if your init writes in chunks.
Until then, prefer write_text for cache creation.

pytest in a fresh subprocess

Spawn the tape with subprocess.run and check returncode.
Pass TAPE_MODULE through the child environment only.
Assert the child wrote the expected JSON tape.

# tests/test_import_open_tape.py
from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
GOLDEN = ROOT / "import_open_tape.json"


def test_import_open_tape_matches_golden(tmp_path: Path) -> None:
    env = os.environ.copy()
    env["HOME"] = str(tmp_path)
    env["TAPE_MODULE"] = "messy_pkg"
    env["TAPE_ROOT"] = str(ROOT)
    env["TAPE_OUT"] = str(tmp_path / "tape.json")
    env["PYTHONPATH"] = str(ROOT) + os.pathsep + env.get("PYTHONPATH", "")
    proc = subprocess.run(
        [sys.executable, str(ROOT / "characterize_import_opens.py")],
        cwd=ROOT,
        env=env,
        capture_output=True,
        text=True,
        check=False,
    )
    assert proc.returncode == 0, proc.stderr
    actual = json.loads((tmp_path / "tape.json").read_text(encoding="utf-8"))
    expected = json.loads(GOLDEN.read_text(encoding="utf-8"))
    assert actual == expected


def test_public_count_stays_zero(tmp_path: Path, monkeypatch) -> None:
    monkeypatch.setenv("HOME", str(tmp_path))
    import importlib

    sys.modules.pop("messy_pkg", None)
    mod = importlib.import_module("messy_pkg")
    assert mod.public_count() == 0
    assert mod.PLUGIN_NAMES == ["default"]
Enter fullscreen mode Exit fullscreen mode

Compare JSON tapes with exact list equality first.
Sorted-set compares will hide registration order regressions.
Order is part of the public import contract.

The second test pins one public function only.
It does not authorize a broader helper extract.
Leave every other symbol frozen during this pass.

Numbered workflow

1. Clone and freeze the tree

Copy the messy repository into a disposable directory.
Record the commit hash before any local edit.
Set HOME and XDG paths to empty temp dirs.

git rev-parse HEAD > TAPE_COMMIT.txt
export HOME="$(mktemp -d)"
export XDG_CACHE_HOME="$HOME/cache"
Enter fullscreen mode Exit fullscreen mode

2. Capture a golden import tape

Run the harness against the current package name.
Commit the JSON tape beside the existing tests.
Treat unexpected extra opens as failing tests later.

TAPE_MODULE=messy_pkg TAPE_ROOT="$PWD" PYTHONPATH="$PWD" python characterize_import_opens.py
git add import_open_tape.json TAPE_COMMIT.txt
Enter fullscreen mode Exit fullscreen mode

3. Add one behavioral assertion

Call one public function after the import completes.
Pin its return value, type, and raised errors.
Leave every other function untouched in this pass.

4. Propose the smallest layout change

Move one submodule, not the entire package tree.
Keep the __init__.py import order identical when possible.
Re-run the tape in a fresh Python process.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.
MonkeyCode provides free model access and a free server option.
Rerun the import tape there after one drafted module move.

5. Accept or revert

Keep the move only when both oracles stay equal.
Equal means identical open order and function result.
Revert the tree when either oracle drifts.

git checkout -- messy_pkg
Enter fullscreen mode Exit fullscreen mode

Decision table

Read the table as a freeze or revert gate.
Do not negotiate away a hash mismatch.
A new cache path is a failed move.

Observation after the move Meaning Action
New open or mkdir event Extra import-time I/O Revert the move
Missing open or mkdir event A write disappeared Revert the move
sha256 mismatch on a kept path Cache format changed Revert unless migrating
Event order swapped Registration reorder Revert the move
Identical tape and public_count Layout change was inert Keep one submodule move

Read a failing event

A new open at order 4 means extra import work.
A missing open means a cache write disappeared.
A hash mismatch means the file format changed.

Mode flips from r to w+ are high severity.
Treat makedirs exist_ok noise as a real event.
Empty directories still change the deployment layout later.

Print the first differing event and stop there.
Do not scan the rest of the tape for comfort.
The first drift is the move to revert.

Who should use this

Use this when a package creates files at import.
Use this before any automated rename across modules.
Use this when plugin registration order affects output.

Who should not use this

Do not use this workflow on binary-only distributions.
Do not use it to redesign a public package API.
Do not run wrappers against shared production home directories.

Skip this tape for pure function libraries without import side effects.
Skip this tape when you must change on-disk formats.
Those changes need explicit migration tests, not open logs.

Teams without a disposable clone should stop here.
Multi-process import races need a different harness.
Windows path casing may require a normalization step.

This tape does not record network or subprocess I/O.
It also skips C extensions that bypass builtins.open.
Socket, mmap, and os.open calls need extra wrappers.

Limits of model-assisted moves

Models collapse several renames into one noisy patch.
The tape rejects those patches without extra debate.
Smallest safe change means one submodule per cycle.

Import-time I/O is part of the package contract.
Pin the open log before any directory rename.
Then move one module and stop for the day.

Commit the JSON tape before the first layout patch.

Top comments (1)

Collapse
 
eternaclarity profile image
Jesse Gamble

Pinning the import-time side effects before the move is a useful guard. Those cold-start writes are exactly the kind of behavior a normal unit suite can miss while a refactor still looks green.